diff --git a/internal/cli/assembled_spell_pipeline_contract_test.go b/internal/cli/assembled_spell_pipeline_contract_test.go index 1f9c274..03d4baf 100644 --- a/internal/cli/assembled_spell_pipeline_contract_test.go +++ b/internal/cli/assembled_spell_pipeline_contract_test.go @@ -213,11 +213,12 @@ func assembledSpellPipeline(t *testing.T, options assembledSpellPipelineOptions) components := productionTestComponents(t) extractor := &assembledSpellExtractor{unknownSpell: options.unknownSpell} if err := pipeline.RegisterExtractor[dnd.SpellList](components.registries.Extractors, pipeline.ModuleSpec{ - Key: assembledSpellExtractorKey, - Stage: pipeline.StageExtract, - Requires: []string{"chunks", "source.transcript"}, - Provides: []string{"dnd.spell_casts"}, - ArtifactKind: dnd.SpellListKind, + Key: assembledSpellExtractorKey, + Stage: pipeline.StageExtract, + ExecutionClass: contracts.ExecutionClassDeterministic, + Requires: []string{"chunks", "source.transcript"}, + Provides: []string{"dnd.spell_casts"}, + ArtifactKind: dnd.SpellListKind, }, func() (contracts.Extractor[dnd.SpellList], error) { return extractor, nil }); err != nil { diff --git a/internal/cli/recompute_execution_contract_test.go b/internal/cli/recompute_execution_contract_test.go index 112c98d..e593e68 100644 --- a/internal/cli/recompute_execution_contract_test.go +++ b/internal/cli/recompute_execution_contract_test.go @@ -116,7 +116,7 @@ func (h *recomputeTestHarness) options() Options { for _, key := range []string{"test/extract/producer", "test/extract/unrelated", "test/extract/middle", "test/extract/dependent"} { moduleKey := key spec := pipeline.ModuleSpec{ - Key: moduleKey, Stage: pipeline.StageExtract, Requires: []string{"chunks"}, Provides: []string{"artifact"}, ArtifactKind: stateTestArtifactKind, + Key: moduleKey, Stage: pipeline.StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunks"}, Provides: []string{"artifact"}, ArtifactKind: stateTestArtifactKind, ReferenceSlots: []contracts.ReferenceSlot{{Name: "upstream", AcceptedMediaTypes: []string{"application/json"}, AcceptedArtifactKinds: []contracts.ArtifactKind{stateTestArtifactKind}}}, } if err := pipeline.RegisterExtractor(opts.Registries.Extractors, spec, func() (contracts.Extractor[stateTestArtifact], error) { @@ -125,7 +125,7 @@ func (h *recomputeTestHarness) options() Options { panic(err) } } - if err := opts.Registries.Outputs.RegisterWithSpec(pipeline.ModuleSpec{Key: "test/recompute-output", Stage: pipeline.StageOutput, Requires: []string{"normalized"}, Provides: []string{"output"}}, func() (contracts.OutputEncoder, error) { + if err := opts.Registries.Outputs.RegisterWithSpec(pipeline.ModuleSpec{Key: "test/recompute-output", Stage: pipeline.StageOutput, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"normalized"}, Provides: []string{"output"}}, func() (contracts.OutputEncoder, error) { return recomputeTestOutput{}, nil }); err != nil { panic(err) diff --git a/internal/cli/reference_contract_test.go b/internal/cli/reference_contract_test.go index 1d8a192..d3d8add 100644 --- a/internal/cli/reference_contract_test.go +++ b/internal/cli/reference_contract_test.go @@ -364,21 +364,21 @@ func referenceContractCatalog(t *testing.T, includeBetaMerger, includeBetaNormal t.Fatal(err) } } - register(registries.Inputs.RegisterBuilderWithSpec(pipeline.ModuleSpec{Key: "reference/input", Stage: pipeline.StageInput, Provides: []string{"source"}}, func(map[string]any) error { return nil }, func(pipeline.BuildRequest) (contracts.InputAdapter, error) { return stateTestInput{}, nil })) - register(registries.Chunkers.RegisterWithSpec(pipeline.ModuleSpec{Key: "reference/chunk", Stage: pipeline.StageChunk, Requires: []string{"source"}, Provides: []string{"chunks"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "chunk-slot"}, {Name: "required-chunk", Required: true}}}, func() (contracts.Chunker, error) { return stateTestChunker{}, nil })) + register(registries.Inputs.RegisterBuilderWithSpec(pipeline.ModuleSpec{Key: "reference/input", Stage: pipeline.StageInput, ExecutionClass: contracts.ExecutionClassDeterministic, Provides: []string{"source"}}, func(map[string]any) error { return nil }, func(pipeline.BuildRequest) (contracts.InputAdapter, error) { return stateTestInput{}, nil })) + register(registries.Chunkers.RegisterWithSpec(pipeline.ModuleSpec{Key: "reference/chunk", Stage: pipeline.StageChunk, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"source"}, Provides: []string{"chunks"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "chunk-slot"}, {Name: "required-chunk", Required: true}}}, func() (contracts.Chunker, error) { return stateTestChunker{}, nil })) register(pipeline.RegisterArtifactCodec(registries.ArtifactCodecs, referenceContractCodecA{})) register(pipeline.RegisterArtifactCodec(registries.ArtifactCodecs, referenceContractCodecB{})) - register(pipeline.RegisterExtractor(registries.Extractors, pipeline.ModuleSpec{Key: "reference/extract-alpha", Stage: pipeline.StageExtract, Requires: []string{"chunks"}, Provides: []string{"artifact"}, ArtifactKind: referenceContractKindAlpha, ReferenceSlots: []contracts.ReferenceSlot{{Name: "shared"}, {Name: "alpha-slot"}, {Name: "required-extract", Required: true}}}, func() (contracts.Extractor[stateTestArtifact], error) { return stateTestExtractor{}, nil })) - register(pipeline.RegisterExtractor(registries.Extractors, pipeline.ModuleSpec{Key: "reference/extract-beta", Stage: pipeline.StageExtract, Requires: []string{"chunks"}, Provides: []string{"artifact"}, ArtifactKind: referenceContractKindBeta, ReferenceSlots: []contracts.ReferenceSlot{{Name: "shared"}, {Name: "beta-slot"}, {Name: "required-extract", Required: true}}}, func() (contracts.Extractor[stateTestArtifact], error) { return stateTestExtractor{}, nil })) - register(pipeline.RegisterMerger(registries.Mergers, pipeline.ModuleSpec{Key: "reference/shared-merge", Stage: pipeline.StageMerge, Requires: []string{"artifact"}, Provides: []string{"merged"}, ArtifactKind: referenceContractKindAlpha, ReferenceSlots: []contracts.ReferenceSlot{{Name: "shared"}, {Name: "alpha-merge"}, {Name: "required-merge", Required: true}}}, func() (contracts.Merger[stateTestArtifact], error) { return stateTestMerger{}, nil })) + register(pipeline.RegisterExtractor(registries.Extractors, pipeline.ModuleSpec{Key: "reference/extract-alpha", Stage: pipeline.StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunks"}, Provides: []string{"artifact"}, ArtifactKind: referenceContractKindAlpha, ReferenceSlots: []contracts.ReferenceSlot{{Name: "shared"}, {Name: "alpha-slot"}, {Name: "required-extract", Required: true}}}, func() (contracts.Extractor[stateTestArtifact], error) { return stateTestExtractor{}, nil })) + register(pipeline.RegisterExtractor(registries.Extractors, pipeline.ModuleSpec{Key: "reference/extract-beta", Stage: pipeline.StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunks"}, Provides: []string{"artifact"}, ArtifactKind: referenceContractKindBeta, ReferenceSlots: []contracts.ReferenceSlot{{Name: "shared"}, {Name: "beta-slot"}, {Name: "required-extract", Required: true}}}, func() (contracts.Extractor[stateTestArtifact], error) { return stateTestExtractor{}, nil })) + register(pipeline.RegisterMerger(registries.Mergers, pipeline.ModuleSpec{Key: "reference/shared-merge", Stage: pipeline.StageMerge, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"artifact"}, Provides: []string{"merged"}, ArtifactKind: referenceContractKindAlpha, ReferenceSlots: []contracts.ReferenceSlot{{Name: "shared"}, {Name: "alpha-merge"}, {Name: "required-merge", Required: true}}}, func() (contracts.Merger[stateTestArtifact], error) { return stateTestMerger{}, nil })) if includeBetaMerger { - register(pipeline.RegisterMerger(registries.Mergers, pipeline.ModuleSpec{Key: "reference/shared-merge", Stage: pipeline.StageMerge, Requires: []string{"artifact"}, Provides: []string{"merged"}, ArtifactKind: referenceContractKindBeta, ReferenceSlots: []contracts.ReferenceSlot{{Name: "shared"}, {Name: "beta-merge"}, {Name: "required-merge", Required: true}}}, func() (contracts.Merger[stateTestArtifact], error) { return stateTestMerger{}, nil })) + register(pipeline.RegisterMerger(registries.Mergers, pipeline.ModuleSpec{Key: "reference/shared-merge", Stage: pipeline.StageMerge, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"artifact"}, Provides: []string{"merged"}, ArtifactKind: referenceContractKindBeta, ReferenceSlots: []contracts.ReferenceSlot{{Name: "shared"}, {Name: "beta-merge"}, {Name: "required-merge", Required: true}}}, func() (contracts.Merger[stateTestArtifact], error) { return stateTestMerger{}, nil })) } - register(pipeline.RegisterNormalizer(registries.Normalizers, pipeline.ModuleSpec{Key: "reference/shared-normalize", Stage: pipeline.StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}, ArtifactKind: referenceContractKindAlpha, ReferenceSlots: []contracts.ReferenceSlot{{Name: "shared"}, {Name: "alpha-normalize"}, {Name: "required-normalize", Required: true}}}, func() (contracts.Normalizer[stateTestArtifact], error) { return stateTestNormalizer{}, nil })) + register(pipeline.RegisterNormalizer(registries.Normalizers, pipeline.ModuleSpec{Key: "reference/shared-normalize", Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"merged"}, Provides: []string{"normalized"}, ArtifactKind: referenceContractKindAlpha, ReferenceSlots: []contracts.ReferenceSlot{{Name: "shared"}, {Name: "alpha-normalize"}, {Name: "required-normalize", Required: true}}}, func() (contracts.Normalizer[stateTestArtifact], error) { return stateTestNormalizer{}, nil })) if includeBetaNormalizer { - register(pipeline.RegisterNormalizer(registries.Normalizers, pipeline.ModuleSpec{Key: "reference/shared-normalize", Stage: pipeline.StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}, ArtifactKind: referenceContractKindBeta, ReferenceSlots: []contracts.ReferenceSlot{{Name: "shared"}, {Name: "beta-normalize"}, {Name: "required-normalize", Required: true}}}, func() (contracts.Normalizer[stateTestArtifact], error) { return stateTestNormalizer{}, nil })) + register(pipeline.RegisterNormalizer(registries.Normalizers, pipeline.ModuleSpec{Key: "reference/shared-normalize", Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"merged"}, Provides: []string{"normalized"}, ArtifactKind: referenceContractKindBeta, ReferenceSlots: []contracts.ReferenceSlot{{Name: "shared"}, {Name: "beta-normalize"}, {Name: "required-normalize", Required: true}}}, func() (contracts.Normalizer[stateTestArtifact], error) { return stateTestNormalizer{}, nil })) } - register(registries.Outputs.RegisterWithSpec(pipeline.ModuleSpec{Key: "reference/output", Stage: pipeline.StageOutput, Requires: []string{"normalized"}, Provides: []string{"output"}}, func() (contracts.OutputEncoder, error) { return stateTestOutput{}, nil })) + register(registries.Outputs.RegisterWithSpec(pipeline.ModuleSpec{Key: "reference/output", Stage: pipeline.StageOutput, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"normalized"}, Provides: []string{"output"}}, func() (contracts.OutputEncoder, error) { return stateTestOutput{}, nil })) return catalogFromRegistries(registries) } diff --git a/internal/cli/run_contract_test.go b/internal/cli/run_contract_test.go index c2109e5..a8c9f72 100644 --- a/internal/cli/run_contract_test.go +++ b/internal/cli/run_contract_test.go @@ -526,7 +526,7 @@ func TestRunFactoryAndPreparationFailuresAreProcessFailures(t *testing.T) { t.Fatal(err) } opts := newStateTestHarness().options() - if err := pipeline.RegisterExtractorBuilder(opts.Registries.Extractors, pipeline.ModuleSpec{Key: "test/failing-extract", Stage: pipeline.StageExtract, Requires: []string{"chunks"}, Provides: []string{"artifact"}, ArtifactKind: stateTestArtifactKind}, func(map[string]any) error { return nil }, func(pipeline.BuildRequest) (contracts.Extractor[stateTestArtifact], error) { + if err := pipeline.RegisterExtractorBuilder(opts.Registries.Extractors, pipeline.ModuleSpec{Key: "test/failing-extract", Stage: pipeline.StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunks"}, Provides: []string{"artifact"}, ArtifactKind: stateTestArtifactKind}, func(map[string]any) error { return nil }, func(pipeline.BuildRequest) (contracts.Extractor[stateTestArtifact], error) { return nil, errors.New("injected extractor construction failure") }); err != nil { t.Fatal(err) diff --git a/internal/cli/state_hardening_test.go b/internal/cli/state_hardening_test.go index 029876c..b8306b8 100644 --- a/internal/cli/state_hardening_test.go +++ b/internal/cli/state_hardening_test.go @@ -823,22 +823,22 @@ func (h *stateTestHarness) options() Options { if err := pipeline.RegisterArtifactCodec(registries.ArtifactCodecs, stateTestCodec{}); err != nil { panic(err) } - if err := registries.Inputs.RegisterBuilderWithSpec(pipeline.ModuleSpec{Key: "test/input", Stage: pipeline.StageInput, Provides: []string{"source"}}, func(map[string]any) error { return nil }, func(pipeline.BuildRequest) (contracts.InputAdapter, error) { return stateTestInput{}, nil }); err != nil { + if err := registries.Inputs.RegisterBuilderWithSpec(pipeline.ModuleSpec{Key: "test/input", Stage: pipeline.StageInput, ExecutionClass: contracts.ExecutionClassDeterministic, Provides: []string{"source"}}, func(map[string]any) error { return nil }, func(pipeline.BuildRequest) (contracts.InputAdapter, error) { return stateTestInput{}, nil }); err != nil { panic(err) } - if err := registries.Chunkers.RegisterBuilderWithSpec(pipeline.ModuleSpec{Key: "test/chunk", Stage: pipeline.StageChunk, Requires: []string{"source"}, Provides: []string{"chunks"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "cache-reference"}}}, func(map[string]any) error { return nil }, func(pipeline.BuildRequest) (contracts.Chunker, error) { return stateTestChunker{h}, nil }); err != nil { + if err := registries.Chunkers.RegisterBuilderWithSpec(pipeline.ModuleSpec{Key: "test/chunk", Stage: pipeline.StageChunk, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"source"}, Provides: []string{"chunks"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "cache-reference"}}}, func(map[string]any) error { return nil }, func(pipeline.BuildRequest) (contracts.Chunker, error) { return stateTestChunker{h}, nil }); err != nil { panic(err) } - if err := pipeline.RegisterExtractor(registries.Extractors, pipeline.ModuleSpec{Key: "test/extract", Stage: pipeline.StageExtract, Requires: []string{"chunks"}, Provides: []string{"artifact"}, ArtifactKind: stateTestArtifactKind}, func() (contracts.Extractor[stateTestArtifact], error) { return stateTestExtractor{h}, nil }); err != nil { + if err := pipeline.RegisterExtractor(registries.Extractors, pipeline.ModuleSpec{Key: "test/extract", Stage: pipeline.StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunks"}, Provides: []string{"artifact"}, ArtifactKind: stateTestArtifactKind}, func() (contracts.Extractor[stateTestArtifact], error) { return stateTestExtractor{h}, nil }); err != nil { panic(err) } - if err := pipeline.RegisterMerger(registries.Mergers, pipeline.ModuleSpec{Key: "test/merge", Stage: pipeline.StageMerge, Requires: []string{"artifact"}, Provides: []string{"merged"}, ArtifactKind: stateTestArtifactKind}, func() (contracts.Merger[stateTestArtifact], error) { return stateTestMerger{harness: h}, nil }); err != nil { + if err := pipeline.RegisterMerger(registries.Mergers, pipeline.ModuleSpec{Key: "test/merge", Stage: pipeline.StageMerge, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"artifact"}, Provides: []string{"merged"}, ArtifactKind: stateTestArtifactKind}, func() (contracts.Merger[stateTestArtifact], error) { return stateTestMerger{harness: h}, nil }); err != nil { panic(err) } - if err := pipeline.RegisterNormalizer(registries.Normalizers, pipeline.ModuleSpec{Key: "test/normalize", Stage: pipeline.StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}, ArtifactKind: stateTestArtifactKind}, func() (contracts.Normalizer[stateTestArtifact], error) { return stateTestNormalizer{harness: h}, nil }); err != nil { + if err := pipeline.RegisterNormalizer(registries.Normalizers, pipeline.ModuleSpec{Key: "test/normalize", Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"merged"}, Provides: []string{"normalized"}, ArtifactKind: stateTestArtifactKind}, func() (contracts.Normalizer[stateTestArtifact], error) { return stateTestNormalizer{harness: h}, nil }); err != nil { panic(err) } - if err := registries.Outputs.RegisterWithSpec(pipeline.ModuleSpec{Key: "test/output", Stage: pipeline.StageOutput, Requires: []string{"normalized"}, Provides: []string{"output"}}, func() (contracts.OutputEncoder, error) { + if err := registries.Outputs.RegisterWithSpec(pipeline.ModuleSpec{Key: "test/output", Stage: pipeline.StageOutput, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"normalized"}, Provides: []string{"output"}}, func() (contracts.OutputEncoder, error) { return stateTestOutput{harness: h, includeWarnings: h.includeWarnings}, nil }); err != nil { panic(err) diff --git a/internal/core/config/effective_config_contract_test.go b/internal/core/config/effective_config_contract_test.go index ecb6157..163f0a6 100644 --- a/internal/core/config/effective_config_contract_test.go +++ b/internal/core/config/effective_config_contract_test.go @@ -485,7 +485,7 @@ func effectiveCatalog(t *testing.T) pipeline.ModuleCatalog { if err := pipeline.RegisterArtifactCodec(catalog.ArtifactCodecs, effectiveCodec{}); err != nil { t.Fatal(err) } - if err := catalog.Inputs.RegisterWithSpec(pipeline.ModuleSpec{Key: "input", Stage: pipeline.StageInput, Provides: []string{"source"}}, func() (contracts.InputAdapter, error) { + if err := catalog.Inputs.RegisterWithSpec(pipeline.ModuleSpec{Key: "input", Stage: pipeline.StageInput, ExecutionClass: contracts.ExecutionClassDeterministic, Provides: []string{"source"}}, func() (contracts.InputAdapter, error) { return effectiveInput{key: "input"}, nil }); err != nil { t.Fatal(err) @@ -493,6 +493,7 @@ func effectiveCatalog(t *testing.T) pipeline.ModuleCatalog { chunkSpec := pipeline.ModuleSpec{ Key: pipeline.DefaultChunkModule, Stage: pipeline.StageChunk, + ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"source"}, Provides: []string{"chunk"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "chunk-ref"}}, @@ -503,32 +504,32 @@ func effectiveCatalog(t *testing.T) pipeline.ModuleCatalog { }); err != nil { t.Fatal(err) } - if err := catalog.Chunkers.RegisterBuilderWithSpec(pipeline.ModuleSpec{Key: "needs-capability", Stage: pipeline.StageChunk, Requires: []string{"missing"}}, chunkOptions, func(pipeline.BuildRequest) (contracts.Chunker, error) { + if err := catalog.Chunkers.RegisterBuilderWithSpec(pipeline.ModuleSpec{Key: "needs-capability", Stage: pipeline.StageChunk, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"missing"}}, chunkOptions, func(pipeline.BuildRequest) (contracts.Chunker, error) { return effectiveChunker{key: "needs-capability"}, nil }); err != nil { t.Fatal(err) } - if err := pipeline.RegisterExtractor(catalog.Extractors, pipeline.ModuleSpec{Key: "extract", Stage: pipeline.StageExtract, ArtifactKind: effectiveArtifactKind, Requires: []string{"chunk"}, Provides: []string{"candidate"}}, func() (contracts.Extractor[effectiveArtifact], error) { + if err := pipeline.RegisterExtractor(catalog.Extractors, pipeline.ModuleSpec{Key: "extract", Stage: pipeline.StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: effectiveArtifactKind, Requires: []string{"chunk"}, Provides: []string{"candidate"}}, func() (contracts.Extractor[effectiveArtifact], error) { return effectiveExtractor{key: "extract"}, nil }); err != nil { t.Fatal(err) } - if err := pipeline.RegisterMerger(catalog.Mergers, pipeline.ModuleSpec{Key: pipeline.DefaultMergeModule, Stage: pipeline.StageMerge, ArtifactKind: effectiveArtifactKind, Requires: []string{"candidate"}, Provides: []string{"merged"}}, func() (contracts.Merger[effectiveArtifact], error) { + if err := pipeline.RegisterMerger(catalog.Mergers, pipeline.ModuleSpec{Key: pipeline.DefaultMergeModule, Stage: pipeline.StageMerge, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: effectiveArtifactKind, Requires: []string{"candidate"}, Provides: []string{"merged"}}, func() (contracts.Merger[effectiveArtifact], error) { return effectiveMerger{key: pipeline.DefaultMergeModule}, nil }); err != nil { t.Fatal(err) } - if err := pipeline.RegisterMerger(catalog.Mergers, pipeline.ModuleSpec{Key: "other-merge", Stage: pipeline.StageMerge, ArtifactKind: "other-kind", Requires: []string{"candidate"}, Provides: []string{"merged"}}, func() (contracts.Merger[effectiveArtifact], error) { + if err := pipeline.RegisterMerger(catalog.Mergers, pipeline.ModuleSpec{Key: "other-merge", Stage: pipeline.StageMerge, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: "other-kind", Requires: []string{"candidate"}, Provides: []string{"merged"}}, func() (contracts.Merger[effectiveArtifact], error) { return effectiveMerger{key: "other-merge"}, nil }); err != nil { t.Fatal(err) } - if err := pipeline.RegisterNormalizer(catalog.Normalizers, pipeline.ModuleSpec{Key: pipeline.DefaultNormalizeModule, Stage: pipeline.StageNormalize, ArtifactKind: effectiveArtifactKind, Requires: []string{"merged"}, Provides: []string{"normalized"}}, func() (contracts.Normalizer[effectiveArtifact], error) { + if err := pipeline.RegisterNormalizer(catalog.Normalizers, pipeline.ModuleSpec{Key: pipeline.DefaultNormalizeModule, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: effectiveArtifactKind, Requires: []string{"merged"}, Provides: []string{"normalized"}}, func() (contracts.Normalizer[effectiveArtifact], error) { return effectiveNormalizer{key: pipeline.DefaultNormalizeModule}, nil }); err != nil { t.Fatal(err) } - if err := catalog.Outputs.RegisterWithSpec(pipeline.ModuleSpec{Key: pipeline.DefaultOutputModule, Stage: pipeline.StageOutput, Requires: []string{"normalized"}}, func() (contracts.OutputEncoder, error) { + if err := catalog.Outputs.RegisterWithSpec(pipeline.ModuleSpec{Key: pipeline.DefaultOutputModule, Stage: pipeline.StageOutput, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"normalized"}}, func() (contracts.OutputEncoder, error) { return effectiveOutput{key: pipeline.DefaultOutputModule}, nil }); err != nil { t.Fatal(err) diff --git a/internal/framework/contracts/contracts.go b/internal/framework/contracts/contracts.go index 98682c3..052e3df 100644 --- a/internal/framework/contracts/contracts.go +++ b/internal/framework/contracts/contracts.go @@ -267,10 +267,6 @@ const ( ExecutionClassLLMBacked ExecutionClass = "llm_backed" ) -type ChunkExecutionClassProvider interface { - ExecutionClass() ExecutionClass -} - type ValidationResult struct { Approved bool `json:"approved"` ReasonCode string `json:"reason_code,omitempty"` diff --git a/internal/framework/pipeline/chunker_registry_test.go b/internal/framework/pipeline/chunker_registry_test.go index 56934c7..92b326a 100644 --- a/internal/framework/pipeline/chunker_registry_test.go +++ b/internal/framework/pipeline/chunker_registry_test.go @@ -101,10 +101,11 @@ func runRegistryBehaviorTests[M any](t *testing.T, testCase registryBehaviorCase t.Run(testCase.name+"/metadata registration and lookup", func(t *testing.T) { registry := testCase.newRegistry() spec := ModuleSpec{ - Key: " " + testCase.key + " ", - Stage: testCase.stage, - Provides: []string{" beta ", "alpha", "", "beta"}, - Requires: []string{" source ", "source", ""}, + Key: " " + testCase.key + " ", + Stage: testCase.stage, + ExecutionClass: contracts.ExecutionClassDeterministic, + Provides: []string{" beta ", "alpha", "", "beta"}, + Requires: []string{" source ", "source", ""}, } if err := testCase.registerWithSpec(registry, spec, testCase.constructor(testCase.key)); err != nil { t.Fatalf("RegisterWithSpec() error = %v, want nil", err) @@ -153,7 +154,7 @@ func runRegistryBehaviorTests[M any](t *testing.T, testCase registryBehaviorCase t.Run(testCase.name+"/wrong stage rejection", func(t *testing.T) { registry := testCase.newRegistry() - err := testCase.registerWithSpec(registry, ModuleSpec{Key: testCase.key, Stage: testCase.wrongStage}, testCase.constructor(testCase.key)) + err := testCase.registerWithSpec(registry, ModuleSpec{Key: testCase.key, Stage: testCase.wrongStage, ExecutionClass: contracts.ExecutionClassDeterministic}, testCase.constructor(testCase.key)) if err == nil { t.Fatal("RegisterWithSpec() error = nil, want error") } @@ -162,6 +163,22 @@ func runRegistryBehaviorTests[M any](t *testing.T, testCase registryBehaviorCase } }) + t.Run(testCase.name+"/execution class rejection", func(t *testing.T) { + for _, spec := range []ModuleSpec{ + {Key: testCase.key, Stage: testCase.stage}, + {Key: testCase.key, Stage: testCase.stage, ExecutionClass: contracts.ExecutionClass("unsupported")}, + } { + registry := testCase.newRegistry() + err := testCase.registerWithSpec(registry, spec, testCase.constructor(testCase.key)) + if err == nil { + t.Fatal("RegisterWithSpec() error = nil, want execution class error") + } + if !strings.Contains(err.Error(), "execution class") { + t.Fatalf("RegisterWithSpec() error = %q, want execution class error", err.Error()) + } + } + }) + t.Run(testCase.name+"/key trimming", func(t *testing.T) { registry := testCase.newRegistry() if err := testCase.register(registry, " "+testCase.key+" ", testCase.constructor(testCase.key)); err != nil { diff --git a/internal/framework/pipeline/input_registry_test.go b/internal/framework/pipeline/input_registry_test.go index f03c09c..19d2715 100644 --- a/internal/framework/pipeline/input_registry_test.go +++ b/internal/framework/pipeline/input_registry_test.go @@ -46,10 +46,11 @@ func TestInputAdapterRegistryRegisterAndBuildTrimKeys(t *testing.T) { func TestInputAdapterRegistryRegisterWithSpecStoresMetadata(t *testing.T) { registry := NewInputAdapterRegistry() spec := ModuleSpec{ - Key: " generic-input ", - Stage: StageInput, - Provides: []string{" parsed-source ", "source-document", "parsed-source", ""}, - Requires: []string{" raw-bytes ", "raw-bytes", ""}, + Key: " generic-input ", + Stage: StageInput, + ExecutionClass: contracts.ExecutionClassDeterministic, + Provides: []string{" parsed-source ", "source-document", "parsed-source", ""}, + Requires: []string{" raw-bytes ", "raw-bytes", ""}, } if err := registry.RegisterWithSpec(spec, fakeInputAdapterConstructor("generic-input")); err != nil { @@ -101,7 +102,7 @@ func TestInputAdapterRegistryRegisterStoresDefaultSpec(t *testing.T) { func TestInputAdapterRegistryRegisterWithSpecRejectsWrongStage(t *testing.T) { registry := NewInputAdapterRegistry() - err := registry.RegisterWithSpec(ModuleSpec{Key: "generic-input", Stage: StageExtract}, fakeInputAdapterConstructor("generic-input")) + err := registry.RegisterWithSpec(ModuleSpec{Key: "generic-input", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic}, fakeInputAdapterConstructor("generic-input")) if err == nil { t.Fatal("RegisterWithSpec() error = nil, want error") diff --git a/internal/framework/pipeline/module.go b/internal/framework/pipeline/module.go index 2d29a2e..e95d7ec 100644 --- a/internal/framework/pipeline/module.go +++ b/internal/framework/pipeline/module.go @@ -32,16 +32,14 @@ type ModuleSpec struct { func defaultModuleSpec(key string, stage ModuleStage) ModuleSpec { return ModuleSpec{ - Key: key, - Stage: stage, + Key: key, + Stage: stage, + ExecutionClass: contracts.ExecutionClassDeterministic, } } func normalizeModuleSpec(spec ModuleSpec) ModuleSpec { executionClass := contracts.ExecutionClass(strings.TrimSpace(string(spec.ExecutionClass))) - if executionClass == "" { - executionClass = contracts.ExecutionClassDeterministic - } return ModuleSpec{ Key: strings.TrimSpace(spec.Key), Stage: spec.Stage, @@ -97,6 +95,12 @@ func validateModuleSpec(kind string, expectedStage ModuleStage, spec ModuleSpec) if spec.Stage != expectedStage { return fmt.Errorf("%s %q must use %q stage, got %q", kind, spec.Key, expectedStage, spec.Stage) } + if spec.ExecutionClass == "" { + return fmt.Errorf("%s %q execution class must not be empty", kind, spec.Key) + } + if spec.ExecutionClass != contracts.ExecutionClassDeterministic && spec.ExecutionClass != contracts.ExecutionClassLLMBacked { + return fmt.Errorf("%s %q has unsupported execution class %q", kind, spec.Key, spec.ExecutionClass) + } if spec.ArtifactKind != "" && spec.Stage != StageExtract && spec.Stage != StageMerge && spec.Stage != StageNormalize { return fmt.Errorf("%s %q must not declare an artifact kind", kind, spec.Key) } diff --git a/internal/framework/pipeline/module_test.go b/internal/framework/pipeline/module_test.go index 2b98f17..cc5c29c 100644 --- a/internal/framework/pipeline/module_test.go +++ b/internal/framework/pipeline/module_test.go @@ -8,15 +8,35 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" ) -func TestNormalizeModuleSpecDefaultsExecutionClass(t *testing.T) { - normalized := normalizeModuleSpec(ModuleSpec{Key: " module ", Stage: StageChunk}) - if normalized.ExecutionClass != contracts.ExecutionClassDeterministic { - t.Fatalf("ExecutionClass = %q, want deterministic compatibility default", normalized.ExecutionClass) +func TestValidateModuleSpecRequiresSupportedExecutionClass(t *testing.T) { + for _, test := range []struct { + name string + class contracts.ExecutionClass + want string + }{ + {name: "missing", want: "execution class"}, + {name: "unsupported", class: "remote", want: "unsupported"}, + {name: "deterministic", class: contracts.ExecutionClassDeterministic}, + {name: "llm backed", class: contracts.ExecutionClassLLMBacked}, + } { + t.Run(test.name, func(t *testing.T) { + spec := normalizeModuleSpec(ModuleSpec{Key: "module", Stage: StageChunk, ExecutionClass: test.class}) + err := validateModuleSpec("chunker", StageChunk, spec) + if test.want == "" && err != nil { + t.Fatalf("validateModuleSpec() error = %v, want nil", err) + } + if test.want != "" && (err == nil || !strings.Contains(err.Error(), test.want)) { + t.Fatalf("validateModuleSpec() error = %v, want %q", err, test.want) + } + }) } +} - cloned := cloneModuleSpec(normalized) - if !reflect.DeepEqual(cloned, normalized) { - t.Fatalf("cloneModuleSpec() = %#v, want %#v", cloned, normalized) +func TestCloneModuleSpecPreservesExecutionClass(t *testing.T) { + spec := normalizeModuleSpec(ModuleSpec{Key: " module ", Stage: StageChunk, ExecutionClass: contracts.ExecutionClassLLMBacked}) + cloned := cloneModuleSpec(spec) + if !reflect.DeepEqual(cloned, spec) { + t.Fatalf("cloneModuleSpec() = %#v, want %#v", cloned, spec) } } @@ -35,8 +55,9 @@ func TestValidateModuleSpecAllowsReferenceSlotsForEligibleStages(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { spec := normalizeModuleSpec(ModuleSpec{ - Key: "module", - Stage: test.stage, + Key: "module", + Stage: test.stage, + ExecutionClass: contracts.ExecutionClassDeterministic, ReferenceSlots: []contracts.ReferenceSlot{ {Name: "roster", Description: "Character roster", MaxBytes: 1024}, }, @@ -63,8 +84,9 @@ func TestValidateModuleSpecRejectsReferenceSlotsForIneligibleStages(t *testing.T for _, test := range tests { t.Run(test.name, func(t *testing.T) { spec := normalizeModuleSpec(ModuleSpec{ - Key: "module", - Stage: test.stage, + Key: "module", + Stage: test.stage, + ExecutionClass: contracts.ExecutionClassDeterministic, ReferenceSlots: []contracts.ReferenceSlot{ {Name: "roster"}, }, @@ -122,6 +144,7 @@ func TestValidateModuleSpecRejectsInvalidReferenceSlotsForEligibleStages(t *test spec := normalizeModuleSpec(ModuleSpec{ Key: "module", Stage: stage.stage, + ExecutionClass: contracts.ExecutionClassDeterministic, ReferenceSlots: invalid.slots, }) err := validateModuleSpec(stage.kind, stage.stage, spec) diff --git a/internal/framework/pipeline/normalizer_registry_test.go b/internal/framework/pipeline/normalizer_registry_test.go index 6bad130..6d56b2b 100644 --- a/internal/framework/pipeline/normalizer_registry_test.go +++ b/internal/framework/pipeline/normalizer_registry_test.go @@ -26,7 +26,7 @@ func TestNormalizerRegistryErasureClonesRetryDirective(t *testing.T) { FallbackWarnings: []contracts.Warning{{Scope: "fallback", ReasonCode: "omitted", Message: "fallback warning"}}, } registry := NewNormalizerRegistry() - if err := RegisterNormalizer(registry, ModuleSpec{Key: "test/retry-normalize", Stage: StageNormalize, ArtifactKind: "test/notes"}, func() (contracts.Normalizer[codecNotes], error) { + if err := RegisterNormalizer(registry, ModuleSpec{Key: "test/retry-normalize", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: "test/notes"}, func() (contracts.Normalizer[codecNotes], error) { return retryingNotesNormalizer{warnings: warnings, retry: retry}, nil }); err != nil { t.Fatalf("RegisterNormalizer() error = %v", err) diff --git a/internal/framework/pipeline/profile_test.go b/internal/framework/pipeline/profile_test.go index 1df59b2..fcd11b3 100644 --- a/internal/framework/pipeline/profile_test.go +++ b/internal/framework/pipeline/profile_test.go @@ -15,11 +15,11 @@ import ( func TestResolvePipelineWithExplicitModules(t *testing.T) { catalog := newProfileCatalog(t) registerProfileSpecs(t, catalog, - ModuleSpec{Key: "window", Stage: StageChunk, Requires: []string{"source"}, Provides: []string{"chunk"}}, - ModuleSpec{Key: "record-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: "ndjson", Stage: StageOutput, Requires: []string{"normalized"}, Provides: []string{"encoded"}}, + 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{ @@ -386,7 +386,8 @@ func TestResolvePipelinePreservesOrderedStepsAndExpandsGeneratedBindings(t *test catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "note-extractor", Stage: StageExtract, ArtifactKind: "test/notes", - Requires: []string{"chunk"}, Provides: []string{"candidate"}, + 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{ @@ -414,7 +415,7 @@ func TestResolvePipelinePreservesOrderedStepsAndExpandsGeneratedBindings(t *test func TestResolvePipelineRejectsGeneratedBindingOrderingAndKind(t *testing.T) { catalog := newProfileCatalogWithOverrides(t, - ModuleSpec{Key: "note-extractor", Stage: StageExtract, ArtifactKind: "test/notes", Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "npcs", AcceptedArtifactKinds: []contracts.ArtifactKind{"test/other"}}}}, + 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{ @@ -449,10 +450,11 @@ func TestResolvePipelineAppliesReferenceBindings(t *testing.T) { profile.Artifacts["events"] = lane catalog := newProfileCatalogWithOverride(t, ModuleSpec{ - Key: "event-extractor", - Stage: StageExtract, - Requires: []string{"chunk"}, - Provides: []string{"candidate"}, + Key: "event-extractor", + Stage: StageExtract, + ExecutionClass: contracts.ExecutionClassDeterministic, + Requires: []string{"chunk"}, + Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{ {Name: "roster", Required: true}, {Name: "lore"}, @@ -495,6 +497,7 @@ func TestResolvePipelineAppliesPipelineReferenceDefaultToChunkTarget(t *testing. catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "generic", Stage: StageChunk, + ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"source"}, Provides: []string{"chunk"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "scene_guide"}}, @@ -523,6 +526,7 @@ func TestResolvePipelineAppliesPipelineReferenceDefaultToExtractorTarget(t *test catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "event-extractor", Stage: StageExtract, + ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "roster"}}, @@ -551,6 +555,7 @@ func TestResolvePipelineAppliesPipelineReferenceDefaultToNormalizerTarget(t *tes catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "noop", Stage: StageNormalize, + ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"merged"}, Provides: []string{"normalized"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "normalization_notes"}}, @@ -579,6 +584,7 @@ func TestResolvePipelineAppliesPipelineReferenceDefaultToMergeTarget(t *testing. catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "appendorder", Stage: StageMerge, + ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"candidate"}, Provides: []string{"merged"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "merge_notes"}}, @@ -608,10 +614,10 @@ func TestResolvePipelineAppliesOnePipelineReferenceDefaultToMultipleTargets(t *t profile := baselineProfile() profile.References = ExternalReferenceMap(map[string]string{"context": "./context.md"}) catalog := newProfileCatalogWithOverrides(t, - ModuleSpec{Key: "generic", Stage: StageChunk, Requires: []string{"source"}, Provides: []string{"chunk"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}}, - ModuleSpec{Key: "event-extractor", Stage: StageExtract, Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}}, - ModuleSpec{Key: "appendorder", Stage: StageMerge, Requires: []string{"candidate"}, Provides: []string{"merged"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}}, - ModuleSpec{Key: "noop", Stage: StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}}, + 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) @@ -629,10 +635,11 @@ func TestResolvePipelineAllowsPipelineReferenceDeclaredOnlyByUnselectedLane(t *t profile := multiLaneProfile() profile.References = ExternalReferenceMap(map[string]string{"notes_context": "./notes.md"}) catalog := newProfileCatalogWithOverride(t, ModuleSpec{ - Key: "note-extractor", - Stage: StageExtract, - Requires: []string{"chunk"}, - Provides: []string{"candidate"}, + Key: "note-extractor", + Stage: StageExtract, + ExecutionClass: contracts.ExecutionClassDeterministic, + Requires: []string{"chunk"}, + Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{ {Name: "notes_context"}, }, @@ -654,10 +661,11 @@ func TestResolvePipelineAllowsPipelineReferenceDeclaredOnlyByUnselectedNormalize lane.Normalize = Binding("note-normalizer") profile.Artifacts["notes"] = lane catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ - Key: "note-normalizer", - Stage: StageNormalize, - Requires: []string{"merged"}, - Provides: []string{"normalized"}, + Key: "note-normalizer", + Stage: StageNormalize, + ExecutionClass: contracts.ExecutionClassDeterministic, + Requires: []string{"merged"}, + Provides: []string{"normalized"}, ReferenceSlots: []contracts.ReferenceSlot{ {Name: "notes_context"}, }, @@ -707,6 +715,7 @@ func TestResolvePipelineRejectsExtractLocalReferenceDeclaredOnlyByNormalizer(t * catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "noop", Stage: StageNormalize, + ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"merged"}, Provides: []string{"normalized"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "normalization_notes"}}, @@ -725,10 +734,11 @@ func TestResolvePipelineRejectsMergeLocalReferenceDeclaredOnlyByNormalizer(t *te lane.Merge.References = ExternalReferenceMap(map[string]string{"normalization_notes": "./normalize.md"}) profile.Artifacts["events"] = lane catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ - Key: "noop", - Stage: StageNormalize, - Requires: []string{"merged"}, - Provides: []string{"normalized"}, + Key: "noop", + Stage: StageNormalize, + ExecutionClass: contracts.ExecutionClassDeterministic, + Requires: []string{"merged"}, + Provides: []string{"normalized"}, ReferenceSlots: []contracts.ReferenceSlot{ {Name: "normalization_notes"}, }, @@ -749,6 +759,7 @@ func TestResolvePipelineRejectsNormalizeLocalReferenceDeclaredOnlyByExtractor(t catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ Key: "event-extractor", Stage: StageExtract, + ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "roster"}}, @@ -765,6 +776,7 @@ 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}}, @@ -781,6 +793,7 @@ 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}}, @@ -806,9 +819,9 @@ func TestResolvePipelineLocalReferencesOverridePipelineDefaultsForEligibleTarget lane.Normalize.References = ExternalReferenceMap(map[string]string{"normalization_notes": "./local-normalize.md"}) profile.Artifacts["events"] = lane catalog := newProfileCatalogWithOverrides(t, - ModuleSpec{Key: "generic", Stage: StageChunk, Requires: []string{"source"}, Provides: []string{"chunk"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}}, - ModuleSpec{Key: "event-extractor", Stage: StageExtract, Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "roster"}}}, - ModuleSpec{Key: "noop", Stage: StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "normalization_notes"}}}, + 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) @@ -823,10 +836,11 @@ func TestResolvePipelineLocalReferencesOverridePipelineDefaultsForEligibleTarget func TestResolvePipelineRequiresBoundReferenceSlotsForSelectedLanes(t *testing.T) { catalog := newProfileCatalogWithOverride(t, ModuleSpec{ - Key: "event-extractor", - Stage: StageExtract, - Requires: []string{"chunk"}, - Provides: []string{"candidate"}, + Key: "event-extractor", + Stage: StageExtract, + ExecutionClass: contracts.ExecutionClassDeterministic, + Requires: []string{"chunk"}, + Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{ {Name: "roster", Required: true}, }, @@ -847,10 +861,11 @@ 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, - Requires: []string{"chunk"}, - Provides: []string{"candidate"}, + Key: "event-extractor", + Stage: StageExtract, + ExecutionClass: contracts.ExecutionClassDeterministic, + Requires: []string{"chunk"}, + Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{ {Name: "roster", Required: true}, }, @@ -876,11 +891,12 @@ func TestResolvePipelineUsesReferenceSlotsFromSpecWithoutConstructingExtractor(t } } if err := RegisterExtractor[codecNotes](catalog.Extractors, ModuleSpec{ - Key: "event-extractor", - Stage: StageExtract, - ArtifactKind: "test/notes", - Requires: []string{"chunk"}, - Provides: []string{"candidate"}, + Key: "event-extractor", + Stage: StageExtract, + ExecutionClass: contracts.ExecutionClassDeterministic, + ArtifactKind: "test/notes", + Requires: []string{"chunk"}, + Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{ {Name: "roster", Required: true}, }, @@ -1038,32 +1054,32 @@ func TestResolvePipelineRejectsMissingCapabilities(t *testing.T) { }{ { name: "input", - spec: ModuleSpec{Key: "text", Stage: StageInput, Requires: []string{"raw"}}, + 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, Requires: []string{"missing"}}, + 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, Requires: []string{"missing"}}, + 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, Requires: []string{"missing"}}, + 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, Requires: []string{"missing"}}, + 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, Requires: []string{"missing"}}, + spec: ModuleSpec{Key: "json", Stage: StageOutput, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"missing"}}, want: []string{"baseline", "output", "json", "missing"}, }, } @@ -1149,7 +1165,7 @@ func TestResolvePipelineDigestChangesWhenBindingChanges(t *testing.T) { right := baselineProfile() right.Chunk = Binding("window") catalog := newProfileCatalog(t) - registerProfileSpecs(t, catalog, ModuleSpec{Key: "window", Stage: StageChunk, Requires: []string{"source"}, Provides: []string{"chunk"}}) + 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 { @@ -1516,14 +1532,14 @@ func emptyProfileCatalog() ModuleCatalog { 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: "event-extractor", Stage: StageExtract, ArtifactKind: "test/notes", Requires: []string{"chunk"}, Provides: []string{"candidate"}}, - ModuleSpec{Key: "note-extractor", Stage: StageExtract, ArtifactKind: "test/notes", Requires: []string{"chunk"}, Provides: []string{"candidate"}}, - ModuleSpec{Key: "appendorder", Stage: StageMerge, ArtifactKind: "test/notes", Requires: []string{"candidate"}, Provides: []string{"merged"}}, - ModuleSpec{Key: "noop", Stage: StageNormalize, ArtifactKind: "test/notes", 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"}}, + 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"}}, } } diff --git a/internal/framework/pipeline/references_test.go b/internal/framework/pipeline/references_test.go index 9479280..c07719f 100644 --- a/internal/framework/pipeline/references_test.go +++ b/internal/framework/pipeline/references_test.go @@ -189,12 +189,12 @@ func TestMaterializeReferencesUsesLaneArtifactVariant(t *testing.T) { {kind: "test/beta", mergeSlot: "beta_merge", normalizeSlot: "beta_normalize"}, {kind: "test/alpha", mergeSlot: "alpha_merge", normalizeSlot: "alpha_normalize"}, } { - if err := RegisterMerger(mergers, ModuleSpec{Key: "shared/merge", Stage: StageMerge, ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.mergeSlot}}}, func() (contracts.Merger[codecNotes], error) { + if err := RegisterMerger(mergers, ModuleSpec{Key: "shared/merge", Stage: StageMerge, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.mergeSlot}}}, func() (contracts.Merger[codecNotes], error) { return typedTestMerger[codecNotes]{key: "shared/merge"}, nil }); err != nil { t.Fatalf("RegisterMerger(%s): %v", item.kind, err) } - if err := RegisterNormalizer(normalizers, ModuleSpec{Key: "shared/normalize", Stage: StageNormalize, ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.normalizeSlot}}}, func() (contracts.Normalizer[codecNotes], error) { + if err := RegisterNormalizer(normalizers, ModuleSpec{Key: "shared/normalize", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.normalizeSlot}}}, func() (contracts.Normalizer[codecNotes], error) { return typedTestNormalizer[codecNotes]{key: "shared/normalize"}, nil }); err != nil { t.Fatalf("RegisterNormalizer(%s): %v", item.kind, err) @@ -531,6 +531,7 @@ func referenceCatalogForTargets(t *testing.T, chunkSlots, extractSlots, mergeSlo ModuleSpec{ Key: "generic", Stage: StageChunk, + ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"source"}, Provides: []string{"chunk"}, ReferenceSlots: chunkSlots, @@ -538,6 +539,7 @@ func referenceCatalogForTargets(t *testing.T, chunkSlots, extractSlots, mergeSlo ModuleSpec{ Key: "event-extractor", Stage: StageExtract, + ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: extractSlots, @@ -545,6 +547,7 @@ func referenceCatalogForTargets(t *testing.T, chunkSlots, extractSlots, mergeSlo ModuleSpec{ Key: "appendorder", Stage: StageMerge, + ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"candidate"}, Provides: []string{"merged"}, ReferenceSlots: mergeSlots, @@ -552,6 +555,7 @@ func referenceCatalogForTargets(t *testing.T, chunkSlots, extractSlots, mergeSlo ModuleSpec{ Key: "noop", Stage: StageNormalize, + ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"merged"}, Provides: []string{"normalized"}, ReferenceSlots: normalizeSlots, diff --git a/internal/framework/pipeline/runner_chunk_plan.go b/internal/framework/pipeline/runner_chunk_plan.go index a53a91b..ad6a0a5 100644 --- a/internal/framework/pipeline/runner_chunk_plan.go +++ b/internal/framework/pipeline/runner_chunk_plan.go @@ -116,7 +116,7 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S return false, nil, terminal.record(nil, fmt.Errorf("clone chunker manifest metadata: %w", metadataErr)) } profile := "" - if provider, ok := chunker.(contracts.ChunkExecutionClassProvider); ok && provider.ExecutionClass() == contracts.ExecutionClassLLMBacked { + if input.pipeline.ChunkExecutionClass == contracts.ExecutionClassLLMBacked { profile = input.pipeline.Chunk.LLMProfile } candidate := ChunkPlanRecord{ diff --git a/internal/framework/pipeline/runner_chunk_plan_test.go b/internal/framework/pipeline/runner_chunk_plan_test.go index bfce124..008cc0d 100644 --- a/internal/framework/pipeline/runner_chunk_plan_test.go +++ b/internal/framework/pipeline/runner_chunk_plan_test.go @@ -67,30 +67,17 @@ type manifestChunker struct { } func (c manifestChunker) ManifestMetadata() map[string]any { return c.metadata } -func (manifestChunker) ExecutionClass() contracts.ExecutionClass { - return contracts.ExecutionClassLLMBacked -} type llmCountingChunker struct { terminalChunker llmCalls *int } -func (llmCountingChunker) ExecutionClass() contracts.ExecutionClass { - return contracts.ExecutionClassLLMBacked -} - func (c llmCountingChunker) Plan(ctx context.Context, request contracts.ChunkRequest) (contracts.ChunkPlanResult, error) { (*c.llmCalls)++ return c.terminalChunker.Plan(ctx, request) } -type deterministicChunker struct{ terminalChunker } - -func (deterministicChunker) ExecutionClass() contracts.ExecutionClass { - return contracts.ExecutionClassDeterministic -} - type retryingChunker struct { key string plan source.ChunkPlan @@ -535,6 +522,7 @@ func TestRunnerPublishesOnlyAcceptedGeneratedPlans(t *testing.T) { func TestRunnerStoresProducerProvenanceAndProducerWarnings(t *testing.T) { prepared, plan := preparedTerminalDebugPipeline(t) prepared.resolved.Chunk.LLMProfile = "chunk-profile" + prepared.resolved.ChunkExecutionClass = contracts.ExecutionClassLLMBacked prepared.resolved.ChunkReferences = ResolvedReferenceTarget{ Stage: StageChunk, ReferenceSet: contracts.ReferenceSet{Slots: map[string]contracts.ResolvedReferenceSlot{ @@ -577,7 +565,8 @@ func TestRunnerRejectsUncloneableModuleManifestMetadata(t *testing.T) { func TestRunnerOmitsProducerProfileForDeterministicChunker(t *testing.T) { prepared, plan := preparedTerminalDebugPipeline(t) prepared.resolved.Chunk.LLMProfile = "configured-but-unused" - prepared.chunker = deterministicChunker{terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan}} + prepared.resolved.ChunkExecutionClass = contracts.ExecutionClassDeterministic + prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan} store := &recordingChunkPlanStore{} output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheRefresh, ChunkPlans: store}) if err != nil { @@ -588,6 +577,22 @@ func TestRunnerOmitsProducerProfileForDeterministicChunker(t *testing.T) { } } +func TestRunnerDoesNotInventProducerProfileForLLMChunker(t *testing.T) { + prepared, plan := preparedTerminalDebugPipeline(t) + prepared.resolved.ChunkExecutionClass = contracts.ExecutionClassLLMBacked + prepared.resolved.Chunk.LLMProfile = "" + prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan} + store := &recordingChunkPlanStore{} + + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheRefresh, ChunkPlans: store}) + if err != nil { + t.Fatal(err) + } + if store.saved.Producer.LLMProfile != "" || output.Manifest.ChunkPlan.ProducerLLMProfile != "" { + t.Fatalf("LLM producer profile = stored %q manifest %q, want empty", store.saved.Producer.LLMProfile, output.Manifest.ChunkPlan.ProducerLLMProfile) + } +} + func TestRunnerRefreshChangesDownstreamChunkFingerprint(t *testing.T) { doc := typedTestDocumentWithUnits(2) prepared := preparedConcurrentPipeline(t, 1) diff --git a/internal/framework/pipeline/typed_resolution_test.go b/internal/framework/pipeline/typed_resolution_test.go index 231bd2b..fd9ced2 100644 --- a/internal/framework/pipeline/typed_resolution_test.go +++ b/internal/framework/pipeline/typed_resolution_test.go @@ -229,7 +229,7 @@ func TestResolveTypedLaneRejectsIncompatibleComposition(t *testing.T) { func TestTypedVariantRegistrationRejectsDuplicates(t *testing.T) { registry := NewMergerRegistry() - spec := ModuleSpec{Key: "typed/merge", Stage: StageMerge, ArtifactKind: "test/notes"} + spec := ModuleSpec{Key: "typed/merge", Stage: StageMerge, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: "test/notes"} constructor := func() (contracts.Merger[codecNotes], error) { return typedTestMerger[codecNotes]{key: "typed/merge"}, nil } @@ -243,7 +243,7 @@ func TestTypedVariantRegistrationRejectsDuplicates(t *testing.T) { func TestConstructorRegistrationsRejectUnconfiguredOptions(t *testing.T) { extractors := NewExtractorRegistry() - if err := RegisterExtractor(extractors, ModuleSpec{Key: "typed/extract", Stage: StageExtract, ArtifactKind: "test/notes"}, func() (contracts.Extractor[codecNotes], error) { + if err := RegisterExtractor(extractors, ModuleSpec{Key: "typed/extract", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: "test/notes"}, func() (contracts.Extractor[codecNotes], error) { return typedTestExtractor[codecNotes]{key: "typed/extract"}, nil }); err != nil { t.Fatalf("RegisterExtractor() error = %v", err) @@ -253,7 +253,7 @@ func TestConstructorRegistrationsRejectUnconfiguredOptions(t *testing.T) { } mergers := NewMergerRegistry() - if err := RegisterMerger(mergers, ModuleSpec{Key: "typed/merge", Stage: StageMerge, ArtifactKind: "test/notes"}, func() (contracts.Merger[codecNotes], error) { + if err := RegisterMerger(mergers, ModuleSpec{Key: "typed/merge", Stage: StageMerge, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: "test/notes"}, func() (contracts.Merger[codecNotes], error) { return typedTestMerger[codecNotes]{key: "typed/merge"}, nil }); err != nil { t.Fatalf("RegisterMerger() error = %v", err) @@ -263,7 +263,7 @@ func TestConstructorRegistrationsRejectUnconfiguredOptions(t *testing.T) { } normalizers := NewNormalizerRegistry() - if err := RegisterNormalizer(normalizers, ModuleSpec{Key: "typed/normalize", Stage: StageNormalize, ArtifactKind: "test/notes"}, func() (contracts.Normalizer[codecNotes], error) { + if err := RegisterNormalizer(normalizers, ModuleSpec{Key: "typed/normalize", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: "test/notes"}, func() (contracts.Normalizer[codecNotes], error) { return typedTestNormalizer[codecNotes]{key: "typed/normalize"}, nil }); err != nil { t.Fatalf("RegisterNormalizer() error = %v", err) @@ -298,7 +298,7 @@ func TestTypedRegistrySpecLookupUsesArtifactKindAndStableCatalogOrder(t *testing mergers := NewMergerRegistry() normalizers := NewNormalizerRegistry() for _, item := range order { - base := ModuleSpec{Key: "typed/shared", ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.slot}}} + base := ModuleSpec{Key: "typed/shared", ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.slot}}} mergeSpec := base mergeSpec.Stage = StageMerge normalizeSpec := base @@ -474,18 +474,18 @@ func typedResolutionCatalog(t *testing.T, options typedCatalogOptions) ModuleCat func mustRegisterTypedTestBase(t *testing.T, catalog ModuleCatalog) { t.Helper() - if err := catalog.Inputs.RegisterWithSpec(ModuleSpec{Key: "typed/input", Stage: StageInput}, func() (contracts.InputAdapter, error) { + if err := catalog.Inputs.RegisterWithSpec(ModuleSpec{Key: "typed/input", Stage: StageInput, ExecutionClass: contracts.ExecutionClassDeterministic}, func() (contracts.InputAdapter, error) { return &typedTestInput{key: "typed/input", doc: typedTestDocument()}, nil }); err != nil { t.Fatalf("register input: %v", err) } - if err := catalog.Chunkers.RegisterWithSpec(ModuleSpec{Key: "typed/chunk", Stage: StageChunk}, func() (contracts.Chunker, error) { + if err := catalog.Chunkers.RegisterWithSpec(ModuleSpec{Key: "typed/chunk", Stage: StageChunk, ExecutionClass: contracts.ExecutionClassDeterministic}, func() (contracts.Chunker, error) { doc := typedTestDocument() return &typedTestChunker{key: "typed/chunk", plan: typedTestPlan(doc)}, nil }); err != nil { t.Fatalf("register chunker: %v", err) } - if err := catalog.Outputs.RegisterWithSpec(ModuleSpec{Key: "typed/output", Stage: StageOutput}, func() (contracts.OutputEncoder, error) { + if err := catalog.Outputs.RegisterWithSpec(ModuleSpec{Key: "typed/output", Stage: StageOutput, ExecutionClass: contracts.ExecutionClassDeterministic}, func() (contracts.OutputEncoder, error) { return &typedTestOutput{key: "typed/output"}, nil }); err != nil { t.Fatalf("register output: %v", err) @@ -509,21 +509,21 @@ func mustRegisterArtifactCodec[T any](t *testing.T, registry *ArtifactCodecRegis func mustRegisterTypedExtractor[T any](t *testing.T, registry *ExtractorRegistry, key string, kind contracts.ArtifactKind, extractor contracts.Extractor[T]) { t.Helper() - if err := RegisterExtractor(registry, ModuleSpec{Key: key, Stage: StageExtract, ArtifactKind: kind}, func() (contracts.Extractor[T], error) { return extractor, nil }); err != nil { + if err := RegisterExtractor(registry, ModuleSpec{Key: key, Stage: StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: kind}, func() (contracts.Extractor[T], error) { return extractor, nil }); err != nil { t.Fatalf("RegisterExtractor() error = %v", err) } } func mustRegisterTypedMerger[T any](t *testing.T, registry *MergerRegistry, kind contracts.ArtifactKind, merger contracts.Merger[T]) { t.Helper() - if err := RegisterMerger(registry, ModuleSpec{Key: "typed/merge", Stage: StageMerge, ArtifactKind: kind}, func() (contracts.Merger[T], error) { return merger, nil }); err != nil { + if err := RegisterMerger(registry, ModuleSpec{Key: "typed/merge", Stage: StageMerge, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: kind}, func() (contracts.Merger[T], error) { return merger, nil }); err != nil { t.Fatalf("RegisterMerger() error = %v", err) } } func mustRegisterTypedNormalizer[T any](t *testing.T, registry *NormalizerRegistry, kind contracts.ArtifactKind, normalizer contracts.Normalizer[T]) { t.Helper() - if err := RegisterNormalizer(registry, ModuleSpec{Key: "typed/normalize", Stage: StageNormalize, ArtifactKind: kind}, func() (contracts.Normalizer[T], error) { return normalizer, nil }); err != nil { + if err := RegisterNormalizer(registry, ModuleSpec{Key: "typed/normalize", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: kind}, func() (contracts.Normalizer[T], error) { return normalizer, nil }); err != nil { t.Fatalf("RegisterNormalizer() error = %v", err) } } diff --git a/internal/modules/dnd/chunk/scenes/chunker.go b/internal/modules/dnd/chunk/scenes/chunker.go index 7005d6f..3066e20 100644 --- a/internal/modules/dnd/chunk/scenes/chunker.go +++ b/internal/modules/dnd/chunk/scenes/chunker.go @@ -47,10 +47,6 @@ func (c *Chunker) Key() string { return Key } -func (*Chunker) ExecutionClass() contracts.ExecutionClass { - return contracts.ExecutionClassLLMBacked -} - func (c *Chunker) ReferenceSlots() []contracts.ReferenceSlot { return shared.ReferenceSlots(referenceSlotDescriptions) } diff --git a/internal/modules/generic/chunk/units/chunker.go b/internal/modules/generic/chunk/units/chunker.go index 67e9757..0125a66 100644 --- a/internal/modules/generic/chunk/units/chunker.go +++ b/internal/modules/generic/chunk/units/chunker.go @@ -38,10 +38,6 @@ func (c *Chunker) Key() string { return Key } -func (*Chunker) ExecutionClass() contracts.ExecutionClass { - return contracts.ExecutionClassDeterministic -} - func (c *Chunker) ReferenceSlots() []contracts.ReferenceSlot { return nil } diff --git a/internal/modules/integration/concurrent_runner_test.go b/internal/modules/integration/concurrent_runner_test.go index 594ecee..40773f7 100644 --- a/internal/modules/integration/concurrent_runner_test.go +++ b/internal/modules/integration/concurrent_runner_test.go @@ -123,11 +123,11 @@ func TestRunnerIndependentlyBoundsWorkersAndProviderCallsAcrossRegisteredModules } client := frameworkllm.NewScheduledClient(instrumentedProvider{tracker: tracker}, scheduler) catalog := dndSpellsTestCatalog(t, dndSpellsCatalogSpecs{}) - if err := catalog.Chunkers.RegisterWithSpec(pipeline.ModuleSpec{Key: concurrentChunkerKey, Stage: pipeline.StageChunk, Requires: []string{"source.transcript"}, Provides: []string{"chunks"}}, func() (contracts.Chunker, error) { return concurrentChunker{}, nil }); err != nil { + if err := catalog.Chunkers.RegisterWithSpec(pipeline.ModuleSpec{Key: concurrentChunkerKey, Stage: pipeline.StageChunk, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"source.transcript"}, Provides: []string{"chunks"}}, func() (contracts.Chunker, error) { return concurrentChunker{}, nil }); err != nil { t.Fatalf("register chunker: %v", err) } validateOptions := func(options map[string]any) error { return pipeline.RejectUnknownOptions(options) } - if err := pipeline.RegisterExtractorBuilder(catalog.Extractors, pipeline.ModuleSpec{Key: concurrentExtractorKey, Stage: pipeline.StageExtract, ArtifactKind: dnd.SpellListKind, Requires: []string{"chunks", "source.transcript"}, Provides: []string{"dnd.spell_casts"}}, validateOptions, func(request pipeline.BuildRequest) (contracts.Extractor[dnd.SpellList], error) { + if err := pipeline.RegisterExtractorBuilder(catalog.Extractors, pipeline.ModuleSpec{Key: concurrentExtractorKey, Stage: pipeline.StageExtract, ExecutionClass: contracts.ExecutionClassLLMBacked, ArtifactKind: dnd.SpellListKind, Requires: []string{"chunks", "source.transcript"}, Provides: []string{"dnd.spell_casts"}}, validateOptions, func(request pipeline.BuildRequest) (contracts.Extractor[dnd.SpellList], error) { return &concurrentExtractor{client: request.Dependencies.LLM, tracker: tracker}, nil }); err != nil { t.Fatalf("register extractor: %v", err) diff --git a/internal/modules/integration/dnd_spells_config_test.go b/internal/modules/integration/dnd_spells_config_test.go index c915183..79645f1 100644 --- a/internal/modules/integration/dnd_spells_config_test.go +++ b/internal/modules/integration/dnd_spells_config_test.go @@ -75,7 +75,7 @@ func dndCapabilityCatalog(t *testing.T, inputSpec, extractorSpec pipeline.Module } chunkers := pipeline.NewChunkerRegistry() - if err := chunkers.RegisterWithSpec(pipeline.ModuleSpec{Key: "fake/chunk", Stage: pipeline.StageChunk, Provides: []string{"chunks"}}, func() (contracts.Chunker, error) { return dndSpellsChunker{}, nil }); err != nil { + if err := chunkers.RegisterWithSpec(pipeline.ModuleSpec{Key: "fake/chunk", Stage: pipeline.StageChunk, ExecutionClass: contracts.ExecutionClassDeterministic, Provides: []string{"chunks"}}, func() (contracts.Chunker, error) { return dndSpellsChunker{}, nil }); err != nil { t.Fatalf("register capability chunker: %v", err) } @@ -95,20 +95,20 @@ func dndCapabilityCatalog(t *testing.T, inputSpec, extractorSpec pipeline.Module mergers := pipeline.NewMergerRegistry() if err := pipeline.RegisterMerger[dnd.SpellList](mergers, pipeline.ModuleSpec{ - Key: pipeline.DefaultMergeModule, Stage: pipeline.StageMerge, ArtifactKind: dnd.SpellListKind, Requires: []string{"dnd.spell_casts"}, + Key: pipeline.DefaultMergeModule, Stage: pipeline.StageMerge, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: dnd.SpellListKind, Requires: []string{"dnd.spell_casts"}, }, func() (contracts.Merger[dnd.SpellList], error) { return appendorder.NewTyped(appendSpellLists) }); err != nil { t.Fatalf("register capability merger: %v", err) } normalizers := pipeline.NewNormalizerRegistry() - if err := pipeline.RegisterNormalizer[dnd.SpellList](normalizers, pipeline.ModuleSpec{Key: pipeline.DefaultNormalizeModule, Stage: pipeline.StageNormalize, ArtifactKind: dnd.SpellListKind}, func() (contracts.Normalizer[dnd.SpellList], error) { + if err := pipeline.RegisterNormalizer[dnd.SpellList](normalizers, pipeline.ModuleSpec{Key: pipeline.DefaultNormalizeModule, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: dnd.SpellListKind}, func() (contracts.Normalizer[dnd.SpellList], error) { return noop.NewTyped[dnd.SpellList](), nil }); err != nil { t.Fatalf("register capability normalizer: %v", err) } outputs := pipeline.NewOutputEncoderRegistry() - if err := outputs.RegisterWithSpec(pipeline.ModuleSpec{Key: pipeline.DefaultOutputModule, Stage: pipeline.StageOutput}, func() (contracts.OutputEncoder, error) { return dndSpellsOutput{}, nil }); err != nil { + if err := outputs.RegisterWithSpec(pipeline.ModuleSpec{Key: pipeline.DefaultOutputModule, Stage: pipeline.StageOutput, ExecutionClass: contracts.ExecutionClassDeterministic}, func() (contracts.OutputEncoder, error) { return dndSpellsOutput{}, nil }); err != nil { t.Fatalf("register capability output: %v", err) } @@ -192,27 +192,30 @@ func dndSpellsTestCatalog(t *testing.T, specs dndSpellsCatalogSpecs) pipeline.Mo } if err := pipeline.RegisterMerger[dnd.SpellList](mergers, pipeline.ModuleSpec{ - Key: pipeline.DefaultMergeModule, - Stage: pipeline.StageMerge, - ArtifactKind: dnd.SpellListKind, - Requires: []string{"dnd.spell_casts"}, + Key: pipeline.DefaultMergeModule, + Stage: pipeline.StageMerge, + ExecutionClass: contracts.ExecutionClassDeterministic, + ArtifactKind: dnd.SpellListKind, + Requires: []string{"dnd.spell_casts"}, }, func() (contracts.Merger[dnd.SpellList], error) { return appendorder.NewTyped(appendSpellLists) }); err != nil { t.Fatalf("register merger: %v", err) } if err := pipeline.RegisterNormalizer[dnd.SpellList](normalizers, pipeline.ModuleSpec{ - Key: pipeline.DefaultNormalizeModule, - Stage: pipeline.StageNormalize, - ArtifactKind: dnd.SpellListKind, + Key: pipeline.DefaultNormalizeModule, + Stage: pipeline.StageNormalize, + ExecutionClass: contracts.ExecutionClassDeterministic, + ArtifactKind: dnd.SpellListKind, }, func() (contracts.Normalizer[dnd.SpellList], error) { return noop.NewTyped[dnd.SpellList](), nil }); err != nil { t.Fatalf("register normalizer: %v", err) } if err := outputs.RegisterWithSpec(pipeline.ModuleSpec{ - Key: pipeline.DefaultOutputModule, - Stage: pipeline.StageOutput, + Key: pipeline.DefaultOutputModule, + Stage: pipeline.StageOutput, + ExecutionClass: contracts.ExecutionClassDeterministic, }, func() (contracts.OutputEncoder, error) { return dndSpellsOutput{}, nil }); err != nil { @@ -250,10 +253,11 @@ func appendSpellLists(values []dnd.SpellList) (dnd.SpellList, error) { func dndSpellsChunkerSpec() pipeline.ModuleSpec { return pipeline.ModuleSpec{ - Key: "fake/chunk", - Stage: pipeline.StageChunk, - Requires: []string{"source.transcript"}, - Provides: []string{"chunks"}, + Key: "fake/chunk", + Stage: pipeline.StageChunk, + ExecutionClass: contracts.ExecutionClassDeterministic, + Requires: []string{"source.transcript"}, + Provides: []string{"chunks"}, } } diff --git a/internal/modules/seriatim/input/transcript/runner_helpers_test.go b/internal/modules/seriatim/input/transcript/runner_helpers_test.go index 6c60e3f..16134fd 100644 --- a/internal/modules/seriatim/input/transcript/runner_helpers_test.go +++ b/internal/modules/seriatim/input/transcript/runner_helpers_test.go @@ -47,14 +47,14 @@ func seriatimTestCatalog(t *testing.T, inputSpec pipeline.ModuleSpec) pipeline.M t.Fatalf("register seriatim input override: %v", err) } - mustRegisterChunker(t, chunkers, pipeline.ModuleSpec{Key: "fake/chunk", Stage: pipeline.StageChunk, Requires: []string{"source.transcript"}, Provides: []string{"chunks"}}) + mustRegisterChunker(t, chunkers, pipeline.ModuleSpec{Key: "fake/chunk", Stage: pipeline.StageChunk, ExecutionClass: contracts.ExecutionClassDeterministic, Requires: []string{"source.transcript"}, Provides: []string{"chunks"}}) mustRegisterExtractor(t, extractors, pipeline.ModuleSpec{ - Key: "fake/extract", Stage: pipeline.StageExtract, ArtifactKind: seriatimArtifactKind, + Key: "fake/extract", Stage: pipeline.StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: seriatimArtifactKind, Requires: []string{"chunks", "transcript.speaker", "transcript.timestamps"}, Provides: []string{"fake.artifacts"}, }) - mustRegisterMerger(t, mergers, pipeline.ModuleSpec{Key: pipeline.DefaultMergeModule, Stage: pipeline.StageMerge, ArtifactKind: seriatimArtifactKind, Requires: []string{"fake.artifacts"}}) - mustRegisterNormalizer(t, normalizers, pipeline.ModuleSpec{Key: pipeline.DefaultNormalizeModule, Stage: pipeline.StageNormalize, ArtifactKind: seriatimArtifactKind}) - mustRegisterOutput(t, outputs, pipeline.ModuleSpec{Key: pipeline.DefaultOutputModule, Stage: pipeline.StageOutput}) + mustRegisterMerger(t, mergers, pipeline.ModuleSpec{Key: pipeline.DefaultMergeModule, Stage: pipeline.StageMerge, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: seriatimArtifactKind, Requires: []string{"fake.artifacts"}}) + mustRegisterNormalizer(t, normalizers, pipeline.ModuleSpec{Key: pipeline.DefaultNormalizeModule, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: seriatimArtifactKind}) + mustRegisterOutput(t, outputs, pipeline.ModuleSpec{Key: pipeline.DefaultOutputModule, Stage: pipeline.StageOutput, ExecutionClass: contracts.ExecutionClassDeterministic}) codecs := pipeline.NewArtifactCodecRegistry() if err := pipeline.RegisterArtifactCodec(codecs, seriatimArtifactCodec{}); err != nil { diff --git a/internal/modules/seriatim/input/transcript/runner_test.go b/internal/modules/seriatim/input/transcript/runner_test.go index d342034..c390de4 100644 --- a/internal/modules/seriatim/input/transcript/runner_test.go +++ b/internal/modules/seriatim/input/transcript/runner_test.go @@ -131,17 +131,17 @@ func seriatimRunnerRegistries(t *testing.T, extractor contracts.Extractor[seriat }); err != nil { t.Fatalf("register chunker: %v", err) } - if err := pipeline.RegisterExtractor[seriatimArtifact](extractors, pipeline.ModuleSpec{Key: "fake/extract", Stage: pipeline.StageExtract, ArtifactKind: seriatimArtifactKind, Requires: []string{"chunks", "transcript.speaker", "transcript.timestamps"}, Provides: []string{"fake.artifacts"}}, func() (contracts.Extractor[seriatimArtifact], error) { + if err := pipeline.RegisterExtractor[seriatimArtifact](extractors, pipeline.ModuleSpec{Key: "fake/extract", Stage: pipeline.StageExtract, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: seriatimArtifactKind, Requires: []string{"chunks", "transcript.speaker", "transcript.timestamps"}, Provides: []string{"fake.artifacts"}}, func() (contracts.Extractor[seriatimArtifact], error) { return extractor, nil }); err != nil { t.Fatalf("register extractor: %v", err) } - if err := pipeline.RegisterMerger[seriatimArtifact](mergers, pipeline.ModuleSpec{Key: pipeline.DefaultMergeModule, Stage: pipeline.StageMerge, ArtifactKind: seriatimArtifactKind}, func() (contracts.Merger[seriatimArtifact], error) { + if err := pipeline.RegisterMerger[seriatimArtifact](mergers, pipeline.ModuleSpec{Key: pipeline.DefaultMergeModule, Stage: pipeline.StageMerge, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: seriatimArtifactKind}, func() (contracts.Merger[seriatimArtifact], error) { return fakeMerger{}, nil }); err != nil { t.Fatalf("register merger: %v", err) } - if err := pipeline.RegisterNormalizer[seriatimArtifact](normalizers, pipeline.ModuleSpec{Key: pipeline.DefaultNormalizeModule, Stage: pipeline.StageNormalize, ArtifactKind: seriatimArtifactKind}, func() (contracts.Normalizer[seriatimArtifact], error) { + if err := pipeline.RegisterNormalizer[seriatimArtifact](normalizers, pipeline.ModuleSpec{Key: pipeline.DefaultNormalizeModule, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: seriatimArtifactKind}, func() (contracts.Normalizer[seriatimArtifact], error) { return fakeNormalizer{}, nil }); err != nil { t.Fatalf("register normalizer: %v", err)