Add resolved pipeline profiles

2026-07-03 15:37:07 +00:00
parent 62bc8983c7
commit 7580230269
4 changed files with 1098 additions and 12 deletions
--- a/internal/framework/pipeline/profile.go
+++ b/internal/framework/pipeline/profile.go
@@ -0,0 +1,404 @@
+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
+}