diff --git a/docs/cli.md b/docs/cli.md index 1612589..f886952 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -20,7 +20,7 @@ a bearer token. ```text notarius help -notarius run --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] [--reference slot=path] [--without-reference slot] +notarius run --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] [--reference selector=path] [--without-reference selector] notarius config validate --config path/to/config.yml [--pipeline pipeline-id] [--only lane-a,lane-b] notarius pipelines list --config path/to/config.yml [--json] ``` @@ -46,34 +46,42 @@ Flags: invocation. - `--llm-profile id`: override every effective module binding to use one LLM profile. -- `--reference slot=path`: bind a reference path to an extractor reference - slot. Repeatable. Use `lane.slot=path` when multiple selected lanes declare - the same slot. -- `--without-reference slot`: remove a configured optional reference binding. - Repeatable. Use `lane.slot` when multiple selected lanes declare the same - slot. +- `--reference selector=path`: bind a reference path to a chunk, extractor, 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`. On success, the command prints the completed pipeline ID, approved and rejected artifact counts, and the output directory. If the run completes with warnings, the warning count is printed to stderr. -Reference flags are resolved against selected artifact lanes before the run -starts. Flat slot names are accepted only when exactly one selected lane -declares that slot. Bound reference files are read before extraction, validated -as UTF-8 text, and passed only to the lane extractor 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. +Reference flags are resolved against selected chunk, extractor, and normalizer +targets before the run starts. Flat slot names are accepted only when exactly +one selected target declares that slot. Bound extractor reference files are read +before extraction, validated as UTF-8 text, and passed only to the lane +extractor 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. Reference binding precedence is: 1. pipeline-level config `references`; -2. lane-level config `references`; +2. target-local config references, including legacy lane-level extractor + `references`; 3. `--reference` run flags; 4. `--without-reference` run flags. -`--reference` binds or replaces one slot for one selected lane. Use -`slot=path` when the selected lanes declare the slot unambiguously: +`--reference` binds or replaces one slot for one selected target. Selectors are: + +- `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`; +- `lane.extract.slot=path`: target a lane extractor; +- `lane.normalize.slot=path`: target a lane normalizer. + +Use `slot=path` when the selected targets declare the slot unambiguously: ```sh go run ./cmd/notarius run dnd-session \ @@ -82,14 +90,14 @@ go run ./cmd/notarius run dnd-session \ --reference roster=./campaign-roster.txt ``` -Use `lane.slot=path` when multiple selected lanes declare the same slot or when -you want to target a specific lane: +Use an explicit selector when multiple selected targets declare the same slot or +when you want to target a specific target: ```sh go run ./cmd/notarius run dnd-session \ --config examples/dnd-spells.config.yml \ --input examples/seriatim-minimal-transcript.json \ - --reference spells.glossary=./campaign-glossary.txt + --reference spells.extract.glossary=./campaign-glossary.txt ``` Use `--without-reference` to remove a configured optional binding for a run: diff --git a/docs/config.md b/docs/config.md index 563e540..69580a2 100644 --- a/docs/config.md +++ b/docs/config.md @@ -145,14 +145,14 @@ module keys. Reference bindings are validated against reference slots declared by eligible chunk, extract, and normalize targets during pipeline resolution. Required slots must be bound after config defaults, target-local references, lane-level -compatibility bindings, and extractor run-time `--reference` or -`--without-reference` overrides are applied. Config-relative paths are resolved -relative to the config file; CLI reference paths are resolved relative to the -current working directory. Materialized bound files must be UTF-8 text and are -currently passed only to lane extractors that declare the slot. Reference media -types are inferred from file extensions, recorded as canonical base media types, -and checked only when a module declares `AcceptedMediaTypes`; unknown extensions -are recorded as `application/octet-stream`. Reference content is not written to +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 and are currently passed +only to lane extractors that declare the slot. Reference media types are +inferred from file extensions, recorded as canonical base media types, and +checked only when a module declares `AcceptedMediaTypes`; unknown extensions are +recorded as `application/octet-stream`. Reference content is not written to diagnostics, logs, errors, or manifests. Pipeline-level `references` are defaults. They are valid when at least one diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 1a18ae5..192ad1e 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -34,10 +34,11 @@ 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 -`--reference` requests override extractor config bindings, and runtime unbinds -remove optional extractor bindings. Flat runtime slot names are resolved only -when exactly one selected extractor lane declares the slot; otherwise the CLI -requires `lane.slot`. Resolution validates bindings against the declaring +`--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. diff --git a/internal/cli/run.go b/internal/cli/run.go index 599a13c..0d0b8e2 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -25,7 +25,7 @@ const defaultOutputRoot = "./notarius-output" const usage = `Usage: notarius help - notarius run --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] [--reference slot=path] [--without-reference slot] + notarius run --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] [--reference selector=path] [--without-reference selector] notarius config validate --config path/to/config.yml [--pipeline pipeline-id] [--only lane-a,lane-b] notarius pipelines list --config path/to/config.yml [--json] ` @@ -97,8 +97,8 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i llmProfile := fs.String("llm-profile", "", "LLM profile override") referenceFlags := stringListFlag{} withoutReferenceFlags := stringListFlag{} - fs.Var(&referenceFlags, "reference", "reference binding, as slot=path or lane.slot=path") - fs.Var(&withoutReferenceFlags, "without-reference", "unbind a reference, as slot or lane.slot") + 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(&withoutReferenceFlags, "without-reference", "unbind a reference, using the same selector forms as --reference") if err := fs.Parse(reorderRunArgs(args)); err != nil { fmt.Fprintf(stderr, "notarius: %v\n", err) return 2 @@ -712,13 +712,17 @@ func (flag *stringListFlag) Set(value string) error { } type cliReferenceRequest struct { - LaneID string - SlotName string + Selector cliReferenceSelector Source string } type cliReferenceUnbindRequest struct { + Selector cliReferenceSelector +} + +type cliReferenceSelector struct { LaneID string + Stage pipeline.ModuleStage SlotName string } @@ -735,13 +739,12 @@ func parseReferenceFlags(values []string) ([]cliReferenceRequest, error) { if strings.TrimSpace(source) == "" { return nil, fmt.Errorf("--reference path must not be empty; use --without-reference to unbind") } - laneID, slotName, err := parseReferenceSelector(name, "--reference") + selector, err := parseReferenceSelector(name, "--reference") if err != nil { return nil, err } requests = append(requests, cliReferenceRequest{ - LaneID: laneID, - SlotName: slotName, + Selector: selector, Source: strings.TrimSpace(source), }) } @@ -755,39 +758,51 @@ func parseReferenceUnbindFlags(values []string) ([]cliReferenceUnbindRequest, er requests := make([]cliReferenceUnbindRequest, 0, len(values)) for _, raw := range values { if strings.Contains(raw, "=") { - return nil, fmt.Errorf("--without-reference must use slot or lane.slot") + return nil, fmt.Errorf("--without-reference must use a reference selector without =path") } - laneID, slotName, err := parseReferenceSelector(raw, "--without-reference") + selector, err := parseReferenceSelector(raw, "--without-reference") if err != nil { return nil, err } requests = append(requests, cliReferenceUnbindRequest{ - LaneID: laneID, - SlotName: slotName, + Selector: selector, }) } return requests, nil } -func parseReferenceSelector(raw string, flagName string) (string, string, error) { +func parseReferenceSelector(raw string, flagName string) (cliReferenceSelector, error) { selector := strings.TrimSpace(raw) if selector == "" { - return "", "", fmt.Errorf("%s reference slot must not be empty", flagName) + return cliReferenceSelector{}, fmt.Errorf("%s reference slot must not be empty", flagName) } - if strings.Count(selector, ".") > 1 { - return "", "", fmt.Errorf("%s must use slot or lane.slot", flagName) - } - laneID := "" - slotName := selector - if strings.Contains(selector, ".") { - before, after, _ := strings.Cut(selector, ".") - laneID = strings.TrimSpace(before) - slotName = strings.TrimSpace(after) - if laneID == "" || slotName == "" { - return "", "", fmt.Errorf("%s must use non-empty lane.slot values", flagName) + parts := strings.Split(selector, ".") + for _, part := range parts { + if strings.TrimSpace(part) == "" { + return cliReferenceSelector{}, fmt.Errorf("%s must use non-empty reference selector values", flagName) } } - return laneID, slotName, nil + switch len(parts) { + case 1: + return cliReferenceSelector{SlotName: strings.TrimSpace(parts[0])}, nil + case 2: + first := strings.TrimSpace(parts[0]) + slotName := strings.TrimSpace(parts[1]) + if first == string(pipeline.StageChunk) { + return cliReferenceSelector{Stage: pipeline.StageChunk, 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) + } + 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) + } } func resolveCLIReferenceRequests( @@ -802,20 +817,21 @@ func resolveCLIReferenceRequests( return nil, nil, nil } - selected, err := selectedReferenceLanes(cfg, pipelineID, only, catalog) + targets, err := selectedReferenceTargets(cfg, pipelineID, only, catalog) if err != nil { return nil, nil, err } overrides := make([]pipeline.ReferenceBinding, 0, len(referenceRequests)) for _, request := range referenceRequests { - laneID, err := resolveCLIReferenceLane(selected, request.LaneID, request.SlotName) + target, err := resolveCLIReferenceTarget(targets, request.Selector) if err != nil { return nil, nil, err } overrides = append(overrides, pipeline.ReferenceBinding{ - LaneID: laneID, - SlotName: request.SlotName, + Stage: target.stage, + LaneID: target.laneID, + SlotName: request.Selector.SlotName, Source: request.Source, BindingSource: contracts.ReferenceBindingSourceCLI, }) @@ -823,25 +839,28 @@ func resolveCLIReferenceRequests( unbinds := make([]pipeline.ReferenceUnbind, 0, len(unbindRequests)) for _, request := range unbindRequests { - laneID, err := resolveCLIReferenceLane(selected, request.LaneID, request.SlotName) + target, err := resolveCLIReferenceTarget(targets, request.Selector) if err != nil { return nil, nil, err } unbinds = append(unbinds, pipeline.ReferenceUnbind{ - LaneID: laneID, - SlotName: request.SlotName, + Stage: target.stage, + LaneID: target.laneID, + SlotName: request.Selector.SlotName, }) } return overrides, unbinds, nil } -type selectedReferenceLane struct { - id string - slots map[string]struct{} +type selectedReferenceTarget struct { + laneID string + stage pipeline.ModuleStage + module string + slots map[string]struct{} } -func selectedReferenceLanes(cfg config.Config, pipelineID string, only []string, catalog pipeline.ModuleCatalog) ([]selectedReferenceLane, error) { +func selectedReferenceTargets(cfg config.Config, pipelineID string, only []string, catalog pipeline.ModuleCatalog) ([]selectedReferenceTarget, error) { profile, ok := lookupCLIReferencePipeline(cfg.Pipelines, pipelineID) if !ok { return nil, fmt.Errorf("pipeline %q is not configured", strings.TrimSpace(pipelineID)) @@ -882,27 +901,55 @@ func selectedReferenceLanes(cfg config.Config, pipelineID string, only []string, } sort.Strings(selectedIDs) - selected := make([]selectedReferenceLane, 0, len(selectedIDs)) + targets := make([]selectedReferenceTarget, 0, 1+len(selectedIDs)*2) + chunk := pipeline.Binding(profile.Chunk.Module) + chunk.Module = strings.TrimSpace(profile.Chunk.Module) + if chunk.Module == "" { + chunk.Module = pipeline.DefaultChunkModule + } + chunkSpec, err := cliReferenceChunkerSpec(catalog, chunk.Module) + if err != nil { + return nil, fmt.Errorf("pipeline %q chunk module %q: %w", strings.TrimSpace(pipelineID), chunk.Module, err) + } + targets = append(targets, selectedReferenceTarget{ + stage: pipeline.StageChunk, + module: chunk.Module, + slots: referenceSlotSet(chunkSpec.ReferenceSlots), + }) + for _, laneID := range selectedIDs { lane := lanesByID[laneID] extractModule := strings.TrimSpace(lane.Extract.Module) if extractModule == "" { return nil, fmt.Errorf("pipeline %q lane %q extract module must not be empty", strings.TrimSpace(pipelineID), laneID) } - if catalog.Extractors == nil { - return nil, fmt.Errorf("pipeline %q lane %q extract module %q: module %q is not registered", strings.TrimSpace(pipelineID), laneID, extractModule, extractModule) + extractSpec, err := cliReferenceExtractorSpec(catalog, extractModule) + if err != nil { + return nil, fmt.Errorf("pipeline %q lane %q extract module %q: %w", strings.TrimSpace(pipelineID), laneID, extractModule, err) } - spec, ok := catalog.Extractors.Spec(extractModule) - if !ok { - return nil, fmt.Errorf("pipeline %q lane %q extract module %q: module %q is not registered", strings.TrimSpace(pipelineID), laneID, extractModule, extractModule) + targets = append(targets, selectedReferenceTarget{ + laneID: laneID, + stage: pipeline.StageExtract, + module: extractModule, + slots: referenceSlotSet(extractSpec.ReferenceSlots), + }) + + normalizeModule := strings.TrimSpace(lane.Normalize.Module) + if normalizeModule == "" { + normalizeModule = pipeline.DefaultNormalizeModule } - slotSet := make(map[string]struct{}, len(spec.ReferenceSlots)) - for _, slot := range spec.ReferenceSlots { - slotSet[slot.Name] = struct{}{} + normalizeSpec, err := cliReferenceNormalizerSpec(catalog, normalizeModule) + if err != nil { + return nil, fmt.Errorf("pipeline %q lane %q normalize module %q: %w", strings.TrimSpace(pipelineID), laneID, normalizeModule, err) } - selected = append(selected, selectedReferenceLane{id: laneID, slots: slotSet}) + targets = append(targets, selectedReferenceTarget{ + laneID: laneID, + stage: pipeline.StageNormalize, + module: normalizeModule, + slots: referenceSlotSet(normalizeSpec.ReferenceSlots), + }) } - return selected, nil + return targets, nil } func lookupCLIReferencePipeline(profiles map[string]pipeline.PipelineProfile, pipelineID string) (pipeline.PipelineProfile, bool) { @@ -915,37 +962,152 @@ func lookupCLIReferencePipeline(profiles map[string]pipeline.PipelineProfile, pi return pipeline.PipelineProfile{}, false } -func resolveCLIReferenceLane(selected []selectedReferenceLane, requestedLaneID string, slotName string) (string, error) { - slotName = strings.TrimSpace(slotName) - requestedLaneID = strings.TrimSpace(requestedLaneID) - if requestedLaneID != "" { - for _, lane := range selected { - if lane.id == requestedLaneID { - if _, ok := lane.slots[slotName]; !ok { - return "", fmt.Errorf("reference slot %q is not declared by selected lane %q", slotName, requestedLaneID) +func cliReferenceChunkerSpec(catalog pipeline.ModuleCatalog, module string) (pipeline.ModuleSpec, error) { + if catalog.Chunkers == nil { + return pipeline.ModuleSpec{}, fmt.Errorf("module %q is not registered", module) + } + spec, ok := catalog.Chunkers.Spec(module) + if !ok { + return pipeline.ModuleSpec{}, fmt.Errorf("module %q is not registered", module) + } + return spec, nil +} + +func cliReferenceExtractorSpec(catalog pipeline.ModuleCatalog, module string) (pipeline.ModuleSpec, error) { + if catalog.Extractors == nil { + return pipeline.ModuleSpec{}, fmt.Errorf("module %q is not registered", module) + } + spec, ok := catalog.Extractors.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) + } + spec, ok := catalog.Normalizers.Spec(module) + if !ok { + return pipeline.ModuleSpec{}, fmt.Errorf("module %q is not registered", module) + } + return spec, nil +} + +func referenceSlotSet(slots []contracts.ReferenceSlot) map[string]struct{} { + slotSet := make(map[string]struct{}, len(slots)) + for _, slot := range slots { + slotSet[slot.Name] = struct{}{} + } + return slotSet +} + +func resolveCLIReferenceTarget(targets []selectedReferenceTarget, selector cliReferenceSelector) (selectedReferenceTarget, error) { + slotName := strings.TrimSpace(selector.SlotName) + if slotName == "" { + return selectedReferenceTarget{}, fmt.Errorf("reference slot must not be empty") + } + if selector.Stage == pipeline.StageChunk { + for _, target := range targets { + if target.stage != pipeline.StageChunk { + continue + } + if _, ok := target.slots[slotName]; !ok { + return selectedReferenceTarget{}, fmt.Errorf("reference slot %q is not declared by chunk module %q", slotName, target.module) + } + return target, nil + } + return selectedReferenceTarget{}, fmt.Errorf("reference chunk target is not selected") + } + if selector.Stage == pipeline.StageExtract || selector.Stage == pipeline.StageNormalize { + for _, target := range targets { + if target.laneID == selector.LaneID && target.stage == selector.Stage { + if _, ok := target.slots[slotName]; !ok { + return selectedReferenceTarget{}, fmt.Errorf("reference slot %q is not declared by selected %s target %q", slotName, selector.Stage, targetLabel(target)) } - return requestedLaneID, nil + return target, nil } } - return "", fmt.Errorf("reference lane %q is not selected", requestedLaneID) + return selectedReferenceTarget{}, fmt.Errorf("reference lane %q is not selected", selector.LaneID) } + if strings.TrimSpace(selector.LaneID) != "" { + return resolveCLIReferenceLaneTarget(targets, strings.TrimSpace(selector.LaneID), slotName) + } + return resolveCLIReferenceFlatTarget(targets, slotName) +} - matches := make([]string, 0, 1) - for _, lane := range selected { - if _, ok := lane.slots[slotName]; ok { - matches = append(matches, lane.id) +func resolveCLIReferenceLaneTarget(targets []selectedReferenceTarget, laneID string, slotName string) (selectedReferenceTarget, error) { + laneSelected := false + matches := make([]selectedReferenceTarget, 0, 2) + for _, target := range targets { + if target.laneID != laneID { + continue + } + laneSelected = true + if _, ok := target.slots[slotName]; ok { + matches = append(matches, target) + } + } + if !laneSelected { + return selectedReferenceTarget{}, fmt.Errorf("reference lane %q is not selected", laneID) + } + switch len(matches) { + case 0: + return selectedReferenceTarget{}, fmt.Errorf("reference slot %q is not declared by selected lane %q", slotName, laneID) + 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) + } +} + +func resolveCLIReferenceFlatTarget(targets []selectedReferenceTarget, slotName string) (selectedReferenceTarget, error) { + matches := make([]selectedReferenceTarget, 0, 2) + for _, target := range targets { + if _, ok := target.slots[slotName]; ok { + matches = append(matches, target) } } switch len(matches) { case 0: - return "", fmt.Errorf("reference slot %q is not declared by any selected lane", slotName) + return selectedReferenceTarget{}, fmt.Errorf("reference slot %q is not declared by any selected reference target", slotName) case 1: return matches[0], nil default: - return "", fmt.Errorf("reference slot %q is declared by multiple selected lanes (%s); use lane.slot", slotName, strings.Join(matches, ", ")) + return selectedReferenceTarget{}, fmt.Errorf("reference slot %q is declared by multiple selected targets (%s); use a more specific selector such as %s", slotName, targetList(matches), selectorSuggestions(matches, slotName)) } } +func targetList(targets []selectedReferenceTarget) string { + labels := make([]string, 0, len(targets)) + for _, target := range targets { + labels = append(labels, targetLabel(target)) + } + sort.Strings(labels) + return strings.Join(labels, ", ") +} + +func targetLabel(target selectedReferenceTarget) string { + if target.stage == pipeline.StageChunk { + return "chunk" + } + return target.laneID + "." + string(target.stage) +} + +func selectorSuggestions(targets []selectedReferenceTarget, slotName string) string { + suggestions := make([]string, 0, len(targets)) + for _, target := range targets { + if target.stage == pipeline.StageChunk { + suggestions = append(suggestions, "chunk."+slotName) + continue + } + suggestions = append(suggestions, target.laneID+"."+string(target.stage)+"."+slotName) + } + sort.Strings(suggestions) + return strings.Join(suggestions, " or ") +} + func sortedPipelineIDs(cfg config.Config) []string { ids := make([]string, 0, len(cfg.Pipelines)) for id := range cfg.Pipelines { diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index e77ba2f..3f3b01e 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -819,6 +819,193 @@ func TestRunPipelineReferenceFlagBindsLaneQualifiedSlot(t *testing.T) { } } +func TestRunPipelineReferenceFlagBindsChunkQualifiedSlot(t *testing.T) { + configPath := writeTestConfig(t, testConfigYAML("example", "events")) + inputPath := filepath.Join(t.TempDir(), "missing.json") + referencePath := writeFile(t, "scenes.md", "Scenes\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", "chunk.scene_guide=" + referencePath, + }, &stdout, &stderr, Options{ + Catalog: fakeCatalog(t, pipeline.ModuleSpec{ + Key: "generic", + Stage: pipeline.StageChunk, + Requires: []string{"source"}, + Provides: []string{"chunks"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "scene_guide"}, + }, + }), + }) + + 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.ChunkReferences.Bindings + want := []pipeline.ReferenceBinding{ + {SlotName: "scene_guide", Source: referencePath, BindingSource: contracts.ReferenceBindingSourceCLI}, + } + if !reflect.DeepEqual(refs, want) { + t.Fatalf("chunk references = %#v, want %#v", refs, want) + } +} + +func TestRunPipelineReferenceFlagBindsExplicitExtractSlot(t *testing.T) { + configPath := writeTestConfig(t, testConfigYAML("example", "events")) + inputPath := filepath.Join(t.TempDir(), "missing.json") + referencePath := writeFile(t, "roster.yml", "Aria\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.extract.roster=" + referencePath, + }, &stdout, &stderr, Options{ + Catalog: fakeCatalog(t, pipeline.ModuleSpec{ + Key: "fake/extract", + Stage: pipeline.StageExtract, + Requires: []string{"chunks"}, + Provides: []string{"artifact"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "roster"}, + }, + }), + }) + + 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].ExtractReferences.Bindings + want := []pipeline.ReferenceBinding{ + {LaneID: "events", SlotName: "roster", Source: referencePath, BindingSource: contracts.ReferenceBindingSourceCLI}, + } + if !reflect.DeepEqual(refs, want) { + t.Fatalf("extract references = %#v, want %#v", refs, want) + } +} + +func TestRunPipelineReferenceFlagBindsExplicitNormalizeSlot(t *testing.T) { + configPath := writeTestConfig(t, testConfigYAML("example", "events")) + inputPath := filepath.Join(t.TempDir(), "missing.json") + referencePath := writeFile(t, "normalize.md", "Normalize\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.normalize.notes=" + referencePath, + }, &stdout, &stderr, Options{ + Catalog: fakeCatalog(t, pipeline.ModuleSpec{ + Key: "noop", + Stage: pipeline.StageNormalize, + Requires: []string{"merged"}, + Provides: []string{"normalized"}, + 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].NormalizeReferences.Bindings + want := []pipeline.ReferenceBinding{ + {LaneID: "events", SlotName: "notes", Source: referencePath, BindingSource: contracts.ReferenceBindingSourceCLI}, + } + if !reflect.DeepEqual(refs, want) { + t.Fatalf("normalize references = %#v, want %#v", refs, want) + } +} + +func TestRunPipelineReferenceFlagBindsFlatSlotAcrossOneTarget(t *testing.T) { + configPath := writeTestConfig(t, testConfigYAML("example", "events")) + inputPath := filepath.Join(t.TempDir(), "missing.json") + referencePath := writeFile(t, "normalize.md", "Normalize\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", "notes=" + referencePath, + }, &stdout, &stderr, Options{ + Catalog: fakeCatalog(t, pipeline.ModuleSpec{ + Key: "noop", + Stage: pipeline.StageNormalize, + Requires: []string{"merged"}, + Provides: []string{"normalized"}, + 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) + if refs := resolved.ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 1 || refs[0].Source != referencePath { + t.Fatalf("normalize references = %#v, want flat binding", refs) + } +} + +func TestRunPipelineReferenceFlagBindsLaneSlotAcrossOneTarget(t *testing.T) { + configPath := writeTestConfig(t, testConfigYAML("example", "events")) + inputPath := filepath.Join(t.TempDir(), "missing.json") + referencePath := writeFile(t, "normalize.md", "Normalize\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.notes=" + referencePath, + }, &stdout, &stderr, Options{ + Catalog: fakeCatalog(t, pipeline.ModuleSpec{ + Key: "noop", + Stage: pipeline.StageNormalize, + Requires: []string{"merged"}, + Provides: []string{"normalized"}, + 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) + if refs := resolved.ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 1 || refs[0].Source != referencePath { + t.Fatalf("normalize references = %#v, want lane-qualified binding", refs) + } +} + func TestRunPipelineReferenceFlagRejectsAmbiguousFlatSlot(t *testing.T) { configPath := writeTestConfig(t, testConfigYAML("example", "events", "notes")) inputPath := writeSeriatimInput(t) @@ -847,11 +1034,99 @@ func TestRunPipelineReferenceFlagRejectsAmbiguousFlatSlot(t *testing.T) { if code != 1 { t.Fatalf("RunWithOptions() code = %d, want 1", code) } - if !strings.Contains(stderr.String(), "multiple selected lanes") || !strings.Contains(stderr.String(), "lane.slot") { + if !strings.Contains(stderr.String(), "multiple selected targets") || !strings.Contains(stderr.String(), "events.extract.roster") || !strings.Contains(stderr.String(), "notes.extract.roster") { t.Fatalf("stderr = %q, want ambiguous reference error", stderr.String()) } } +func TestRunPipelineReferenceFlagRejectsAmbiguousFlatSlotAcrossChunkAndExtract(t *testing.T) { + configPath := writeTestConfig(t, testConfigYAML("example", "events")) + inputPath := writeSeriatimInput(t) + diagnosticsDir := t.TempDir() + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{ + "run", "example", + "--config", configPath, + "--input", inputPath, + "--diagnostics-dir", diagnosticsDir, + "--reference", "context=./context.md", + }, &stdout, &stderr, Options{ + Catalog: fakeCatalog(t, + pipeline.ModuleSpec{ + Key: "generic", + Stage: pipeline.StageChunk, + Requires: []string{"source"}, + Provides: []string{"chunks"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "context"}, + }, + }, + pipeline.ModuleSpec{ + Key: "fake/extract", + Stage: pipeline.StageExtract, + Requires: []string{"chunks"}, + Provides: []string{"artifact"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "context"}, + }, + }, + ), + }) + + if code != 1 { + t.Fatalf("RunWithOptions() code = %d, want 1", code) + } + if !strings.Contains(stderr.String(), "multiple selected targets") || !strings.Contains(stderr.String(), "chunk.context") || !strings.Contains(stderr.String(), "events.extract.context") { + t.Fatalf("stderr = %q, want ambiguous chunk/extract reference error", stderr.String()) + } +} + +func TestRunPipelineReferenceFlagRejectsAmbiguousLaneSlotAcrossExtractAndNormalize(t *testing.T) { + configPath := writeTestConfig(t, testConfigYAML("example", "events")) + inputPath := writeSeriatimInput(t) + 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.context=./context.md", + }, &stdout, &stderr, Options{ + Catalog: fakeCatalog(t, + pipeline.ModuleSpec{ + Key: "fake/extract", + Stage: pipeline.StageExtract, + Requires: []string{"chunks"}, + Provides: []string{"artifact"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "context"}, + }, + }, + pipeline.ModuleSpec{ + Key: "noop", + Stage: pipeline.StageNormalize, + Requires: []string{"merged"}, + Provides: []string{"normalized"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "context"}, + }, + }, + ), + }) + + if code != 1 { + t.Fatalf("RunWithOptions() code = %d, want 1", code) + } + if !strings.Contains(stderr.String(), "multiple selected targets in lane") || !strings.Contains(stderr.String(), "events.extract.context") || !strings.Contains(stderr.String(), "events.normalize.context") { + t.Fatalf("stderr = %q, want ambiguous lane reference error", stderr.String()) + } +} + func TestRunPipelineReferenceFlagsRejectMalformedValues(t *testing.T) { tests := []struct { name string @@ -861,8 +1136,9 @@ 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: "too many selector parts", args: []string{"--reference", "a.b.c=./roster.yml"}, want: "lane.slot"}, - {name: "unbind with equals", args: []string{"--without-reference", "roster=./roster.yml"}, want: "slot or lane.slot"}, + {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: "unbind with equals", args: []string{"--without-reference", "roster=./roster.yml"}, want: "without =path"}, } for _, test := range tests { @@ -885,6 +1161,72 @@ 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", + })) + inputPath := filepath.Join(t.TempDir(), "missing.json") + diagnosticsDir := t.TempDir() + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{ + "run", "example", + "--config", configPath, + "--input", inputPath, + "--diagnostics-dir", diagnosticsDir, + "--without-reference", "chunk.context", + "--without-reference", "events.extract.roster", + "--without-reference", "events.normalize.notes", + }, &stdout, &stderr, Options{ + Catalog: fakeCatalog(t, + pipeline.ModuleSpec{ + Key: "generic", + Stage: pipeline.StageChunk, + Requires: []string{"source"}, + Provides: []string{"chunks"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "context"}, + }, + }, + pipeline.ModuleSpec{ + Key: "fake/extract", + Stage: pipeline.StageExtract, + Requires: []string{"chunks"}, + Provides: []string{"artifact"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "roster"}, + }, + }, + pipeline.ModuleSpec{ + Key: "noop", + Stage: pipeline.StageNormalize, + Requires: []string{"merged"}, + Provides: []string{"normalized"}, + 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) + 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 TestRunPipelineWithoutReferenceRemovesOptionalConfigBinding(t *testing.T) { configPath := writeTestConfig(t, testConfigYAMLWithReferences("example", "events", map[string]string{"roster": "./config-roster.yml"})) inputPath := filepath.Join(t.TempDir(), "missing.json") @@ -919,6 +1261,107 @@ func TestRunPipelineWithoutReferenceRemovesOptionalConfigBinding(t *testing.T) { } } +func TestRunPipelineWithoutReferenceFailsWhenRequiredChunkSlotWouldBeMissing(t *testing.T) { + configPath := writeTestConfig(t, testConfigYAMLWithPipelineReferences("example", "events", map[string]string{"context": "./config-context.md"})) + inputPath := writeSeriatimInput(t) + diagnosticsDir := t.TempDir() + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{ + "run", "example", + "--config", configPath, + "--input", inputPath, + "--diagnostics-dir", diagnosticsDir, + "--without-reference", "chunk.context", + }, &stdout, &stderr, Options{ + Catalog: fakeCatalog(t, pipeline.ModuleSpec{ + Key: "generic", + Stage: pipeline.StageChunk, + Requires: []string{"source"}, + Provides: []string{"chunks"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "context", Required: true}, + }, + }), + }) + + if code != 1 { + t.Fatalf("RunWithOptions() code = %d, want 1", code) + } + if !strings.Contains(stderr.String(), "required reference slot") || !strings.Contains(stderr.String(), "context") { + t.Fatalf("stderr = %q, want required chunk reference error", stderr.String()) + } +} + +func TestRunPipelineWithoutReferenceFailsWhenRequiredNormalizeSlotWouldBeMissing(t *testing.T) { + configPath := writeTestConfig(t, testConfigYAMLWithPipelineReferences("example", "events", map[string]string{"notes": "./config-notes.md"})) + inputPath := writeSeriatimInput(t) + diagnosticsDir := t.TempDir() + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{ + "run", "example", + "--config", configPath, + "--input", inputPath, + "--diagnostics-dir", diagnosticsDir, + "--without-reference", "events.normalize.notes", + }, &stdout, &stderr, Options{ + Catalog: fakeCatalog(t, pipeline.ModuleSpec{ + Key: "noop", + Stage: pipeline.StageNormalize, + Requires: []string{"merged"}, + Provides: []string{"normalized"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "notes", Required: true}, + }, + }), + }) + + if code != 1 { + t.Fatalf("RunWithOptions() code = %d, want 1", code) + } + if !strings.Contains(stderr.String(), "required reference slot") || !strings.Contains(stderr.String(), "notes") { + t.Fatalf("stderr = %q, want required normalize reference error", stderr.String()) + } +} + +func TestRunPipelineReferenceFlagRejectsUnselectedLaneSelector(t *testing.T) { + configPath := writeTestConfig(t, testConfigYAML("example", "events", "notes")) + inputPath := writeSeriatimInput(t) + referencePath := writeFile(t, "notes.yml", "Notes\n") + diagnosticsDir := t.TempDir() + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{ + "run", "example", + "--config", configPath, + "--input", inputPath, + "--only", "events", + "--diagnostics-dir", diagnosticsDir, + "--reference", "notes.extract.roster=" + referencePath, + }, &stdout, &stderr, Options{ + Catalog: fakeCatalog(t, pipeline.ModuleSpec{ + Key: "fake/extract", + Stage: pipeline.StageExtract, + Requires: []string{"chunks"}, + Provides: []string{"artifact"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "roster"}, + }, + }), + }) + + if code != 1 { + t.Fatalf("RunWithOptions() code = %d, want 1", code) + } + if !strings.Contains(stderr.String(), `reference lane "notes" is not selected`) { + t.Fatalf("stderr = %q, want unselected lane error", stderr.String()) + } +} + func TestRunPipelineWithoutReferenceFailsWhenRequiredSlotWouldBeMissing(t *testing.T) { configPath := writeTestConfig(t, testConfigYAMLWithReferences("example", "events", map[string]string{"roster": "./config-roster.yml"})) inputPath := writeSeriatimInput(t) @@ -1609,6 +2052,27 @@ func testConfigYAMLWithReferences(pipelineID string, laneID string, references m return b.String() } +func testConfigYAMLWithPipelineReferences(pipelineID string, laneID string, references map[string]string) string { + var b strings.Builder + b.WriteString("version: 1\n") + b.WriteString("pipelines:\n") + b.WriteString(" " + pipelineID + ":\n") + b.WriteString(" input: fake/input\n") + b.WriteString(" references:\n") + keys := make([]string, 0, len(references)) + for key := range references { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + b.WriteString(" " + key + ": " + references[key] + "\n") + } + b.WriteString(" artifacts:\n") + b.WriteString(" " + laneID + ":\n") + b.WriteString(" extract: fake/extract\n") + return b.String() +} + func testConfigYAMLWithReferencesAndDiagnostics(pipelineID string, laneID string, diagnosticsDir string, references map[string]string) string { var b strings.Builder b.WriteString("version: 1\n") diff --git a/internal/framework/pipeline/profile.go b/internal/framework/pipeline/profile.go index 574c3dd..b2a6fd7 100644 --- a/internal/framework/pipeline/profile.go +++ b/internal/framework/pipeline/profile.go @@ -50,15 +50,17 @@ type ResolveOptions struct { } type ReferenceBinding struct { - LaneID string `json:"lane_id,omitempty"` - SlotName string `json:"slot_name"` - Source string `json:"source"` - BindingSource string `json:"binding_source,omitempty"` + Stage ModuleStage `json:"stage,omitempty"` + LaneID string `json:"lane_id,omitempty"` + SlotName string `json:"slot_name"` + Source string `json:"source"` + BindingSource string `json:"binding_source,omitempty"` } type ReferenceUnbind struct { - LaneID string `json:"lane_id"` - SlotName string `json:"slot_name"` + Stage ModuleStage `json:"stage,omitempty"` + LaneID string `json:"lane_id,omitempty"` + SlotName string `json:"slot_name"` } type ResolvedReferenceTarget struct { @@ -424,23 +426,12 @@ func resolveReferenceTargetBindings(target referenceResolutionTarget) ([]Referen } } - if target.Stage != StageExtract { - for _, slot := range target.Slots { - if slot.Required { - if _, ok := bindings[slot.Name]; !ok { - return nil, fmt.Errorf("%s required reference slot %q is not bound", referenceTargetErrorContext(target), slot.Name) - } - } - } - return sortedReferenceBindings(bindings), nil - } - for _, override := range target.Options.ReferenceOverrides { - optionLaneID := strings.TrimSpace(override.LaneID) - if optionLaneID == "" { - return nil, fmt.Errorf("pipeline %q reference override lane id must not be empty", target.PipelineID) + match, err := referenceOverrideMatchesTarget(target, override) + if err != nil { + return nil, err } - if optionLaneID != target.LaneID { + if !match { continue } source := override.BindingSource @@ -453,11 +444,11 @@ func resolveReferenceTargetBindings(target referenceResolutionTarget) ([]Referen } for _, unbind := range target.Options.ReferenceUnbinds { - optionLaneID := strings.TrimSpace(unbind.LaneID) - if optionLaneID == "" { - return nil, fmt.Errorf("pipeline %q reference unbind lane id must not be empty", target.PipelineID) + match, err := referenceUnbindMatchesTarget(target, unbind) + if err != nil { + return nil, err } - if optionLaneID != target.LaneID { + if !match { continue } slotName := strings.TrimSpace(unbind.SlotName) @@ -481,6 +472,44 @@ func resolveReferenceTargetBindings(target referenceResolutionTarget) ([]Referen return sortedReferenceBindings(bindings), nil } +func referenceOverrideMatchesTarget(target referenceResolutionTarget, override ReferenceBinding) (bool, error) { + stage, laneID, err := normalizeReferenceOptionTarget(target.PipelineID, "override", override.Stage, override.LaneID) + if err != nil { + return false, err + } + return stage == target.Stage && laneID == target.LaneID, nil +} + +func referenceUnbindMatchesTarget(target referenceResolutionTarget, unbind ReferenceUnbind) (bool, error) { + stage, laneID, err := normalizeReferenceOptionTarget(target.PipelineID, "unbind", unbind.Stage, unbind.LaneID) + if err != nil { + return false, err + } + return stage == target.Stage && laneID == target.LaneID, nil +} + +func normalizeReferenceOptionTarget(pipelineID string, operation string, stage ModuleStage, laneID string) (ModuleStage, string, error) { + stage = ModuleStage(strings.TrimSpace(string(stage))) + if stage == "" { + stage = StageExtract + } + laneID = strings.TrimSpace(laneID) + + switch stage { + case StageChunk: + if laneID != "" { + return "", "", fmt.Errorf("pipeline %q reference %s for chunk must not include a lane id", pipelineID, operation) + } + case StageExtract, StageNormalize: + if laneID == "" { + return "", "", fmt.Errorf("pipeline %q reference %s lane id must not be empty", pipelineID, operation) + } + default: + return "", "", fmt.Errorf("pipeline %q reference %s stage %q is not supported", pipelineID, operation, stage) + } + return stage, laneID, nil +} + func sortedReferenceBindings(bindings map[string]ReferenceBinding) []ReferenceBinding { keys := sortedReferenceBindingKeys(bindings) resolved := make([]ReferenceBinding, 0, len(keys))