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

Add type-safe artifact lane resolution

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-17 05:57:31 +00:00
parent fc1b57bde2
commit 1c84d19e5f
51 changed files with 1402 additions and 338 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -5,6 +5,7 @@ import (
"encoding/hex"
"encoding/json"
"fmt"
"reflect"
"sort"
"strings"
@@ -102,14 +103,19 @@ type ResolvedReferenceTarget struct {
}
type ResolvedArtifactLane struct {
ID string
Extract ModuleBinding
Merge ModuleBinding
Normalize ModuleBinding
Validators []ModuleBinding
ExtractReferences ResolvedReferenceTarget `json:"extract_references"`
MergeReferences ResolvedReferenceTarget `json:"merge_references"`
NormalizeReferences ResolvedReferenceTarget `json:"normalize_references"`
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"`
}
type ResolvedValidatorChain struct {
@@ -122,6 +128,8 @@ type ResolvedValidatorChain struct {
type ResolvedValidator struct {
Binding ModuleBinding `json:"binding"`
ExecutionClass contracts.ExecutionClass `json:"execution_class"`
Target ValidatorTarget `json:"target,omitempty"`
ArtifactKind contracts.ArtifactKind `json:"artifact_kind,omitempty"`
}
type ResolvedPipeline struct {
@@ -216,7 +224,7 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo
ChunkReferences: referenceTarget(StageChunk, "", chunk.Module, chunkReferences),
Output: resolveBinding(profile.Output, DefaultOutputModule),
}
chunkValidatorChain, err := resolveValidatorChain(pipelineID, "", StageChunk, chunk.Module, chunk.Validators, catalog)
chunkValidatorChain, err := resolveValidatorChain(pipelineID, "", StageChunk, chunk.Module, chunk.Validators, "", nil, catalog)
if err != nil {
return ResolvedPipeline{}, err
}
@@ -279,6 +287,10 @@ func resolveArtifactLane(
if missing, ok := capabilities.missing(extractSpec.Requires); ok {
return ResolvedArtifactLane{}, nil, nil, capabilityError(pipelineID, laneID, StageExtract, lane.Extract.Module, missing)
}
artifactType, err := resolveArtifactIdentity(pipelineID, laneID, &lane, extractSpec, catalog)
if err != nil {
return ResolvedArtifactLane{}, nil, nil, err
}
extractReferences := mergeReferenceMaps(profile.References, lane.Extract.References)
references, err := resolveReferenceTargetBindings(referenceResolutionTarget{
PipelineID: pipelineID,
@@ -296,7 +308,7 @@ func resolveArtifactLane(
lane.ExtractReferences = referenceTarget(StageExtract, laneID, lane.Extract.Module, references)
capabilities.add(extractSpec.Provides...)
mergeSpec, err := mergerSpec(catalog, lane.Merge.Module)
mergeSpec, err := mergerSpecForArtifact(catalog, lane.Merge.Module, lane.ArtifactKind, artifactType)
if err != nil {
return ResolvedArtifactLane{}, nil, nil, moduleLookupError(pipelineID, laneID, StageMerge, lane.Merge.Module, err)
}
@@ -319,7 +331,7 @@ func resolveArtifactLane(
lane.MergeReferences = referenceTarget(StageMerge, laneID, lane.Merge.Module, mergeReferences)
capabilities.add(mergeSpec.Provides...)
normalizeSpec, err := normalizerSpec(catalog, lane.Normalize.Module)
normalizeSpec, err := normalizerSpecForArtifact(catalog, lane.Normalize.Module, lane.ArtifactKind, artifactType)
if err != nil {
return ResolvedArtifactLane{}, nil, nil, moduleLookupError(pipelineID, laneID, StageNormalize, lane.Normalize.Module, err)
}
@@ -346,15 +358,15 @@ func resolveArtifactLane(
return ResolvedArtifactLane{}, nil, nil, configuredValidatorsError(pipelineID, laneID)
}
extractValidatorChain, err := resolveValidatorChain(pipelineID, laneID, StageExtract, lane.Extract.Module, lane.Extract.Validators, catalog)
extractValidatorChain, err := resolveValidatorChain(pipelineID, laneID, StageExtract, lane.Extract.Module, lane.Extract.Validators, lane.ArtifactKind, artifactType, catalog)
if err != nil {
return ResolvedArtifactLane{}, nil, nil, err
}
mergeValidatorChain, err := resolveValidatorChain(pipelineID, laneID, StageMerge, lane.Merge.Module, lane.Merge.Validators, catalog)
mergeValidatorChain, err := resolveValidatorChain(pipelineID, laneID, StageMerge, lane.Merge.Module, lane.Merge.Validators, lane.ArtifactKind, artifactType, catalog)
if err != nil {
return ResolvedArtifactLane{}, nil, nil, err
}
normalizeValidatorChain, err := resolveValidatorChain(pipelineID, laneID, StageNormalize, lane.Normalize.Module, lane.Normalize.Validators, catalog)
normalizeValidatorChain, err := resolveValidatorChain(pipelineID, laneID, StageNormalize, lane.Normalize.Module, lane.Normalize.Validators, lane.ArtifactKind, artifactType, catalog)
if err != nil {
return ResolvedArtifactLane{}, nil, nil, err
}
@@ -367,7 +379,128 @@ func configuredValidatorsError(pipelineID string, laneID string) error {
return fmt.Errorf("pipeline %q lane %q validators are not supported at artifact lane level; use extract.validators, merge.validators, or normalize.validators", pipelineID, laneID)
}
func resolveValidatorChain(pipelineID string, laneID string, stage ModuleStage, module string, override ValidatorOverride, catalog ModuleCatalog) (ResolvedValidatorChain, error) {
func resolveArtifactIdentity(pipelineID, laneID string, lane *ResolvedArtifactLane, extractSpec ModuleSpec, catalog ModuleCatalog) (reflect.Type, error) {
if extractSpec.ArtifactKind == "" {
return nil, nil
}
if catalog.Extractors == nil {
return nil, fmt.Errorf("pipeline %q lane %q extractor registry must not be nil", pipelineID, laneID)
}
extractor, ok := catalog.Extractors.typedEntry(lane.Extract.Module)
if !ok {
return nil, fmt.Errorf("pipeline %q lane %q extract module %q declares artifact kind %q without a typed registration", pipelineID, laneID, lane.Extract.Module, extractSpec.ArtifactKind)
}
if catalog.ArtifactCodecs == nil {
return nil, fmt.Errorf("pipeline %q lane %q artifact codec registry must not be nil for kind %q", pipelineID, laneID, extractSpec.ArtifactKind)
}
codecSpec, ok := catalog.ArtifactCodecs.Spec(extractSpec.ArtifactKind)
if !ok {
return nil, fmt.Errorf("pipeline %q lane %q artifact codec %q is not registered", pipelineID, laneID, extractSpec.ArtifactKind)
}
codecType, ok := catalog.ArtifactCodecs.valueType(extractSpec.ArtifactKind)
if !ok {
return nil, fmt.Errorf("pipeline %q lane %q artifact codec %q has no Go type", pipelineID, laneID, extractSpec.ArtifactKind)
}
if extractor.valueType != codecType {
return nil, artifactTypeMismatchError(pipelineID, laneID, StageExtract, lane.Extract.Module, extractSpec.ArtifactKind, codecType, extractor.valueType)
}
lane.ArtifactKind = codecSpec.Kind
lane.ArtifactSchemaID = codecSpec.Schema.ID
lane.ArtifactSchemaName = codecSpec.Schema.Name
lane.ArtifactSchemaVersion = codecSpec.Schema.Version
lane.ArtifactSchemaDigest = codecSpec.SchemaDigest
return codecType, nil
}
func mergerSpecForArtifact(catalog ModuleCatalog, key string, kind contracts.ArtifactKind, expectedType reflect.Type) (ModuleSpec, error) {
if kind == "" {
return mergerSpec(catalog, key)
}
if catalog.Mergers == nil {
return ModuleSpec{}, fmt.Errorf("module %q is not registered", key)
}
entry, ok := catalog.Mergers.typedEntry(key, kind)
if !ok {
return ModuleSpec{}, missingArtifactVariantError("merger", key, kind, catalog.Mergers.registeredKinds(key))
}
if entry.valueType != expectedType {
return ModuleSpec{}, fmt.Errorf("artifact kind %q requires Go type %s, but merger %q variant uses %s", kind, typeName(expectedType), key, typeName(entry.valueType))
}
return cloneModuleSpec(entry.spec), nil
}
func normalizerSpecForArtifact(catalog ModuleCatalog, key string, kind contracts.ArtifactKind, expectedType reflect.Type) (ModuleSpec, error) {
if kind == "" {
return normalizerSpec(catalog, key)
}
if catalog.Normalizers == nil {
return ModuleSpec{}, fmt.Errorf("module %q is not registered", key)
}
entry, ok := catalog.Normalizers.typedEntry(key, kind)
if !ok {
return ModuleSpec{}, missingArtifactVariantError("normalizer", key, kind, catalog.Normalizers.registeredKinds(key))
}
if entry.valueType != expectedType {
return ModuleSpec{}, fmt.Errorf("artifact kind %q requires Go type %s, but normalizer %q variant uses %s", kind, typeName(expectedType), key, typeName(entry.valueType))
}
return cloneModuleSpec(entry.spec), nil
}
func validatorSpecForTarget(registry *ValidatorRegistry, stage ModuleStage, key string, kind contracts.ArtifactKind, expectedType reflect.Type) (ValidatorSpec, ValidatorTarget, error) {
key = strings.TrimSpace(key)
if stage == StageChunk {
if entry, ok := registry.chunkEntry(key); ok {
return entry.spec, ValidatorTargetChunk, nil
}
if entry, ok := registry.serializedEntry(key); ok && entry.spec.SupportsChunks {
return entry.spec.ValidatorSpec, ValidatorTargetSerialized, nil
}
if spec, ok := registry.Spec(key); ok {
return spec, "", nil
}
return ValidatorSpec{}, "", fmt.Errorf("references unknown validator %q for chunk target", key)
}
if kind == "" {
if spec, ok := registry.Spec(key); ok {
return spec, "", nil
}
return ValidatorSpec{}, "", fmt.Errorf("references unknown validator %q on legacy raw path", key)
}
if entry, ok := registry.typedEntry(key, kind); ok {
if entry.valueType != expectedType {
return ValidatorSpec{}, "", fmt.Errorf("artifact kind %q requires Go type %s, but validator %q variant uses %s", kind, typeName(expectedType), key, typeName(entry.valueType))
}
return entry.spec, ValidatorTargetTyped, nil
}
if entry, ok := registry.serializedEntry(key); ok && entry.spec.SupportsArtifacts {
return entry.spec.ValidatorSpec, ValidatorTargetSerialized, nil
}
return ValidatorSpec{}, "", missingArtifactVariantError("validator", key, kind, registry.registeredTypedKinds(key))
}
func missingArtifactVariantError(moduleType, key string, kind contracts.ArtifactKind, registered []contracts.ArtifactKind) error {
if len(registered) == 0 {
return fmt.Errorf("%s %q has no typed variant for artifact kind %q", moduleType, key, kind)
}
values := make([]string, len(registered))
for i, value := range registered {
values[i] = string(value)
}
return fmt.Errorf("%s %q has no typed variant for artifact kind %q; registered kinds: %s", moduleType, key, kind, strings.Join(values, ", "))
}
func artifactTypeMismatchError(pipelineID, laneID string, stage ModuleStage, module string, kind contracts.ArtifactKind, expected, actual reflect.Type) error {
return fmt.Errorf("pipeline %q lane %q %s module %q artifact kind %q requires Go type %s, got %s", pipelineID, laneID, stage, module, kind, typeName(expected), typeName(actual))
}
func typeName(value reflect.Type) string {
if value == nil {
return "<nil>"
}
return value.String()
}
func resolveValidatorChain(pipelineID string, laneID string, stage ModuleStage, module string, override ValidatorOverride, artifactKind contracts.ArtifactKind, artifactType reflect.Type, catalog ModuleCatalog) (ResolvedValidatorChain, error) {
chain := ResolvedValidatorChain{
Stage: stage,
LaneID: strings.TrimSpace(laneID),
@@ -396,9 +529,9 @@ func resolveValidatorChain(pipelineID string, laneID string, stage ModuleStage,
}
chain.Validators = make([]ResolvedValidator, 0, len(bindings))
for _, validator := range bindings {
spec, ok := catalog.Validators.Spec(validator.Module)
if !ok {
return ResolvedValidatorChain{}, fmt.Errorf("pipeline %q %s validator chain for module %q references unknown validator %q", pipelineID, stage, chain.ModuleKey, validator.Module)
spec, target, err := validatorSpecForTarget(catalog.Validators, stage, validator.Module, artifactKind, artifactType)
if err != nil {
return ResolvedValidatorChain{}, fmt.Errorf("pipeline %q %s validator chain for module %q: %w", pipelineID, stage, chain.ModuleKey, err)
}
if strings.TrimSpace(validator.LLMProfile) != "" && spec.ExecutionClass != contracts.ExecutionClassLLMBacked {
return ResolvedValidatorChain{}, fmt.Errorf("pipeline %q %s validator chain for module %q assigns llm_profile to deterministic validator %q", pipelineID, stage, chain.ModuleKey, validator.Module)
@@ -406,6 +539,8 @@ func resolveValidatorChain(pipelineID string, laneID string, stage ModuleStage,
chain.Validators = append(chain.Validators, ResolvedValidator{
Binding: cloneModuleBinding(validator),
ExecutionClass: spec.ExecutionClass,
Target: target,
ArtifactKind: artifactKind,
})
}
return chain, nil
@@ -436,6 +571,8 @@ func cloneResolvedValidators(validators []ResolvedValidator) []ResolvedValidator
out[i] = ResolvedValidator{
Binding: cloneModuleBinding(validator.Binding),
ExecutionClass: validator.ExecutionClass,
Target: validator.Target,
ArtifactKind: validator.ArtifactKind,
}
}
return out
@@ -498,7 +635,13 @@ func validatePipelineReferenceDefaults(
}
merge := resolveBinding(laneProfile.Merge, DefaultMergeModule)
mergeSpec, err := mergerSpec(catalog, merge.Module)
var artifactType reflect.Type
if extractSpec.ArtifactKind != "" && catalog.Extractors != nil {
if entry, ok := catalog.Extractors.typedEntry(extract.Module); ok {
artifactType = entry.valueType
}
}
mergeSpec, err := mergerSpecForArtifact(catalog, merge.Module, extractSpec.ArtifactKind, artifactType)
if err != nil {
return moduleLookupError(pipelineID, laneID, StageMerge, merge.Module, err)
}
@@ -507,7 +650,7 @@ func validatePipelineReferenceDefaults(
}
normalize := resolveBinding(laneProfile.Normalize, DefaultNormalizeModule)
normalizeSpec, err := normalizerSpec(catalog, normalize.Module)
normalizeSpec, err := normalizerSpecForArtifact(catalog, normalize.Module, extractSpec.ArtifactKind, artifactType)
if err != nil {
return moduleLookupError(pipelineID, laneID, StageNormalize, normalize.Module, err)
}