diff --git a/docs/cli.md b/docs/cli.md index e39f3af..dd9ef45 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -49,8 +49,8 @@ Flags: use one Scriptorium profile ID. - `--session-id id`: pass a stable prompt session identifier through LLM-backed module calls. -- `--reference selector=path`: bind a reference path to a chunk, extractor, or - normalizer reference slot. Repeatable. +- `--reference selector=path`: bind a reference path to a chunk, extractor, + merger, or normalizer reference slot. Repeatable. - `--without-reference selector`: remove a configured optional reference binding. Repeatable. It accepts the same selector forms as `--reference`, without `=path`. @@ -59,12 +59,12 @@ On success, the command prints the completed pipeline ID, normalized output and rejected output counts, and the output directory. If the run completes with warnings, the warning count is printed to stderr. -Reference flags are resolved against selected chunk, extractor, and normalizer +Reference flags are resolved against selected chunk, extractor, merger, and normalizer targets before the run starts. Flat slot names are accepted only when exactly one selected target declares that slot. Bound reference files are read before pipeline work starts, validated as UTF-8 text, and recorded as provenance for the target that declares the slot. Runtime reference content is passed to the -chunker, extractor, or normalizer target that declares the slot. Notarius infers +chunker, extractor, merger, or normalizer target that declares the slot. Notarius infers reference media types from file extensions for provenance and for optional slot checks. Reference content is not written to diagnostics, logs, errors, or manifests. @@ -81,9 +81,11 @@ Reference binding precedence is: - `slot=path`: valid when exactly one selected target declares `slot`; - `chunk.slot=path`: target the chunker; -- `lane.slot=path`: valid when exactly one selected extractor or normalizer in - that lane declares `slot`; +- `merge.slot=path`: valid when exactly one selected merger declares `slot`; +- `lane.slot=path`: valid when exactly one selected extractor, merger, or + normalizer in that lane declares `slot`; - `lane.extract.slot=path`: target a lane extractor; +- `lane.merge.slot=path`: target a lane merger; - `lane.normalize.slot=path`: target a lane normalizer. Use `slot=path` when the selected targets declare the slot unambiguously: @@ -105,7 +107,7 @@ go run ./cmd/notarius run dnd-session \ --reference spells.extract.glossary=./campaign-glossary.txt ``` -The same grammar can target chunk and normalize slots when the configured +The same grammar can target chunk, merge, and normalize slots when the configured modules declare them: ```sh @@ -113,6 +115,7 @@ go run ./cmd/notarius run dnd-session \ --config path/to/config.yml \ --input examples/seriatim-minimal-transcript.json \ --reference chunk.scene_guide=./campaign-scenes.txt \ + --reference spells.merge.merge_notes=./merge-notes.txt \ --reference spells.normalize.normalization_notes=./normalization-notes.txt ``` diff --git a/docs/config.md b/docs/config.md index d56d10a..fd724b7 100644 --- a/docs/config.md +++ b/docs/config.md @@ -141,13 +141,13 @@ against the production module catalog and fail fast for unknown or incompatible module keys. Reference bindings are validated against reference slots declared by eligible -chunk, extract, and normalize targets during pipeline resolution. Required slots +chunk, extract, merge, and normalize targets during pipeline resolution. Required slots must be bound after config defaults, target-local references, lane-level compatibility bindings, and run-time `--reference` or `--without-reference` overrides are applied. Config-relative paths are resolved relative to the config file; CLI reference paths are resolved relative to the current working directory. Materialized bound files must be UTF-8 text. Materialized reference -provenance is recorded for chunk, extractor, and normalizer targets, and runtime +provenance is recorded for chunk, extractor, merger, and normalizer targets, and runtime reference content is passed to the target that declares the slot. Reference media types are inferred from file extensions, recorded as canonical base media types, and checked only when a module declares `AcceptedMediaTypes`; unknown @@ -156,8 +156,8 @@ written to diagnostics, logs, errors, or manifests. Pipeline-level `references` are defaults. They are valid when at least one eligible target in the full configured pipeline declares the slot, including -chunk, extractor, and normalizer targets. During a run, they apply only to the -selected targets that declare the slot: +chunk, extractor, merger, and normalizer targets. During a run, they apply only +to the selected targets that declare the slot: ```yaml pipelines: @@ -192,15 +192,17 @@ pipelines: party: ./campaign/session-party.txt ``` -`chunk.references` and `normalize.references` are accepted in object-form -bindings. They override pipeline-level defaults for slots declared by the chunk -or normalizer module. Extractor-local references apply only to the extractor, -and normalizer-local references apply only to the normalizer. +`chunk.references`, `merge.references`, and `normalize.references` are accepted +in object-form bindings. They override pipeline-level defaults for slots +declared by that target module. Extractor-local references apply only to the +extractor, merger-local references apply only to the merger, and +normalizer-local references apply only to the normalizer. Target-local reference fields use the same map shape at: - `pipelines..chunk.references` - `pipelines..artifacts..extract.references` +- `pipelines..artifacts..merge.references` - `pipelines..artifacts..normalize.references` Each binding is valid only when that target module declares the slot. @@ -226,13 +228,16 @@ Binding fields: - `module`: module key. - `llm_profile`: optional Scriptorium profile ID. Empty or omitted lets the Scriptorium prompt default select the profile. +- `retries`: non-negative retry count for runtime stages that support retries. + The current runner preserves this value in resolved config. - `options`: optional module-specific settings. - `references`: optional reference bindings. Supported only for `chunk`, - `extract`, and `normalize` bindings. `input`, `merge`, validator, and + `extract`, `merge`, and `normalize` bindings. `input`, validator, and `output` bindings reject this field during validation. The `--llm-profile` run flag overrides every effective LLM-capable module -binding to use one Scriptorium profile ID. +binding to use one Scriptorium profile ID: chunk, every selected lane extract, +merge, and normalize binding. ## Implemented Production Modules diff --git a/docs/internal/llm.md b/docs/internal/llm.md index 7e4e6aa..16d90c4 100644 --- a/docs/internal/llm.md +++ b/docs/internal/llm.md @@ -41,6 +41,10 @@ The runtime records the actual selected Scriptorium profile, provider, and model used during execution. Manifest population does not rely on a precomputed profile ID before pipeline execution. +Explicit profile validation and `--llm-profile` overrides apply to LLM-capable +pipeline stages: chunk, extract, merge, and normalize. Input, output, and +validator bindings are not part of the current production LLM profile scope. + ## Scriptorium Adapter `ScriptoriumClient` implements `contracts.StructuredLLMClient` by converting diff --git a/docs/internal/modules.md b/docs/internal/modules.md index fa2dddb..c95eb34 100644 --- a/docs/internal/modules.md +++ b/docs/internal/modules.md @@ -20,7 +20,7 @@ A production module package should provide: Module specs should describe capabilities accurately. Resolution uses specs to reject incompatible pipelines before execution. -Chunk, extract, and normalize modules that accept auxiliary reference material +Chunk, extract, merge, and normalize modules that accept auxiliary reference material must declare slots through both `ReferenceSlots()` and `ModuleSpec().ReferenceSlots`. The runtime slot list and registry metadata should match so config validation can inspect slots without constructing module @@ -31,10 +31,10 @@ must still be UTF-8 text. When a slot declares accepted media types, Notarius compares the canonical base media type inferred from the file extension, case-insensitively and without parameters. -The resolver materializes reference content for chunk, extractor, and +The resolver materializes reference content for chunk, extractor, merger, and normalizer targets. Runtime delivery uses `contracts.ChunkRequest.References`, -`contracts.ExtractionRequest.References`, and -`contracts.NormalizeRequest.References`. Reference material is not source +`contracts.ExtractionRequest.References`, `contracts.MergeRequest.References`, +and `contracts.NormalizeRequest.References`. Reference material is not source evidence and must not be converted into `SourceRef` values. If a module prompt uses references, pass them as prompt input materials through the structured LLM request. Prompt metadata hashes remain based on prompt asset source, not @@ -48,7 +48,7 @@ Common D&D prompt fragments, reference slot helpers, prompt input assembly, and reference rendering live under `internal/modules/sharedassets/dnd`. Module contracts should expose prompt IDs, versions, input material names, and non-secret prompt/schema hashes through manifest metadata; they should not -expose Scriptorium public types through chunk, extract, or normalize contracts. +expose Scriptorium public types through chunk, extract, merge, or normalize contracts. Chunk modules receive the structured LLM client, configured Scriptorium profile ID, prompt session ID, and raw source input material through @@ -60,6 +60,10 @@ Normalize modules receive the structured LLM client, configured Scriptorium profile ID, prompt session ID, and reference material through `contracts.NormalizeRequest` when they need model-backed reconciliation. +Merge modules receive the structured LLM client, configured Scriptorium profile +ID, prompt session ID, raw source input material, and reference material through +`contracts.MergeRequest` when they need model-backed merge behavior. + ## `seriatim` Input Package: `internal/modules/input/seriatim` diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 17fe044..55ccfa6 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -33,18 +33,19 @@ The CLI writes the resolved pipeline and digest to diagnostics. Pipeline profiles and artifact lanes may include reference binding maps keyed by reference slot name. During resolution, pipeline-level bindings act as defaults -for selected chunk, extractor, and normalizer targets that declare the slot; -target-local bindings override or add bindings for that target. Runtime +for selected chunk, extractor, merger, and normalizer targets that declare the +slot; target-local bindings override or add bindings for that target. Runtime `--reference` requests override target config bindings, and runtime unbinds remove optional target bindings. Flat runtime slot names are resolved only when exactly one selected target declares the slot; otherwise the CLI requires a more -specific selector such as `chunk.slot`, `lane.extract.slot`, or -`lane.normalize.slot`. Resolution validates bindings against the declaring -target specs and stores the bindings in target-aware resolved reference holders. -It does not read reference files or include reference bytes in source digests. +specific selector such as `chunk.slot`, `lane.extract.slot`, +`lane.merge.slot`, or `lane.normalize.slot`. Resolution validates bindings +against the declaring target specs and stores the bindings in target-aware +resolved reference holders. It does not read reference files or include +reference bytes in source digests. -During run preparation, resolved file references for chunk, extractor, and -normalizer targets are materialized before any LLM-backed pipeline work. Config +During run preparation, resolved file references for chunk, extractor, merger, +and normalizer targets are materialized before any LLM-backed pipeline work. Config bindings resolve relative to the config file, and CLI bindings resolve relative to the current working directory. Materialization accepts UTF-8 text files, computes `sha256:` content digests, records file origins, infers canonical base @@ -55,8 +56,8 @@ empty bound files. Media-type acceptance is checked only when a slot declares manifests. The CLI writes provenance-only resolved reference diagnostics, and the run manifest records target-stage reference provenance separately from source digests. Runtime reference content is passed to the matching chunker, -extractor, or normalizer request. LLM-backed modules pass that material onward -as named Scriptorium prompt inputs. +extractor, merger, or normalizer request. LLM-backed modules pass that material +onward as named Scriptorium prompt inputs. The CLI carries raw input bytes into `pipeline.RunInput`. Input adapters parse those bytes into the source document, while LLM-backed modules that need the @@ -65,9 +66,9 @@ origin metadata. The raw input payload is not written to manifests or default diagnostics. The CLI also carries an optional run `session_id`. The runner makes it available -to chunk, extract, and normalize requests; LLM-backed modules forward it through -their structured completion requests so Scriptorium can include it in prompt -execution metadata. +to chunk, extract, merge, and normalize requests; LLM-backed modules forward it +through their structured completion requests so Scriptorium can include it in +prompt execution metadata. ## Registries And Module Specs @@ -83,10 +84,9 @@ Every production module registers a `ModuleSpec` with: - `Provides`: capabilities added after that module runs; - `Requires`: capabilities that must already be available. -Chunk, extract, and normalize specs may also declare reference slots. Slot +Chunk, extract, merge, and normalize specs may also declare reference slots. Slot declarations are available from registry metadata without constructing module -instances. Input, merge, validate, and output specs must not declare reference -slots. +instances. Input, validate, and output specs must not declare reference slots. Capability checks prevent incompatible pipeline composition before a run starts. diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 5df3506..ded07ae 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -118,16 +118,18 @@ Symptoms include: Fix: -- Confirm the selected chunker, extractor, or normalizer declares the slot. The +- Confirm the selected chunker, extractor, merger, or normalizer declares the slot. The implemented `dnd/scenes` chunker and `dnd/spells` extractor declare optional `roster` and `glossary` slots. - Use a specific selector when more than one selected target declares the same - slot. Examples include `chunk.context=./context.txt`, - `spells.extract.context=./extract-context.txt`, and + slot: `chunk.context=./context.txt`, + `spells.extract.context=./extract-context.txt`, + `spells.merge.context=./merge-context.txt`, or `spells.normalize.context=./normalize-context.txt`. -- `lane.slot=path` is valid only when exactly one selected extractor or - normalizer in that lane declares the slot. If both do, use - `lane.extract.slot=path` or `lane.normalize.slot=path`. +- `lane.slot=path` is valid only when exactly one selected extractor, merger, + or normalizer in that lane declares the slot. If more than one does, use + `lane.extract.slot=path`, `lane.merge.slot=path`, or + `lane.normalize.slot=path`. - Use `--without-reference selector` to remove optional config bindings; do not pass an empty `--reference selector=`. - Check whether a path came from config or CLI. Config paths are relative to diff --git a/internal/cli/run.go b/internal/cli/run.go index 5a11107..b66610a 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -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)) } } diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index bd7c848..51b3ffb 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -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) } diff --git a/internal/core/config/effective_config.go b/internal/core/config/effective_config.go index 94cf6e8..68c22cb 100644 --- a/internal/core/config/effective_config.go +++ b/internal/core/config/effective_config.go @@ -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 } diff --git a/internal/core/config/effective_config_test.go b/internal/core/config/effective_config_test.go index c0847c4..d0eedb9 100644 --- a/internal/core/config/effective_config_test.go +++ b/internal/core/config/effective_config_test.go @@ -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 } diff --git a/internal/core/config/file_config.go b/internal/core/config/file_config.go index 8d14f6f..cec535a 100644 --- a/internal/core/config/file_config.go +++ b/internal/core/config/file_config.go @@ -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), } diff --git a/internal/core/config/file_config_test.go b/internal/core/config/file_config_test.go index a2e86da..12a9d3b 100644 --- a/internal/core/config/file_config_test.go +++ b/internal/core/config/file_config_test.go @@ -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) } diff --git a/internal/core/config/redaction.go b/internal/core/config/redaction.go index 7c1437b..b07619b 100644 --- a/internal/core/config/redaction.go +++ b/internal/core/config/redaction.go @@ -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)) diff --git a/internal/core/config/validation.go b/internal/core/config/validation.go index 907d5a3..deac65a 100644 --- a/internal/core/config/validation.go +++ b/internal/core/config/validation.go @@ -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 } diff --git a/internal/core/config/validation_test.go b/internal/core/config/validation_test.go index 98b03cd..b674a54 100644 --- a/internal/core/config/validation_test.go +++ b/internal/core/config/validation_test.go @@ -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 { diff --git a/internal/framework/contracts/contracts.go b/internal/framework/contracts/contracts.go index 2948cfa..5892b85 100644 --- a/internal/framework/contracts/contracts.go +++ b/internal/framework/contracts/contracts.go @@ -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"` diff --git a/internal/framework/pipeline/module.go b/internal/framework/pipeline/module.go index e356a1c..f4cf397 100644 --- a/internal/framework/pipeline/module.go +++ b/internal/framework/pipeline/module.go @@ -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 { diff --git a/internal/framework/pipeline/module_test.go b/internal/framework/pipeline/module_test.go index 8926c18..acb6b3d 100644 --- a/internal/framework/pipeline/module_test.go +++ b/internal/framework/pipeline/module_test.go @@ -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}, } diff --git a/internal/framework/pipeline/profile.go b/internal/framework/pipeline/profile.go index 5bfb5f2..61a11ef 100644 --- a/internal/framework/pipeline/profile.go +++ b/internal/framework/pipeline/profile.go @@ -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), } diff --git a/internal/framework/pipeline/profile_test.go b/internal/framework/pipeline/profile_test.go index d32029c..08f0077 100644 --- a/internal/framework/pipeline/profile_test.go +++ b/internal/framework/pipeline/profile_test.go @@ -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"] diff --git a/internal/framework/pipeline/references.go b/internal/framework/pipeline/references.go index 59ff30a..9f7d49a 100644 --- a/internal/framework/pipeline/references.go +++ b/internal/framework/pipeline/references.go @@ -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 diff --git a/internal/framework/pipeline/references_test.go b/internal/framework/pipeline/references_test.go index a7356e6..b045a4f 100644 --- a/internal/framework/pipeline/references_test.go +++ b/internal/framework/pipeline/references_test.go @@ -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, diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index fe91137..1b48b9f 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -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, diff --git a/internal/framework/pipeline/runner_test.go b/internal/framework/pipeline/runner_test.go index 0517ad6..d8bc16a 100644 --- a/internal/framework/pipeline/runner_test.go +++ b/internal/framework/pipeline/runner_test.go @@ -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), }, }, diff --git a/internal/framework/pipeline/testdata/walking_skeleton_output.json b/internal/framework/pipeline/testdata/walking_skeleton_output.json index 7b9c3cc..7594653 100644 --- a/internal/framework/pipeline/testdata/walking_skeleton_output.json +++ b/internal/framework/pipeline/testdata/walking_skeleton_output.json @@ -1,7 +1,7 @@ { "manifest": { "pipeline_id": "walking-skeleton", - "pipeline_digest": "sha256:437d7ff486c336ddba38654ea00b7ef9dd6e49ce09f468698202ad8fc459bdcd", + "pipeline_digest": "sha256:75c1f6d64d86666734ccf175cdb82ae90e0ba5be3194be952850d7c00c66e615", "validation_status": "approved", "artifact_lanes": [ {