From 7580230269c55eb1b165cbf0c8393cb0ab141386 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Fri, 3 Jul 2026 15:37:07 +0000 Subject: [PATCH] Add resolved pipeline profiles --- internal/core/artifacts/artifacts.go | 35 +- internal/core/artifacts/artifacts_test.go | 41 ++ internal/framework/pipeline/profile.go | 404 +++++++++++++ internal/framework/pipeline/profile_test.go | 630 ++++++++++++++++++++ 4 files changed, 1098 insertions(+), 12 deletions(-) create mode 100644 internal/framework/pipeline/profile.go create mode 100644 internal/framework/pipeline/profile_test.go diff --git a/internal/core/artifacts/artifacts.go b/internal/core/artifacts/artifacts.go index e68230f..7f69d2e 100644 --- a/internal/core/artifacts/artifacts.go +++ b/internal/core/artifacts/artifacts.go @@ -33,19 +33,30 @@ type RejectedArtifact struct { Message string `json:"message"` } +type ArtifactLaneManifest struct { + ID string `json:"id"` + Extractor string `json:"extractor"` + Merger string `json:"merger"` + Normalizer string `json:"normalizer"` + Validators []string `json:"validators,omitempty"` +} + type RunManifest struct { - RunID string `json:"run_id,omitempty"` - InputModule string `json:"input_module,omitempty"` - Chunker string `json:"chunker,omitempty"` - SourceDigests []string `json:"source_digests,omitempty"` - Extractors []string `json:"extractors,omitempty"` - Merger string `json:"merger,omitempty"` - Normalizer string `json:"normalizer,omitempty"` - OutputEncoder string `json:"output_encoder,omitempty"` - SchemaVersion string `json:"schema_version,omitempty"` - ValidationStatus string `json:"validation_status,omitempty"` - StartedAt *time.Time `json:"started_at,omitempty"` - CompletedAt *time.Time `json:"completed_at,omitempty"` + RunID string `json:"run_id,omitempty"` + PipelineID string `json:"pipeline_id,omitempty"` + PipelineDigest string `json:"pipeline_digest,omitempty"` + InputModule string `json:"input_module,omitempty"` + Chunker string `json:"chunker,omitempty"` + SourceDigests []string `json:"source_digests,omitempty"` + Extractors []string `json:"extractors,omitempty"` + Merger string `json:"merger,omitempty"` + Normalizer string `json:"normalizer,omitempty"` + OutputEncoder string `json:"output_encoder,omitempty"` + ArtifactLanes []ArtifactLaneManifest `json:"artifact_lanes,omitempty"` + SchemaVersion string `json:"schema_version,omitempty"` + ValidationStatus string `json:"validation_status,omitempty"` + StartedAt *time.Time `json:"started_at,omitempty"` + CompletedAt *time.Time `json:"completed_at,omitempty"` } func ArtifactFromCandidate(candidate ArtifactCandidate) Artifact { diff --git a/internal/core/artifacts/artifacts_test.go b/internal/core/artifacts/artifacts_test.go index ff1425a..fe2eb04 100644 --- a/internal/core/artifacts/artifacts_test.go +++ b/internal/core/artifacts/artifacts_test.go @@ -123,6 +123,47 @@ func TestRunManifestOmitsEmptyOptionalFields(t *testing.T) { } } +func TestRunManifestIncludesPipelineAndArtifactLaneFields(t *testing.T) { + manifest := RunManifest{ + PipelineID: "pipeline-1", + PipelineDigest: "sha256:abc123", + ArtifactLanes: []ArtifactLaneManifest{ + { + ID: "spells", + Extractor: "spell-extractor", + Merger: "appendorder", + Normalizer: "noop", + Validators: []string{"grounded"}, + }, + }, + } + + gotJSON, err := json.Marshal(manifest) + if err != nil { + t.Fatalf("json.Marshal() error = %v", err) + } + + var got map[string]any + if err := json.Unmarshal(gotJSON, &got); err != nil { + t.Fatalf("json.Unmarshal() error = %v", err) + } + + assertHasKeys(t, got, "pipeline_id", "pipeline_digest", "artifact_lanes") + + lanes, ok := got["artifact_lanes"].([]any) + if !ok { + t.Fatalf("artifact_lanes = %#v, want array", got["artifact_lanes"]) + } + if len(lanes) != 1 { + t.Fatalf("len(artifact_lanes) = %d, want 1", len(lanes)) + } + lane, ok := lanes[0].(map[string]any) + if !ok { + t.Fatalf("artifact_lanes[0] = %#v, want object", lanes[0]) + } + assertHasKeys(t, lane, "id", "extractor", "merger", "normalizer", "validators") +} + func assertHasKeys(t *testing.T, values map[string]any, keys ...string) { t.Helper() diff --git a/internal/framework/pipeline/profile.go b/internal/framework/pipeline/profile.go new file mode 100644 index 0000000..b667e49 --- /dev/null +++ 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 +} diff --git a/internal/framework/pipeline/profile_test.go b/internal/framework/pipeline/profile_test.go new file mode 100644 index 0000000..e5adf50 --- /dev/null +++ b/internal/framework/pipeline/profile_test.go @@ -0,0 +1,630 @@ +package pipeline + +import ( + "context" + "encoding/json" + "reflect" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +func TestResolvePipelineWithExplicitModules(t *testing.T) { + catalog := newProfileCatalog(t) + registerProfileSpecs(t, catalog, + ModuleSpec{Key: "window", Stage: StageChunk, Requires: []string{"source"}, Provides: []string{"chunk"}}, + ModuleSpec{Key: "npc-extractor", Stage: StageExtract, Requires: []string{"chunk"}, Provides: []string{"candidate"}}, + ModuleSpec{Key: "dedupe", Stage: StageMerge, Requires: []string{"candidate"}, Provides: []string{"merged"}}, + ModuleSpec{Key: "canonical", Stage: StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}}, + ModuleSpec{Key: "schema-check", Stage: StageValidate, Requires: []string{"normalized"}, Provides: []string{"validated"}}, + ModuleSpec{Key: "ndjson", Stage: StageOutput, Requires: []string{"validated"}, Provides: []string{"encoded"}}, + ) + + resolved, err := ResolvePipeline(PipelineProfile{ + ID: " campaign ", + Input: ModuleBinding{Module: " text ", LLMProfile: " fast "}, + Chunk: ModuleBinding{Module: " window ", Options: map[string]any{ + "size": 10, + }}, + Artifacts: map[string]ArtifactLaneProfile{ + " npcs ": { + Extract: ModuleBinding{Module: " npc-extractor ", LLMProfile: " careful "}, + Merge: Binding(" dedupe "), + Normalize: Binding(" canonical "), + Validators: []ModuleBinding{Binding(" schema-check ")}, + }, + }, + Output: Binding(" ndjson "), + }, ResolveOptions{}, catalog) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + + if resolved.ID != "campaign" { + t.Fatalf("ID = %q, want campaign", resolved.ID) + } + if !reflect.DeepEqual(resolved.Input, ModuleBinding{Module: "text", LLMProfile: "fast"}) { + t.Fatalf("Input = %#v, want trimmed explicit input", resolved.Input) + } + if resolved.Chunk.Module != "window" || resolved.Chunk.LLMProfile != DefaultLLMProfile { + t.Fatalf("Chunk = %#v, want explicit module and default LLM profile", resolved.Chunk) + } + if resolved.Chunk.Options["size"] != 10 { + t.Fatalf("Chunk.Options = %#v, want size option", resolved.Chunk.Options) + } + if len(resolved.ArtifactLanes) != 1 { + t.Fatalf("len(ArtifactLanes) = %d, want 1", len(resolved.ArtifactLanes)) + } + lane := resolved.ArtifactLanes[0] + if lane.ID != "npcs" { + t.Fatalf("lane.ID = %q, want npcs", lane.ID) + } + if !reflect.DeepEqual(lane.Extract, ModuleBinding{Module: "npc-extractor", LLMProfile: "careful"}) { + t.Fatalf("lane.Extract = %#v, want explicit extractor", lane.Extract) + } + if lane.Merge.Module != "dedupe" || lane.Normalize.Module != "canonical" { + t.Fatalf("lane merge/normalize = %#v/%#v, want explicit modules", lane.Merge, lane.Normalize) + } + if len(lane.Validators) != 1 || lane.Validators[0].Module != "schema-check" { + t.Fatalf("lane.Validators = %#v, want schema-check", lane.Validators) + } + if resolved.Output.Module != "ndjson" { + t.Fatalf("Output.Module = %q, want ndjson", resolved.Output.Module) + } + if !strings.HasPrefix(resolved.Digest, "sha256:") { + t.Fatalf("Digest = %q, want sha256 digest", resolved.Digest) + } +} + +func TestResolvePipelineAppliesDefaults(t *testing.T) { + resolved, err := ResolvePipeline(PipelineProfile{ + ID: "defaulted", + Input: Binding("text"), + Artifacts: map[string]ArtifactLaneProfile{ + "spells": {Extract: Binding("spell-extractor")}, + }, + }, ResolveOptions{}, newProfileCatalog(t)) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + + if resolved.Input.LLMProfile != DefaultLLMProfile { + t.Fatalf("Input.LLMProfile = %q, want %q", resolved.Input.LLMProfile, DefaultLLMProfile) + } + if !reflect.DeepEqual(resolved.Chunk, ModuleBinding{Module: DefaultChunkModule, LLMProfile: DefaultLLMProfile}) { + t.Fatalf("Chunk = %#v, want default chunk binding", resolved.Chunk) + } + if !reflect.DeepEqual(resolved.Output, ModuleBinding{Module: DefaultOutputModule, LLMProfile: DefaultLLMProfile}) { + t.Fatalf("Output = %#v, want default output binding", resolved.Output) + } + lane := resolved.ArtifactLanes[0] + if !reflect.DeepEqual(lane.Merge, ModuleBinding{Module: DefaultMergeModule, LLMProfile: DefaultLLMProfile}) { + t.Fatalf("Merge = %#v, want default merge binding", lane.Merge) + } + if !reflect.DeepEqual(lane.Normalize, ModuleBinding{Module: DefaultNormalizeModule, LLMProfile: DefaultLLMProfile}) { + t.Fatalf("Normalize = %#v, want default normalize binding", lane.Normalize) + } + if lane.Extract.LLMProfile != DefaultLLMProfile { + t.Fatalf("Extract.LLMProfile = %q, want %q", lane.Extract.LLMProfile, DefaultLLMProfile) + } +} + +func TestResolvePipelineSelectsOnlyRequestedLanes(t *testing.T) { + profile := multiLaneProfile() + resolved, err := ResolvePipeline(profile, ResolveOptions{Only: []string{" treasure ", "spells", "treasure"}}, newProfileCatalog(t)) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + + got := laneIDs(resolved.ArtifactLanes) + want := []string{"spells", "treasure"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("lane IDs = %#v, want %#v", got, want) + } +} + +func TestResolvePipelineRejectsUnknownOnlyLane(t *testing.T) { + _, err := ResolvePipeline(multiLaneProfile(), ResolveOptions{Only: []string{"missing"}}, newProfileCatalog(t)) + if err == nil { + t.Fatal("ResolvePipeline() error = nil, want error") + } + assertErrorContains(t, err, "pipeline", "missing", "not declared") +} + +func TestResolvePipelineRejectsEmptyOnlyLane(t *testing.T) { + _, err := ResolvePipeline(multiLaneProfile(), ResolveOptions{Only: []string{" \t"}}, newProfileCatalog(t)) + if err == nil { + t.Fatal("ResolvePipeline() error = nil, want error") + } + assertErrorContains(t, err, "pipeline", "artifact lane", "empty") +} + +func TestResolvePipelineRejectsEmptyArtifactSet(t *testing.T) { + _, err := ResolvePipeline(PipelineProfile{ + ID: "empty", + Input: Binding("text"), + Artifacts: map[string]ArtifactLaneProfile{}, + }, ResolveOptions{}, newProfileCatalog(t)) + if err == nil { + t.Fatal("ResolvePipeline() error = nil, want error") + } + assertErrorContains(t, err, "empty", "artifact lane") +} + +func TestResolvePipelineRejectsEmptyPipelineID(t *testing.T) { + _, err := ResolvePipeline(PipelineProfile{ + ID: " ", + Input: Binding("text"), + Artifacts: map[string]ArtifactLaneProfile{ + "spells": {Extract: Binding("spell-extractor")}, + }, + }, ResolveOptions{}, newProfileCatalog(t)) + if err == nil { + t.Fatal("ResolvePipeline() error = nil, want error") + } + assertErrorContains(t, err, "pipeline id", "empty") +} + +func TestResolvePipelineRejectsMissingInput(t *testing.T) { + _, err := ResolvePipeline(PipelineProfile{ + ID: "missing-input", + Artifacts: map[string]ArtifactLaneProfile{ + "spells": {Extract: Binding("spell-extractor")}, + }, + }, ResolveOptions{}, newProfileCatalog(t)) + if err == nil { + t.Fatal("ResolvePipeline() error = nil, want error") + } + assertErrorContains(t, err, "missing-input", "input", "empty") +} + +func TestResolvePipelineRejectsUnknownModuleKeys(t *testing.T) { + tests := []struct { + name string + profile PipelineProfile + want []string + }{ + { + name: "input", + profile: PipelineProfile{ + ID: "unknown-input", + Input: Binding("missing-input"), + Artifacts: map[string]ArtifactLaneProfile{ + "spells": {Extract: Binding("spell-extractor")}, + }, + }, + want: []string{"unknown-input", "input", "missing-input"}, + }, + { + name: "chunk", + profile: withProfileChange(func(profile PipelineProfile) PipelineProfile { + profile.Chunk = Binding("missing-chunk") + return profile + }), + want: []string{"baseline", "chunk", "missing-chunk"}, + }, + { + name: "extract", + profile: withProfileChange(func(profile PipelineProfile) PipelineProfile { + lane := profile.Artifacts["spells"] + lane.Extract = Binding("missing-extractor") + profile.Artifacts["spells"] = lane + return profile + }), + want: []string{"baseline", "spells", "extract", "missing-extractor"}, + }, + { + name: "merge", + profile: withProfileChange(func(profile PipelineProfile) PipelineProfile { + lane := profile.Artifacts["spells"] + lane.Merge = Binding("missing-merge") + profile.Artifacts["spells"] = lane + return profile + }), + want: []string{"baseline", "spells", "merge", "missing-merge"}, + }, + { + name: "normalize", + profile: withProfileChange(func(profile PipelineProfile) PipelineProfile { + lane := profile.Artifacts["spells"] + lane.Normalize = Binding("missing-normalize") + profile.Artifacts["spells"] = lane + return profile + }), + want: []string{"baseline", "spells", "normalize", "missing-normalize"}, + }, + { + name: "validate", + profile: withProfileChange(func(profile PipelineProfile) PipelineProfile { + lane := profile.Artifacts["spells"] + lane.Validators = []ModuleBinding{Binding("missing-validator")} + profile.Artifacts["spells"] = lane + return profile + }), + want: []string{"baseline", "spells", "validate", "missing-validator"}, + }, + { + name: "output", + profile: withProfileChange(func(profile PipelineProfile) PipelineProfile { + profile.Output = Binding("missing-output") + return profile + }), + want: []string{"baseline", "output", "missing-output"}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + _, err := ResolvePipeline(test.profile, ResolveOptions{}, newProfileCatalog(t)) + if err == nil { + t.Fatal("ResolvePipeline() error = nil, want error") + } + assertErrorContains(t, err, test.want...) + }) + } +} + +func TestResolvePipelineRejectsMissingCapabilities(t *testing.T) { + tests := []struct { + name string + spec ModuleSpec + want []string + }{ + { + name: "input", + spec: ModuleSpec{Key: "text", Stage: StageInput, Requires: []string{"raw"}}, + want: []string{"baseline", "input", "text", "raw"}, + }, + { + name: "chunk", + spec: ModuleSpec{Key: "generic", Stage: StageChunk, Requires: []string{"missing"}}, + want: []string{"baseline", "chunk", "generic", "missing"}, + }, + { + name: "extract", + spec: ModuleSpec{Key: "spell-extractor", Stage: StageExtract, Requires: []string{"missing"}}, + want: []string{"baseline", "spells", "extract", "spell-extractor", "missing"}, + }, + { + name: "merge", + spec: ModuleSpec{Key: "appendorder", Stage: StageMerge, Requires: []string{"missing"}}, + want: []string{"baseline", "spells", "merge", "appendorder", "missing"}, + }, + { + name: "normalize", + spec: ModuleSpec{Key: "noop", Stage: StageNormalize, Requires: []string{"missing"}}, + want: []string{"baseline", "spells", "normalize", "noop", "missing"}, + }, + { + name: "validate", + spec: ModuleSpec{Key: "grounded", Stage: StageValidate, Requires: []string{"missing"}}, + want: []string{"baseline", "spells", "validate", "grounded", "missing"}, + }, + { + name: "output", + spec: ModuleSpec{Key: "json", Stage: StageOutput, Requires: []string{"missing"}}, + want: []string{"baseline", "output", "json", "missing"}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + catalog := newProfileCatalogWithOverride(t, test.spec) + profile := baselineProfile() + lane := profile.Artifacts["spells"] + lane.Validators = []ModuleBinding{Binding("grounded")} + profile.Artifacts["spells"] = lane + + _, err := ResolvePipeline(profile, ResolveOptions{}, catalog) + if err == nil { + t.Fatal("ResolvePipeline() error = nil, want error") + } + assertErrorContains(t, err, test.want...) + }) + } +} + +func TestResolvePipelineOrdersLanesDeterministically(t *testing.T) { + resolved, err := ResolvePipeline(multiLaneProfile(), ResolveOptions{}, newProfileCatalog(t)) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + + got := laneIDs(resolved.ArtifactLanes) + want := []string{"items", "spells", "treasure"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("lane IDs = %#v, want %#v", got, want) + } +} + +func TestResolvePipelineDigestIsDeterministicForEquivalentMaps(t *testing.T) { + left := PipelineProfile{ + ID: "digest", + Input: Binding("text"), + Output: Binding("json"), + Chunk: ModuleBinding{Module: "generic", Options: map[string]any{"b": 2, "a": 1}}, + Artifacts: map[string]ArtifactLaneProfile{ + "spells": {Extract: Binding("spell-extractor")}, + "items": {Extract: Binding("item-extractor")}, + }, + } + right := PipelineProfile{ + ID: "digest", + Input: Binding("text"), + Output: Binding("json"), + Chunk: ModuleBinding{Module: "generic", Options: map[string]any{"a": 1, "b": 2}}, + Artifacts: map[string]ArtifactLaneProfile{ + "items": {Extract: Binding("item-extractor")}, + "spells": {Extract: Binding("spell-extractor")}, + }, + } + + leftResolved, err := ResolvePipeline(left, ResolveOptions{}, newProfileCatalog(t)) + if err != nil { + t.Fatalf("ResolvePipeline(left) error = %v, want nil", err) + } + rightResolved, err := ResolvePipeline(right, ResolveOptions{}, newProfileCatalog(t)) + if err != nil { + t.Fatalf("ResolvePipeline(right) error = %v, want nil", err) + } + + if leftResolved.Digest != rightResolved.Digest { + t.Fatalf("digests differ for equivalent profiles: %q != %q", leftResolved.Digest, rightResolved.Digest) + } +} + +func TestResolvePipelineDigestChangesWhenBindingChanges(t *testing.T) { + left := baselineProfile() + right := baselineProfile() + right.Chunk = Binding("window") + catalog := newProfileCatalog(t) + registerProfileSpecs(t, catalog, ModuleSpec{Key: "window", Stage: StageChunk, Requires: []string{"source"}, Provides: []string{"chunk"}}) + + leftResolved, err := ResolvePipeline(left, ResolveOptions{}, catalog) + if err != nil { + t.Fatalf("ResolvePipeline(left) error = %v, want nil", err) + } + rightResolved, err := ResolvePipeline(right, ResolveOptions{}, catalog) + if err != nil { + t.Fatalf("ResolvePipeline(right) error = %v, want nil", err) + } + + if leftResolved.Digest == rightResolved.Digest { + t.Fatalf("digest = %q for both profiles, want changed digest", leftResolved.Digest) + } +} + +func TestBindingTrimsModuleAndLeavesResolutionFieldsEmpty(t *testing.T) { + binding := Binding(" module ") + if binding.Module != "module" { + t.Fatalf("Module = %q, want module", binding.Module) + } + if binding.LLMProfile != "" { + t.Fatalf("LLMProfile = %q, want empty", binding.LLMProfile) + } + if binding.Options != nil { + t.Fatalf("Options = %#v, want nil", binding.Options) + } +} + +func TestResolvedPipelineDigestExcludesDigestField(t *testing.T) { + resolved, err := ResolvePipeline(baselineProfile(), ResolveOptions{}, newProfileCatalog(t)) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + changed := resolved + changed.Digest = "sha256:changed" + + leftDigest, err := resolvedPipelineDigest(resolved) + if err != nil { + t.Fatalf("resolvedPipelineDigest(resolved) error = %v, want nil", err) + } + rightDigest, err := resolvedPipelineDigest(changed) + if err != nil { + t.Fatalf("resolvedPipelineDigest(changed) error = %v, want nil", err) + } + if leftDigest != rightDigest { + t.Fatalf("digest with changed digest field = %q, want %q", rightDigest, leftDigest) + } +} + +func baselineProfile() PipelineProfile { + return PipelineProfile{ + ID: "baseline", + Input: Binding("text"), + Artifacts: map[string]ArtifactLaneProfile{ + "spells": {Extract: Binding("spell-extractor")}, + }, + } +} + +func multiLaneProfile() PipelineProfile { + profile := baselineProfile() + profile.ID = "multi" + profile.Artifacts = map[string]ArtifactLaneProfile{ + "treasure": {Extract: Binding("item-extractor")}, + "spells": {Extract: Binding("spell-extractor")}, + "items": {Extract: Binding("item-extractor")}, + } + return profile +} + +func withProfileChange(change func(PipelineProfile) PipelineProfile) PipelineProfile { + return change(baselineProfile()) +} + +func laneIDs(lanes []ResolvedArtifactLane) []string { + ids := make([]string, 0, len(lanes)) + for _, lane := range lanes { + ids = append(ids, lane.ID) + } + return ids +} + +func assertErrorContains(t *testing.T, err error, values ...string) { + t.Helper() + + message := err.Error() + for _, value := range values { + if !strings.Contains(message, value) { + t.Fatalf("error = %q, want substring %q", message, value) + } + } +} + +func newProfileCatalog(t *testing.T) ModuleCatalog { + t.Helper() + + catalog := emptyProfileCatalog() + registerProfileSpecs(t, catalog, defaultProfileSpecs()...) + return catalog +} + +func newProfileCatalogWithOverride(t *testing.T, override ModuleSpec) ModuleCatalog { + t.Helper() + + specs := defaultProfileSpecs() + for index, spec := range specs { + if spec.Stage == override.Stage && spec.Key == override.Key { + specs[index] = override + catalog := emptyProfileCatalog() + registerProfileSpecs(t, catalog, specs...) + return catalog + } + } + + catalog := emptyProfileCatalog() + registerProfileSpecs(t, catalog, specs...) + registerProfileSpecs(t, catalog, override) + return catalog +} + +func emptyProfileCatalog() ModuleCatalog { + return ModuleCatalog{ + Inputs: NewInputAdapterRegistry(), + Chunkers: NewChunkerRegistry(), + Extractors: NewExtractorRegistry(), + Mergers: NewMergerRegistry(), + Normalizers: NewNormalizerRegistry(), + Validators: NewValidatorRegistry(), + Outputs: NewOutputEncoderRegistry(), + } +} + +func defaultProfileSpecs() []ModuleSpec { + return []ModuleSpec{ + ModuleSpec{Key: "text", Stage: StageInput, Provides: []string{"source"}}, + ModuleSpec{Key: "generic", Stage: StageChunk, Requires: []string{"source"}, Provides: []string{"chunk"}}, + ModuleSpec{Key: "spell-extractor", Stage: StageExtract, Requires: []string{"chunk"}, Provides: []string{"candidate"}}, + ModuleSpec{Key: "item-extractor", Stage: StageExtract, Requires: []string{"chunk"}, Provides: []string{"candidate"}}, + ModuleSpec{Key: "appendorder", Stage: StageMerge, Requires: []string{"candidate"}, Provides: []string{"merged"}}, + ModuleSpec{Key: "noop", Stage: StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}}, + ModuleSpec{Key: "grounded", Stage: StageValidate, Requires: []string{"normalized"}, Provides: []string{"validated"}}, + ModuleSpec{Key: "json", Stage: StageOutput, Requires: []string{"normalized"}, Provides: []string{"encoded"}}, + } +} + +func registerProfileSpecs(t *testing.T, catalog ModuleCatalog, specs ...ModuleSpec) { + t.Helper() + + for _, spec := range specs { + switch spec.Stage { + case StageInput: + if err := catalog.Inputs.RegisterWithSpec(spec, profileInputConstructor(spec.Key)); err != nil { + t.Fatalf("register input spec %#v: %v", spec, err) + } + case StageChunk: + if err := catalog.Chunkers.RegisterWithSpec(spec, profileChunkerConstructor(spec.Key)); err != nil { + t.Fatalf("register chunk spec %#v: %v", spec, err) + } + case StageExtract: + if err := catalog.Extractors.RegisterWithSpec(spec, profileExtractorConstructor(spec.Key)); err != nil { + t.Fatalf("register extractor spec %#v: %v", spec, err) + } + case StageMerge: + if err := catalog.Mergers.RegisterWithSpec(spec, profileMergerConstructor(spec.Key)); err != nil { + t.Fatalf("register merger spec %#v: %v", spec, err) + } + case StageNormalize: + if err := catalog.Normalizers.RegisterWithSpec(spec, profileNormalizerConstructor(spec.Key)); err != nil { + t.Fatalf("register normalizer spec %#v: %v", spec, err) + } + case StageValidate: + if err := catalog.Validators.RegisterWithSpec(spec, profileValidatorConstructor(spec.Key)); err != nil { + t.Fatalf("register validator spec %#v: %v", spec, err) + } + case StageOutput: + if err := catalog.Outputs.RegisterWithSpec(spec, profileOutputConstructor(spec.Key)); err != nil { + t.Fatalf("register output spec %#v: %v", spec, err) + } + default: + t.Fatalf("unsupported spec stage %q", spec.Stage) + } + } +} + +func profileInputConstructor(key string) InputAdapterConstructor { + return func() (contracts.InputAdapter, error) { + return profileInputAdapter{key: key}, nil + } +} + +type profileInputAdapter struct { + key string +} + +func (adapter profileInputAdapter) Key() string { + return adapter.key +} + +func (adapter profileInputAdapter) Parse(ctx context.Context, req contracts.ParseRequest) (*source.SourceDocument, error) { + return &source.SourceDocument{}, nil +} + +func profileChunkerConstructor(key string) ChunkerConstructor { + return func() (contracts.Chunker, error) { + return registryChunker{key: key}, nil + } +} + +func profileExtractorConstructor(key string) ExtractorConstructor { + return func() (contracts.Extractor, error) { + return registryFakeExtractor{key: key}, nil + } +} + +func profileMergerConstructor(key string) MergerConstructor { + return func() (contracts.Merger, error) { + return registryMerger{key: key}, nil + } +} + +func profileNormalizerConstructor(key string) NormalizerConstructor { + return func() (contracts.Normalizer, error) { + return registryNormalizer{key: key}, nil + } +} + +func profileValidatorConstructor(key string) ValidatorConstructor { + return func() (contracts.Validator, error) { + return registryValidator{name: key}, nil + } +} + +func profileOutputConstructor(key string) OutputEncoderConstructor { + return func() (contracts.OutputEncoder, error) { + return registryOutputEncoder{key: key}, nil + } +} + +func TestResolvedPipelineCanMarshalToCanonicalJSON(t *testing.T) { + resolved, err := ResolvePipeline(baselineProfile(), ResolveOptions{}, newProfileCatalog(t)) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + if _, err := json.Marshal(resolved); err != nil { + t.Fatalf("json.Marshal(resolved) error = %v, want nil", err) + } +}