package pipeline import ( "crypto/sha256" "encoding/hex" "encoding/json" "fmt" "reflect" "sort" "strings" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" ) const ( DefaultChunkModule = "generic" DefaultMergeModule = "appendorder" DefaultNormalizeModule = "noop" DefaultOutputModule = "json" DefaultLLMProfile = "default" ) 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]ReferenceSource `json:"references,omitempty"` Validators ValidatorOverride `json:"validators,omitempty"` } // ArtifactReference identifies a normalized artifact produced by an earlier // ordered step. It is a selector only; it does not contain artifact bytes. type ArtifactReference struct { Step string `json:"step"` Lane string `json:"lane"` } // ReferenceSource is the discriminated source form used by configured // reference maps. Exactly one of Path and Artifact is set after resolution. type ReferenceSource struct { Path string `json:"path,omitempty"` Artifact *ArtifactReference `json:"artifact,omitempty"` } func ExternalReference(path string) ReferenceSource { return ReferenceSource{Path: strings.TrimSpace(path)} } func ExternalReferenceMap(values map[string]string) map[string]ReferenceSource { out := make(map[string]ReferenceSource, len(values)) for key, value := range values { out[key] = ExternalReference(value) } return out } func GeneratedReference(step, lane string) ReferenceSource { return ReferenceSource{Artifact: &ArtifactReference{Step: strings.TrimSpace(step), Lane: strings.TrimSpace(lane)}} } func (source ReferenceSource) IsGenerated() bool { return source.Artifact != nil } func (source ReferenceSource) MarshalJSON() ([]byte, error) { if source.Path != "" && source.Artifact != nil { return nil, fmt.Errorf("reference source must not contain both an external path and an artifact selector") } if source.Artifact != nil { return json.Marshal(struct { Artifact *ArtifactReference `json:"artifact"` }{Artifact: source.Artifact}) } return json.Marshal(source.Path) } type ValidatorOverride struct { Set bool `json:"set,omitempty"` Validators []ModuleBinding `json:"validators,omitempty"` } func (binding ModuleBinding) MarshalJSON() ([]byte, error) { type moduleBindingJSON 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]ReferenceSource `json:"references,omitempty"` Validators *[]ModuleBinding `json:"validators,omitempty"` } out := moduleBindingJSON{ Module: binding.Module, LLMProfile: binding.LLMProfile, Retries: binding.Retries, Options: binding.Options, References: binding.References, } if binding.Validators.Set { validators := cloneModuleBindings(binding.Validators.Validators) out.Validators = &validators } return json.Marshal(out) } type ArtifactLaneProfile struct { Extract ModuleBinding `json:"extract"` Merge ModuleBinding `json:"merge,omitempty"` Normalize ModuleBinding `json:"normalize,omitempty"` Validators []ModuleBinding `json:"validators,omitempty"` References map[string]ReferenceSource `json:"references,omitempty"` } type PipelineStepProfile struct { ID string `json:"id"` Artifacts map[string]ArtifactLaneProfile `json:"artifacts"` References map[string]ReferenceSource `json:"references,omitempty"` } type PipelineProfile struct { ID string `json:"id"` Input ModuleBinding `json:"input"` Chunk ModuleBinding `json:"chunk,omitempty"` Artifacts map[string]ArtifactLaneProfile `json:"artifacts"` Steps []PipelineStepProfile `json:"steps,omitempty"` Output ModuleBinding `json:"output,omitempty"` References map[string]ReferenceSource `json:"references,omitempty"` } type ResolveOptions struct { Only []string ReferenceOverrides []ReferenceBinding ReferenceUnbinds []ReferenceUnbind } type ReferenceBinding struct { Stage ModuleStage `json:"stage,omitempty"` LaneID string `json:"lane_id,omitempty"` SlotName string `json:"slot_name"` Source string `json:"source,omitempty"` Artifact *ArtifactReference `json:"artifact,omitempty"` BindingSource string `json:"binding_source,omitempty"` } type ReferenceUnbind struct { Stage ModuleStage `json:"stage,omitempty"` LaneID string `json:"lane_id,omitempty"` SlotName string `json:"slot_name"` } type ResolvedReferenceTarget struct { Stage ModuleStage `json:"stage"` StepID string `json:"step_id,omitempty"` LaneID string `json:"lane_id,omitempty"` Module string `json:"module"` Bindings []ReferenceBinding `json:"bindings,omitempty"` ReferenceSet contracts.ReferenceSet `json:"-"` } type ResolvedArtifactLane struct { StepID string ID string ArtifactKind contracts.ArtifactKind `json:"artifact_kind,omitempty"` ArtifactSchemaID string `json:"artifact_schema_id,omitempty"` ArtifactSchemaName string `json:"artifact_schema_name,omitempty"` ArtifactSchemaVersion string `json:"artifact_schema_version,omitempty"` ArtifactSchemaDigest string `json:"artifact_schema_digest,omitempty"` Extract ModuleBinding Merge ModuleBinding Normalize ModuleBinding Validators []ModuleBinding ExtractReferences ResolvedReferenceTarget `json:"extract_references"` MergeReferences ResolvedReferenceTarget `json:"merge_references"` NormalizeReferences ResolvedReferenceTarget `json:"normalize_references"` } type ResolvedPipelineStep struct { ID string `json:"id"` ArtifactLanes []ResolvedArtifactLane `json:"artifact_lanes"` } type ResolvedValidatorChain struct { Stage ModuleStage `json:"stage"` LaneID string `json:"lane_id,omitempty"` ModuleKey string `json:"module_key"` Validators []ResolvedValidator `json:"validators"` } type ResolvedValidator struct { Binding ModuleBinding `json:"binding"` ExecutionClass contracts.ExecutionClass `json:"execution_class"` Target ValidatorTarget `json:"target,omitempty"` ArtifactKind contracts.ArtifactKind `json:"artifact_kind,omitempty"` } type ResolvedPipeline struct { ID string Digest string Input ModuleBinding Chunk ModuleBinding ChunkReferences ResolvedReferenceTarget `json:"chunk_references"` Steps []ResolvedPipelineStep ValidatorChains []ResolvedValidatorChain `json:"validator_chains"` Output ModuleBinding } // AllArtifactLanes returns lanes in deterministic step order for read-only // discovery and identity construction. func (resolved ResolvedPipeline) AllArtifactLanes() []ResolvedArtifactLane { var lanes []ResolvedArtifactLane for _, step := range resolved.Steps { lanes = append(lanes, step.ArtifactLanes...) } return lanes } type ModuleCatalog struct { Inputs *InputAdapterRegistry Chunkers *ChunkerRegistry ArtifactCodecs *ArtifactCodecRegistry Extractors *ExtractorRegistry Mergers *MergerRegistry Normalizers *NormalizerRegistry Validators *ValidatorRegistry ValidatorChains *ValidatorChainRegistry Outputs *OutputEncoderRegistry } func Binding(module string) ModuleBinding { return ModuleBinding{Module: strings.TrimSpace(module)} } func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog ModuleCatalog) (ResolvedPipeline, error) { pipelineID := strings.TrimSpace(profile.ID) if pipelineID == "" { return ResolvedPipeline{}, fmt.Errorf("pipeline id must not be empty") } explicitSteps := profile.Steps != nil if len(profile.Artifacts) > 0 && explicitSteps { return ResolvedPipeline{}, fmt.Errorf("pipeline %q must not declare both artifacts and steps", pipelineID) } if explicitSteps && len(options.Only) > 0 { return ResolvedPipeline{}, fmt.Errorf("pipeline %q --only is not supported for explicit ordered steps", pipelineID) } steps := profile.Steps if !explicitSteps { steps = []PipelineStepProfile{{ID: "default", Artifacts: profile.Artifacts}} } if explicitSteps && len(steps) == 0 { return ResolvedPipeline{}, fmt.Errorf("pipeline %q must declare at least one ordered step", pipelineID) } if len(steps) == 0 { return ResolvedPipeline{}, fmt.Errorf("pipeline %q must declare at least one artifact lane", pipelineID) } input := resolveBinding(profile.Input, "") if input.Module == "" { return ResolvedPipeline{}, fmt.Errorf("pipeline %q input module must not be empty", pipelineID) } inputModuleSpec, err := inputSpec(catalog, input.Module) if err != nil { return ResolvedPipeline{}, moduleLookupError(pipelineID, "", StageInput, input.Module, err) } capabilities := newCapabilitySet() if missing, ok := capabilities.missing(inputModuleSpec.Requires); ok { return ResolvedPipeline{}, capabilityError(pipelineID, "", StageInput, input.Module, missing) } capabilities.add(inputModuleSpec.Provides...) chunk := resolveBinding(profile.Chunk, DefaultChunkModule) chunkSpec, err := chunkerSpec(catalog, chunk.Module) if err != nil { return ResolvedPipeline{}, moduleLookupError(pipelineID, "", StageChunk, chunk.Module, err) } if missing, ok := capabilities.missing(chunkSpec.Requires); ok { return ResolvedPipeline{}, capabilityError(pipelineID, "", StageChunk, chunk.Module, missing) } capabilities.add(chunkSpec.Provides...) if !explicitSteps { if err := validatePipelineReferenceDefaults(pipelineID, profile.References, chunkSpec, profile.Artifacts, catalog); err != nil { return ResolvedPipeline{}, err } } else { allLanes := make(map[string]ArtifactLaneProfile) for _, step := range profile.Steps { for laneID, lane := range step.Artifacts { allLanes[strings.TrimSpace(laneID)] = lane } } if err := validatePipelineReferenceDefaults(pipelineID, profile.References, chunkSpec, allLanes, catalog); err != nil { return ResolvedPipeline{}, err } } for _, source := range profile.References { if source.Artifact != nil { return ResolvedPipeline{}, fmt.Errorf("pipeline %q pipeline-level references may not use generated artifact selectors", pipelineID) } } chunkReferences, err := resolveReferenceTargetBindings(referenceResolutionTarget{ PipelineID: pipelineID, Stage: StageChunk, Module: chunk.Module, Slots: chunkSpec.ReferenceSlots, PipelineReferences: profile.References, LocalReferences: chunk.References, Options: options, }) if err != nil { return ResolvedPipeline{}, err } resolved := ResolvedPipeline{ ID: pipelineID, Input: input, Chunk: chunk, ChunkReferences: referenceTarget(StageChunk, "", chunk.Module, chunkReferences), Output: resolveBinding(profile.Output, DefaultOutputModule), } chunkValidatorChain, err := resolveValidatorChain(pipelineID, "", StageChunk, chunk.Module, chunk.Validators, "", nil, catalog) if err != nil { return ResolvedPipeline{}, err } resolved.ValidatorChains = append(resolved.ValidatorChains, chunkValidatorChain) outputCapabilities := capabilities.clone() seenResolvedLanes := make(map[string]string) seenResolvedSteps := make(map[string]struct{}, len(steps)) for _, step := range steps { stepID := strings.TrimSpace(step.ID) if stepID == "" { return ResolvedPipeline{}, fmt.Errorf("pipeline %q step id must not be empty", pipelineID) } if _, exists := seenResolvedSteps[stepID]; exists { return ResolvedPipeline{}, fmt.Errorf("pipeline %q step id %q is duplicated after trimming", pipelineID, stepID) } seenResolvedSteps[stepID] = struct{}{} laneOptions := ResolveOptions{} if !explicitSteps { laneOptions = options } lanesByID, selectedLaneIDs, err := selectedArtifactLanes(pipelineID, step.Artifacts, laneOptions) if err != nil { return ResolvedPipeline{}, err } if len(selectedLaneIDs) == 0 { return ResolvedPipeline{}, fmt.Errorf("pipeline %q step %q must declare at least one artifact lane", pipelineID, stepID) } if referenceSourceMapsConflict(profile.References, step.References) { return ResolvedPipeline{}, fmt.Errorf("pipeline %q step %q has a generated and external reference collision", pipelineID, stepID) } if err := validatePipelineReferenceDefaults(pipelineID, step.References, chunkSpec, lanesByID, catalog); err != nil { return ResolvedPipeline{}, err } stepReferences := mergeReferenceMaps(profile.References, step.References) resolvedStep := ResolvedPipelineStep{ID: stepID} for _, laneID := range selectedLaneIDs { if previousStep, exists := seenResolvedLanes[laneID]; exists { return ResolvedPipeline{}, fmt.Errorf("pipeline %q artifact lane id %q is duplicated across steps %q and %q", pipelineID, laneID, previousStep, stepID) } seenResolvedLanes[laneID] = stepID laneProfile := lanesByID[laneID] lane, validatorChains, laneCapabilities, err := resolveArtifactLane(pipelineID, stepID, laneID, laneProfile, stepReferences, options, capabilities, catalog) if err != nil { return ResolvedPipeline{}, err } resolvedStep.ArtifactLanes = append(resolvedStep.ArtifactLanes, lane) resolved.ValidatorChains = append(resolved.ValidatorChains, validatorChains...) outputCapabilities.addSet(laneCapabilities) } resolved.Steps = append(resolved.Steps, resolvedStep) } if err := validateGeneratedBindings(pipelineID, resolved, catalog); err != nil { return ResolvedPipeline{}, err } outputSpec, err := outputSpec(catalog, resolved.Output.Module) if err != nil { return ResolvedPipeline{}, moduleLookupError(pipelineID, "", StageOutput, resolved.Output.Module, err) } if missing, ok := outputCapabilities.missing(outputSpec.Requires); ok { return ResolvedPipeline{}, capabilityError(pipelineID, "", StageOutput, resolved.Output.Module, missing) } if err := validateResolvedOptions(resolved, catalog); err != nil { return ResolvedPipeline{}, err } digest, err := resolvedPipelineDigest(resolved) if err != nil { return ResolvedPipeline{}, fmt.Errorf("pipeline %q digest: %w", pipelineID, err) } resolved.Digest = digest return resolved, nil } func resolveArtifactLane( pipelineID string, stepID string, laneID string, profile ArtifactLaneProfile, pipelineReferences map[string]ReferenceSource, options ResolveOptions, inherited capabilitySet, catalog ModuleCatalog, ) (ResolvedArtifactLane, []ResolvedValidatorChain, capabilitySet, error) { lane := ResolvedArtifactLane{ StepID: strings.TrimSpace(stepID), ID: laneID, Extract: resolveBinding(profile.Extract, ""), Merge: resolveBinding(profile.Merge, DefaultMergeModule), Normalize: resolveBinding(profile.Normalize, DefaultNormalizeModule), Validators: resolveBindings(profile.Validators, ""), } if lane.Extract.Module == "" { return ResolvedArtifactLane{}, nil, nil, fmt.Errorf("pipeline %q lane %q extract module must not be empty", pipelineID, laneID) } capabilities := inherited.clone() extractSpec, err := extractorSpec(catalog, lane.Extract.Module) if err != nil { return ResolvedArtifactLane{}, nil, nil, moduleLookupError(pipelineID, laneID, StageExtract, lane.Extract.Module, err) } if missing, ok := capabilities.missing(extractSpec.Requires); ok { return ResolvedArtifactLane{}, nil, nil, capabilityError(pipelineID, laneID, StageExtract, lane.Extract.Module, missing) } artifactType, err := resolveArtifactIdentity(pipelineID, laneID, &lane, extractSpec, catalog) if err != nil { return ResolvedArtifactLane{}, nil, nil, err } if referenceSourceMapsConflict(profile.References, lane.Extract.References) { return ResolvedArtifactLane{}, nil, nil, fmt.Errorf("pipeline %q step %q lane %q extract has a generated and external reference collision", pipelineID, stepID, laneID) } extractReferences := mergeReferenceMaps(profile.References, lane.Extract.References) references, err := resolveReferenceTargetBindings(referenceResolutionTarget{ PipelineID: pipelineID, LaneID: laneID, Stage: StageExtract, Module: lane.Extract.Module, Slots: extractSpec.ReferenceSlots, PipelineReferences: pipelineReferences, LocalReferences: extractReferences, Options: options, }) if err != nil { return ResolvedArtifactLane{}, nil, nil, err } lane.ExtractReferences = referenceTarget(StageExtract, laneID, lane.Extract.Module, references) lane.ExtractReferences.StepID = strings.TrimSpace(stepID) capabilities.add(extractSpec.Provides...) mergeSpec, err := mergerSpecForArtifact(catalog, lane.Merge.Module, lane.ArtifactKind, artifactType) if err != nil { return ResolvedArtifactLane{}, nil, nil, moduleLookupError(pipelineID, laneID, StageMerge, lane.Merge.Module, err) } if missing, ok := capabilities.missing(mergeSpec.Requires); ok { return ResolvedArtifactLane{}, nil, 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, nil, err } lane.MergeReferences = referenceTarget(StageMerge, laneID, lane.Merge.Module, mergeReferences) lane.MergeReferences.StepID = strings.TrimSpace(stepID) capabilities.add(mergeSpec.Provides...) normalizeSpec, err := normalizerSpecForArtifact(catalog, lane.Normalize.Module, lane.ArtifactKind, artifactType) if err != nil { return ResolvedArtifactLane{}, nil, nil, moduleLookupError(pipelineID, laneID, StageNormalize, lane.Normalize.Module, err) } if missing, ok := capabilities.missing(normalizeSpec.Requires); ok { return ResolvedArtifactLane{}, nil, nil, capabilityError(pipelineID, laneID, StageNormalize, lane.Normalize.Module, missing) } normalizeReferences, err := resolveReferenceTargetBindings(referenceResolutionTarget{ PipelineID: pipelineID, LaneID: laneID, Stage: StageNormalize, Module: lane.Normalize.Module, Slots: normalizeSpec.ReferenceSlots, PipelineReferences: pipelineReferences, LocalReferences: lane.Normalize.References, Options: options, }) if err != nil { return ResolvedArtifactLane{}, nil, nil, err } lane.NormalizeReferences = referenceTarget(StageNormalize, laneID, lane.Normalize.Module, normalizeReferences) lane.NormalizeReferences.StepID = strings.TrimSpace(stepID) capabilities.add(normalizeSpec.Provides...) if len(lane.Validators) > 0 { return ResolvedArtifactLane{}, nil, nil, configuredValidatorsError(pipelineID, laneID) } extractValidatorChain, err := resolveValidatorChain(pipelineID, laneID, StageExtract, lane.Extract.Module, lane.Extract.Validators, lane.ArtifactKind, artifactType, catalog) if err != nil { return ResolvedArtifactLane{}, nil, nil, err } mergeValidatorChain, err := resolveValidatorChain(pipelineID, laneID, StageMerge, lane.Merge.Module, lane.Merge.Validators, lane.ArtifactKind, artifactType, catalog) if err != nil { return ResolvedArtifactLane{}, nil, nil, err } normalizeValidatorChain, err := resolveValidatorChain(pipelineID, laneID, StageNormalize, lane.Normalize.Module, lane.Normalize.Validators, lane.ArtifactKind, artifactType, catalog) if err != nil { return ResolvedArtifactLane{}, nil, nil, err } validatorChains := []ResolvedValidatorChain{extractValidatorChain, mergeValidatorChain, normalizeValidatorChain} return lane, validatorChains, capabilities, nil } func configuredValidatorsError(pipelineID string, laneID string) error { return fmt.Errorf("pipeline %q lane %q validators are not supported at artifact lane level; use extract.validators, merge.validators, or normalize.validators", pipelineID, laneID) } func validateGeneratedBindings(pipelineID string, resolved ResolvedPipeline, catalog ModuleCatalog) error { stepIndex := make(map[string]int, len(resolved.Steps)) laneIndex := make(map[string]ResolvedArtifactLane) for index, step := range resolved.Steps { if _, exists := stepIndex[step.ID]; exists { return fmt.Errorf("pipeline %q step id %q is ambiguous", pipelineID, step.ID) } stepIndex[step.ID] = index for _, lane := range step.ArtifactLanes { laneIndex[step.ID+"\x00"+lane.ID] = lane } } for consumerIndex, step := range resolved.Steps { for _, lane := range step.ArtifactLanes { for _, target := range []ResolvedReferenceTarget{lane.ExtractReferences, lane.MergeReferences, lane.NormalizeReferences} { slots, err := resolvedReferenceSlots(target, lane.ArtifactKind, catalog) if err != nil { return fmt.Errorf("pipeline %q step %q lane %q: %w", pipelineID, step.ID, lane.ID, err) } slotByName := make(map[string]contracts.ReferenceSlot, len(slots)) for _, slot := range slots { slotByName[slot.Name] = slot } for _, binding := range target.Bindings { if binding.Artifact == nil { continue } producerStep, ok := stepIndex[strings.TrimSpace(binding.Artifact.Step)] if !ok { return fmt.Errorf("pipeline %q step %q lane %q %s reference slot %q producer step %q is not declared", pipelineID, step.ID, lane.ID, target.Stage, binding.SlotName, binding.Artifact.Step) } if producerStep >= consumerIndex { return fmt.Errorf("pipeline %q step %q lane %q reference slot %q producer %q must be an earlier step", pipelineID, step.ID, lane.ID, binding.SlotName, binding.Artifact.Step) } producer, ok := laneIndex[strings.TrimSpace(binding.Artifact.Step)+"\x00"+strings.TrimSpace(binding.Artifact.Lane)] if !ok { return fmt.Errorf("pipeline %q step %q lane %q reference slot %q producer lane %q is not selected", pipelineID, step.ID, lane.ID, binding.SlotName, binding.Artifact.Lane) } slot, ok := slotByName[strings.TrimSpace(binding.SlotName)] if !ok { return fmt.Errorf("pipeline %q step %q lane %q %s reference slot %q is not declared", pipelineID, step.ID, lane.ID, target.Stage, binding.SlotName) } if len(slot.AcceptedArtifactKinds) == 0 { return fmt.Errorf("pipeline %q step %q lane %q reference slot %q does not accept generated artifacts", pipelineID, step.ID, lane.ID, binding.SlotName) } accepted := false for _, kind := range slot.AcceptedArtifactKinds { if kind == producer.ArtifactKind { accepted = true break } } if !accepted { return fmt.Errorf("pipeline %q step %q lane %q reference slot %q does not accept artifact kind %q", pipelineID, step.ID, lane.ID, binding.SlotName, producer.ArtifactKind) } codecSpec, ok := catalog.ArtifactCodecs.Spec(producer.ArtifactKind) if !ok { return fmt.Errorf("pipeline %q step %q lane %q producer %q artifact codec %q is not registered", pipelineID, step.ID, lane.ID, producer.ID, producer.ArtifactKind) } if !referenceMediaTypeAccepted(codecSpec.MediaType, slot.AcceptedMediaTypes) { return fmt.Errorf("pipeline %q step %q lane %q reference slot %q does not accept producer media type %q", pipelineID, step.ID, lane.ID, binding.SlotName, codecSpec.MediaType) } } } } } return nil } func resolvedReferenceSlots(target ResolvedReferenceTarget, artifactKind contracts.ArtifactKind, catalog ModuleCatalog) ([]contracts.ReferenceSlot, error) { spec, err := referenceTargetSpec(target, artifactKind, catalog) if err != nil { return nil, err } return contracts.CloneReferenceSlots(spec.ReferenceSlots), nil } func resolveArtifactIdentity(pipelineID, laneID string, lane *ResolvedArtifactLane, extractSpec ModuleSpec, catalog ModuleCatalog) (reflect.Type, error) { if extractSpec.ArtifactKind == "" { return nil, fmt.Errorf("pipeline %q lane %q extract module %q does not declare an artifact kind", pipelineID, laneID, lane.Extract.Module) } if catalog.Extractors == nil { return nil, fmt.Errorf("pipeline %q lane %q extractor registry must not be nil", pipelineID, laneID) } extractor, ok := catalog.Extractors.typedEntry(lane.Extract.Module) if !ok { return nil, fmt.Errorf("pipeline %q lane %q extract module %q declares artifact kind %q without a typed registration", pipelineID, laneID, lane.Extract.Module, extractSpec.ArtifactKind) } if catalog.ArtifactCodecs == nil { return nil, fmt.Errorf("pipeline %q lane %q artifact codec registry must not be nil for kind %q", pipelineID, laneID, extractSpec.ArtifactKind) } codecSpec, ok := catalog.ArtifactCodecs.Spec(extractSpec.ArtifactKind) if !ok { return nil, fmt.Errorf("pipeline %q lane %q artifact codec %q is not registered", pipelineID, laneID, extractSpec.ArtifactKind) } codecType, ok := catalog.ArtifactCodecs.valueType(extractSpec.ArtifactKind) if !ok { return nil, fmt.Errorf("pipeline %q lane %q artifact codec %q has no Go type", pipelineID, laneID, extractSpec.ArtifactKind) } if extractor.valueType != codecType { return nil, artifactTypeMismatchError(pipelineID, laneID, StageExtract, lane.Extract.Module, extractSpec.ArtifactKind, codecType, extractor.valueType) } lane.ArtifactKind = codecSpec.Kind lane.ArtifactSchemaID = codecSpec.Schema.ID lane.ArtifactSchemaName = codecSpec.Schema.Name lane.ArtifactSchemaVersion = codecSpec.Schema.Version lane.ArtifactSchemaDigest = codecSpec.SchemaDigest return codecType, nil } func mergerSpecForArtifact(catalog ModuleCatalog, key string, kind contracts.ArtifactKind, expectedType reflect.Type) (ModuleSpec, error) { if kind == "" { return ModuleSpec{}, fmt.Errorf("merger %q cannot be resolved without an artifact kind", key) } if catalog.Mergers == nil { return ModuleSpec{}, fmt.Errorf("module %q is not registered", key) } entry, ok := catalog.Mergers.typedEntry(key, kind) if !ok { return ModuleSpec{}, missingArtifactVariantError("merger", key, kind, catalog.Mergers.registeredKinds(key)) } if entry.valueType != expectedType { return ModuleSpec{}, fmt.Errorf("artifact kind %q requires Go type %s, but merger %q variant uses %s", kind, typeName(expectedType), key, typeName(entry.valueType)) } return cloneModuleSpec(entry.spec), nil } func normalizerSpecForArtifact(catalog ModuleCatalog, key string, kind contracts.ArtifactKind, expectedType reflect.Type) (ModuleSpec, error) { if kind == "" { return ModuleSpec{}, fmt.Errorf("normalizer %q cannot be resolved without an artifact kind", key) } if catalog.Normalizers == nil { return ModuleSpec{}, fmt.Errorf("module %q is not registered", key) } entry, ok := catalog.Normalizers.typedEntry(key, kind) if !ok { return ModuleSpec{}, missingArtifactVariantError("normalizer", key, kind, catalog.Normalizers.registeredKinds(key)) } if entry.valueType != expectedType { return ModuleSpec{}, fmt.Errorf("artifact kind %q requires Go type %s, but normalizer %q variant uses %s", kind, typeName(expectedType), key, typeName(entry.valueType)) } return cloneModuleSpec(entry.spec), nil } func validatorSpecForTarget(registry *ValidatorRegistry, stage ModuleStage, key string, kind contracts.ArtifactKind, expectedType reflect.Type) (ValidatorSpec, ValidatorTarget, error) { key = strings.TrimSpace(key) if stage == StageChunk { if entry, ok := registry.chunkEntry(key); ok { return entry.spec, ValidatorTargetChunk, nil } if entry, ok := registry.serializedEntry(key); ok && entry.spec.SupportsChunks { return entry.spec.ValidatorSpec, ValidatorTargetSerialized, nil } if spec, ok := registry.Spec(key); ok { return spec, "", nil } return ValidatorSpec{}, "", fmt.Errorf("references unknown validator %q for chunk target", key) } if kind == "" { if spec, ok := registry.Spec(key); ok { return spec, "", nil } return ValidatorSpec{}, "", fmt.Errorf("references unknown validator %q without an artifact kind", key) } if entry, ok := registry.typedEntry(key, kind); ok { if entry.valueType != expectedType { return ValidatorSpec{}, "", fmt.Errorf("artifact kind %q requires Go type %s, but validator %q variant uses %s", kind, typeName(expectedType), key, typeName(entry.valueType)) } return entry.spec, ValidatorTargetTyped, nil } if entry, ok := registry.serializedEntry(key); ok && entry.spec.SupportsArtifacts { return entry.spec.ValidatorSpec, ValidatorTargetSerialized, nil } if _, ok := registry.Spec(key); !ok { return ValidatorSpec{}, "", fmt.Errorf("references unknown validator %q for artifact kind %q", key, kind) } return ValidatorSpec{}, "", missingArtifactVariantError("validator", key, kind, registry.registeredTypedKinds(key)) } func missingArtifactVariantError(moduleType, key string, kind contracts.ArtifactKind, registered []contracts.ArtifactKind) error { if len(registered) == 0 { return fmt.Errorf("%s %q has no typed variant for artifact kind %q", moduleType, key, kind) } values := make([]string, len(registered)) for i, value := range registered { values[i] = string(value) } return fmt.Errorf("%s %q has no typed variant for artifact kind %q; registered kinds: %s", moduleType, key, kind, strings.Join(values, ", ")) } func artifactTypeMismatchError(pipelineID, laneID string, stage ModuleStage, module string, kind contracts.ArtifactKind, expected, actual reflect.Type) error { return fmt.Errorf("pipeline %q lane %q %s module %q artifact kind %q requires Go type %s, got %s", pipelineID, laneID, stage, module, kind, typeName(expected), typeName(actual)) } func typeName(value reflect.Type) string { if value == nil { return "" } return value.String() } func resolveValidatorChain(pipelineID string, laneID string, stage ModuleStage, module string, override ValidatorOverride, artifactKind contracts.ArtifactKind, artifactType reflect.Type, catalog ModuleCatalog) (ResolvedValidatorChain, error) { chain := ResolvedValidatorChain{ Stage: stage, LaneID: strings.TrimSpace(laneID), ModuleKey: strings.TrimSpace(module), } if chain.ModuleKey == "" { return ResolvedValidatorChain{}, fmt.Errorf("pipeline %q validator chain %q module key must not be empty", pipelineID, stage) } switch stage { case StageChunk, StageExtract, StageMerge, StageNormalize: default: return ResolvedValidatorChain{}, fmt.Errorf("pipeline %q validator chain stage %q is not supported", pipelineID, stage) } var bindings []ModuleBinding if override.Set { bindings = cloneModuleBindings(override.Validators) } else if catalog.ValidatorChains != nil { bindings = catalog.ValidatorChains.Validators(stage, chain.ModuleKey) } if len(bindings) == 0 { return chain, nil } if catalog.Validators == nil { return ResolvedValidatorChain{}, fmt.Errorf("pipeline %q validator registry must not be nil for %s validator chain on module %q", pipelineID, stage, chain.ModuleKey) } chain.Validators = make([]ResolvedValidator, 0, len(bindings)) for _, validator := range bindings { spec, target, err := validatorSpecForTarget(catalog.Validators, stage, validator.Module, artifactKind, artifactType) if err != nil { return ResolvedValidatorChain{}, fmt.Errorf("pipeline %q %s validator chain for module %q: %w", pipelineID, stage, chain.ModuleKey, err) } if strings.TrimSpace(validator.LLMProfile) != "" && spec.ExecutionClass != contracts.ExecutionClassLLMBacked { return ResolvedValidatorChain{}, fmt.Errorf("pipeline %q %s validator chain for module %q assigns llm_profile to deterministic validator %q", pipelineID, stage, chain.ModuleKey, validator.Module) } chain.Validators = append(chain.Validators, ResolvedValidator{ Binding: cloneModuleBinding(validator), ExecutionClass: spec.ExecutionClass, Target: target, ArtifactKind: artifactKind, }) } return chain, nil } func cloneResolvedValidatorChains(chains []ResolvedValidatorChain) []ResolvedValidatorChain { if len(chains) == 0 { return nil } out := make([]ResolvedValidatorChain, len(chains)) for i, chain := range chains { out[i] = ResolvedValidatorChain{ Stage: chain.Stage, LaneID: strings.TrimSpace(chain.LaneID), ModuleKey: strings.TrimSpace(chain.ModuleKey), Validators: cloneResolvedValidators(chain.Validators), } } return out } func cloneResolvedValidators(validators []ResolvedValidator) []ResolvedValidator { if len(validators) == 0 { return nil } out := make([]ResolvedValidator, len(validators)) for i, validator := range validators { out[i] = ResolvedValidator{ Binding: cloneModuleBinding(validator.Binding), ExecutionClass: validator.ExecutionClass, Target: validator.Target, ArtifactKind: validator.ArtifactKind, } } return out } func referenceTarget(stage ModuleStage, laneID string, module string, bindings []ReferenceBinding) ResolvedReferenceTarget { return ResolvedReferenceTarget{ Stage: stage, LaneID: strings.TrimSpace(laneID), Module: strings.TrimSpace(module), Bindings: append([]ReferenceBinding(nil), bindings...), } } func mergeReferenceMaps(base map[string]ReferenceSource, override map[string]ReferenceSource) map[string]ReferenceSource { if len(base) == 0 && len(override) == 0 { return nil } out := make(map[string]ReferenceSource, len(base)+len(override)) for key, value := range base { out[key] = cloneReferenceSource(value) } for key, value := range override { out[key] = cloneReferenceSource(value) } return out } func referenceSourceMapsConflict(base, override map[string]ReferenceSource) bool { for rawKey, left := range base { key := strings.TrimSpace(rawKey) for rawOverrideKey, right := range override { if key == strings.TrimSpace(rawOverrideKey) && left.IsGenerated() != right.IsGenerated() { return true } } } return false } func validatePipelineReferenceDefaults( pipelineID string, pipelineReferences map[string]ReferenceSource, chunkSpec ModuleSpec, lanesByID map[string]ArtifactLaneProfile, catalog ModuleCatalog, ) error { normalizedPipelineReferences, err := normalizedReferenceMap(pipelineReferences, fmt.Sprintf("pipeline %q reference slot", pipelineID)) if err != nil { return err } if len(normalizedPipelineReferences) == 0 { return nil } declaredByAnyTarget := make(map[string]struct{}, len(normalizedPipelineReferences)) for _, slot := range chunkSpec.ReferenceSlots { declaredByAnyTarget[slot.Name] = struct{}{} } for _, laneID := range sortedArtifactLaneProfileKeys(lanesByID) { laneProfile := lanesByID[laneID] extract := resolveBinding(laneProfile.Extract, "") if extract.Module == "" { return fmt.Errorf("pipeline %q lane %q extract module must not be empty", pipelineID, laneID) } extractSpec, err := extractorSpec(catalog, extract.Module) if err != nil { return moduleLookupError(pipelineID, laneID, StageExtract, extract.Module, err) } for _, slot := range extractSpec.ReferenceSlots { declaredByAnyTarget[slot.Name] = struct{}{} } merge := resolveBinding(laneProfile.Merge, DefaultMergeModule) var artifactType reflect.Type artifactKind := extractSpec.ArtifactKind if artifactKind != "" && catalog.Extractors != nil { if entry, ok := catalog.Extractors.typedEntry(extract.Module); ok { artifactType = entry.valueType } } mergeSpec, err := mergerSpecForArtifact(catalog, merge.Module, artifactKind, artifactType) 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 := normalizerSpecForArtifact(catalog, normalize.Module, artifactKind, artifactType) if err != nil { return moduleLookupError(pipelineID, laneID, StageNormalize, normalize.Module, err) } for _, slot := range normalizeSpec.ReferenceSlots { declaredByAnyTarget[slot.Name] = struct{}{} } } for _, slotName := range sortedStringMapKeys(normalizedPipelineReferences) { if _, ok := declaredByAnyTarget[slotName]; !ok { return fmt.Errorf("pipeline %q reference slot %q is not declared by any eligible reference target", pipelineID, slotName) } } return nil } type referenceResolutionTarget struct { PipelineID string LaneID string Stage ModuleStage Module string Slots []contracts.ReferenceSlot PipelineReferences map[string]ReferenceSource LocalReferences map[string]ReferenceSource Options ResolveOptions } func resolveReferenceTargetBindings(target referenceResolutionTarget) ([]ReferenceBinding, error) { slotByName := make(map[string]contracts.ReferenceSlot, len(target.Slots)) for _, slot := range target.Slots { slotByName[slot.Name] = slot } bindings := make(map[string]ReferenceBinding) addBinding := func(slotName string, source ReferenceSource, bindingSource string) error { slotName = strings.TrimSpace(slotName) if slotName == "" { return fmt.Errorf("%s reference slot name must not be empty", referenceTargetErrorContext(target)) } if source.Artifact == nil { source.Path = strings.TrimSpace(source.Path) } if source.Artifact == nil && source.Path == "" { return fmt.Errorf("%s reference slot %q source must not be empty", referenceTargetErrorContext(target), slotName) } if source.Artifact != nil { if strings.TrimSpace(source.Path) != "" { return fmt.Errorf("%s reference slot %q must contain exactly one source form", referenceTargetErrorContext(target), slotName) } artifact := *source.Artifact artifact.Step = strings.TrimSpace(artifact.Step) artifact.Lane = strings.TrimSpace(artifact.Lane) if artifact.Step == "" || artifact.Lane == "" { return fmt.Errorf("%s reference slot %q artifact selector step and lane must not be empty", referenceTargetErrorContext(target), slotName) } source.Artifact = &artifact } if _, ok := slotByName[slotName]; !ok { return fmt.Errorf("%s reference slot %q is not declared by %s module %q", referenceTargetErrorContext(target), slotName, target.Stage, target.Module) } if previous, exists := bindings[slotName]; exists && (previous.Artifact != nil || source.Artifact != nil) { return fmt.Errorf("%s reference slot %q has conflicting generated and external bindings", referenceTargetErrorContext(target), slotName) } binding := ReferenceBinding{ LaneID: target.LaneID, SlotName: slotName, BindingSource: bindingSource, } if source.Artifact != nil { binding.Artifact = source.Artifact } else { binding.Source = source.Path } bindings[slotName] = binding return nil } normalizedPipelineReferences, err := normalizedReferenceMap(target.PipelineReferences, fmt.Sprintf("pipeline %q reference slot", target.PipelineID)) if err != nil { return nil, err } for _, slotName := range sortedStringMapKeys(normalizedPipelineReferences) { if _, ok := slotByName[slotName]; !ok { continue } if err := addBinding(slotName, normalizedPipelineReferences[slotName], contracts.ReferenceBindingSourceConfig); err != nil { return nil, err } } normalizedLocalReferences, err := normalizedReferenceMap(target.LocalReferences, referenceTargetSlotLabel(target)) if err != nil { return nil, err } for _, slotName := range sortedStringMapKeys(normalizedLocalReferences) { if err := addBinding(slotName, normalizedLocalReferences[slotName], contracts.ReferenceBindingSourceConfig); err != nil { return nil, err } } for _, override := range target.Options.ReferenceOverrides { match, err := referenceOverrideMatchesTarget(target, override) if err != nil { return nil, err } if !match { continue } source := override.BindingSource if strings.TrimSpace(source) == "" { source = contracts.ReferenceBindingSourceCLI } if err := addBinding(override.SlotName, ExternalReference(override.Source), strings.TrimSpace(source)); err != nil { return nil, err } } for _, unbind := range target.Options.ReferenceUnbinds { match, err := referenceUnbindMatchesTarget(target, unbind) if err != nil { return nil, err } if !match { continue } slotName := strings.TrimSpace(unbind.SlotName) if slotName == "" { return nil, fmt.Errorf("%s reference unbind slot name must not be empty", referenceTargetErrorContext(target)) } if _, ok := slotByName[slotName]; !ok { return nil, fmt.Errorf("%s reference slot %q is not declared", referenceTargetErrorContext(target), slotName) } if binding, ok := bindings[slotName]; ok && binding.Artifact == nil { delete(bindings, slotName) } } 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 } 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, StageMerge, 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)) for _, slotName := range keys { resolved = append(resolved, bindings[slotName]) } return resolved } func referenceTargetErrorContext(target referenceResolutionTarget) string { if target.LaneID != "" { return fmt.Sprintf("pipeline %q lane %q %s module %q", target.PipelineID, target.LaneID, target.Stage, target.Module) } return fmt.Sprintf("pipeline %q %s module %q", target.PipelineID, target.Stage, target.Module) } func referenceTargetSlotLabel(target referenceResolutionTarget) string { if target.LaneID != "" { return fmt.Sprintf("pipeline %q lane %q %s reference slot", target.PipelineID, target.LaneID, target.Stage) } return fmt.Sprintf("pipeline %q %s reference slot", target.PipelineID, target.Stage) } func normalizedReferenceMap(values map[string]ReferenceSource, keyName string) (map[string]ReferenceSource, error) { if len(values) == 0 { return nil, nil } out := make(map[string]ReferenceSource, len(values)) for rawSlotName, rawSource := range values { slotName := strings.TrimSpace(rawSlotName) if slotName == "" { return nil, fmt.Errorf("%s must not be empty", keyName) } if _, ok := out[slotName]; ok { return nil, fmt.Errorf("%s %q is duplicated after trimming", keyName, slotName) } source := cloneReferenceSource(rawSource) if source.Artifact == nil && strings.TrimSpace(source.Path) == "" { return nil, fmt.Errorf("%s %q source must not be empty", keyName, slotName) } out[slotName] = source } return out, nil } func sortedArtifactLaneProfileKeys(values map[string]ArtifactLaneProfile) []string { if len(values) == 0 { return nil } keys := make([]string, 0, len(values)) for key := range values { keys = append(keys, key) } sort.Strings(keys) return keys } func sortedStringMapKeys(values map[string]ReferenceSource) []string { if len(values) == 0 { return nil } keys := make([]string, 0, len(values)) for key := range values { keys = append(keys, key) } sort.Strings(keys) return keys } func sortedReferenceBindingKeys(values map[string]ReferenceBinding) []string { if len(values) == 0 { return nil } keys := make([]string, 0, len(values)) for key := range values { keys = append(keys, key) } sort.Strings(keys) return keys } func resolveBinding(binding ModuleBinding, defaultModule string) ModuleBinding { module := strings.TrimSpace(binding.Module) if module == "" { module = defaultModule } llmProfile := strings.TrimSpace(binding.LLMProfile) return ModuleBinding{ Module: module, LLMProfile: llmProfile, Retries: binding.Retries, Options: cloneOptions(binding.Options), References: normalizeReferenceMap(binding.References), Validators: cloneValidatorOverride(binding.Validators), } } func resolveBindings(bindings []ModuleBinding, defaultModule string) []ModuleBinding { if len(bindings) == 0 { return nil } resolved := make([]ModuleBinding, 0, len(bindings)) for _, binding := range bindings { resolvedBinding := resolveBinding(binding, defaultModule) resolved = append(resolved, resolvedBinding) } return resolved } func cloneOptions(options map[string]any) map[string]any { if len(options) == 0 { return nil } copied := make(map[string]any, len(options)) for key, value := range options { copied[key] = value } return copied } func normalizeReferenceMap(values map[string]ReferenceSource) map[string]ReferenceSource { if len(values) == 0 { return nil } out := make(map[string]ReferenceSource, len(values)) keys := make([]string, 0, len(values)) rawByNormalized := make(map[string]string, len(values)) for rawKey := range values { key := strings.TrimSpace(rawKey) rawByNormalized[key] = rawKey keys = append(keys, key) } sort.Strings(keys) for _, key := range keys { out[key] = cloneReferenceSource(values[rawByNormalized[key]]) } return out } func selectedArtifactLanes(pipelineID string, artifacts map[string]ArtifactLaneProfile, options ResolveOptions) (map[string]ArtifactLaneProfile, []string, error) { lanesByID := make(map[string]ArtifactLaneProfile, len(artifacts)) for rawLaneID, lane := range artifacts { laneID := strings.TrimSpace(rawLaneID) if laneID == "" { return nil, nil, fmt.Errorf("pipeline %q artifact lane id must not be empty", pipelineID) } if _, ok := lanesByID[laneID]; ok { return nil, nil, fmt.Errorf("pipeline %q artifact lane %q is duplicated after trimming", pipelineID, laneID) } lanesByID[laneID] = lane } if len(options.Only) == 0 { keys := make([]string, 0, len(lanesByID)) for laneID := range lanesByID { keys = append(keys, laneID) } sort.Strings(keys) return lanesByID, keys, nil } selected := make(map[string]struct{}, len(options.Only)) for _, rawLaneID := range options.Only { laneID := strings.TrimSpace(rawLaneID) if laneID == "" { return nil, nil, fmt.Errorf("pipeline %q selected artifact lane id must not be empty", pipelineID) } if _, ok := lanesByID[laneID]; !ok { return nil, nil, fmt.Errorf("pipeline %q selected artifact lane %q is not declared", pipelineID, laneID) } selected[laneID] = struct{}{} } keys := make([]string, 0, len(selected)) for laneID := range selected { keys = append(keys, laneID) } sort.Strings(keys) return lanesByID, keys, nil } func resolvedPipelineDigest(resolved ResolvedPipeline) (string, error) { withoutDigest := struct { ID string Input ModuleBinding Chunk ModuleBinding ChunkReferences ResolvedReferenceTarget Steps []ResolvedPipelineStep ValidatorChains []ResolvedValidatorChain Output ModuleBinding }{ ID: resolved.ID, Input: resolved.Input, Chunk: resolved.Chunk, ChunkReferences: resolved.ChunkReferences, Steps: resolved.Steps, ValidatorChains: resolved.ValidatorChains, Output: resolved.Output, } encoded, err := json.Marshal(withoutDigest) if err != nil { return "", err } sum := sha256.Sum256(encoded) return "sha256:" + hex.EncodeToString(sum[:]), nil } func inputSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) { return registrySpec(catalog.Inputs, key) } func chunkerSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) { return registrySpec(catalog.Chunkers, key) } func extractorSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) { return registrySpec(catalog.Extractors, key) } func outputSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) { return registrySpec(catalog.Outputs, key) } type specRegistry interface { Spec(key string) (ModuleSpec, bool) } func registrySpec(registry specRegistry, key string) (ModuleSpec, error) { if registry == nil { return ModuleSpec{}, fmt.Errorf("module %q is not registered", key) } spec, ok := registry.Spec(key) if !ok { return ModuleSpec{}, fmt.Errorf("module %q is not registered", key) } return spec, nil } func moduleLookupError(pipelineID, laneID string, stage ModuleStage, module string, err error) error { if laneID != "" { return fmt.Errorf("pipeline %q lane %q %s module %q: %w", pipelineID, laneID, stage, module, err) } return fmt.Errorf("pipeline %q %s module %q: %w", pipelineID, stage, module, err) } func capabilityError(pipelineID, laneID string, stage ModuleStage, module, capability string) error { if laneID != "" { return fmt.Errorf("pipeline %q lane %q %s module %q requires missing capability %q", pipelineID, laneID, stage, module, capability) } return fmt.Errorf("pipeline %q %s module %q requires missing capability %q", pipelineID, stage, module, capability) } type capabilitySet map[string]struct{} func newCapabilitySet() capabilitySet { return make(capabilitySet) } func (set capabilitySet) clone() capabilitySet { copied := make(capabilitySet, len(set)) for capability := range set { copied[capability] = struct{}{} } return copied } func (set capabilitySet) add(values ...string) { for _, value := range values { set[value] = struct{}{} } } func (set capabilitySet) addSet(other capabilitySet) { for value := range other { set[value] = struct{}{} } } func (set capabilitySet) missing(required []string) (string, bool) { for _, capability := range required { if _, ok := set[capability]; !ok { return capability, true } } return "", false }