eric/notarius
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Require explicit module execution classes

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-03 17:00:08 +00:00
parent ce857966f1
commit 58815aaf33
23 changed files with 246 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
internal/cli/assembled_spell_pipeline_contract_test.go
View File

@@ -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 {

4
internal/cli/recompute_execution_contract_test.go
View File

@@ -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)

18
internal/cli/reference_contract_test.go
View File

@@ -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)
}

2
internal/cli/run_contract_test.go
View File

@@ -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)

12
internal/cli/state_hardening_test.go
View File

@@ -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)

15
internal/core/config/effective_config_contract_test.go
View File

@@ -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)

4
internal/framework/contracts/contracts.go
View File

@@ -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"`

27
internal/framework/pipeline/chunker_registry_test.go
View File

@@ -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 {

11
internal/framework/pipeline/input_registry_test.go
View File

@@ -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")

14
internal/framework/pipeline/module.go
View File

@@ -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)
}

45
internal/framework/pipeline/module_test.go
View File

@@ -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)

2
internal/framework/pipeline/normalizer_registry_test.go
View File

@@ -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)

132
internal/framework/pipeline/profile_test.go
View File

@@ -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"}},
}
}

8
internal/framework/pipeline/references_test.go
View File

@@ -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,

2
internal/framework/pipeline/runner_chunk_plan.go
View File

@@ -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{

33
internal/framework/pipeline/runner_chunk_plan_test.go
View File

@@ -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)

22
internal/framework/pipeline/typed_resolution_test.go
View File

@@ -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)
}
}

4
internal/modules/dnd/chunk/scenes/chunker.go
View File

@@ -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)
}

4
internal/modules/generic/chunk/units/chunker.go
View File

@@ -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
}

4
internal/modules/integration/concurrent_runner_test.go
View File

@@ -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)

38
internal/modules/integration/dnd_spells_config_test.go
View File

@@ -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"},
}
}

10
internal/modules/seriatim/input/transcript/runner_helpers_test.go
View File

@@ -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 {

6
internal/modules/seriatim/input/transcript/runner_test.go
View File

@@ -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)