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

Add execution metadata to module specifications

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-03 16:49:51 +00:00
parent a3bd0c1867
commit ce857966f1
39 changed files with 331 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -115,10 +115,11 @@ func runRegistryBehaviorTests[M any](t *testing.T, testCase registryBehaviorCase
t.Fatal("Spec() ok = false, want true")
}
want := ModuleSpec{
Key: testCase.key,
Stage: testCase.stage,
Provides: []string{"alpha", "beta"},
Requires: []string{"source"},
Key: testCase.key,
Stage: testCase.stage,
ExecutionClass: contracts.ExecutionClassDeterministic,
Provides: []string{"alpha", "beta"},
Requires: []string{"source"},
}
if !reflect.DeepEqual(got, want) {
t.Fatalf("Spec() = %#v, want %#v", got, want)
@@ -144,7 +145,7 @@ func runRegistryBehaviorTests[M any](t *testing.T, testCase registryBehaviorCase
if !ok {
t.Fatal("Spec() ok = false, want true")
}
want := ModuleSpec{Key: testCase.key, Stage: testCase.stage}
want := ModuleSpec{Key: testCase.key, Stage: testCase.stage, ExecutionClass: contracts.ExecutionClassDeterministic}
if !reflect.DeepEqual(spec, want) {
t.Fatalf("Spec() = %#v, want %#v", spec, want)
}

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

@@ -61,10 +61,11 @@ func TestInputAdapterRegistryRegisterWithSpecStoresMetadata(t *testing.T) {
t.Fatal("Spec() ok = false, want true")
}
want := ModuleSpec{
Key: "generic-input",
Stage: StageInput,
Provides: []string{"parsed-source", "source-document"},
Requires: []string{"raw-bytes"},
Key: "generic-input",
Stage: StageInput,
ExecutionClass: contracts.ExecutionClassDeterministic,
Provides: []string{"parsed-source", "source-document"},
Requires: []string{"raw-bytes"},
}
if !reflect.DeepEqual(got, want) {
t.Fatalf("Spec() = %#v, want %#v", got, want)
@@ -91,7 +92,7 @@ func TestInputAdapterRegistryRegisterStoresDefaultSpec(t *testing.T) {
if !ok {
t.Fatal("Spec() ok = false, want true")
}
want := ModuleSpec{Key: "generic-input", Stage: StageInput}
want := ModuleSpec{Key: "generic-input", Stage: StageInput, ExecutionClass: contracts.ExecutionClassDeterministic}
if !reflect.DeepEqual(got, want) {
t.Fatalf("Spec() = %#v, want %#v", got, want)
}

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

@@ -23,6 +23,7 @@ const (
type ModuleSpec struct {
Key string
Stage ModuleStage
ExecutionClass contracts.ExecutionClass
ArtifactKind contracts.ArtifactKind
Provides []string
Requires []string
@@ -37,9 +38,14 @@ func defaultModuleSpec(key string, stage ModuleStage) ModuleSpec {
}
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,
ExecutionClass: executionClass,
ArtifactKind: normalizeArtifactKind(spec.ArtifactKind),
Provides: normalizeCapabilities(spec.Provides),
Requires: normalizeCapabilities(spec.Requires),
@@ -76,6 +82,7 @@ func cloneModuleSpec(spec ModuleSpec) ModuleSpec {
return ModuleSpec{
Key: spec.Key,
Stage: spec.Stage,
ExecutionClass: spec.ExecutionClass,
ArtifactKind: spec.ArtifactKind,
Provides: append([]string(nil), spec.Provides...),
Requires: append([]string(nil), spec.Requires...),

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

@@ -1,12 +1,25 @@
package pipeline
import (
"reflect"
"strings"
"testing"
"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)
}
cloned := cloneModuleSpec(normalized)
if !reflect.DeepEqual(cloned, normalized) {
t.Fatalf("cloneModuleSpec() = %#v, want %#v", cloned, normalized)
}
}
func TestValidateModuleSpecAllowsReferenceSlotsForEligibleStages(t *testing.T) {
tests := []struct {
name string

140
internal/framework/pipeline/profile.go
View File

@@ -156,20 +156,23 @@ type ResolvedReferenceTarget struct {
}
type ResolvedArtifactLane struct {
StepID string
ID string
ArtifactKind contracts.ArtifactKind `json:"artifact_kind,omitempty"`
ArtifactSchemaID string `json:"artifact_schema_id,omitempty"`
ArtifactSchemaName string `json:"artifact_schema_name,omitempty"`
ArtifactSchemaVersion string `json:"artifact_schema_version,omitempty"`
ArtifactSchemaDigest string `json:"artifact_schema_digest,omitempty"`
Extract ModuleBinding
Merge ModuleBinding
Normalize ModuleBinding
Validators []ModuleBinding
ExtractReferences ResolvedReferenceTarget `json:"extract_references"`
MergeReferences ResolvedReferenceTarget `json:"merge_references"`
NormalizeReferences ResolvedReferenceTarget `json:"normalize_references"`
StepID string
ID string
ArtifactKind contracts.ArtifactKind `json:"artifact_kind,omitempty"`
ArtifactSchemaID string `json:"artifact_schema_id,omitempty"`
ArtifactSchemaName string `json:"artifact_schema_name,omitempty"`
ArtifactSchemaVersion string `json:"artifact_schema_version,omitempty"`
ArtifactSchemaDigest string `json:"artifact_schema_digest,omitempty"`
Extract ModuleBinding
ExtractExecutionClass contracts.ExecutionClass `json:"extract_execution_class"`
Merge ModuleBinding
MergeExecutionClass contracts.ExecutionClass `json:"merge_execution_class"`
Normalize ModuleBinding
NormalizeExecutionClass contracts.ExecutionClass `json:"normalize_execution_class"`
Validators []ModuleBinding
ExtractReferences ResolvedReferenceTarget `json:"extract_references"`
MergeReferences ResolvedReferenceTarget `json:"merge_references"`
NormalizeReferences ResolvedReferenceTarget `json:"normalize_references"`
}
type ResolvedPipelineStep struct {
@@ -192,14 +195,17 @@ type ResolvedValidator struct {
}
type ResolvedPipeline struct {
ID string
Digest string
Input ModuleBinding
Chunk ModuleBinding
ChunkReferences ResolvedReferenceTarget `json:"chunk_references"`
Steps []ResolvedPipelineStep
ValidatorChains []ResolvedValidatorChain `json:"validator_chains"`
Output ModuleBinding
ID string
Digest string
Input ModuleBinding
InputExecutionClass contracts.ExecutionClass `json:"input_execution_class"`
Chunk ModuleBinding
ChunkExecutionClass contracts.ExecutionClass `json:"chunk_execution_class"`
ChunkReferences ResolvedReferenceTarget `json:"chunk_references"`
Steps []ResolvedPipelineStep
ValidatorChains []ResolvedValidatorChain `json:"validator_chains"`
Output ModuleBinding
OutputExecutionClass contracts.ExecutionClass `json:"output_execution_class"`
}
// AllArtifactLanes returns lanes in deterministic step order for read-only
@@ -225,6 +231,46 @@ type ModuleCatalog struct {
Outputs *OutputEncoderRegistry
}
// ExecutionClass returns the registered execution class for a module selected
// by stage and key without constructing the module.
func (catalog ModuleCatalog) ExecutionClass(stage ModuleStage, key string) (contracts.ExecutionClass, bool) {
var executionClass contracts.ExecutionClass
var ok bool
switch stage {
case StageInput:
var spec ModuleSpec
spec, ok = catalog.Inputs.Spec(key)
executionClass = spec.ExecutionClass
case StageChunk:
var spec ModuleSpec
spec, ok = catalog.Chunkers.Spec(key)
executionClass = spec.ExecutionClass
case StageExtract:
var spec ModuleSpec
spec, ok = catalog.Extractors.Spec(key)
executionClass = spec.ExecutionClass
case StageMerge:
var spec ModuleSpec
spec, ok = catalog.Mergers.Spec(key)
executionClass = spec.ExecutionClass
case StageNormalize:
var spec ModuleSpec
spec, ok = catalog.Normalizers.Spec(key)
executionClass = spec.ExecutionClass
case StageValidate:
var spec ValidatorSpec
spec, ok = catalog.Validators.Spec(key)
executionClass = spec.ExecutionClass
case StageOutput:
var spec ModuleSpec
spec, ok = catalog.Outputs.Spec(key)
executionClass = spec.ExecutionClass
}
return executionClass, ok
}
func Binding(module string) ModuleBinding {
return ModuleBinding{Module: strings.TrimSpace(module)}
}
@@ -316,11 +362,13 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo
return ResolvedPipeline{}, err
}
resolved := ResolvedPipeline{
ID: pipelineID,
Input: input,
Chunk: chunk,
ChunkReferences: referenceTarget(StageChunk, "", chunk.Module, chunkReferences),
Output: resolveBinding(profile.Output, DefaultOutputModule),
ID: pipelineID,
Input: input,
InputExecutionClass: inputModuleSpec.ExecutionClass,
Chunk: chunk,
ChunkExecutionClass: chunkSpec.ExecutionClass,
ChunkReferences: referenceTarget(StageChunk, "", chunk.Module, chunkReferences),
Output: resolveBinding(profile.Output, DefaultOutputModule),
}
chunkValidatorChain, err := resolveValidatorChain(pipelineID, "", StageChunk, chunk.Module, chunk.Validators, "", nil, catalog)
if err != nil {
@@ -386,6 +434,7 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo
if missing, ok := outputCapabilities.missing(outputSpec.Requires); ok {
return ResolvedPipeline{}, capabilityError(pipelineID, "", StageOutput, resolved.Output.Module, missing)
}
resolved.OutputExecutionClass = outputSpec.ExecutionClass
if err := validateResolvedOptions(resolved, catalog, configuredLaneIDs); err != nil {
return ResolvedPipeline{}, err
}
@@ -475,6 +524,7 @@ func resolveArtifactLane(
}
lane.ExtractReferences = referenceTarget(StageExtract, laneID, lane.Extract.Module, references)
lane.ExtractReferences.StepID = strings.TrimSpace(stepID)
lane.ExtractExecutionClass = extractSpec.ExecutionClass
capabilities.add(extractSpec.Provides...)
mergeSpec, err := mergerSpecForArtifact(catalog, lane.Merge.Module, lane.ArtifactKind, artifactType)
@@ -499,6 +549,7 @@ func resolveArtifactLane(
}
lane.MergeReferences = referenceTarget(StageMerge, laneID, lane.Merge.Module, mergeReferences)
lane.MergeReferences.StepID = strings.TrimSpace(stepID)
lane.MergeExecutionClass = mergeSpec.ExecutionClass
capabilities.add(mergeSpec.Provides...)
normalizeSpec, err := normalizerSpecForArtifact(catalog, lane.Normalize.Module, lane.ArtifactKind, artifactType)
@@ -523,6 +574,7 @@ func resolveArtifactLane(
}
lane.NormalizeReferences = referenceTarget(StageNormalize, laneID, lane.Normalize.Module, normalizeReferences)
lane.NormalizeReferences.StepID = strings.TrimSpace(stepID)
lane.NormalizeExecutionClass = normalizeSpec.ExecutionClass
capabilities.add(normalizeSpec.Provides...)
if len(lane.Validators) > 0 {
@@ -1307,21 +1359,27 @@ func selectedArtifactLanes(pipelineID string, artifacts map[string]ArtifactLaneP
func resolvedPipelineDigest(resolved ResolvedPipeline) (string, error) {
withoutDigest := struct {
ID string
Input ModuleBinding
Chunk ModuleBinding
ChunkReferences ResolvedReferenceTarget
Steps []ResolvedPipelineStep
ValidatorChains []ResolvedValidatorChain
Output ModuleBinding
ID string
Input ModuleBinding
InputExecutionClass contracts.ExecutionClass
Chunk ModuleBinding
ChunkExecutionClass contracts.ExecutionClass
ChunkReferences ResolvedReferenceTarget
Steps []ResolvedPipelineStep
ValidatorChains []ResolvedValidatorChain
Output ModuleBinding
OutputExecutionClass contracts.ExecutionClass
}{
ID: resolved.ID,
Input: resolved.Input,
Chunk: resolved.Chunk,
ChunkReferences: resolved.ChunkReferences,
Steps: resolved.Steps,
ValidatorChains: resolved.ValidatorChains,
Output: resolved.Output,
ID: resolved.ID,
Input: resolved.Input,
InputExecutionClass: resolved.InputExecutionClass,
Chunk: resolved.Chunk,
ChunkExecutionClass: resolved.ChunkExecutionClass,
ChunkReferences: resolved.ChunkReferences,
Steps: resolved.Steps,
ValidatorChains: resolved.ValidatorChains,
Output: resolved.Output,
OutputExecutionClass: resolved.OutputExecutionClass,
}
encoded, err := json.Marshal(withoutDigest)
if err != nil {

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

@@ -47,12 +47,18 @@ func TestResolvePipelineWithExplicitModules(t *testing.T) {
if !reflect.DeepEqual(resolved.Input, ModuleBinding{Module: "text", LLMProfile: "fast"}) {
t.Fatalf("Input = %#v, want trimmed explicit input", resolved.Input)
}
if resolved.InputExecutionClass != contracts.ExecutionClassDeterministic {
t.Fatalf("InputExecutionClass = %q, want deterministic", resolved.InputExecutionClass)
}
if resolved.Chunk.Module != "window" || resolved.Chunk.LLMProfile != "" {
t.Fatalf("Chunk = %#v, want explicit module and empty LLM profile", resolved.Chunk)
}
if resolved.Chunk.Options["size"] != 10 {
t.Fatalf("Chunk.Options = %#v, want size option", resolved.Chunk.Options)
}
if resolved.ChunkExecutionClass != contracts.ExecutionClassDeterministic {
t.Fatalf("ChunkExecutionClass = %q, want deterministic", resolved.ChunkExecutionClass)
}
if len(resolved.Steps) != 1 || resolved.Steps[0].ID != "default" || len(resolved.Steps[0].ArtifactLanes) != 1 {
t.Fatalf("resolved steps = %#v, want one default step with one lane", resolved.Steps)
}
@@ -63,6 +69,9 @@ func TestResolvePipelineWithExplicitModules(t *testing.T) {
if !reflect.DeepEqual(lane.Extract, ModuleBinding{Module: "record-extractor", LLMProfile: "careful"}) {
t.Fatalf("lane.Extract = %#v, want explicit extractor", lane.Extract)
}
if lane.ExtractExecutionClass != contracts.ExecutionClassDeterministic || lane.MergeExecutionClass != contracts.ExecutionClassDeterministic || lane.NormalizeExecutionClass != contracts.ExecutionClassDeterministic {
t.Fatalf("lane execution classes = %q/%q/%q, want deterministic", lane.ExtractExecutionClass, lane.MergeExecutionClass, lane.NormalizeExecutionClass)
}
if lane.Merge.Module != "dedupe" || lane.Normalize.Module != "canonical" {
t.Fatalf("lane merge/normalize = %#v/%#v, want explicit modules", lane.Merge, lane.Normalize)
}
@@ -72,11 +81,51 @@ func TestResolvePipelineWithExplicitModules(t *testing.T) {
if resolved.Output.Module != "ndjson" {
t.Fatalf("Output.Module = %q, want ndjson", resolved.Output.Module)
}
if resolved.OutputExecutionClass != contracts.ExecutionClassDeterministic {
t.Fatalf("OutputExecutionClass = %q, want deterministic", resolved.OutputExecutionClass)
}
if !strings.HasPrefix(resolved.Digest, "sha256:") {
t.Fatalf("Digest = %q, want sha256 digest", resolved.Digest)
}
}
func TestModuleCatalogExecutionClassLooksUpRegisteredMetadata(t *testing.T) {
catalog := newProfileCatalogWithOverrides(t,
ModuleSpec{Key: "llm-input", Stage: StageInput, ExecutionClass: contracts.ExecutionClassLLMBacked},
ModuleSpec{Key: "llm-chunk", Stage: StageChunk, ExecutionClass: contracts.ExecutionClassLLMBacked},
ModuleSpec{Key: "llm-extract", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassLLMBacked},
ModuleSpec{Key: "llm-merge", Stage: StageMerge, ExecutionClass: contracts.ExecutionClassLLMBacked},
ModuleSpec{Key: "llm-normalize", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked},
ModuleSpec{Key: "llm-output", Stage: StageOutput, ExecutionClass: contracts.ExecutionClassLLMBacked},
)
registerProfileValidatorSpec(t, catalog, ValidatorSpec{Key: "llm-validator", ExecutionClass: contracts.ExecutionClassLLMBacked})
for _, test := range []struct {
stage ModuleStage
key string
want contracts.ExecutionClass
}{
{stage: StageInput, key: "llm-input", want: contracts.ExecutionClassLLMBacked},
{stage: StageChunk, key: "llm-chunk", want: contracts.ExecutionClassLLMBacked},
{stage: StageExtract, key: "llm-extract", want: contracts.ExecutionClassLLMBacked},
{stage: StageMerge, key: "llm-merge", want: contracts.ExecutionClassLLMBacked},
{stage: StageNormalize, key: "llm-normalize", want: contracts.ExecutionClassLLMBacked},
{stage: StageValidate, key: "llm-validator", want: contracts.ExecutionClassLLMBacked},
{stage: StageOutput, key: "llm-output", want: contracts.ExecutionClassLLMBacked},
} {
t.Run(string(test.stage), func(t *testing.T) {
got, ok := catalog.ExecutionClass(test.stage, test.key)
if !ok || got != test.want {
t.Fatalf("ExecutionClass(%q, %q) = %q, %t; want %q, true", test.stage, test.key, got, ok, test.want)
}
})
}
if _, ok := catalog.ExecutionClass(StageExtract, "missing"); ok {
t.Fatal("ExecutionClass() found an unregistered module")
}
}
func TestResolvePipelineAppliesDefaults(t *testing.T) {
resolved, err := ResolvePipeline(PipelineProfile{
ID: "defaulted",
@@ -1566,7 +1615,13 @@ func TestResolvedPipelineCanMarshalToCanonicalJSON(t *testing.T) {
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
if _, err := json.Marshal(resolved); err != nil {
encoded, err := json.Marshal(resolved)
if err != nil {
t.Fatalf("json.Marshal(resolved) error = %v, want nil", err)
}
for _, field := range []string{"input_execution_class", "chunk_execution_class", "extract_execution_class", "merge_execution_class", "normalize_execution_class", "output_execution_class"} {
if !strings.Contains(string(encoded), `"`+field+`":"deterministic"`) {
t.Fatalf("resolved JSON does not retain %q: %s", field, encoded)
}
}
}