package pipeline import ( "crypto/sha256" "encoding/hex" "encoding/json" "fmt" "sort" "strings" ) const ( DefaultChunkModule = "generic" DefaultMergeModule = "appendorder" DefaultNormalizeModule = "noop" DefaultOutputModule = "json" DefaultLLMProfile = "default" ) type ModuleBinding struct { Module string `json:"module"` LLMProfile string `json:"llm_profile,omitempty"` Options map[string]any `json:"options,omitempty"` } type ArtifactLaneProfile struct { Extract ModuleBinding `json:"extract"` Merge ModuleBinding `json:"merge,omitempty"` Normalize ModuleBinding `json:"normalize,omitempty"` Validators []ModuleBinding `json:"validators,omitempty"` } type PipelineProfile struct { ID string `json:"id"` Input ModuleBinding `json:"input"` Chunk ModuleBinding `json:"chunk,omitempty"` Artifacts map[string]ArtifactLaneProfile `json:"artifacts"` Output ModuleBinding `json:"output,omitempty"` } type ResolveOptions struct { Only []string } type ResolvedArtifactLane struct { ID string Extract ModuleBinding Merge ModuleBinding Normalize ModuleBinding Validators []ModuleBinding } type ResolvedPipeline struct { ID string Digest string Input ModuleBinding Chunk ModuleBinding ArtifactLanes []ResolvedArtifactLane Output ModuleBinding } type ModuleCatalog struct { Inputs *InputAdapterRegistry Chunkers *ChunkerRegistry Extractors *ExtractorRegistry Mergers *MergerRegistry Normalizers *NormalizerRegistry Validators *ValidatorRegistry 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") } if len(profile.Artifacts) == 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...) lanesByID, selectedLaneIDs, err := selectedArtifactLanes(pipelineID, profile.Artifacts, options) if err != nil { return ResolvedPipeline{}, err } if len(selectedLaneIDs) == 0 { return ResolvedPipeline{}, fmt.Errorf("pipeline %q must select at least one artifact lane", pipelineID) } resolved := ResolvedPipeline{ ID: pipelineID, Input: input, Chunk: chunk, Output: resolveBinding(profile.Output, DefaultOutputModule), } outputCapabilities := capabilities.clone() for _, laneID := range selectedLaneIDs { laneProfile := lanesByID[laneID] lane, laneCapabilities, err := resolveArtifactLane(pipelineID, laneID, laneProfile, capabilities, catalog) if err != nil { return ResolvedPipeline{}, err } resolved.ArtifactLanes = append(resolved.ArtifactLanes, lane) outputCapabilities.addSet(laneCapabilities) } 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) } 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, laneID string, profile ArtifactLaneProfile, inherited capabilitySet, catalog ModuleCatalog) (ResolvedArtifactLane, capabilitySet, error) { lane := ResolvedArtifactLane{ 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, 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, moduleLookupError(pipelineID, laneID, StageExtract, lane.Extract.Module, err) } if missing, ok := capabilities.missing(extractSpec.Requires); ok { return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageExtract, lane.Extract.Module, missing) } capabilities.add(extractSpec.Provides...) mergeSpec, err := mergerSpec(catalog, lane.Merge.Module) if err != nil { return ResolvedArtifactLane{}, nil, moduleLookupError(pipelineID, laneID, StageMerge, lane.Merge.Module, err) } if missing, ok := capabilities.missing(mergeSpec.Requires); ok { return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageMerge, lane.Merge.Module, missing) } capabilities.add(mergeSpec.Provides...) normalizeSpec, err := normalizerSpec(catalog, lane.Normalize.Module) if err != nil { return ResolvedArtifactLane{}, nil, moduleLookupError(pipelineID, laneID, StageNormalize, lane.Normalize.Module, err) } if missing, ok := capabilities.missing(normalizeSpec.Requires); ok { return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageNormalize, lane.Normalize.Module, missing) } capabilities.add(normalizeSpec.Provides...) for _, validator := range lane.Validators { validatorSpec, err := validatorSpec(catalog, validator.Module) if err != nil { return ResolvedArtifactLane{}, nil, moduleLookupError(pipelineID, laneID, StageValidate, validator.Module, err) } if missing, ok := capabilities.missing(validatorSpec.Requires); ok { return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageValidate, validator.Module, missing) } capabilities.add(validatorSpec.Provides...) } return lane, capabilities, nil } func resolveBinding(binding ModuleBinding, defaultModule string) ModuleBinding { module := strings.TrimSpace(binding.Module) if module == "" { module = defaultModule } llmProfile := strings.TrimSpace(binding.LLMProfile) if llmProfile == "" { llmProfile = DefaultLLMProfile } return ModuleBinding{ Module: module, LLMProfile: llmProfile, Options: cloneOptions(binding.Options), } } 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 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 ArtifactLanes []ResolvedArtifactLane Output ModuleBinding }{ ID: resolved.ID, Input: resolved.Input, Chunk: resolved.Chunk, ArtifactLanes: resolved.ArtifactLanes, 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 mergerSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) { return registrySpec(catalog.Mergers, key) } func normalizerSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) { return registrySpec(catalog.Normalizers, key) } func validatorSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) { return registrySpec(catalog.Validators, 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 }