Support target-aware reference selectors
This commit is contained in:
@@ -25,7 +25,7 @@ const defaultOutputRoot = "./notarius-output"
|
||||
|
||||
const usage = `Usage:
|
||||
notarius help
|
||||
notarius run <pipeline-id> --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] [--reference slot=path] [--without-reference slot]
|
||||
notarius run <pipeline-id> --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 {
|
||||
|
||||
Reference in New Issue
Block a user