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

Add merge references and retry config

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-07 19:10:55 +00:00
parent c05ecb58d8
commit bcedf19a08
25 changed files with 466 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
internal/cli/run.go
View File

@@ -99,7 +99,7 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i
referenceFlags := stringListFlag{}
withoutReferenceFlags := stringListFlag{}
fs.Var(&sessionID, "session-id", "prompt session identifier")
fs.Var(&referenceFlags, "reference", "reference binding, as slot=path, chunk.slot=path, lane.slot=path, lane.extract.slot=path, or lane.normalize.slot=path")
fs.Var(&referenceFlags, "reference", "reference binding, as slot=path, chunk.slot=path, merge.slot=path, lane.slot=path, lane.extract.slot=path, lane.merge.slot=path, or lane.normalize.slot=path")
fs.Var(&withoutReferenceFlags, "without-reference", "unbind a reference, using the same selector forms as --reference")
if err := validateRunFlagValues(args); err != nil {
fmt.Fprintf(stderr, "notarius: %v\n", err)
@@ -492,6 +492,7 @@ func effectiveLLMProfileIDs(resolved pipeline.ResolvedPipeline) []string {
add(resolved.Chunk)
for _, lane := range resolved.ArtifactLanes {
add(lane.Extract)
add(lane.Merge)
add(lane.Normalize)
}
ids := make([]string, 0, len(seen))
@@ -834,17 +835,20 @@ func parseReferenceSelector(raw string, flagName string) (cliReferenceSelector,
if first == string(pipeline.StageChunk) {
return cliReferenceSelector{Stage: pipeline.StageChunk, SlotName: slotName}, nil
}
if first == string(pipeline.StageMerge) {
return cliReferenceSelector{Stage: pipeline.StageMerge, SlotName: slotName}, nil
}
return cliReferenceSelector{LaneID: first, SlotName: slotName}, nil
case 3:
laneID := strings.TrimSpace(parts[0])
stage := pipeline.ModuleStage(strings.TrimSpace(parts[1]))
slotName := strings.TrimSpace(parts[2])
if stage != pipeline.StageExtract && stage != pipeline.StageNormalize {
return cliReferenceSelector{}, fmt.Errorf("%s lane-qualified selector must use lane.extract.slot or lane.normalize.slot", flagName)
if stage != pipeline.StageExtract && stage != pipeline.StageMerge && stage != pipeline.StageNormalize {
return cliReferenceSelector{}, fmt.Errorf("%s lane-qualified selector must use lane.extract.slot, lane.merge.slot, or lane.normalize.slot", flagName)
}
return cliReferenceSelector{LaneID: laneID, Stage: stage, SlotName: slotName}, nil
default:
return cliReferenceSelector{}, fmt.Errorf("%s must use slot, chunk.slot, lane.slot, lane.extract.slot, or lane.normalize.slot", flagName)
return cliReferenceSelector{}, fmt.Errorf("%s must use slot, chunk.slot, merge.slot, lane.slot, lane.extract.slot, lane.merge.slot, or lane.normalize.slot", flagName)
}
}
@@ -944,7 +948,7 @@ func selectedReferenceTargets(cfg config.Config, pipelineID string, only []strin
}
sort.Strings(selectedIDs)
targets := make([]selectedReferenceTarget, 0, 1+len(selectedIDs)*2)
targets := make([]selectedReferenceTarget, 0, 1+len(selectedIDs)*3)
chunk := pipeline.Binding(profile.Chunk.Module)
chunk.Module = strings.TrimSpace(profile.Chunk.Module)
if chunk.Module == "" {
@@ -977,6 +981,21 @@ func selectedReferenceTargets(cfg config.Config, pipelineID string, only []strin
slots: referenceSlotSet(extractSpec.ReferenceSlots),
})
mergeModule := strings.TrimSpace(lane.Merge.Module)
if mergeModule == "" {
mergeModule = pipeline.DefaultMergeModule
}
mergeSpec, err := cliReferenceMergerSpec(catalog, mergeModule)
if err != nil {
return nil, fmt.Errorf("pipeline %q lane %q merge module %q: %w", strings.TrimSpace(pipelineID), laneID, mergeModule, err)
}
targets = append(targets, selectedReferenceTarget{
laneID: laneID,
stage: pipeline.StageMerge,
module: mergeModule,
slots: referenceSlotSet(mergeSpec.ReferenceSlots),
})
normalizeModule := strings.TrimSpace(lane.Normalize.Module)
if normalizeModule == "" {
normalizeModule = pipeline.DefaultNormalizeModule
@@ -1027,6 +1046,17 @@ func cliReferenceExtractorSpec(catalog pipeline.ModuleCatalog, module string) (p
return spec, nil
}
func cliReferenceMergerSpec(catalog pipeline.ModuleCatalog, module string) (pipeline.ModuleSpec, error) {
if catalog.Mergers == nil {
return pipeline.ModuleSpec{}, fmt.Errorf("module %q is not registered", module)
}
spec, ok := catalog.Mergers.Spec(module)
if !ok {
return pipeline.ModuleSpec{}, fmt.Errorf("module %q is not registered", module)
}
return spec, nil
}
func cliReferenceNormalizerSpec(catalog pipeline.ModuleCatalog, module string) (pipeline.ModuleSpec, error) {
if catalog.Normalizers == nil {
return pipeline.ModuleSpec{}, fmt.Errorf("module %q is not registered", module)
@@ -1063,7 +1093,10 @@ func resolveCLIReferenceTarget(targets []selectedReferenceTarget, selector cliRe
}
return selectedReferenceTarget{}, fmt.Errorf("reference chunk target is not selected")
}
if selector.Stage == pipeline.StageExtract || selector.Stage == pipeline.StageNormalize {
if selector.Stage == pipeline.StageExtract || selector.Stage == pipeline.StageMerge || selector.Stage == pipeline.StageNormalize {
if selector.LaneID == "" && selector.Stage == pipeline.StageMerge {
return resolveCLIReferenceStageTarget(targets, selector.Stage, slotName)
}
for _, target := range targets {
if target.laneID == selector.LaneID && target.stage == selector.Stage {
if _, ok := target.slots[slotName]; !ok {
@@ -1080,6 +1113,26 @@ func resolveCLIReferenceTarget(targets []selectedReferenceTarget, selector cliRe
return resolveCLIReferenceFlatTarget(targets, slotName)
}
func resolveCLIReferenceStageTarget(targets []selectedReferenceTarget, stage pipeline.ModuleStage, slotName string) (selectedReferenceTarget, error) {
matches := make([]selectedReferenceTarget, 0, 2)
for _, target := range targets {
if target.stage != stage {
continue
}
if _, ok := target.slots[slotName]; ok {
matches = append(matches, target)
}
}
switch len(matches) {
case 0:
return selectedReferenceTarget{}, fmt.Errorf("reference slot %q is not declared by any selected %s target", slotName, stage)
case 1:
return matches[0], nil
default:
return selectedReferenceTarget{}, fmt.Errorf("reference slot %q is declared by multiple selected %s targets (%s); use a more specific selector such as %s", slotName, stage, targetList(matches), selectorSuggestions(matches, slotName))
}
}
func resolveCLIReferenceLaneTarget(targets []selectedReferenceTarget, laneID string, slotName string) (selectedReferenceTarget, error) {
laneSelected := false
matches := make([]selectedReferenceTarget, 0, 2)
@@ -1101,7 +1154,7 @@ func resolveCLIReferenceLaneTarget(targets []selectedReferenceTarget, laneID str
case 1:
return matches[0], nil
default:
return selectedReferenceTarget{}, fmt.Errorf("reference slot %q is declared by multiple selected targets in lane %q (%s); use %s.extract.%s or %s.normalize.%s", slotName, laneID, targetList(matches), laneID, slotName, laneID, slotName)
return selectedReferenceTarget{}, fmt.Errorf("reference slot %q is declared by multiple selected targets in lane %q (%s); use a more specific selector such as %s", slotName, laneID, targetList(matches), selectorSuggestions(matches, slotName))
}
}

111
internal/cli/run_test.go
View File

@@ -816,7 +816,7 @@ pipelines:
}
}
func TestRunConfigValidateIgnoresNonLLMStageScriptoriumProfiles(t *testing.T) {
func TestRunConfigValidateIncludesMergeAndIgnoresNonLLMStageScriptoriumProfiles(t *testing.T) {
profilePath := writeScriptoriumProfileFile(t, "known", "http://profile.test/v1", "test-model")
configPath := writeTestConfig(t, `version: 2
scriptorium:
@@ -843,11 +843,11 @@ pipelines:
Catalog: fakeCatalog(t),
})
if code != 0 {
t.Fatalf("RunWithOptions() code = %d, stderr=%q, want success", code, stderr.String())
if code != 1 {
t.Fatalf("RunWithOptions() code = %d, want failure for missing merge profile", code)
}
if strings.Contains(stderr.String(), "missing-") {
t.Fatalf("stderr = %q, want non-LLM stage profiles ignored", stderr.String())
if !strings.Contains(stderr.String(), "Scriptorium profile") || !strings.Contains(stderr.String(), "missing-merge") {
t.Fatalf("stderr = %q, want missing merge profile error", stderr.String())
}
}
@@ -867,7 +867,7 @@ func TestEffectiveLLMProfileIDsUsesLLMCapableStagesOnly(t *testing.T) {
}
got := effectiveLLMProfileIDs(resolved)
want := []string{"chunk-profile", "extract-profile", "normalize-profile"}
want := []string{"chunk-profile", "extract-profile", "merge-profile", "normalize-profile"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("effectiveLLMProfileIDs() = %#v, want %#v", got, want)
}
@@ -1166,6 +1166,81 @@ func TestRunPipelineReferenceFlagBindsExplicitNormalizeSlot(t *testing.T) {
}
}
func TestRunPipelineReferenceFlagBindsExplicitMergeSlot(t *testing.T) {
configPath := writeTestConfig(t, testConfigYAML("example", "events"))
inputPath := filepath.Join(t.TempDir(), "missing.json")
referencePath := writeFile(t, "merge.md", "Merge notes\n")
diagnosticsDir := t.TempDir()
var stdout bytes.Buffer
var stderr bytes.Buffer
code := RunWithOptions([]string{
"run", "example",
"--config", configPath,
"--input", inputPath,
"--diagnostics-dir", diagnosticsDir,
"--reference", "events.merge.notes=" + referencePath,
}, &stdout, &stderr, Options{
Catalog: fakeCatalog(t, pipeline.ModuleSpec{
Key: "appendorder",
Stage: pipeline.StageMerge,
Requires: []string{"artifact"},
Provides: []string{"merged"},
ReferenceSlots: []contracts.ReferenceSlot{
{Name: "notes"},
},
}),
})
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].MergeReferences.Bindings
want := []pipeline.ReferenceBinding{
{LaneID: "events", SlotName: "notes", Source: referencePath, BindingSource: contracts.ReferenceBindingSourceCLI},
}
if !reflect.DeepEqual(refs, want) {
t.Fatalf("merge references = %#v, want %#v", refs, want)
}
}
func TestRunPipelineReferenceFlagBindsUnambiguousMergeSlot(t *testing.T) {
configPath := writeTestConfig(t, testConfigYAML("example", "events"))
inputPath := filepath.Join(t.TempDir(), "missing.json")
referencePath := writeFile(t, "merge.md", "Merge notes\n")
diagnosticsDir := t.TempDir()
var stdout bytes.Buffer
var stderr bytes.Buffer
code := RunWithOptions([]string{
"run", "example",
"--config", configPath,
"--input", inputPath,
"--diagnostics-dir", diagnosticsDir,
"--reference", "merge.notes=" + referencePath,
}, &stdout, &stderr, Options{
Catalog: fakeCatalog(t, pipeline.ModuleSpec{
Key: "appendorder",
Stage: pipeline.StageMerge,
Requires: []string{"artifact"},
Provides: []string{"merged"},
ReferenceSlots: []contracts.ReferenceSlot{
{Name: "notes"},
},
}),
})
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].MergeReferences.Bindings
if len(refs) != 1 || refs[0].Source != referencePath || refs[0].LaneID != "events" {
t.Fatalf("merge references = %#v, want unambiguous merge binding", refs)
}
}
func TestRunPipelineReferenceFlagBindsFlatSlotAcrossOneTarget(t *testing.T) {
configPath := writeTestConfig(t, testConfigYAML("example", "events"))
inputPath := filepath.Join(t.TempDir(), "missing.json")
@@ -1366,8 +1441,8 @@ func TestRunPipelineReferenceFlagsRejectMalformedValues(t *testing.T) {
{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: "unsupported explicit stage", args: []string{"--reference", "a.b.c=./roster.yml"}, want: "lane.extract.slot or lane.normalize.slot"},
{name: "too many selector parts", args: []string{"--reference", "a.b.c.d=./roster.yml"}, want: "slot, chunk.slot, lane.slot, lane.extract.slot, or lane.normalize.slot"},
{name: "unsupported explicit stage", args: []string{"--reference", "a.b.c=./roster.yml"}, want: "lane.extract.slot, lane.merge.slot, or lane.normalize.slot"},
{name: "too many selector parts", args: []string{"--reference", "a.b.c.d=./roster.yml"}, want: "slot, chunk.slot, merge.slot, lane.slot, lane.extract.slot, lane.merge.slot, or lane.normalize.slot"},
{name: "unbind with equals", args: []string{"--without-reference", "roster=./roster.yml"}, want: "without =path"},
}
@@ -1393,9 +1468,10 @@ func TestRunPipelineReferenceFlagsRejectMalformedValues(t *testing.T) {
func TestRunPipelineWithoutReferenceRemovesConfigBindingsForEligibleTargets(t *testing.T) {
configPath := writeTestConfig(t, testConfigYAMLWithPipelineReferences("example", "events", map[string]string{
"context": "./config-context.md",
"notes": "./config-notes.md",
"roster": "./config-roster.yml",
"context": "./config-context.md",
"merge_notes": "./config-merge.md",
"notes": "./config-notes.md",
"roster": "./config-roster.yml",
}))
inputPath := filepath.Join(t.TempDir(), "missing.json")
diagnosticsDir := t.TempDir()
@@ -1409,6 +1485,7 @@ func TestRunPipelineWithoutReferenceRemovesConfigBindingsForEligibleTargets(t *t
"--diagnostics-dir", diagnosticsDir,
"--without-reference", "chunk.context",
"--without-reference", "events.extract.roster",
"--without-reference", "events.merge.merge_notes",
"--without-reference", "events.normalize.notes",
}, &stdout, &stderr, Options{
Catalog: fakeCatalog(t,
@@ -1421,6 +1498,15 @@ func TestRunPipelineWithoutReferenceRemovesConfigBindingsForEligibleTargets(t *t
{Name: "context"},
},
},
pipeline.ModuleSpec{
Key: "appendorder",
Stage: pipeline.StageMerge,
Requires: []string{"artifact"},
Provides: []string{"merged"},
ReferenceSlots: []contracts.ReferenceSlot{
{Name: "merge_notes"},
},
},
pipeline.ModuleSpec{
Key: "fake/extract",
Stage: pipeline.StageExtract,
@@ -1452,6 +1538,9 @@ func TestRunPipelineWithoutReferenceRemovesConfigBindingsForEligibleTargets(t *t
if refs := resolved.ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 {
t.Fatalf("extract references = %#v, want none", refs)
}
if refs := resolved.ArtifactLanes[0].MergeReferences.Bindings; len(refs) != 0 {
t.Fatalf("merge references = %#v, want none", refs)
}
if refs := resolved.ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 0 {
t.Fatalf("normalize references = %#v, want none", refs)
}

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

@@ -68,6 +68,7 @@ func applyLLMProfileOverride(profile *pipeline.PipelineProfile, profileID string
profile.Chunk.LLMProfile = profileID
for laneID, lane := range profile.Artifacts {
lane.Extract.LLMProfile = profileID
lane.Merge.LLMProfile = profileID
lane.Normalize.LLMProfile = profileID
profile.Artifacts[laneID] = lane
}

6
internal/core/config/effective_config_test.go
View File

@@ -193,8 +193,8 @@ func TestResolveLLMProfileOverrideAppliesBeforeDigest(t *testing.T) {
t.Fatalf("output profile = %q, want original output-profile", effective.ResolvedPipeline.Output.LLMProfile)
}
eventLane := effective.ResolvedPipeline.ArtifactLanes[0]
if eventLane.Merge.LLMProfile != "merge-profile" {
t.Fatalf("merge profile = %q, want original merge-profile", eventLane.Merge.LLMProfile)
if eventLane.Merge.LLMProfile != "runtime" {
t.Fatalf("merge profile = %q, want runtime", eventLane.Merge.LLMProfile)
}
if len(eventLane.Validators) != 1 || eventLane.Validators[0].LLMProfile != "validator-profile" {
t.Fatalf("validator profiles = %#v, want original validator-profile", eventLane.Validators)
@@ -204,7 +204,7 @@ func TestResolveLLMProfileOverrideAppliesBeforeDigest(t *testing.T) {
func llmCapableBindings(resolved pipeline.ResolvedPipeline) []pipeline.ModuleBinding {
bindings := []pipeline.ModuleBinding{resolved.Chunk}
for _, lane := range resolved.ArtifactLanes {
bindings = append(bindings, lane.Extract, lane.Normalize)
bindings = append(bindings, lane.Extract, lane.Merge, lane.Normalize)
}
return bindings
}

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

@@ -53,6 +53,7 @@ type FileDiagnosticsConfig struct {
type fileModuleBinding struct {
Module string
LLMProfile string
Retries int
Options map[string]any
References map[string]string
}
@@ -83,6 +84,12 @@ func (b *fileModuleBinding) UnmarshalYAML(node *yaml.Node) error {
return err
}
b.LLMProfile = strings.TrimSpace(llmProfile)
case "retries":
var retries int
if err := valueNode.Decode(&retries); err != nil {
return err
}
b.Retries = retries
case "options":
var options map[string]any
if err := valueNode.Decode(&options); err != nil {
@@ -109,6 +116,7 @@ func (b fileModuleBinding) toPipelineBinding() pipeline.ModuleBinding {
return pipeline.ModuleBinding{
Module: strings.TrimSpace(b.Module),
LLMProfile: strings.TrimSpace(b.LLMProfile),
Retries: b.Retries,
Options: cloneOptions(b.Options),
References: normalizedStringMap(b.References),
}

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

@@ -127,6 +127,7 @@ pipelines:
input: fake/input
chunk:
module: generic
retries: 2
options:
size: 10
flags:
@@ -138,9 +139,12 @@ pipelines:
extract:
module: fake/extract
llm_profile: fast
retries: 3
options:
temperature: 0
merge: appendorder
merge:
module: appendorder
retries: 1
normalize:
module: noop
output: json
@@ -153,6 +157,9 @@ pipelines:
if profile.Chunk.Module != "generic" {
t.Fatalf("unexpected chunk binding: %+v", profile.Chunk)
}
if profile.Chunk.Retries != 2 {
t.Fatalf("chunk retries = %d, want 2", profile.Chunk.Retries)
}
if profile.Chunk.Options["size"] != 10 {
t.Fatalf("expected chunk options to preserve scalar, got %#v", profile.Chunk.Options)
}
@@ -168,6 +175,9 @@ pipelines:
if lane.Extract.Module != "fake/extract" || lane.Extract.LLMProfile != "fast" {
t.Fatalf("unexpected extract binding: %+v", lane.Extract)
}
if lane.Extract.Retries != 3 || lane.Merge.Retries != 1 {
t.Fatalf("unexpected retries: extract=%d merge=%d", lane.Extract.Retries, lane.Merge.Retries)
}
if lane.Extract.Options["temperature"] != 0 {
t.Fatalf("expected object options, got %#v", lane.Extract.Options)
}
@@ -224,6 +234,10 @@ pipelines:
references:
roster: ./legacy-roster.yml
lore: ./lore.md
merge:
module: appendorder
references:
" merge_notes ": " ./merge.md "
normalize:
module: noop
references:
@@ -246,6 +260,9 @@ pipelines:
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.Merge.References, map[string]string{"merge_notes": "./merge.md"}) {
t.Fatalf("merge references = %#v, want trimmed map", lane.Merge.References)
}
if !reflect.DeepEqual(lane.Normalize.References, map[string]string{"normalization_notes": "./normalization.md"}) {
t.Fatalf("normalize references = %#v, want trimmed map", lane.Normalize.References)
}

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

@@ -42,6 +42,7 @@ func cloneResolvedArtifactLane(in pipeline.ResolvedArtifactLane) pipeline.Resolv
out.Merge = cloneModuleBinding(in.Merge)
out.Normalize = cloneModuleBinding(in.Normalize)
out.ExtractReferences = pipeline.CloneReferenceTarget(in.ExtractReferences)
out.MergeReferences = pipeline.CloneReferenceTarget(in.MergeReferences)
out.NormalizeReferences = pipeline.CloneReferenceTarget(in.NormalizeReferences)
if len(in.Validators) > 0 {
out.Validators = make([]pipeline.ModuleBinding, len(in.Validators))

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

@@ -78,7 +78,7 @@ func validatePipelineProfiles(profiles map[string]pipeline.PipelineProfile) erro
if err := validateBinding(id, laneID, "extract", lane.Extract, true); err != nil {
return err
}
if err := validateBinding(id, laneID, "merge", lane.Merge, false); err != nil {
if err := validateBinding(id, laneID, "merge", lane.Merge, true); err != nil {
return err
}
if err := validateBinding(id, laneID, "normalize", lane.Normalize, true); err != nil {
@@ -104,6 +104,12 @@ func validateBinding(
if err := validateBindingLLMProfile(pipelineID, laneID, slot, binding); err != nil {
return err
}
if binding.Retries < 0 {
if laneID != "" {
return fmt.Errorf("pipeline %q lane %q %s retries must be greater than or equal to zero", pipelineID, laneID, slot)
}
return fmt.Errorf("pipeline %q %s retries must be greater than or equal to zero", pipelineID, slot)
}
if len(binding.References) == 0 {
return nil
}

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

@@ -54,6 +54,18 @@ func TestValidateRejectsInvalidNumericFields(t *testing.T) {
},
want: "total LLM concurrency",
},
{
name: "negative retries",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
lane := profile.Artifacts["events"]
lane.Merge.Retries = -1
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
return cfg
},
want: "retries",
},
}
for _, tc := range tests {
@@ -238,18 +250,6 @@ func TestValidateRejectsReferencesOnUnsupportedBindings(t *testing.T) {
},
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 {

4
internal/framework/contracts/contracts.go
View File

@@ -223,6 +223,10 @@ type MergeRequest struct {
Source *source.SourceDocument `json:"-"`
LaneID string `json:"lane_id"`
ExtractOutputs []ExtractOutput `json:"extract_outputs"`
SourceInput LLMInputMaterial `json:"source_input,omitempty"`
SessionID string `json:"session_id,omitempty"`
References ReferenceSet `json:"references,omitempty"`
LLMClient StructuredLLMClient `json:"-"`
LLMProfile string `json:"llm_profile,omitempty"`
Options map[string]any `json:"options,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`

2
internal/framework/pipeline/module.go
View File

@@ -97,7 +97,7 @@ func validateModuleSpec(kind string, expectedStage ModuleStage, spec ModuleSpec)
}
func referenceSlotStage(stage ModuleStage) bool {
return stage == StageChunk || stage == StageExtract || stage == StageNormalize
return stage == StageChunk || stage == StageExtract || stage == StageMerge || stage == StageNormalize
}
func sortedRegistryKeys[C any](constructors map[string]C) []string {

3
internal/framework/pipeline/module_test.go
View File

@@ -15,6 +15,7 @@ func TestValidateModuleSpecAllowsReferenceSlotsForEligibleStages(t *testing.T) {
}{
{name: "chunker", kind: "chunker", stage: StageChunk},
{name: "extractor", kind: "extractor", stage: StageExtract},
{name: "merger", kind: "merger", stage: StageMerge},
{name: "normalizer", kind: "normalizer", stage: StageNormalize},
}
@@ -42,7 +43,6 @@ func TestValidateModuleSpecRejectsReferenceSlotsForIneligibleStages(t *testing.T
stage ModuleStage
}{
{name: "input", kind: "input adapter", stage: StageInput},
{name: "merge", kind: "merger", stage: StageMerge},
{name: "validate", kind: "validator", stage: StageValidate},
{name: "output", kind: "output encoder", stage: StageOutput},
}
@@ -99,6 +99,7 @@ func TestValidateModuleSpecRejectsInvalidReferenceSlotsForEligibleStages(t *test
}{
{name: "chunk", kind: "chunker", stage: StageChunk},
{name: "extract", kind: "extractor", stage: StageExtract},
{name: "merge", kind: "merger", stage: StageMerge},
{name: "normalize", kind: "normalizer", stage: StageNormalize},
}

28
internal/framework/pipeline/profile.go
View File

@@ -22,6 +22,7 @@ const (
type ModuleBinding struct {
Module string `json:"module"`
LLMProfile string `json:"llm_profile,omitempty"`
Retries int `json:"retries,omitempty"`
Options map[string]any `json:"options,omitempty"`
References map[string]string `json:"references,omitempty"`
}
@@ -78,6 +79,7 @@ type ResolvedArtifactLane struct {
Normalize ModuleBinding
Validators []ModuleBinding
ExtractReferences ResolvedReferenceTarget `json:"extract_references"`
MergeReferences ResolvedReferenceTarget `json:"merge_references"`
NormalizeReferences ResolvedReferenceTarget `json:"normalize_references"`
}
@@ -251,6 +253,20 @@ func resolveArtifactLane(
if missing, ok := capabilities.missing(mergeSpec.Requires); ok {
return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageMerge, lane.Merge.Module, missing)
}
mergeReferences, err := resolveReferenceTargetBindings(referenceResolutionTarget{
PipelineID: pipelineID,
LaneID: laneID,
Stage: StageMerge,
Module: lane.Merge.Module,
Slots: mergeSpec.ReferenceSlots,
PipelineReferences: pipelineReferences,
LocalReferences: lane.Merge.References,
Options: options,
})
if err != nil {
return ResolvedArtifactLane{}, nil, err
}
lane.MergeReferences = referenceTarget(StageMerge, laneID, lane.Merge.Module, mergeReferences)
capabilities.add(mergeSpec.Provides...)
normalizeSpec, err := normalizerSpec(catalog, lane.Normalize.Module)
@@ -346,6 +362,15 @@ func validatePipelineReferenceDefaults(
declaredByAnyTarget[slot.Name] = struct{}{}
}
merge := resolveBinding(laneProfile.Merge, DefaultMergeModule)
mergeSpec, err := mergerSpec(catalog, merge.Module)
if err != nil {
return moduleLookupError(pipelineID, laneID, StageMerge, merge.Module, err)
}
for _, slot := range mergeSpec.ReferenceSlots {
declaredByAnyTarget[slot.Name] = struct{}{}
}
normalize := resolveBinding(laneProfile.Normalize, DefaultNormalizeModule)
normalizeSpec, err := normalizerSpec(catalog, normalize.Module)
if err != nil {
@@ -500,7 +525,7 @@ func normalizeReferenceOptionTarget(pipelineID string, operation string, stage M
if laneID != "" {
return "", "", fmt.Errorf("pipeline %q reference %s for chunk must not include a lane id", pipelineID, operation)
}
case StageExtract, StageNormalize:
case StageExtract, StageMerge, StageNormalize:
if laneID == "" {
return "", "", fmt.Errorf("pipeline %q reference %s lane id must not be empty", pipelineID, operation)
}
@@ -600,6 +625,7 @@ func resolveBinding(binding ModuleBinding, defaultModule string) ModuleBinding {
return ModuleBinding{
Module: module,
LLMProfile: llmProfile,
Retries: binding.Retries,
Options: cloneOptions(binding.Options),
References: normalizeReferenceMap(binding.References),
}

58
internal/framework/pipeline/profile_test.go
View File

@@ -160,6 +160,9 @@ func TestResolvePipelineAppliesReferenceBindings(t *testing.T) {
if events.NormalizeReferences.Stage != StageNormalize || events.NormalizeReferences.LaneID != "events" || events.NormalizeReferences.Module != DefaultNormalizeModule {
t.Fatalf("normalize reference target = %#v, want event normalizer target", events.NormalizeReferences)
}
if events.MergeReferences.Stage != StageMerge || events.MergeReferences.LaneID != "events" || events.MergeReferences.Module != DefaultMergeModule {
t.Fatalf("merge reference target = %#v, want event merger target", events.MergeReferences)
}
if resolved.ChunkReferences.Stage != StageChunk || resolved.ChunkReferences.Module != DefaultChunkModule {
t.Fatalf("chunk reference target = %#v, want chunk target", resolved.ChunkReferences)
}
@@ -260,12 +263,44 @@ func TestResolvePipelineAppliesPipelineReferenceDefaultToNormalizerTarget(t *tes
}
}
func TestResolvePipelineAppliesPipelineReferenceDefaultToMergeTarget(t *testing.T) {
profile := baselineProfile()
profile.References = map[string]string{"merge_notes": "./merge.md"}
catalog := newProfileCatalogWithOverrides(t, ModuleSpec{
Key: "appendorder",
Stage: StageMerge,
Requires: []string{"candidate"},
Provides: []string{"merged"},
ReferenceSlots: []contracts.ReferenceSlot{{Name: "merge_notes"}},
})
resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog)
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
want := []ReferenceBinding{{LaneID: "events", SlotName: "merge_notes", Source: "./merge.md", BindingSource: contracts.ReferenceBindingSourceConfig}}
if !reflect.DeepEqual(resolved.ArtifactLanes[0].MergeReferences.Bindings, want) {
t.Fatalf("merge references = %#v, want %#v", resolved.ArtifactLanes[0].MergeReferences.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)
}
if refs := resolved.ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 0 {
t.Fatalf("normalize 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: "appendorder", Stage: StageMerge, Requires: []string{"candidate"}, Provides: []string{"merged"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}},
ModuleSpec{Key: "noop", Stage: StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}},
)
@@ -276,6 +311,7 @@ func TestResolvePipelineAppliesOnePipelineReferenceDefaultToMultipleTargets(t *t
assertBindingSource(t, resolved.ChunkReferences.Bindings, "context", "./context.md")
assertBindingSource(t, resolved.ArtifactLanes[0].ExtractReferences.Bindings, "context", "./context.md")
assertBindingSource(t, resolved.ArtifactLanes[0].MergeReferences.Bindings, "context", "./context.md")
assertBindingSource(t, resolved.ArtifactLanes[0].NormalizeReferences.Bindings, "context", "./context.md")
}
@@ -373,6 +409,28 @@ func TestResolvePipelineRejectsExtractLocalReferenceDeclaredOnlyByNormalizer(t *
assertErrorContains(t, err, "baseline", "events", "extract", "event-extractor", "normalization_notes", "not declared")
}
func TestResolvePipelineRejectsMergeLocalReferenceDeclaredOnlyByNormalizer(t *testing.T) {
profile := baselineProfile()
lane := profile.Artifacts["events"]
lane.Merge.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, "merge", "normalization_notes", "not declared")
}
func TestResolvePipelineRejectsNormalizeLocalReferenceDeclaredOnlyByExtractor(t *testing.T) {
profile := baselineProfile()
lane := profile.Artifacts["events"]

11
internal/framework/pipeline/references.go
View File

@@ -44,6 +44,7 @@ func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, opt
for i, lane := range resolved.ArtifactLanes {
materializedLane := lane
materializedLane.ExtractReferences = CloneReferenceTarget(lane.ExtractReferences)
materializedLane.MergeReferences = CloneReferenceTarget(lane.MergeReferences)
materializedLane.NormalizeReferences = CloneReferenceTarget(lane.NormalizeReferences)
extractReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.ExtractReferences, catalog, options)
if err != nil {
@@ -52,6 +53,13 @@ func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, opt
materializedLane.ExtractReferences.ReferenceSet = extractReferenceSet
warnings = append(warnings, laneWarnings...)
mergeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.MergeReferences, catalog, options)
if err != nil {
return ResolvedPipeline{}, nil, err
}
materializedLane.MergeReferences.ReferenceSet = mergeReferenceSet
warnings = append(warnings, laneWarnings...)
normalizeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.NormalizeReferences, catalog, options)
if err != nil {
return ResolvedPipeline{}, nil, err
@@ -140,6 +148,8 @@ func referenceTargetSpec(target ResolvedReferenceTarget, catalog ModuleCatalog)
return registrySpec(catalog.Chunkers, target.Module)
case StageExtract:
return registrySpec(catalog.Extractors, target.Module)
case StageMerge:
return registrySpec(catalog.Mergers, target.Module)
case StageNormalize:
return registrySpec(catalog.Normalizers, target.Module)
default:
@@ -291,6 +301,7 @@ func ReferenceProvenance(resolved ResolvedPipeline) []artifacts.ReferenceProvena
provenance = append(provenance, referenceTargetProvenance(resolved.ChunkReferences)...)
for _, lane := range resolved.ArtifactLanes {
provenance = append(provenance, referenceTargetProvenance(lane.ExtractReferences)...)
provenance = append(provenance, referenceTargetProvenance(lane.MergeReferences)...)
provenance = append(provenance, referenceTargetProvenance(lane.NormalizeReferences)...)
}
return provenance

44
internal/framework/pipeline/references_test.go
View File

@@ -109,17 +109,20 @@ func TestMaterializeReferencesStoresSetsAndProvenanceForAllTargets(t *testing.T)
configDir := t.TempDir()
writeReferenceFile(t, filepath.Join(configDir, "chunk.txt"), []byte("chunk text"))
writeReferenceFile(t, filepath.Join(configDir, "extract.txt"), []byte("extract text"))
writeReferenceFile(t, filepath.Join(configDir, "merge.txt"), []byte("merge text"))
writeReferenceFile(t, filepath.Join(configDir, "normalize.txt"), []byte("normalize text"))
profile := baselineProfile()
profile.References = map[string]string{
"scene_guide": "chunk.txt",
"roster": "extract.txt",
"merge_notes": "merge.txt",
"normalization_notes": "normalize.txt",
}
catalog := referenceCatalogForTargets(t,
[]contracts.ReferenceSlot{{Name: "scene_guide"}},
[]contracts.ReferenceSlot{{Name: "roster"}},
[]contracts.ReferenceSlot{{Name: "merge_notes"}},
[]contracts.ReferenceSlot{{Name: "normalization_notes"}},
)
resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog)
@@ -145,14 +148,18 @@ func TestMaterializeReferencesStoresSetsAndProvenanceForAllTargets(t *testing.T)
if string(extractItem.Content) != "extract text" {
t.Fatalf("extract content = %q, want extract text", extractItem.Content)
}
mergeItem := materialized.ArtifactLanes[0].MergeReferences.ReferenceSet.Slots["merge_notes"].Items[0]
if string(mergeItem.Content) != "merge text" {
t.Fatalf("merge content = %q, want merge text", mergeItem.Content)
}
normalizeItem := materialized.ArtifactLanes[0].NormalizeReferences.ReferenceSet.Slots["normalization_notes"].Items[0]
if string(normalizeItem.Content) != "normalize text" {
t.Fatalf("normalize content = %q, want normalize text", normalizeItem.Content)
}
provenance := ReferenceProvenance(materialized)
if len(provenance) != 3 {
t.Fatalf("ReferenceProvenance() = %#v, want three entries", provenance)
if len(provenance) != 4 {
t.Fatalf("ReferenceProvenance() = %#v, want four entries", provenance)
}
if provenance[0].Stage != string(StageChunk) || provenance[0].LaneID != "" || provenance[0].SlotName != "scene_guide" || provenance[0].Digest != chunkItem.Digest {
t.Fatalf("ReferenceProvenance()[0] = %#v, want chunk scene guide provenance", provenance[0])
@@ -160,7 +167,10 @@ func TestMaterializeReferencesStoresSetsAndProvenanceForAllTargets(t *testing.T)
if provenance[1].Stage != string(StageExtract) || provenance[1].LaneID != "events" || provenance[1].SlotName != "roster" || provenance[1].Digest != extractItem.Digest {
t.Fatalf("ReferenceProvenance()[1] = %#v, want extract roster provenance", provenance[1])
}
if provenance[2].Stage != string(StageNormalize) || provenance[2].LaneID != "events" || provenance[2].SlotName != "normalization_notes" || provenance[2].Digest != normalizeItem.Digest {
if provenance[2].Stage != string(StageMerge) || provenance[2].LaneID != "events" || provenance[2].SlotName != "merge_notes" || provenance[2].Digest != mergeItem.Digest {
t.Fatalf("ReferenceProvenance()[2] = %#v, want merge notes provenance", provenance[2])
}
if provenance[3].Stage != string(StageNormalize) || provenance[3].LaneID != "events" || provenance[3].SlotName != "normalization_notes" || provenance[3].Digest != normalizeItem.Digest {
t.Fatalf("ReferenceProvenance()[2] = %#v, want normalize notes provenance", provenance[2])
}
}
@@ -185,7 +195,7 @@ func TestMaterializeReferencesRejectsNonUTF8ContentForChunkTarget(t *testing.T)
writeReferenceFile(t, path, []byte{0xff, 0xfe})
resolved := resolvedPipelineWithTargetReference(t, StageChunk, "", "scene_guide", "bad.txt", contracts.ReferenceBindingSourceConfig, contracts.ReferenceSlot{Name: "scene_guide"})
_, _, err := MaterializeReferences(resolved, referenceCatalogForTargets(t, []contracts.ReferenceSlot{{Name: "scene_guide"}}, nil, nil), ReferenceMaterializationOptions{
_, _, err := MaterializeReferences(resolved, referenceCatalogForTargets(t, []contracts.ReferenceSlot{{Name: "scene_guide"}}, nil, nil, nil), ReferenceMaterializationOptions{
ConfigPath: filepath.Join(configDir, "config.yml"),
})
if err == nil || !strings.Contains(err.Error(), "chunk") || !strings.Contains(err.Error(), "UTF-8") || !strings.Contains(err.Error(), "scene_guide") || !strings.Contains(err.Error(), path) {
@@ -309,7 +319,7 @@ func TestMaterializeReferencesRejectsUnacceptedMediaTypeForNormalizeTarget(t *te
slot := contracts.ReferenceSlot{Name: "normalization_notes", AcceptedMediaTypes: []string{"text/markdown"}}
resolved := resolvedPipelineWithTargetReference(t, StageNormalize, "events", "normalization_notes", "notes.json", contracts.ReferenceBindingSourceConfig, slot)
_, _, err := MaterializeReferences(resolved, referenceCatalogForTargets(t, nil, nil, []contracts.ReferenceSlot{slot}), ReferenceMaterializationOptions{
_, _, err := MaterializeReferences(resolved, referenceCatalogForTargets(t, nil, nil, nil, []contracts.ReferenceSlot{slot}), ReferenceMaterializationOptions{
ConfigPath: filepath.Join(configDir, "config.yml"),
})
if err == nil || !strings.Contains(err.Error(), "normalize") || !strings.Contains(err.Error(), "media type") || !strings.Contains(err.Error(), "application/json") || !strings.Contains(err.Error(), "normalization_notes") {
@@ -353,6 +363,7 @@ func TestMaterializeReferencesWarningScopesIncludeTargetContext(t *testing.T) {
catalog := referenceCatalogForTargets(t,
[]contracts.ReferenceSlot{{Name: "scene_guide"}},
[]contracts.ReferenceSlot{{Name: "roster"}},
nil,
[]contracts.ReferenceSlot{{Name: "normalization_notes"}},
)
resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog)
@@ -407,6 +418,10 @@ func resolvedPipelineWithTargetReference(t *testing.T, stage ModuleStage, laneID
lane := profile.Artifacts[laneID]
lane.References = map[string]string{slotName: source}
profile.Artifacts[laneID] = lane
case StageMerge:
lane := profile.Artifacts[laneID]
lane.Merge.References = map[string]string{slotName: source}
profile.Artifacts[laneID] = lane
case StageNormalize:
lane := profile.Artifacts[laneID]
lane.Normalize.References = map[string]string{slotName: source}
@@ -425,6 +440,8 @@ func resolvedPipelineWithTargetReference(t *testing.T, stage ModuleStage, laneID
resolved.ChunkReferences.Bindings[0].BindingSource = bindingSource
case StageExtract:
resolved.ArtifactLanes[0].ExtractReferences.Bindings[0].BindingSource = bindingSource
case StageMerge:
resolved.ArtifactLanes[0].MergeReferences.Bindings[0].BindingSource = bindingSource
case StageNormalize:
resolved.ArtifactLanes[0].NormalizeReferences.Bindings[0].BindingSource = bindingSource
}
@@ -441,18 +458,20 @@ func referenceCatalogForStage(t *testing.T, stage ModuleStage, slots []contracts
t.Helper()
switch stage {
case StageChunk:
return referenceCatalogForTargets(t, slots, nil, nil)
return referenceCatalogForTargets(t, slots, nil, nil, nil)
case StageExtract:
return referenceCatalogForTargets(t, nil, slots, nil)
return referenceCatalogForTargets(t, nil, slots, nil, nil)
case StageMerge:
return referenceCatalogForTargets(t, nil, nil, slots, nil)
case StageNormalize:
return referenceCatalogForTargets(t, nil, nil, slots)
return referenceCatalogForTargets(t, nil, nil, nil, slots)
default:
t.Fatalf("unsupported reference target stage %q", stage)
return ModuleCatalog{}
}
}
func referenceCatalogForTargets(t *testing.T, chunkSlots, extractSlots, normalizeSlots []contracts.ReferenceSlot) ModuleCatalog {
func referenceCatalogForTargets(t *testing.T, chunkSlots, extractSlots, mergeSlots, normalizeSlots []contracts.ReferenceSlot) ModuleCatalog {
t.Helper()
return newProfileCatalogWithOverrides(t,
ModuleSpec{
@@ -469,6 +488,13 @@ func referenceCatalogForTargets(t *testing.T, chunkSlots, extractSlots, normaliz
Provides: []string{"candidate"},
ReferenceSlots: extractSlots,
},
ModuleSpec{
Key: "appendorder",
Stage: StageMerge,
Requires: []string{"candidate"},
Provides: []string{"merged"},
ReferenceSlots: mergeSlots,
},
ModuleSpec{
Key: "noop",
Stage: StageNormalize,

4
internal/framework/pipeline/runner.go
View File

@@ -212,6 +212,10 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source
Source: doc,
LaneID: lane.ID,
ExtractOutputs: cloneExtractOutputs(extractOutputs),
SourceInput: sourceInput.Clone(),
SessionID: sessionID,
References: CloneReferenceSet(lane.MergeReferences.ReferenceSet),
LLMClient: input.LLMClient,
LLMProfile: lane.Merge.LLMProfile,
Options: cloneOptions(lane.Merge.Options),
Metadata: input.Metadata,

54
internal/framework/pipeline/runner_test.go
View File

@@ -500,6 +500,9 @@ func TestRunExecutesChunksAndPassesChunkAndLLMClient(t *testing.T) {
if len(normalizer.requests) != 1 || normalizer.requests[0].LLMClient == nil {
t.Fatalf("normalizer LLM client = %#v, want client on normalize request", normalizer.requests)
}
if len(modules.mergers["merge"].requests) != 1 || modules.mergers["merge"].requests[0].LLMClient == nil {
t.Fatalf("merger LLM client = %#v, want client on merge request", modules.mergers["merge"].requests)
}
if extractor.seenMetadata[0]["request"] != "test" {
t.Fatalf("seen metadata = %#v, want request metadata", extractor.seenMetadata)
}
@@ -534,6 +537,7 @@ func TestRunPassesSourceInputAndSessionIDToPromptCapableStages(t *testing.T) {
{name: "chunk", material: modules.chunker.requests[0].SourceInput, sessionID: modules.chunker.requests[0].SessionID},
{name: "extract first", material: modules.extractors["extract-alpha"].requests[0].SourceInput, sessionID: modules.extractors["extract-alpha"].requests[0].SessionID},
{name: "extract second", material: modules.extractors["extract-alpha"].requests[1].SourceInput, sessionID: modules.extractors["extract-alpha"].requests[1].SessionID},
{name: "merge", material: modules.mergers["merge"].requests[0].SourceInput, sessionID: modules.mergers["merge"].requests[0].SessionID},
{name: "normalize", material: modules.normalizers["normalize"].requests[0].SourceInput, sessionID: modules.normalizers["normalize"].requests[0].SessionID},
}
for _, req := range requests {
@@ -688,6 +692,27 @@ func TestRunPassesLaneReferencesToExtractorRequests(t *testing.T) {
}
}
func TestRunPassesMergeReferencesToMergerRequest(t *testing.T) {
modules := defaultRunnerModules()
pipeline := resolvedPipeline()
pipeline.ArtifactLanes[0].MergeReferences.ReferenceSet = testReferenceSet("merge_notes", "merge reference text")
_, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline})
if err != nil {
t.Fatalf("Run() error = %v, want nil", err)
}
req := modules.mergers["merge"].requests[0]
item := req.References.Slots["merge_notes"].Items[0]
if string(item.Content) != "merge reference text" {
t.Fatalf("merge reference content = %q, want merge reference text", item.Content)
}
item.Content[0] = 'M'
if got := string(pipeline.ArtifactLanes[0].MergeReferences.ReferenceSet.Slots["merge_notes"].Items[0].Content); got != "merge reference text" {
t.Fatalf("runner mutated merge reference set content = %q", got)
}
}
func TestRunPassesChunkReferencesToChunkerRequest(t *testing.T) {
modules := defaultRunnerModules()
pipeline := resolvedPipeline()
@@ -984,6 +1009,24 @@ func TestRunManifestIncludesPipelineAndLaneDetails(t *testing.T) {
},
},
}
resolved.ArtifactLanes[0].MergeReferences.ReferenceSet = contracts.ReferenceSet{
Slots: map[string]contracts.ResolvedReferenceSlot{
"merge_notes": {
Slot: contracts.ReferenceSlot{Name: "merge_notes"},
Items: []contracts.ReferenceItem{
{
SlotName: "merge_notes",
MediaType: "text/plain; charset=utf-8",
Content: []byte("merge reference content"),
Digest: "sha256:merge-reference",
Origin: contracts.ReferenceOrigin{Type: "file", URI: "file:///tmp/merge.txt"},
SizeBytes: int64(len("merge reference content")),
BindingSource: contracts.ReferenceBindingSourceConfig,
},
},
},
},
}
resolved.ArtifactLanes[0].NormalizeReferences.ReferenceSet = contracts.ReferenceSet{
Slots: map[string]contracts.ResolvedReferenceSlot{
"normalization_notes": {
@@ -1018,8 +1061,8 @@ func TestRunManifestIncludesPipelineAndLaneDetails(t *testing.T) {
if !reflect.DeepEqual(manifest.SourceDigests, []string{"sha256:source"}) {
t.Fatalf("SourceDigests = %#v, want source digest", manifest.SourceDigests)
}
if len(manifest.References) != 3 {
t.Fatalf("References = %#v, want three reference provenance entries", manifest.References)
if len(manifest.References) != 4 {
t.Fatalf("References = %#v, want four reference provenance entries", manifest.References)
}
chunkReference := manifest.References[0]
if chunkReference.Stage != string(StageChunk) || chunkReference.LaneID != "" || chunkReference.SlotName != "scene_guide" || chunkReference.Digest != "sha256:chunk-reference" {
@@ -1035,7 +1078,11 @@ func TestRunManifestIncludesPipelineAndLaneDetails(t *testing.T) {
if extractReference.OriginType != "file" || extractReference.OriginURI != "file:///tmp/roster.txt" || extractReference.MediaType != "text/plain; charset=utf-8" || extractReference.SizeBytes != int64(len("reference content")) || extractReference.BindingSource != contracts.ReferenceBindingSourceConfig {
t.Fatalf("extract reference provenance = %#v, want origin/media/size/source", extractReference)
}
normalizeReference := manifest.References[2]
mergeReference := manifest.References[2]
if mergeReference.Stage != string(StageMerge) || mergeReference.LaneID != "alpha" || mergeReference.SlotName != "merge_notes" || mergeReference.Digest != "sha256:merge-reference" {
t.Fatalf("merge reference provenance = %#v, want lane slot digest", mergeReference)
}
normalizeReference := manifest.References[3]
if normalizeReference.Stage != string(StageNormalize) || normalizeReference.LaneID != "alpha" || normalizeReference.SlotName != "normalization_notes" || normalizeReference.Digest != "sha256:normalize-reference" {
t.Fatalf("normalize reference provenance = %#v, want lane slot digest", normalizeReference)
}
@@ -1187,6 +1234,7 @@ func resolvedPipeline() ResolvedPipeline {
Merge: Binding("merge"),
Normalize: Binding("normalize"),
ExtractReferences: referenceTarget(StageExtract, "alpha", "extract-alpha", nil),
MergeReferences: referenceTarget(StageMerge, "alpha", "merge", nil),
NormalizeReferences: referenceTarget(StageNormalize, "alpha", "normalize", nil),
},
},

2
internal/framework/pipeline/testdata/walking_skeleton_output.json vendored
View File

@@ -1,7 +1,7 @@
{
"manifest": {
"pipeline_id": "walking-skeleton",
"pipeline_digest": "sha256:437d7ff486c336ddba38654ea00b7ef9dd6e49ce09f468698202ad8fc459bdcd",
"pipeline_digest": "sha256:75c1f6d64d86666734ccf175cdb82ae90e0ba5be3194be952850d7c00c66e615",
"validation_status": "approved",
"artifact_lanes": [
{