package pipeline import ( "context" "encoding/json" "errors" "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, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"source"}, Provides: []string{"chunk"}}, ModuleSpec{Key: "record-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunk"}, Provides: []string{"candidate"}}, ModuleSpec{Key: "dedupe", Stage: StageMerge, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"candidate"}, Provides: []string{"merged"}}, ModuleSpec{Key: "canonical", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"merged"}, Provides: []string{"normalized"}}, ModuleSpec{Key: "ndjson", Stage: StageOutput, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"normalized"}, Provides: []string{"encoded"}}, ) resolved, err := ResolvePipeline(PipelineProfile{ ID: " campaign ", Input: Binding(" text "), Chunk: ModuleBinding{Module: " window ", Options: map[string]any{ "size": 10, }}, Artifacts: map[string]ArtifactLaneProfile{ " records ": { Extract: Binding(" record-extractor "), Merge: Binding(" dedupe "), Normalize: Binding(" canonical "), }, }, 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"}) { t.Fatalf("Input = %#v, want trimmed explicit input", resolved.Input) } if resolved.InputExecutionClass != contracts.ExecutionClassDeterministic { t.Fatalf("InputExecutionClass = %q, want deterministic", resolved.InputExecutionClass) } if resolved.Chunk.Module != "window" || resolved.Chunk.LLMProfile != "" { t.Fatalf("Chunk = %#v, want explicit module and empty LLM profile", resolved.Chunk) } if resolved.Chunk.Options["size"] != 10 { t.Fatalf("Chunk.Options = %#v, want size option", resolved.Chunk.Options) } if resolved.ChunkExecutionClass != contracts.ExecutionClassDeterministic { t.Fatalf("ChunkExecutionClass = %q, want deterministic", resolved.ChunkExecutionClass) } if len(resolved.Steps) != 1 || resolved.Steps[0].ID != "default" || len(resolved.Steps[0].ArtifactLanes) != 1 { t.Fatalf("resolved steps = %#v, want one default step with one lane", resolved.Steps) } lane := resolved.Steps[0].ArtifactLanes[0] if lane.ID != "records" { t.Fatalf("lane.ID = %q, want records", lane.ID) } if !reflect.DeepEqual(lane.Extract, ModuleBinding{Module: "record-extractor"}) { t.Fatalf("lane.Extract = %#v, want explicit extractor", lane.Extract) } if lane.ExtractExecutionClass != contracts.ExecutionClassDeterministic || lane.MergeExecutionClass != contracts.ExecutionClassDeterministic || lane.NormalizeExecutionClass != contracts.ExecutionClassDeterministic { t.Fatalf("lane execution classes = %q/%q/%q, want deterministic", lane.ExtractExecutionClass, lane.MergeExecutionClass, lane.NormalizeExecutionClass) } 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) != 0 { t.Fatalf("lane.Validators = %#v, want none", lane.Validators) } if resolved.Output.Module != "ndjson" { t.Fatalf("Output.Module = %q, want ndjson", resolved.Output.Module) } if resolved.OutputExecutionClass != contracts.ExecutionClassDeterministic { t.Fatalf("OutputExecutionClass = %q, want deterministic", resolved.OutputExecutionClass) } if !strings.HasPrefix(resolved.Digest, "sha256:") { t.Fatalf("Digest = %q, want sha256 digest", resolved.Digest) } } func TestModuleCatalogExecutionClassLooksUpRegisteredMetadata(t *testing.T) { catalog := newProfileCatalogWithOverrides(t, ModuleSpec{Key: "llm-input", Stage: StageInput, ExecutionClass: contracts.ExecutionClassLLMBacked}, ModuleSpec{Key: "llm-chunk", Stage: StageChunk, ExecutionClass: contracts.ExecutionClassLLMBacked}, ModuleSpec{Key: "llm-extract", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassLLMBacked}, ModuleSpec{Key: "llm-merge", Stage: StageMerge, ExecutionClass: contracts.ExecutionClassLLMBacked}, ModuleSpec{Key: "llm-normalize", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked}, ModuleSpec{Key: "llm-output", Stage: StageOutput, ExecutionClass: contracts.ExecutionClassLLMBacked}, ) registerProfileValidatorSpec(t, catalog, ValidatorSpec{Key: "llm-validator", ExecutionClass: contracts.ExecutionClassLLMBacked}) for _, test := range []struct { stage ModuleStage key string want contracts.ExecutionClass }{ {stage: StageInput, key: "llm-input", want: contracts.ExecutionClassLLMBacked}, {stage: StageChunk, key: "llm-chunk", want: contracts.ExecutionClassLLMBacked}, {stage: StageExtract, key: "llm-extract", want: contracts.ExecutionClassLLMBacked}, {stage: StageMerge, key: "llm-merge", want: contracts.ExecutionClassLLMBacked}, {stage: StageNormalize, key: "llm-normalize", want: contracts.ExecutionClassLLMBacked}, {stage: StageValidate, key: "llm-validator", want: contracts.ExecutionClassLLMBacked}, {stage: StageOutput, key: "llm-output", want: contracts.ExecutionClassLLMBacked}, } { t.Run(string(test.stage), func(t *testing.T) { got, ok := catalog.ExecutionClass(test.stage, test.key) if !ok || got != test.want { t.Fatalf("ExecutionClass(%q, %q) = %q, %t; want %q, true", test.stage, test.key, got, ok, test.want) } }) } if _, ok := catalog.ExecutionClass(StageExtract, "missing"); ok { t.Fatal("ExecutionClass() found an unregistered module") } } func TestResolvePipelineAppliesDefaults(t *testing.T) { resolved, err := ResolvePipeline(PipelineProfile{ ID: "defaulted", Input: Binding("text"), Artifacts: map[string]ArtifactLaneProfile{ "events": {Extract: Binding("event-extractor")}, }, }, ResolveOptions{}, newProfileCatalog(t)) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } if resolved.Input.LLMProfile != "" { t.Fatalf("Input.LLMProfile = %q, want empty", resolved.Input.LLMProfile) } if !reflect.DeepEqual(resolved.Chunk, ModuleBinding{Module: DefaultChunkModule}) { t.Fatalf("Chunk = %#v, want default chunk binding", resolved.Chunk) } if !reflect.DeepEqual(resolved.Output, ModuleBinding{Module: DefaultOutputModule}) { t.Fatalf("Output = %#v, want default output binding", resolved.Output) } lane := resolved.Steps[0].ArtifactLanes[0] if !reflect.DeepEqual(lane.Merge, ModuleBinding{Module: DefaultMergeModule}) { t.Fatalf("Merge = %#v, want default merge binding", lane.Merge) } if !reflect.DeepEqual(lane.Normalize, ModuleBinding{Module: DefaultNormalizeModule}) { t.Fatalf("Normalize = %#v, want default normalize binding", lane.Normalize) } if lane.Extract.LLMProfile != "" { t.Fatalf("Extract.LLMProfile = %q, want empty", lane.Extract.LLMProfile) } } func TestResolvePipelineAppliesEffectiveLLMProfiles(t *testing.T) { for _, test := range []struct { name string profile PipelineProfile options ResolveOptions want map[string]string }{ { name: "runtime override", profile: func() PipelineProfile { profile := llmProfilePipeline() profile.LLMProfile = " pipeline " profile.Chunk.LLMProfile = "binding" return profile }(), options: ResolveOptions{LLMProfileOverride: " runtime "}, want: llmProfileValues("runtime"), }, { name: "binding exception", profile: func() PipelineProfile { profile := llmProfilePipeline() profile.LLMProfile = "pipeline" lane := profile.Artifacts["events"] lane.Extract.LLMProfile = "extract" lane.Extract.Validators = ValidatorOverride{ Set: true, Validators: []ModuleBinding{{Module: "llm-validator", LLMProfile: "validator"}}, } profile.Artifacts["events"] = lane return profile }(), want: func() map[string]string { values := llmProfileValues("pipeline") values["extract"] = "extract" values["validator:extract:events"] = "validator" return values }(), }, { name: "pipeline default", profile: func() PipelineProfile { profile := llmProfilePipeline() profile.LLMProfile = "pipeline" return profile }(), want: llmProfileValues("pipeline"), }, { name: "prompt fallback", profile: llmProfilePipeline(), want: llmProfileValues(""), }, } { t.Run(test.name, func(t *testing.T) { resolved, err := ResolvePipeline(test.profile, test.options, llmProfileCatalog(t)) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } if got := resolvedLLMProfileValues(resolved); !reflect.DeepEqual(got, test.want) { t.Fatalf("resolved profiles = %#v, want %#v", got, test.want) } }) } } func TestResolvePipelineAppliesProfilesOnlyToSelectedLLMBackedBindings(t *testing.T) { profile := llmProfilePipeline() profile.LLMProfile = "pipeline" profile.Artifacts["notes"] = ArtifactLaneProfile{Extract: Binding("llm-extractor")} resolved, err := ResolvePipeline(profile, ResolveOptions{Only: []string{"events"}}, llmProfileCatalog(t)) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } if got := laneIDs(resolved.Steps[0].ArtifactLanes); !reflect.DeepEqual(got, []string{"events"}) { t.Fatalf("selected lanes = %#v, want events only", got) } if got := resolvedLLMProfileValues(resolved); !reflect.DeepEqual(got, llmProfileValues("pipeline")) { t.Fatalf("resolved profiles = %#v, want selected LLM bindings only", got) } } func TestResolvePipelineLeavesUnusedProfilesOffDeterministicBindings(t *testing.T) { profile := baselineProfile() profile.LLMProfile = "unused" resolved, err := ResolvePipeline(profile, ResolveOptions{LLMProfileOverride: "also-unused"}, newProfileCatalog(t)) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } if resolved.Input.LLMProfile != "" || resolved.Chunk.LLMProfile != "" || resolved.Output.LLMProfile != "" { t.Fatalf("deterministic pipeline profiles = input %q chunk %q output %q, want empty", resolved.Input.LLMProfile, resolved.Chunk.LLMProfile, resolved.Output.LLMProfile) } lane := resolved.Steps[0].ArtifactLanes[0] if lane.Extract.LLMProfile != "" || lane.Merge.LLMProfile != "" || lane.Normalize.LLMProfile != "" { t.Fatalf("deterministic lane profiles = extract %q merge %q normalize %q, want empty", lane.Extract.LLMProfile, lane.Merge.LLMProfile, lane.Normalize.LLMProfile) } } func TestResolvePipelineRejectsLLMProfileForDeterministicModule(t *testing.T) { for _, test := range []struct { name string mutate func(*PipelineProfile) }{ {name: "input", mutate: func(profile *PipelineProfile) { profile.Input.LLMProfile = "invalid" }}, {name: "chunk", mutate: func(profile *PipelineProfile) { profile.Chunk.LLMProfile = "invalid" }}, {name: "extract", mutate: func(profile *PipelineProfile) { lane := profile.Artifacts["events"] lane.Extract.LLMProfile = "invalid" profile.Artifacts["events"] = lane }}, {name: "merge", mutate: func(profile *PipelineProfile) { lane := profile.Artifacts["events"] lane.Merge.LLMProfile = "invalid" profile.Artifacts["events"] = lane }}, {name: "normalize", mutate: func(profile *PipelineProfile) { lane := profile.Artifacts["events"] lane.Normalize.LLMProfile = "invalid" profile.Artifacts["events"] = lane }}, {name: "output", mutate: func(profile *PipelineProfile) { profile.Output.LLMProfile = "invalid" }}, } { t.Run(test.name, func(t *testing.T) { profile := baselineProfile() test.mutate(&profile) _, err := ResolvePipeline(profile, ResolveOptions{}, newProfileCatalog(t)) if err == nil || !strings.Contains(err.Error(), "llm_profile") || !strings.Contains(err.Error(), "deterministic") { t.Fatalf("ResolvePipeline() error = %v, want deterministic profile rejection", err) } }) } } func TestResolvePipelineDigestUsesEffectiveLLMProfiles(t *testing.T) { inherited := llmProfilePipeline() inherited.LLMProfile = "shared" explicit := llmProfilePipeline() explicit.Input.LLMProfile = "shared" explicit.Chunk.LLMProfile = "shared" explicit.Output.LLMProfile = "shared" lane := explicit.Artifacts["events"] lane.Extract.LLMProfile = "shared" lane.Merge.LLMProfile = "shared" lane.Normalize.LLMProfile = "shared" explicit.Artifacts["events"] = lane inheritedResolved, err := ResolvePipeline(inherited, ResolveOptions{}, llmProfileCatalogWithoutValidatorChains(t)) if err != nil { t.Fatalf("ResolvePipeline(inherited) error = %v, want nil", err) } explicitResolved, err := ResolvePipeline(explicit, ResolveOptions{}, llmProfileCatalogWithoutValidatorChains(t)) if err != nil { t.Fatalf("ResolvePipeline(explicit) error = %v, want nil", err) } if inheritedResolved.Digest != explicitResolved.Digest { t.Fatalf("effective profile digests differ: %q != %q", inheritedResolved.Digest, explicitResolved.Digest) } explicit.Chunk.LLMProfile = "different" changedResolved, err := ResolvePipeline(explicit, ResolveOptions{}, llmProfileCatalogWithoutValidatorChains(t)) if err != nil { t.Fatalf("ResolvePipeline(changed) error = %v, want nil", err) } if explicitResolved.Digest == changedResolved.Digest { t.Fatalf("digest = %q after effective profile change, want different", explicitResolved.Digest) } } func TestResolvePipelineRecordsValidatorChains(t *testing.T) { catalog := newProfileCatalog(t) if err := catalog.ValidatorChains.Register(ValidatorChainMapping{ Stage: StageExtract, Module: "event-extractor", Validators: []ModuleBinding{Binding("grounded")}, }); err != nil { t.Fatalf("register validator chain: %v", err) } resolved, err := ResolvePipeline(PipelineProfile{ ID: "validated", Input: Binding("text"), Artifacts: map[string]ArtifactLaneProfile{ "events": {Extract: Binding("event-extractor")}, }, }, ResolveOptions{}, catalog) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } if len(resolved.ValidatorChains) != 4 { t.Fatalf("len(ValidatorChains) = %d, want chunk plus lane extract/merge/normalize", len(resolved.ValidatorChains)) } extractChain := findResolvedValidatorChain(resolved.ValidatorChains, StageExtract, "events", "event-extractor") if extractChain == nil { t.Fatal("extract validator chain not found") } if len(extractChain.Validators) != 1 { t.Fatalf("extract validators = %#v, want one validator", extractChain.Validators) } if extractChain.Validators[0].Binding.Module != "grounded" { t.Fatalf("extract validator key = %q, want grounded", extractChain.Validators[0].Binding.Module) } if extractChain.Validators[0].ExecutionClass != contracts.ExecutionClassDeterministic { t.Fatalf("extract validator execution class = %q, want deterministic", extractChain.Validators[0].ExecutionClass) } chunkChain := findResolvedValidatorChain(resolved.ValidatorChains, StageChunk, "", DefaultChunkModule) if chunkChain == nil { t.Fatal("chunk validator chain not found") } if len(chunkChain.Validators) != 0 { t.Fatalf("chunk validators = %#v, want explicit empty chain", chunkChain.Validators) } } func TestResolvePipelineRejectsUnknownDefaultValidator(t *testing.T) { catalog := newProfileCatalog(t) if err := catalog.ValidatorChains.Register(ValidatorChainMapping{ Stage: StageNormalize, Module: DefaultNormalizeModule, Validators: []ModuleBinding{Binding("missing-validator")}, }); err != nil { t.Fatalf("register validator chain: %v", err) } _, err := ResolvePipeline(PipelineProfile{ ID: "invalid-chain", Input: Binding("text"), Artifacts: map[string]ArtifactLaneProfile{ "events": {Extract: Binding("event-extractor")}, }, }, ResolveOptions{}, catalog) if err == nil { t.Fatal("ResolvePipeline() error = nil, want unknown validator error") } if !strings.Contains(err.Error(), "missing-validator") { t.Fatalf("ResolvePipeline() error = %q, want missing validator context", err.Error()) } } func TestResolvePipelineValidatorOverrideReplacesDefaultChain(t *testing.T) { catalog := newProfileCatalog(t) registerProfileValidatorSpec(t, catalog, ValidatorSpec{Key: "second-validator", ExecutionClass: contracts.ExecutionClassLLMBacked}) if err := catalog.ValidatorChains.Register(ValidatorChainMapping{ Stage: StageExtract, Module: "event-extractor", Validators: []ModuleBinding{Binding("grounded")}, }); err != nil { t.Fatalf("register validator chain: %v", err) } profile := PipelineProfile{ ID: "validated", Input: Binding("text"), Artifacts: map[string]ArtifactLaneProfile{ "events": { Extract: ModuleBinding{ Module: "event-extractor", Validators: ValidatorOverride{ Set: true, Validators: []ModuleBinding{ {Module: "second-validator", LLMProfile: "careful"}, Binding("grounded"), }, }, }, }, }, } resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } extractChain := findResolvedValidatorChain(resolved.ValidatorChains, StageExtract, "events", "event-extractor") if extractChain == nil { t.Fatal("extract validator chain not found") } if len(extractChain.Validators) != 2 { t.Fatalf("extract validators = %#v, want explicit two-validator override", extractChain.Validators) } if extractChain.Validators[0].Binding.Module != "second-validator" || extractChain.Validators[0].Binding.LLMProfile != "careful" { t.Fatalf("first validator = %#v, want explicit LLM-backed validator first", extractChain.Validators[0]) } if extractChain.Validators[1].Binding.Module != "grounded" { t.Fatalf("second validator = %#v, want grounded second", extractChain.Validators[1]) } } func TestResolvePipelineExplicitEmptyValidatorOverrideSuppressesDefaultChain(t *testing.T) { catalog := newProfileCatalog(t) if err := catalog.ValidatorChains.Register(ValidatorChainMapping{ Stage: StageExtract, Module: "event-extractor", Validators: []ModuleBinding{Binding("grounded")}, }); err != nil { t.Fatalf("register validator chain: %v", err) } profile := PipelineProfile{ ID: "validated", Input: Binding("text"), Artifacts: map[string]ArtifactLaneProfile{ "events": { Extract: ModuleBinding{ Module: "event-extractor", Validators: ValidatorOverride{Set: true}, }, }, }, } resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } extractChain := findResolvedValidatorChain(resolved.ValidatorChains, StageExtract, "events", "event-extractor") if extractChain == nil { t.Fatal("extract validator chain not found") } if len(extractChain.Validators) != 0 { t.Fatalf("extract validators = %#v, want explicit empty override", extractChain.Validators) } } func TestResolvePipelineRejectsUnknownOverrideValidator(t *testing.T) { _, err := ResolvePipeline(PipelineProfile{ ID: "validated", Input: Binding("text"), Artifacts: map[string]ArtifactLaneProfile{ "events": { Extract: ModuleBinding{ Module: "event-extractor", Validators: ValidatorOverride{ Set: true, Validators: []ModuleBinding{Binding("missing-validator")}, }, }, }, }, }, ResolveOptions{}, newProfileCatalog(t)) if err == nil { t.Fatal("ResolvePipeline() error = nil, want unknown validator error") } if !strings.Contains(err.Error(), "missing-validator") { t.Fatalf("ResolvePipeline() error = %q, want missing validator context", err.Error()) } } func TestResolvePipelineRejectsLLMProfileForDeterministicValidator(t *testing.T) { _, err := ResolvePipeline(PipelineProfile{ ID: "validated", Input: Binding("text"), Artifacts: map[string]ArtifactLaneProfile{ "events": { Extract: ModuleBinding{ Module: "event-extractor", Validators: ValidatorOverride{ Set: true, Validators: []ModuleBinding{ {Module: "grounded", LLMProfile: "careful"}, }, }, }, }, }, }, ResolveOptions{}, newProfileCatalog(t)) if err == nil { t.Fatal("ResolvePipeline() error = nil, want deterministic validator profile error") } if !strings.Contains(err.Error(), "grounded") || !strings.Contains(err.Error(), "llm_profile") { t.Fatalf("ResolvePipeline() error = %q, want validator profile context", err.Error()) } } func TestResolvePipelineSelectsOnlyRequestedLanes(t *testing.T) { profile := multiLaneProfile() resolved, err := ResolvePipeline(profile, ResolveOptions{Only: []string{" summaries ", "events", "summaries"}}, newProfileCatalog(t)) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } got := laneIDs(resolved.Steps[0].ArtifactLanes) want := []string{"events", "summaries"} if !reflect.DeepEqual(got, want) { t.Fatalf("lane IDs = %#v, want %#v", got, want) } } func TestResolvePipelinePreservesOrderedStepsAndExpandsGeneratedBindings(t *testing.T) { catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "note-extractor", Stage: StageExtract, ArtifactKind: "test/notes", ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "npcs", AcceptedArtifactKinds: []contracts.ArtifactKind{"test/notes"}, AcceptedMediaTypes: []string{"application/json"}}}, }) resolved, err := ResolvePipeline(PipelineProfile{ ID: "ordered", Input: Binding("text"), Steps: []PipelineStepProfile{ {ID: "produce", Artifacts: map[string]ArtifactLaneProfile{"npcs": {Extract: Binding("event-extractor")}}}, {ID: "consume", References: map[string]ReferenceSource{"npcs": GeneratedReference("produce", "npcs")}, Artifacts: map[string]ArtifactLaneProfile{"events": {Extract: Binding("note-extractor")}}}, }, }, ResolveOptions{}, catalog) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } if got := []string{resolved.Steps[0].ID, resolved.Steps[1].ID}; !reflect.DeepEqual(got, []string{"produce", "consume"}) { t.Fatalf("step order = %#v", got) } binding := resolved.Steps[1].ArtifactLanes[0].ExtractReferences.Bindings[0] if binding.Artifact == nil || binding.Artifact.Step != "produce" || binding.Artifact.Lane != "npcs" { t.Fatalf("generated binding = %#v", binding) } if len(resolved.AllArtifactLanes()) != 2 || resolved.Steps[1].ArtifactLanes[0].StepID != "consume" { t.Fatalf("resolved lane topology = %#v", resolved.Steps) } } func TestResolvePipelineRejectsGeneratedBindingOrderingAndKind(t *testing.T) { catalog := newProfileCatalogWithOverrides(t, ModuleSpec{Key: "note-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: "test/notes", Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "npcs", AcceptedArtifactKinds: []contracts.ArtifactKind{"test/other"}}}}, ) _, err := ResolvePipeline(PipelineProfile{ ID: "invalid-order", Input: Binding("text"), Steps: []PipelineStepProfile{ {ID: "first", References: map[string]ReferenceSource{"npcs": GeneratedReference("later", "npcs")}, Artifacts: map[string]ArtifactLaneProfile{"events": {Extract: Binding("note-extractor")}}}, {ID: "later", Artifacts: map[string]ArtifactLaneProfile{"npcs": {Extract: Binding("event-extractor")}}}, }, }, ResolveOptions{}, catalog) if err == nil || !strings.Contains(err.Error(), "earlier step") { t.Fatalf("ResolvePipeline() error = %v, want ordering failure", err) } _, err = ResolvePipeline(PipelineProfile{ ID: "invalid-kind", Input: Binding("text"), Steps: []PipelineStepProfile{ {ID: "produce", Artifacts: map[string]ArtifactLaneProfile{"npcs": {Extract: Binding("event-extractor")}}}, {ID: "consume", References: map[string]ReferenceSource{"npcs": GeneratedReference("produce", "npcs")}, Artifacts: map[string]ArtifactLaneProfile{"events": {Extract: Binding("note-extractor")}}}, }, }, ResolveOptions{}, catalog) if err == nil || !strings.Contains(err.Error(), "does not accept artifact kind") { t.Fatalf("ResolvePipeline() error = %v, want generated kind failure", err) } } func TestResolvePipelineAppliesReferenceBindings(t *testing.T) { profile := multiLaneProfile() profile.References = ExternalReferenceMap(map[string]string{ " roster ": " ./shared-roster.yml ", }) lane := profile.Artifacts["events"] lane.References = ExternalReferenceMap(map[string]string{ "roster": "./lane-roster.yml", " lore ": " ./lore.md ", }) profile.Artifacts["events"] = lane catalog := newProfileCatalogWithOverride(t, ModuleSpec{ Key: "event-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{ {Name: "roster", Required: true}, {Name: "lore"}, }, }) resolved, err := ResolvePipeline(profile, ResolveOptions{Only: []string{"events", "summaries"}}, catalog) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } events := resolvedLane(t, resolved.Steps[0].ArtifactLanes, "events") if events.ExtractReferences.Stage != StageExtract || events.ExtractReferences.LaneID != "events" || events.ExtractReferences.Module != "event-extractor" { t.Fatalf("extract reference target = %#v, want event extractor target", events.ExtractReferences) } if events.NormalizeReferences.Stage != StageNormalize || events.NormalizeReferences.LaneID != "events" || events.NormalizeReferences.Module != DefaultNormalizeModule { t.Fatalf("normalize reference target = %#v, want event normalizer target", events.NormalizeReferences) } if events.MergeReferences.Stage != StageMerge || events.MergeReferences.LaneID != "events" || events.MergeReferences.Module != DefaultMergeModule { t.Fatalf("merge reference target = %#v, want event merger target", events.MergeReferences) } if resolved.ChunkReferences.Stage != StageChunk || resolved.ChunkReferences.Module != DefaultChunkModule { t.Fatalf("chunk reference target = %#v, want chunk target", resolved.ChunkReferences) } want := []ReferenceBinding{ {LaneID: "events", SlotName: "lore", Source: "./lore.md", BindingSource: contracts.ReferenceBindingSourceConfig}, {LaneID: "events", SlotName: "roster", Source: "./lane-roster.yml", BindingSource: contracts.ReferenceBindingSourceConfig}, } if !reflect.DeepEqual(events.ExtractReferences.Bindings, want) { t.Fatalf("events references = %#v, want %#v", events.ExtractReferences.Bindings, want) } summaries := resolvedLane(t, resolved.Steps[0].ArtifactLanes, "summaries") if len(summaries.ExtractReferences.Bindings) != 0 { t.Fatalf("summaries references = %#v, want none", summaries.ExtractReferences.Bindings) } } func TestResolvePipelineAppliesPipelineReferenceDefaultToChunkTarget(t *testing.T) { profile := baselineProfile() profile.References = ExternalReferenceMap(map[string]string{"scene_guide": "./scenes.md"}) catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "generic", Stage: StageChunk, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"source"}, Provides: []string{"chunk"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "scene_guide"}}, }) resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } want := []ReferenceBinding{{SlotName: "scene_guide", Source: "./scenes.md", BindingSource: contracts.ReferenceBindingSourceConfig}} if !reflect.DeepEqual(resolved.ChunkReferences.Bindings, want) { t.Fatalf("chunk references = %#v, want %#v", resolved.ChunkReferences.Bindings, want) } if refs := resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 { t.Fatalf("extract references = %#v, want none", refs) } if refs := resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 0 { t.Fatalf("normalize references = %#v, want none", refs) } } func TestResolvePipelineAppliesPipelineReferenceDefaultToExtractorTarget(t *testing.T) { profile := baselineProfile() profile.References = ExternalReferenceMap(map[string]string{"roster": "./roster.yml"}) catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "event-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "roster"}}, }) resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } want := []ReferenceBinding{{LaneID: "events", SlotName: "roster", Source: "./roster.yml", BindingSource: contracts.ReferenceBindingSourceConfig}} if !reflect.DeepEqual(resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings, want) { t.Fatalf("extract references = %#v, want %#v", resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings, want) } if refs := resolved.ChunkReferences.Bindings; len(refs) != 0 { t.Fatalf("chunk references = %#v, want none", refs) } if refs := resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 0 { t.Fatalf("normalize references = %#v, want none", refs) } } func TestResolvePipelineAppliesPipelineReferenceDefaultToNormalizerTarget(t *testing.T) { profile := baselineProfile() profile.References = ExternalReferenceMap(map[string]string{"normalization_notes": "./normalize.md"}) catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "noop", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"merged"}, Provides: []string{"normalized"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "normalization_notes"}}, }) resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } want := []ReferenceBinding{{LaneID: "events", SlotName: "normalization_notes", Source: "./normalize.md", BindingSource: contracts.ReferenceBindingSourceConfig}} if !reflect.DeepEqual(resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings, want) { t.Fatalf("normalize references = %#v, want %#v", resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings, want) } if refs := resolved.ChunkReferences.Bindings; len(refs) != 0 { t.Fatalf("chunk references = %#v, want none", refs) } if refs := resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 { t.Fatalf("extract references = %#v, want none", refs) } } func TestResolvePipelineAppliesPipelineReferenceDefaultToMergeTarget(t *testing.T) { profile := baselineProfile() profile.References = ExternalReferenceMap(map[string]string{"merge_notes": "./merge.md"}) catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "appendorder", Stage: StageMerge, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"candidate"}, Provides: []string{"merged"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "merge_notes"}}, }) resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } want := []ReferenceBinding{{LaneID: "events", SlotName: "merge_notes", Source: "./merge.md", BindingSource: contracts.ReferenceBindingSourceConfig}} if !reflect.DeepEqual(resolved.Steps[0].ArtifactLanes[0].MergeReferences.Bindings, want) { t.Fatalf("merge references = %#v, want %#v", resolved.Steps[0].ArtifactLanes[0].MergeReferences.Bindings, want) } if refs := resolved.ChunkReferences.Bindings; len(refs) != 0 { t.Fatalf("chunk references = %#v, want none", refs) } if refs := resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 { t.Fatalf("extract references = %#v, want none", refs) } if refs := resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 0 { t.Fatalf("normalize references = %#v, want none", refs) } } func TestResolvePipelineAppliesOnePipelineReferenceDefaultToMultipleTargets(t *testing.T) { profile := baselineProfile() profile.References = ExternalReferenceMap(map[string]string{"context": "./context.md"}) catalog := newProfileCatalogWithOverrides(t, ModuleSpec{Key: "generic", Stage: StageChunk, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"source"}, Provides: []string{"chunk"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}}, ModuleSpec{Key: "event-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}}, ModuleSpec{Key: "appendorder", Stage: StageMerge, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"candidate"}, Provides: []string{"merged"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}}, ModuleSpec{Key: "noop", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"merged"}, Provides: []string{"normalized"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}}, ) resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } assertBindingSource(t, resolved.ChunkReferences.Bindings, "context", "./context.md") assertBindingSource(t, resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings, "context", "./context.md") assertBindingSource(t, resolved.Steps[0].ArtifactLanes[0].MergeReferences.Bindings, "context", "./context.md") assertBindingSource(t, resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings, "context", "./context.md") } func TestResolvePipelineAllowsPipelineReferenceDeclaredOnlyByUnselectedLane(t *testing.T) { profile := multiLaneProfile() profile.References = ExternalReferenceMap(map[string]string{"notes_context": "./notes.md"}) catalog := newProfileCatalogWithOverride(t, ModuleSpec{ Key: "note-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{ {Name: "notes_context"}, }, }) resolved, err := ResolvePipeline(profile, ResolveOptions{Only: []string{"events"}}, catalog) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } if refs := resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 { t.Fatalf("selected lane references = %#v, want none", refs) } } func TestResolvePipelineAllowsPipelineReferenceDeclaredOnlyByUnselectedNormalizer(t *testing.T) { profile := multiLaneProfile() profile.References = ExternalReferenceMap(map[string]string{"notes_context": "./notes.md"}) lane := profile.Artifacts["notes"] lane.Normalize = Binding("note-normalizer") profile.Artifacts["notes"] = lane catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "note-normalizer", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"merged"}, Provides: []string{"normalized"}, ReferenceSlots: []contracts.ReferenceSlot{ {Name: "notes_context"}, }, }) resolved, err := ResolvePipeline(profile, ResolveOptions{Only: []string{"events"}}, catalog) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } if refs := resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 { t.Fatalf("selected extract references = %#v, want none", refs) } if refs := resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 0 { t.Fatalf("selected normalize references = %#v, want none", refs) } } func TestResolvePipelineRejectsPipelineReferenceNotDeclaredByAnyLane(t *testing.T) { profile := multiLaneProfile() profile.References = ExternalReferenceMap(map[string]string{"missing": "./missing.md"}) _, err := ResolvePipeline(profile, ResolveOptions{Only: []string{"events"}}, newProfileCatalog(t)) if err == nil { t.Fatal("ResolvePipeline() error = nil, want error") } assertErrorContains(t, err, "multi", "reference slot", "missing", "not declared") } func TestResolvePipelineRejectsUndeclaredReferenceSlot(t *testing.T) { profile := baselineProfile() lane := profile.Artifacts["events"] lane.References = ExternalReferenceMap(map[string]string{"missing": "./missing.yml"}) profile.Artifacts["events"] = lane _, err := ResolvePipeline(profile, ResolveOptions{}, newProfileCatalog(t)) if err == nil { t.Fatal("ResolvePipeline() error = nil, want error") } assertErrorContains(t, err, "events", "missing", "not declared") } func TestResolvePipelineRejectsExtractLocalReferenceDeclaredOnlyByNormalizer(t *testing.T) { profile := baselineProfile() lane := profile.Artifacts["events"] lane.Extract.References = ExternalReferenceMap(map[string]string{"normalization_notes": "./normalize.md"}) profile.Artifacts["events"] = lane catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "noop", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"merged"}, Provides: []string{"normalized"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "normalization_notes"}}, }) _, err := ResolvePipeline(profile, ResolveOptions{}, catalog) if err == nil { t.Fatal("ResolvePipeline() error = nil, want error") } assertErrorContains(t, err, "baseline", "events", "extract", "event-extractor", "normalization_notes", "not declared") } func TestResolvePipelineRejectsMergeLocalReferenceDeclaredOnlyByNormalizer(t *testing.T) { profile := baselineProfile() lane := profile.Artifacts["events"] lane.Merge.References = ExternalReferenceMap(map[string]string{"normalization_notes": "./normalize.md"}) profile.Artifacts["events"] = lane catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "noop", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"merged"}, Provides: []string{"normalized"}, ReferenceSlots: []contracts.ReferenceSlot{ {Name: "normalization_notes"}, }, }) _, err := ResolvePipeline(profile, ResolveOptions{}, catalog) if err == nil { t.Fatal("ResolvePipeline() error = nil, want error") } assertErrorContains(t, err, "merge", "normalization_notes", "not declared") } func TestResolvePipelineRejectsNormalizeLocalReferenceDeclaredOnlyByExtractor(t *testing.T) { profile := baselineProfile() lane := profile.Artifacts["events"] lane.Normalize.References = ExternalReferenceMap(map[string]string{"roster": "./roster.yml"}) profile.Artifacts["events"] = lane catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "event-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "roster"}}, }) _, err := ResolvePipeline(profile, ResolveOptions{}, catalog) if err == nil { t.Fatal("ResolvePipeline() error = nil, want error") } assertErrorContains(t, err, "baseline", "events", "normalize", "noop", "roster", "not declared") } func TestResolvePipelineRequiresBoundChunkReference(t *testing.T) { catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "generic", Stage: StageChunk, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"source"}, Provides: []string{"chunk"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "scene_guide", Required: true}}, }) _, err := ResolvePipeline(baselineProfile(), ResolveOptions{}, catalog) if err == nil { t.Fatal("ResolvePipeline() error = nil, want error") } assertErrorContains(t, err, "baseline", "chunk", "generic", "required", "scene_guide", "not bound") } func TestResolvePipelineRequiresBoundNormalizeReference(t *testing.T) { catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "noop", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"merged"}, Provides: []string{"normalized"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "normalization_notes", Required: true}}, }) _, err := ResolvePipeline(baselineProfile(), ResolveOptions{}, catalog) if err == nil { t.Fatal("ResolvePipeline() error = nil, want error") } assertErrorContains(t, err, "baseline", "events", "normalize", "noop", "required", "normalization_notes", "not bound") } func TestResolvePipelineLocalReferencesOverridePipelineDefaultsForEligibleTargets(t *testing.T) { profile := baselineProfile() profile.References = ExternalReferenceMap(map[string]string{ "context": "./shared-context.md", "roster": "./shared-roster.yml", "normalization_notes": "./shared-normalize.md", }) profile.Chunk.References = ExternalReferenceMap(map[string]string{"context": "./chunk-context.md"}) lane := profile.Artifacts["events"] lane.Extract.References = ExternalReferenceMap(map[string]string{"roster": "./extract-roster.yml"}) lane.Normalize.References = ExternalReferenceMap(map[string]string{"normalization_notes": "./local-normalize.md"}) profile.Artifacts["events"] = lane catalog := newProfileCatalogWithOverrides(t, ModuleSpec{Key: "generic", Stage: StageChunk, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"source"}, Provides: []string{"chunk"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}}, ModuleSpec{Key: "event-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "roster"}}}, ModuleSpec{Key: "noop", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"merged"}, Provides: []string{"normalized"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "normalization_notes"}}}, ) resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } assertBindingSource(t, resolved.ChunkReferences.Bindings, "context", "./chunk-context.md") assertBindingSource(t, resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings, "roster", "./extract-roster.yml") assertBindingSource(t, resolved.Steps[0].ArtifactLanes[0].NormalizeReferences.Bindings, "normalization_notes", "./local-normalize.md") } func TestResolvePipelineRequiresBoundReferenceSlotsForSelectedLanes(t *testing.T) { catalog := newProfileCatalogWithOverride(t, ModuleSpec{ Key: "event-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{ {Name: "roster", Required: true}, }, }) if _, err := ResolvePipeline(multiLaneProfile(), ResolveOptions{Only: []string{"notes"}}, catalog); err != nil { t.Fatalf("ResolvePipeline(unselected required slot) error = %v, want nil", err) } _, err := ResolvePipeline(multiLaneProfile(), ResolveOptions{Only: []string{"events"}}, catalog) if err == nil { t.Fatal("ResolvePipeline(selected required slot) error = nil, want error") } assertErrorContains(t, err, "events", "required", "roster", "not bound") } func TestResolvePipelineReferenceUnbindCanLeaveRequiredSlotMissing(t *testing.T) { profile := baselineProfile() profile.References = ExternalReferenceMap(map[string]string{"roster": "./roster.yml"}) catalog := newProfileCatalogWithOverride(t, ModuleSpec{ Key: "event-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{ {Name: "roster", Required: true}, }, }) _, err := ResolvePipeline(profile, ResolveOptions{ ReferenceUnbinds: []ReferenceUnbind{{LaneID: "events", SlotName: "roster"}}, }, catalog) if err == nil { t.Fatal("ResolvePipeline() error = nil, want error") } assertErrorContains(t, err, "events", "required", "roster", "not bound") } func TestResolvePipelineUsesReferenceSlotsFromSpecWithoutConstructingExtractor(t *testing.T) { profile := baselineProfile() profile.References = ExternalReferenceMap(map[string]string{"roster": "./roster.yml"}) catalog := emptyProfileCatalog() mustRegisterArtifactCodec(t, catalog.ArtifactCodecs, notesCodec()) for _, spec := range defaultProfileSpecs() { if spec.Key != "event-extractor" { registerProfileSpecs(t, catalog, spec) } } if err := RegisterExtractor[codecNotes](catalog.Extractors, ModuleSpec{ Key: "event-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: "test/notes", Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{ {Name: "roster", Required: true}, }, }, func() (contracts.Extractor[codecNotes], error) { return nil, errors.New("constructor should not run") }); err != nil { t.Fatalf("RegisterWithSpec() error = %v, want nil", err) } resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } if got := resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings[0].Source; got != "./roster.yml" { t.Fatalf("reference source = %q, want ./roster.yml", got) } } 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{ "events": {Extract: Binding("event-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{ "events": {Extract: Binding("event-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{ "events": {Extract: Binding("event-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["events"] lane.Extract = Binding("missing-extractor") profile.Artifacts["events"] = lane return profile }), want: []string{"baseline", "events", "extract", "missing-extractor"}, }, { name: "merge", profile: withProfileChange(func(profile PipelineProfile) PipelineProfile { lane := profile.Artifacts["events"] lane.Merge = Binding("missing-merge") profile.Artifacts["events"] = lane return profile }), want: []string{"baseline", "events", "merge", "missing-merge"}, }, { name: "normalize", profile: withProfileChange(func(profile PipelineProfile) PipelineProfile { lane := profile.Artifacts["events"] lane.Normalize = Binding("missing-normalize") profile.Artifacts["events"] = lane return profile }), want: []string{"baseline", "events", "normalize", "missing-normalize"}, }, { 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, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"raw"}}, want: []string{"baseline", "input", "text", "raw"}, }, { name: "chunk", spec: ModuleSpec{Key: "generic", Stage: StageChunk, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"missing"}}, want: []string{"baseline", "chunk", "generic", "missing"}, }, { name: "extract", spec: ModuleSpec{Key: "event-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"missing"}}, want: []string{"baseline", "events", "extract", "event-extractor", "missing"}, }, { name: "merge", spec: ModuleSpec{Key: "appendorder", Stage: StageMerge, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"missing"}}, want: []string{"baseline", "events", "merge", "appendorder", "missing"}, }, { name: "normalize", spec: ModuleSpec{Key: "noop", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"missing"}}, want: []string{"baseline", "events", "normalize", "noop", "missing"}, }, { name: "output", spec: ModuleSpec{Key: "json", Stage: StageOutput, ExecutionClass: contracts.ExecutionClassDeterministic, 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() _, err := ResolvePipeline(profile, ResolveOptions{}, catalog) if err == nil { t.Fatal("ResolvePipeline() error = nil, want error") } assertErrorContains(t, err, test.want...) }) } } func TestResolvePipelineRejectsConfiguredValidators(t *testing.T) { profile := baselineProfile() lane := profile.Artifacts["events"] lane.Validators = []ModuleBinding{Binding("grounded")} profile.Artifacts["events"] = lane _, err := ResolvePipeline(profile, ResolveOptions{}, newProfileCatalog(t)) if err == nil { t.Fatal("ResolvePipeline() error = nil, want error") } assertErrorContains(t, err, "baseline", "events", "validators", "extract.validators") } 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.Steps[0].ArtifactLanes) want := []string{"events", "notes", "summaries"} 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{ "events": {Extract: Binding("event-extractor")}, "notes": {Extract: Binding("note-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{ "notes": {Extract: Binding("note-extractor")}, "events": {Extract: Binding("event-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, ExecutionClass: contracts.ExecutionClassDeterministic, 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 TestResolvePipelineDigestIncludesEffectiveValidatorChain(t *testing.T) { resolve := func(t *testing.T, validators []ModuleBinding, explicitEmpty bool) ResolvedPipeline { t.Helper() catalog := newProfileCatalog(t) registerProfileValidatorSpec(t, catalog, ValidatorSpec{Key: "second-validator", ExecutionClass: contracts.ExecutionClassDeterministic}) if len(validators) > 0 { if err := catalog.ValidatorChains.Register(ValidatorChainMapping{ Stage: StageExtract, Module: "event-extractor", Validators: validators, }); err != nil { t.Fatalf("register validator chain: %v", err) } } profile := baselineProfile() if explicitEmpty { lane := profile.Artifacts["events"] lane.Extract.Validators = ValidatorOverride{Set: true} profile.Artifacts["events"] = lane } resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) if err != nil { t.Fatalf("ResolvePipeline() error = %v, want nil", err) } return resolved } inherited := resolve(t, []ModuleBinding{Binding("grounded"), Binding("second-validator")}, false) repeated := resolve(t, []ModuleBinding{Binding("grounded"), Binding("second-validator")}, false) if inherited.Digest != repeated.Digest { t.Fatalf("same resolved validator policy produced digests %q and %q", inherited.Digest, repeated.Digest) } tests := []struct { name string left ResolvedPipeline right ResolvedPipeline }{ { name: "different default", left: resolve(t, []ModuleBinding{Binding("grounded")}, false), right: resolve(t, []ModuleBinding{Binding("second-validator")}, false), }, { name: "added validator", left: resolve(t, nil, false), right: resolve(t, []ModuleBinding{Binding("grounded")}, false), }, { name: "removed validator", left: resolve(t, []ModuleBinding{Binding("grounded")}, false), right: resolve(t, nil, false), }, { name: "reordered chain", left: inherited, right: resolve(t, []ModuleBinding{Binding("second-validator"), Binding("grounded")}, false), }, { name: "explicit empty", left: inherited, right: resolve(t, []ModuleBinding{Binding("grounded"), Binding("second-validator")}, true), }, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { if tc.left.Digest == tc.right.Digest { t.Fatalf("digest = %q for both validator policies, want changed digest", tc.left.Digest) } }) } } func TestResolvedPipelineDigestIncludesCompleteValidatorPolicy(t *testing.T) { base := validatorDigestFixture() baseDigest, err := resolvedPipelineDigest(base) if err != nil { t.Fatalf("resolvedPipelineDigest(base) error = %v, want nil", err) } tests := []struct { name string mutate func(*ResolvedPipeline) }{ {name: "stage", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Stage = StageMerge }}, {name: "lane", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].LaneID = "other" }}, {name: "owner module", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].ModuleKey = "other-extractor" }}, {name: "validator order", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Validators[0], value.ValidatorChains[0].Validators[1] = value.ValidatorChains[0].Validators[1], value.ValidatorChains[0].Validators[0] }}, {name: "validator module", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Validators[0].Binding.Module = "other-validator" }}, {name: "llm profile", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Validators[0].Binding.LLMProfile = "fast" }}, {name: "retries", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Validators[0].Binding.Retries++ }}, {name: "options", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Validators[0].Binding.Options["threshold"] = 0.75 }}, {name: "references", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Validators[0].Binding.References["rules"] = ExternalReference("other.md") }}, {name: "execution class", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Validators[0].ExecutionClass = contracts.ExecutionClassDeterministic }}, {name: "target", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Validators[0].Target = ValidatorTargetSerialized }}, {name: "artifact kind", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Validators[0].ArtifactKind = "test/other" }}, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { changed := validatorDigestFixture() tc.mutate(&changed) digest, err := resolvedPipelineDigest(changed) if err != nil { t.Fatalf("resolvedPipelineDigest(changed) error = %v, want nil", err) } if digest == baseDigest { t.Fatalf("digest = %q after %s change, want different digest", digest, tc.name) } }) } } func TestResolvedPipelineDigestCanonicalizesValidatorBindingMaps(t *testing.T) { left := validatorDigestFixture() right := validatorDigestFixture() right.ValidatorChains[0].Validators[0].Binding.Options = map[string]any{} right.ValidatorChains[0].Validators[0].Binding.Options["threshold"] = 0.5 right.ValidatorChains[0].Validators[0].Binding.Options["mode"] = "strict" right.ValidatorChains[0].Validators[0].Binding.References = ExternalReferenceMap(map[string]string{}) right.ValidatorChains[0].Validators[0].Binding.References["examples"] = ExternalReference("examples.md") right.ValidatorChains[0].Validators[0].Binding.References["rules"] = ExternalReference("rules.md") leftDigest, err := resolvedPipelineDigest(left) if err != nil { t.Fatalf("resolvedPipelineDigest(left) error = %v, want nil", err) } rightDigest, err := resolvedPipelineDigest(right) if err != nil { t.Fatalf("resolvedPipelineDigest(right) error = %v, want nil", err) } if leftDigest != rightDigest { t.Fatalf("equivalent validator maps produced digests %q and %q", leftDigest, rightDigest) } } func validatorDigestFixture() ResolvedPipeline { return ResolvedPipeline{ ID: "validator-policy", ValidatorChains: []ResolvedValidatorChain{{ Stage: StageExtract, LaneID: "events", ModuleKey: "event-extractor", Validators: []ResolvedValidator{ { Binding: ModuleBinding{ Module: "semantic-validator", LLMProfile: "careful", Retries: 2, Options: map[string]any{"mode": "strict", "threshold": 0.5}, References: ExternalReferenceMap(map[string]string{"rules": "rules.md", "examples": "examples.md"}), }, ExecutionClass: contracts.ExecutionClassLLMBacked, Target: ValidatorTargetTyped, ArtifactKind: "test/notes", }, { Binding: Binding("grounded"), ExecutionClass: contracts.ExecutionClassDeterministic, Target: ValidatorTargetTyped, ArtifactKind: "test/notes", }, }, }}, } } 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{ "events": {Extract: Binding("event-extractor")}, }, } } func multiLaneProfile() PipelineProfile { profile := baselineProfile() profile.ID = "multi" profile.Artifacts = map[string]ArtifactLaneProfile{ "summaries": {Extract: Binding("note-extractor")}, "events": {Extract: Binding("event-extractor")}, "notes": {Extract: Binding("note-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 resolvedLane(t *testing.T, lanes []ResolvedArtifactLane, laneID string) ResolvedArtifactLane { t.Helper() for _, lane := range lanes { if lane.ID == laneID { return lane } } t.Fatalf("lane %q not found in %#v", laneID, laneIDs(lanes)) return ResolvedArtifactLane{} } 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 assertBindingSource(t *testing.T, bindings []ReferenceBinding, slotName string, source string) { t.Helper() for _, binding := range bindings { if binding.SlotName != slotName { continue } if binding.Source != source { t.Fatalf("binding %q source = %q, want %q in %#v", slotName, binding.Source, source, bindings) } return } t.Fatalf("binding %q not found in %#v", slotName, bindings) } func findResolvedValidatorChain(chains []ResolvedValidatorChain, stage ModuleStage, laneID string, module string) *ResolvedValidatorChain { for i := range chains { if chains[i].Stage == stage && chains[i].LaneID == laneID && chains[i].ModuleKey == module { return &chains[i] } } return nil } func newProfileCatalog(t *testing.T) ModuleCatalog { t.Helper() catalog := emptyProfileCatalog() mustRegisterArtifactCodec(t, catalog.ArtifactCodecs, notesCodec()) registerProfileSpecs(t, catalog, defaultProfileSpecs()...) return catalog } func newProfileCatalogWithOverride(t *testing.T, override ModuleSpec) ModuleCatalog { t.Helper() return newProfileCatalogWithOverrides(t, override) } func newProfileCatalogWithOverrides(t *testing.T, overrides ...ModuleSpec) ModuleCatalog { t.Helper() specs := defaultProfileSpecs() for _, override := range overrides { if override.ArtifactKind == "" && (override.Stage == StageExtract || override.Stage == StageMerge || override.Stage == StageNormalize) { override.ArtifactKind = "test/notes" } replaced := false for index, spec := range specs { if spec.Stage == override.Stage && spec.Key == override.Key { specs[index] = override replaced = true break } } if !replaced { specs = append(specs, override) } } catalog := emptyProfileCatalog() mustRegisterArtifactCodec(t, catalog.ArtifactCodecs, notesCodec()) registerProfileSpecs(t, catalog, specs...) return catalog } func emptyProfileCatalog() ModuleCatalog { return ModuleCatalog{ Inputs: NewInputAdapterRegistry(), Chunkers: NewChunkerRegistry(), ArtifactCodecs: NewArtifactCodecRegistry(), Extractors: NewExtractorRegistry(), Mergers: NewMergerRegistry(), Normalizers: NewNormalizerRegistry(), Validators: NewValidatorRegistry(), ValidatorChains: NewValidatorChainRegistry(), Outputs: NewOutputEncoderRegistry(), } } func defaultProfileSpecs() []ModuleSpec { return []ModuleSpec{ ModuleSpec{Key: "text", Stage: StageInput, ExecutionClass: contracts.ExecutionClassDeterministic, Provides: []string{"source"}}, ModuleSpec{Key: "generic", Stage: StageChunk, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"source"}, Provides: []string{"chunk"}}, ModuleSpec{Key: "event-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: "test/notes", Requires: []string{"chunk"}, Provides: []string{"candidate"}}, ModuleSpec{Key: "note-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: "test/notes", Requires: []string{"chunk"}, Provides: []string{"candidate"}}, ModuleSpec{Key: "appendorder", Stage: StageMerge, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: "test/notes", Requires: []string{"candidate"}, Provides: []string{"merged"}}, ModuleSpec{Key: "noop", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: "test/notes", Requires: []string{"merged"}, Provides: []string{"normalized"}}, ModuleSpec{Key: "grounded", Stage: StageValidate, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"normalized"}, Provides: []string{"validated"}}, ModuleSpec{Key: "json", Stage: StageOutput, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"normalized"}, Provides: []string{"encoded"}}, } } func llmProfileCatalog(t *testing.T) ModuleCatalog { t.Helper() catalog := newProfileCatalogWithOverrides(t, ModuleSpec{Key: "llm-input", Stage: StageInput, ExecutionClass: contracts.ExecutionClassLLMBacked, Provides: []string{"source"}}, ModuleSpec{Key: "llm-chunk", Stage: StageChunk, ExecutionClass: contracts.ExecutionClassLLMBacked, Requires: []string{"source"}, Provides: []string{"chunk"}}, ModuleSpec{Key: "llm-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassLLMBacked, ArtifactKind: "test/notes", Requires: []string{"chunk"}, Provides: []string{"candidate"}}, ModuleSpec{Key: "llm-merge", Stage: StageMerge, ExecutionClass: contracts.ExecutionClassLLMBacked, ArtifactKind: "test/notes", Requires: []string{"candidate"}, Provides: []string{"merged"}}, ModuleSpec{Key: "llm-normalize", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, ArtifactKind: "test/notes", Requires: []string{"merged"}, Provides: []string{"normalized"}}, ModuleSpec{Key: "llm-output", Stage: StageOutput, ExecutionClass: contracts.ExecutionClassLLMBacked, Requires: []string{"normalized"}, Provides: []string{"encoded"}}, ) if err := RegisterChunkValidator(catalog.Validators, ValidatorSpec{Key: "llm-chunk-validator", ExecutionClass: contracts.ExecutionClassLLMBacked}, func() (contracts.ChunkValidator, error) { return typedTestChunkValidator{key: "llm-chunk-validator"}, nil }); err != nil { t.Fatalf("register chunk validator: %v", err) } registerProfileValidatorSpec(t, catalog, ValidatorSpec{Key: "llm-validator", ExecutionClass: contracts.ExecutionClassLLMBacked}) for _, mapping := range []ValidatorChainMapping{ {Stage: StageChunk, Module: "llm-chunk", Validators: []ModuleBinding{Binding("llm-chunk-validator")}}, {Stage: StageExtract, Module: "llm-extractor", Validators: []ModuleBinding{Binding("llm-validator")}}, {Stage: StageMerge, Module: "llm-merge", Validators: []ModuleBinding{Binding("llm-validator")}}, {Stage: StageNormalize, Module: "llm-normalize", Validators: []ModuleBinding{Binding("llm-validator")}}, } { if err := catalog.ValidatorChains.Register(mapping); err != nil { t.Fatalf("register validator chain %#v: %v", mapping, err) } } return catalog } func llmProfileCatalogWithoutValidatorChains(t *testing.T) ModuleCatalog { catalog := llmProfileCatalog(t) catalog.ValidatorChains = NewValidatorChainRegistry() return catalog } func llmProfilePipeline() PipelineProfile { return PipelineProfile{ ID: "llm-profile", Input: Binding("llm-input"), Chunk: Binding("llm-chunk"), Output: Binding("llm-output"), Artifacts: map[string]ArtifactLaneProfile{ "events": { Extract: Binding("llm-extractor"), Merge: Binding("llm-merge"), Normalize: Binding("llm-normalize"), }, }, } } func llmProfileValues(profile string) map[string]string { return map[string]string{ "input": profile, "chunk": profile, "extract": profile, "merge": profile, "normalize": profile, "output": profile, "validator:chunk:": profile, "validator:extract:events": profile, "validator:merge:events": profile, "validator:normalize:events": profile, } } func resolvedLLMProfileValues(resolved ResolvedPipeline) map[string]string { values := map[string]string{ "input": resolved.Input.LLMProfile, "chunk": resolved.Chunk.LLMProfile, "output": resolved.Output.LLMProfile, } lane := resolved.Steps[0].ArtifactLanes[0] values["extract"] = lane.Extract.LLMProfile values["merge"] = lane.Merge.LLMProfile values["normalize"] = lane.Normalize.LLMProfile for _, chain := range resolved.ValidatorChains { for _, validator := range chain.Validators { if validator.ExecutionClass == contracts.ExecutionClassLLMBacked { values["validator:"+string(chain.Stage)+":"+chain.LaneID] = validator.Binding.LLMProfile } } } return values } func registerProfileSpecs(t *testing.T, catalog ModuleCatalog, specs ...ModuleSpec) { t.Helper() for _, spec := range specs { if spec.ArtifactKind == "" && (spec.Stage == StageExtract || spec.Stage == StageMerge || spec.Stage == StageNormalize) { spec.ArtifactKind = "test/notes" } 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: validateOptions := func(options map[string]any) error { return RejectUnknownOptions(options, "a", "b", "size") } if err := catalog.Chunkers.RegisterBuilderWithSpec(spec, validateOptions, func(BuildRequest) (contracts.Chunker, error) { return &typedTestChunker{key: spec.Key}, nil }); err != nil { t.Fatalf("register chunk spec %#v: %v", spec, err) } case StageExtract: if err := RegisterExtractor(catalog.Extractors, spec, func() (contracts.Extractor[codecNotes], error) { return typedTestExtractor[codecNotes]{key: spec.Key}, nil }); err != nil { t.Fatalf("register extractor spec %#v: %v", spec, err) } case StageMerge: if err := RegisterMerger(catalog.Mergers, spec, func() (contracts.Merger[codecNotes], error) { return typedTestMerger[codecNotes]{key: spec.Key}, nil }); err != nil { t.Fatalf("register merger spec %#v: %v", spec, err) } case StageNormalize: if err := RegisterNormalizer(catalog.Normalizers, spec, func() (contracts.Normalizer[codecNotes], error) { return typedTestNormalizer[codecNotes]{key: spec.Key}, nil }); err != nil { t.Fatalf("register normalizer spec %#v: %v", spec, err) } case StageValidate: validatorSpec := ValidatorSpec{Key: spec.Key, ExecutionClass: contracts.ExecutionClassDeterministic} if err := RegisterTypedValidator(catalog.Validators, "test/notes", validatorSpec, func() (contracts.TypedValidator[codecNotes], error) { return typedTestValidator[codecNotes]{key: spec.Key}, nil }); 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 registerProfileValidatorSpec(t *testing.T, catalog ModuleCatalog, spec ValidatorSpec) { t.Helper() if err := RegisterTypedValidator(catalog.Validators, "test/notes", spec, func() (contracts.TypedValidator[codecNotes], error) { return typedTestValidator[codecNotes]{key: spec.Key}, nil }); err != nil { t.Fatalf("register validator spec %#v: %v", spec, err) } } 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 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) } encoded, err := json.Marshal(resolved) if err != nil { t.Fatalf("json.Marshal(resolved) error = %v, want nil", err) } for _, field := range []string{"input_execution_class", "chunk_execution_class", "extract_execution_class", "merge_execution_class", "normalize_execution_class", "output_execution_class"} { if !strings.Contains(string(encoded), `"`+field+`":"deterministic"`) { t.Fatalf("resolved JSON does not retain %q: %s", field, encoded) } } }