2026-07-03 15:37:07 +00:00
package pipeline
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
2026-07-17 05:57:31 +00:00
"reflect"
2026-07-03 15:37:07 +00:00
"sort"
"strings"
2026-07-05 14:21:36 +00:00
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
2026-07-03 15:37:07 +00:00
)
const (
DefaultChunkModule = "generic"
DefaultMergeModule = "appendorder"
DefaultNormalizeModule = "noop"
DefaultOutputModule = "json"
DefaultLLMProfile = "default"
)
type ModuleBinding struct {
2026-08-25 19:39:04 +00:00
Module string ` json:"module" `
LLMProfile string ` json:"llm_profile,omitempty" `
StructuredOutputRepairAttempts * int ` json:"structured_output_repair_attempts,omitempty" `
2026-08-26 23:34:06 +00:00
ValidationPolicy * ValidationPolicyOverride ` json:"validation_policy,omitempty" `
2026-08-25 19:39:04 +00:00
Retries int ` json:"retries,omitempty" `
Options map [ string ] any ` json:"options,omitempty" `
References map [ string ] ReferenceSource ` json:"references,omitempty" `
Validators ValidatorOverride ` json:"validators,omitempty" `
2026-07-21 21:05:28 +00:00
}
// ArtifactReference identifies a normalized artifact produced by an earlier
// ordered step. It is a selector only; it does not contain artifact bytes.
type ArtifactReference struct {
Step string ` json:"step" `
Lane string ` json:"lane" `
}
// ReferenceSource is the discriminated source form used by configured
// reference maps. Exactly one of Path and Artifact is set after resolution.
type ReferenceSource struct {
Path string ` json:"path,omitempty" `
Artifact * ArtifactReference ` json:"artifact,omitempty" `
}
func ExternalReference ( path string ) ReferenceSource {
return ReferenceSource { Path : strings . TrimSpace ( path ) }
}
func ExternalReferenceMap ( values map [ string ] string ) map [ string ] ReferenceSource {
out := make ( map [ string ] ReferenceSource , len ( values ) )
for key , value := range values {
out [ key ] = ExternalReference ( value )
}
return out
}
func GeneratedReference ( step , lane string ) ReferenceSource {
return ReferenceSource { Artifact : & ArtifactReference { Step : strings . TrimSpace ( step ) , Lane : strings . TrimSpace ( lane ) } }
}
func ( source ReferenceSource ) IsGenerated ( ) bool { return source . Artifact != nil }
func ( source ReferenceSource ) MarshalJSON ( ) ( [ ] byte , error ) {
if source . Path != "" && source . Artifact != nil {
return nil , fmt . Errorf ( "reference source must not contain both an external path and an artifact selector" )
}
if source . Artifact != nil {
return json . Marshal ( struct {
Artifact * ArtifactReference ` json:"artifact" `
} { Artifact : source . Artifact } )
}
return json . Marshal ( source . Path )
2026-07-07 21:21:12 +00:00
}
type ValidatorOverride struct {
Set bool ` json:"set,omitempty" `
Validators [ ] ModuleBinding ` json:"validators,omitempty" `
}
func ( binding ModuleBinding ) MarshalJSON ( ) ( [ ] byte , error ) {
type moduleBindingJSON struct {
2026-08-25 19:47:42 +00:00
Module string ` json:"module" `
LLMProfile string ` json:"llm_profile,omitempty" `
StructuredOutputRepairAttempts * int ` json:"structured_output_repair_attempts,omitempty" `
2026-08-26 23:34:06 +00:00
ValidationPolicy * ValidationPolicyOverride ` json:"validation_policy,omitempty" `
2026-08-25 19:47:42 +00:00
Retries int ` json:"retries,omitempty" `
Options map [ string ] any ` json:"options,omitempty" `
References map [ string ] ReferenceSource ` json:"references,omitempty" `
Validators * [ ] ModuleBinding ` json:"validators,omitempty" `
2026-07-07 21:21:12 +00:00
}
out := moduleBindingJSON {
2026-08-25 19:47:42 +00:00
Module : binding . Module ,
LLMProfile : binding . LLMProfile ,
StructuredOutputRepairAttempts : binding . StructuredOutputRepairAttempts ,
2026-08-26 23:34:06 +00:00
ValidationPolicy : cloneValidationPolicyOverride ( binding . ValidationPolicy ) ,
2026-08-25 19:47:42 +00:00
Retries : binding . Retries ,
Options : binding . Options ,
References : binding . References ,
2026-07-07 21:21:12 +00:00
}
if binding . Validators . Set {
validators := cloneModuleBindings ( binding . Validators . Validators )
out . Validators = & validators
}
return json . Marshal ( out )
2026-07-03 15:37:07 +00:00
}
2026-08-25 19:47:42 +00:00
func ( binding * ModuleBinding ) UnmarshalJSON ( data [ ] byte ) error {
if binding == nil {
return fmt . Errorf ( "module binding must not be nil" )
}
type moduleBindingJSON struct {
Module string ` json:"module" `
LLMProfile string ` json:"llm_profile,omitempty" `
StructuredOutputRepairAttempts * int ` json:"structured_output_repair_attempts,omitempty" `
2026-08-26 23:34:06 +00:00
ValidationPolicy * ValidationPolicyOverride ` json:"validation_policy,omitempty" `
2026-08-25 19:47:42 +00:00
Retries int ` json:"retries,omitempty" `
Options map [ string ] any ` json:"options,omitempty" `
References map [ string ] ReferenceSource ` json:"references,omitempty" `
Validators json . RawMessage ` json:"validators,omitempty" `
}
var decoded moduleBindingJSON
if err := json . Unmarshal ( data , & decoded ) ; err != nil {
return err
}
out := ModuleBinding {
Module : decoded . Module ,
LLMProfile : decoded . LLMProfile ,
StructuredOutputRepairAttempts : decoded . StructuredOutputRepairAttempts ,
2026-08-26 23:34:06 +00:00
ValidationPolicy : cloneValidationPolicyOverride ( decoded . ValidationPolicy ) ,
2026-08-25 19:47:42 +00:00
Retries : decoded . Retries ,
Options : decoded . Options ,
References : decoded . References ,
}
if len ( decoded . Validators ) > 0 && string ( decoded . Validators ) != "null" {
var validators [ ] ModuleBinding
if err := json . Unmarshal ( decoded . Validators , & validators ) ; err != nil {
return fmt . Errorf ( "decode module binding validators: %w" , err )
}
out . Validators = ValidatorOverride { Set : true , Validators : validators }
}
* binding = cloneModuleBinding ( out )
return nil
}
2026-07-03 15:37:07 +00:00
type ArtifactLaneProfile struct {
2026-07-21 21:05:28 +00:00
Extract ModuleBinding ` json:"extract" `
Merge ModuleBinding ` json:"merge,omitempty" `
Normalize ModuleBinding ` json:"normalize,omitempty" `
Validators [ ] ModuleBinding ` json:"validators,omitempty" `
References map [ string ] ReferenceSource ` json:"references,omitempty" `
}
type PipelineStepProfile struct {
ID string ` json:"id" `
Artifacts map [ string ] ArtifactLaneProfile ` json:"artifacts" `
References map [ string ] ReferenceSource ` json:"references,omitempty" `
2026-07-03 15:37:07 +00:00
}
type PipelineProfile struct {
2026-08-25 19:47:42 +00:00
ID string ` json:"id" `
LLMProfile string ` json:"llm_profile,omitempty" `
StructuredOutputRepairAttempts * int ` json:"structured_output_repair_attempts,omitempty" `
2026-08-26 23:34:06 +00:00
ValidationPolicy * ValidationPolicyOverride ` json:"validation_policy,omitempty" `
2026-08-25 19:47:42 +00:00
Input ModuleBinding ` json:"input" `
Chunk ModuleBinding ` json:"chunk,omitempty" `
Artifacts map [ string ] ArtifactLaneProfile ` json:"artifacts" `
Steps [ ] PipelineStepProfile ` json:"steps,omitempty" `
Output ModuleBinding ` json:"output,omitempty" `
References map [ string ] ReferenceSource ` json:"references,omitempty" `
2026-07-03 15:37:07 +00:00
}
type ResolveOptions struct {
2026-07-05 14:21:36 +00:00
Only [ ] string
2026-08-03 17:09:16 +00:00
LLMProfileOverride string
2026-07-05 14:21:36 +00:00
ReferenceOverrides [ ] ReferenceBinding
ReferenceUnbinds [ ] ReferenceUnbind
}
type ReferenceBinding struct {
2026-07-21 21:05:28 +00:00
Stage ModuleStage ` json:"stage,omitempty" `
LaneID string ` json:"lane_id,omitempty" `
SlotName string ` json:"slot_name" `
Source string ` json:"source,omitempty" `
Artifact * ArtifactReference ` json:"artifact,omitempty" `
BindingSource string ` json:"binding_source,omitempty" `
2026-07-05 14:21:36 +00:00
}
type ReferenceUnbind struct {
2026-07-05 16:31:23 +00:00
Stage ModuleStage ` json:"stage,omitempty" `
LaneID string ` json:"lane_id,omitempty" `
SlotName string ` json:"slot_name" `
2026-07-03 15:37:07 +00:00
}
2026-07-05 16:12:38 +00:00
type ResolvedReferenceTarget struct {
Stage ModuleStage ` json:"stage" `
2026-07-21 21:05:28 +00:00
StepID string ` json:"step_id,omitempty" `
2026-07-05 16:12:38 +00:00
LaneID string ` json:"lane_id,omitempty" `
Module string ` json:"module" `
Bindings [ ] ReferenceBinding ` json:"bindings,omitempty" `
2026-07-05 14:32:35 +00:00
ReferenceSet contracts . ReferenceSet ` json:"-" `
2026-07-03 15:37:07 +00:00
}
2026-07-05 16:12:38 +00:00
type ResolvedArtifactLane struct {
2026-08-26 23:42:11 +00:00
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" `
ExtractCorrectionProtocol contracts . CorrectionProtocol ` json:"extract_correction_protocol,omitempty" `
ExtractValidationPolicy ValidationPolicy ` json:"extract_validation_policy" `
Merge ModuleBinding
MergeExecutionClass contracts . ExecutionClass ` json:"merge_execution_class" `
MergeCorrectionProtocol contracts . CorrectionProtocol ` json:"merge_correction_protocol,omitempty" `
MergeValidationPolicy ValidationPolicy ` json:"merge_validation_policy" `
Normalize ModuleBinding
NormalizeExecutionClass contracts . ExecutionClass ` json:"normalize_execution_class" `
NormalizeCorrectionProtocol contracts . CorrectionProtocol ` json:"normalize_correction_protocol,omitempty" `
NormalizeValidationPolicy ValidationPolicy ` json:"normalize_validation_policy" `
Validators [ ] ModuleBinding
ExtractReferences ResolvedReferenceTarget ` json:"extract_references" `
MergeReferences ResolvedReferenceTarget ` json:"merge_references" `
NormalizeReferences ResolvedReferenceTarget ` json:"normalize_references" `
2026-07-05 16:12:38 +00:00
}
2026-07-21 21:05:28 +00:00
type ResolvedPipelineStep struct {
ID string ` json:"id" `
ArtifactLanes [ ] ResolvedArtifactLane ` json:"artifact_lanes" `
}
2026-07-07 21:10:29 +00:00
type ResolvedValidatorChain struct {
Stage ModuleStage ` json:"stage" `
LaneID string ` json:"lane_id,omitempty" `
ModuleKey string ` json:"module_key" `
Validators [ ] ResolvedValidator ` json:"validators" `
}
type ResolvedValidator struct {
Binding ModuleBinding ` json:"binding" `
ExecutionClass contracts . ExecutionClass ` json:"execution_class" `
2026-07-17 05:57:31 +00:00
Target ValidatorTarget ` json:"target,omitempty" `
ArtifactKind contracts . ArtifactKind ` json:"artifact_kind,omitempty" `
2026-07-07 21:10:29 +00:00
}
2026-07-03 15:37:07 +00:00
type ResolvedPipeline struct {
2026-08-26 23:34:06 +00:00
ID string
Digest string
ConfiguredValidationPolicy * ValidationPolicyOverride ` json:"configured_validation_policy,omitempty" `
Input ModuleBinding
InputExecutionClass contracts . ExecutionClass ` json:"input_execution_class" `
Chunk ModuleBinding
2026-08-26 23:42:11 +00:00
ChunkExecutionClass contracts . ExecutionClass ` json:"chunk_execution_class" `
ChunkCorrectionProtocol contracts . CorrectionProtocol ` json:"chunk_correction_protocol,omitempty" `
ChunkValidationPolicy ValidationPolicy ` json:"chunk_validation_policy" `
ChunkReferences ResolvedReferenceTarget ` json:"chunk_references" `
2026-08-26 23:34:06 +00:00
Steps [ ] ResolvedPipelineStep
ValidatorChains [ ] ResolvedValidatorChain ` json:"validator_chains" `
Output ModuleBinding
OutputExecutionClass contracts . ExecutionClass ` json:"output_execution_class" `
2026-07-03 15:37:07 +00:00
}
2026-07-21 21:05:28 +00:00
// AllArtifactLanes returns lanes in deterministic step order for read-only
// discovery and identity construction.
func ( resolved ResolvedPipeline ) AllArtifactLanes ( ) [ ] ResolvedArtifactLane {
var lanes [ ] ResolvedArtifactLane
for _ , step := range resolved . Steps {
lanes = append ( lanes , step . ArtifactLanes ... )
}
return lanes
}
2026-07-03 15:37:07 +00:00
type ModuleCatalog struct {
2026-07-27 18:04:12 +00:00
Inputs * InputAdapterRegistry
Chunkers * ChunkerRegistry
ArtifactCodecs * ArtifactCodecRegistry
ArtifactEvidence * ArtifactEvidenceRegistry
Extractors * ExtractorRegistry
Mergers * MergerRegistry
Normalizers * NormalizerRegistry
Validators * ValidatorRegistry
ValidatorChains * ValidatorChainRegistry
Outputs * OutputEncoderRegistry
2026-07-03 15:37:07 +00:00
}
2026-08-03 16:49:51 +00:00
// 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
}
2026-07-03 15:37:07 +00:00
func Binding ( module string ) ModuleBinding {
return ModuleBinding { Module : strings . TrimSpace ( module ) }
}
func ResolvePipeline ( profile PipelineProfile , options ResolveOptions , catalog ModuleCatalog ) ( ResolvedPipeline , error ) {
pipelineID := strings . TrimSpace ( profile . ID )
if pipelineID == "" {
return ResolvedPipeline { } , fmt . Errorf ( "pipeline id must not be empty" )
}
2026-07-21 21:05:28 +00:00
explicitSteps := profile . Steps != nil
if len ( profile . Artifacts ) > 0 && explicitSteps {
return ResolvedPipeline { } , fmt . Errorf ( "pipeline %q must not declare both artifacts and steps" , pipelineID )
}
if explicitSteps && len ( options . Only ) > 0 {
return ResolvedPipeline { } , fmt . Errorf ( "pipeline %q --only is not supported for explicit ordered steps" , pipelineID )
}
steps := profile . Steps
if ! explicitSteps {
steps = [ ] PipelineStepProfile { { ID : "default" , Artifacts : profile . Artifacts } }
}
if explicitSteps && len ( steps ) == 0 {
return ResolvedPipeline { } , fmt . Errorf ( "pipeline %q must declare at least one ordered step" , pipelineID )
}
if len ( steps ) == 0 {
2026-07-03 15:37:07 +00:00
return ResolvedPipeline { } , fmt . Errorf ( "pipeline %q must declare at least one artifact lane" , pipelineID )
}
2026-07-27 18:04:12 +00:00
configuredLaneIDs , err := configuredProfileLaneIDs ( pipelineID , steps )
if err != nil {
return ResolvedPipeline { } , err
}
2026-07-03 15:37:07 +00:00
2026-08-09 00:29:04 +00:00
input , err := resolveBinding ( profile . Input , "" , fmt . Sprintf ( "pipeline %q input reference slot" , pipelineID ) )
if err != nil {
return ResolvedPipeline { } , err
}
2026-07-03 15:37:07 +00:00
if input . Module == "" {
return ResolvedPipeline { } , fmt . Errorf ( "pipeline %q input module must not be empty" , pipelineID )
}
inputModuleSpec , err := inputSpec ( catalog , input . Module )
if err != nil {
return ResolvedPipeline { } , moduleLookupError ( pipelineID , "" , StageInput , input . Module , err )
}
capabilities := newCapabilitySet ( )
if missing , ok := capabilities . missing ( inputModuleSpec . Requires ) ; ok {
return ResolvedPipeline { } , capabilityError ( pipelineID , "" , StageInput , input . Module , missing )
}
capabilities . add ( inputModuleSpec . Provides ... )
2026-08-09 00:29:04 +00:00
chunk , err := resolveBinding ( profile . Chunk , DefaultChunkModule , fmt . Sprintf ( "pipeline %q chunk reference slot" , pipelineID ) )
if err != nil {
return ResolvedPipeline { } , err
}
2026-07-03 15:37:07 +00:00
chunkSpec , err := chunkerSpec ( catalog , chunk . Module )
if err != nil {
return ResolvedPipeline { } , moduleLookupError ( pipelineID , "" , StageChunk , chunk . Module , err )
}
if missing , ok := capabilities . missing ( chunkSpec . Requires ) ; ok {
return ResolvedPipeline { } , capabilityError ( pipelineID , "" , StageChunk , chunk . Module , missing )
}
capabilities . add ( chunkSpec . Provides ... )
2026-07-21 21:05:28 +00:00
if ! explicitSteps {
if err := validatePipelineReferenceDefaults ( pipelineID , profile . References , chunkSpec , profile . Artifacts , catalog ) ; err != nil {
return ResolvedPipeline { } , err
}
} else {
allLanes := make ( map [ string ] ArtifactLaneProfile )
for _ , step := range profile . Steps {
for laneID , lane := range step . Artifacts {
allLanes [ strings . TrimSpace ( laneID ) ] = lane
}
}
if err := validatePipelineReferenceDefaults ( pipelineID , profile . References , chunkSpec , allLanes , catalog ) ; err != nil {
return ResolvedPipeline { } , err
}
2026-07-03 15:37:07 +00:00
}
2026-07-21 21:05:28 +00:00
for _ , source := range profile . References {
if source . Artifact != nil {
return ResolvedPipeline { } , fmt . Errorf ( "pipeline %q pipeline-level references may not use generated artifact selectors" , pipelineID )
}
2026-07-05 10:53:16 -05:00
}
2026-07-03 15:37:07 +00:00
2026-07-05 16:24:49 +00:00
chunkReferences , err := resolveReferenceTargetBindings ( referenceResolutionTarget {
PipelineID : pipelineID ,
Stage : StageChunk ,
Module : chunk . Module ,
Slots : chunkSpec . ReferenceSlots ,
PipelineReferences : profile . References ,
LocalReferences : chunk . References ,
Options : options ,
} )
if err != nil {
return ResolvedPipeline { } , err
}
2026-08-09 00:29:04 +00:00
output , err := resolveBinding ( profile . Output , DefaultOutputModule , fmt . Sprintf ( "pipeline %q output reference slot" , pipelineID ) )
if err != nil {
return ResolvedPipeline { } , err
}
2026-07-03 15:37:07 +00:00
resolved := ResolvedPipeline {
2026-08-26 23:34:06 +00:00
ID : pipelineID ,
ConfiguredValidationPolicy : cloneValidationPolicyOverride ( profile . ValidationPolicy ) ,
Input : input ,
InputExecutionClass : inputModuleSpec . ExecutionClass ,
Chunk : chunk ,
ChunkExecutionClass : chunkSpec . ExecutionClass ,
2026-08-26 23:42:11 +00:00
ChunkCorrectionProtocol : chunkSpec . CorrectionProtocol ,
2026-08-26 23:34:06 +00:00
ChunkReferences : referenceTarget ( StageChunk , "" , chunk . Module , chunkReferences ) ,
Output : output ,
2026-07-03 15:37:07 +00:00
}
2026-07-17 05:57:31 +00:00
chunkValidatorChain , err := resolveValidatorChain ( pipelineID , "" , StageChunk , chunk . Module , chunk . Validators , "" , nil , catalog )
2026-07-07 21:10:29 +00:00
if err != nil {
return ResolvedPipeline { } , err
}
resolved . ValidatorChains = append ( resolved . ValidatorChains , chunkValidatorChain )
2026-07-03 15:37:07 +00:00
outputCapabilities := capabilities . clone ( )
2026-07-21 21:05:28 +00:00
seenResolvedLanes := make ( map [ string ] string )
seenResolvedSteps := make ( map [ string ] struct { } , len ( steps ) )
2026-07-03 15:37:07 +00:00
2026-07-21 21:05:28 +00:00
for _ , step := range steps {
stepID := strings . TrimSpace ( step . ID )
if stepID == "" {
return ResolvedPipeline { } , fmt . Errorf ( "pipeline %q step id must not be empty" , pipelineID )
}
if _ , exists := seenResolvedSteps [ stepID ] ; exists {
return ResolvedPipeline { } , fmt . Errorf ( "pipeline %q step id %q is duplicated after trimming" , pipelineID , stepID )
}
seenResolvedSteps [ stepID ] = struct { } { }
laneOptions := ResolveOptions { }
if ! explicitSteps {
laneOptions = options
}
lanesByID , selectedLaneIDs , err := selectedArtifactLanes ( pipelineID , step . Artifacts , laneOptions )
2026-07-03 15:37:07 +00:00
if err != nil {
return ResolvedPipeline { } , err
}
2026-07-21 21:05:28 +00:00
if len ( selectedLaneIDs ) == 0 {
return ResolvedPipeline { } , fmt . Errorf ( "pipeline %q step %q must declare at least one artifact lane" , pipelineID , stepID )
}
if referenceSourceMapsConflict ( profile . References , step . References ) {
return ResolvedPipeline { } , fmt . Errorf ( "pipeline %q step %q has a generated and external reference collision" , pipelineID , stepID )
}
if err := validatePipelineReferenceDefaults ( pipelineID , step . References , chunkSpec , lanesByID , catalog ) ; err != nil {
return ResolvedPipeline { } , err
}
stepReferences := mergeReferenceMaps ( profile . References , step . References )
resolvedStep := ResolvedPipelineStep { ID : stepID }
for _ , laneID := range selectedLaneIDs {
if previousStep , exists := seenResolvedLanes [ laneID ] ; exists {
return ResolvedPipeline { } , fmt . Errorf ( "pipeline %q artifact lane id %q is duplicated across steps %q and %q" , pipelineID , laneID , previousStep , stepID )
}
seenResolvedLanes [ laneID ] = stepID
laneProfile := lanesByID [ laneID ]
lane , validatorChains , laneCapabilities , err := resolveArtifactLane ( pipelineID , stepID , laneID , laneProfile , stepReferences , options , capabilities , catalog )
if err != nil {
return ResolvedPipeline { } , err
}
resolvedStep . ArtifactLanes = append ( resolvedStep . ArtifactLanes , lane )
resolved . ValidatorChains = append ( resolved . ValidatorChains , validatorChains ... )
outputCapabilities . addSet ( laneCapabilities )
}
resolved . Steps = append ( resolved . Steps , resolvedStep )
}
if err := validateGeneratedBindings ( pipelineID , resolved , catalog ) ; err != nil {
return ResolvedPipeline { } , err
2026-07-03 15:37:07 +00:00
}
outputSpec , err := outputSpec ( catalog , resolved . Output . Module )
if err != nil {
return ResolvedPipeline { } , moduleLookupError ( pipelineID , "" , StageOutput , resolved . Output . Module , err )
}
if missing , ok := outputCapabilities . missing ( outputSpec . Requires ) ; ok {
return ResolvedPipeline { } , capabilityError ( pipelineID , "" , StageOutput , resolved . Output . Module , missing )
}
2026-08-03 16:49:51 +00:00
resolved . OutputExecutionClass = outputSpec . ExecutionClass
2026-08-03 17:09:16 +00:00
if err := applyEffectiveLLMProfiles ( & resolved , profile . LLMProfile , options . LLMProfileOverride ) ; err != nil {
return ResolvedPipeline { } , err
}
2026-08-25 19:55:39 +00:00
if err := applyEffectiveStructuredOutputRepairAttempts ( & resolved , profile . StructuredOutputRepairAttempts ) ; err != nil {
return ResolvedPipeline { } , err
}
2026-08-26 23:34:06 +00:00
if err := applyEffectiveValidationPolicies ( & resolved , profile . ValidationPolicy ) ; err != nil {
return ResolvedPipeline { } , err
}
2026-07-27 18:04:12 +00:00
if err := validateResolvedOptions ( resolved , catalog , configuredLaneIDs ) ; err != nil {
2026-07-17 06:18:46 +00:00
return ResolvedPipeline { } , err
}
2026-07-03 15:37:07 +00:00
digest , err := resolvedPipelineDigest ( resolved )
if err != nil {
return ResolvedPipeline { } , fmt . Errorf ( "pipeline %q digest: %w" , pipelineID , err )
}
resolved . Digest = digest
return resolved , nil
}
2026-07-27 18:04:12 +00:00
// configuredProfileLaneIDs validates the complete configured lane identity
// set before invocation-level selection removes legacy-profile lanes.
func configuredProfileLaneIDs ( pipelineID string , steps [ ] PipelineStepProfile ) ( [ ] string , error ) {
seen := make ( map [ string ] string )
var laneIDs [ ] string
for _ , step := range steps {
_ , ids , err := selectedArtifactLanes ( pipelineID , step . Artifacts , ResolveOptions { } )
if err != nil {
return nil , err
}
stepID := strings . TrimSpace ( step . ID )
for _ , laneID := range ids {
if previous , ok := seen [ laneID ] ; ok {
return nil , fmt . Errorf ( "pipeline %q artifact lane id %q is duplicated across steps %q and %q" , pipelineID , laneID , previous , stepID )
}
seen [ laneID ] = stepID
laneIDs = append ( laneIDs , laneID )
}
}
sort . Strings ( laneIDs )
return laneIDs , nil
}
2026-07-05 14:21:36 +00:00
func resolveArtifactLane (
pipelineID string ,
2026-07-21 21:05:28 +00:00
stepID string ,
2026-07-05 14:21:36 +00:00
laneID string ,
profile ArtifactLaneProfile ,
2026-07-21 21:05:28 +00:00
pipelineReferences map [ string ] ReferenceSource ,
2026-07-05 14:21:36 +00:00
options ResolveOptions ,
inherited capabilitySet ,
catalog ModuleCatalog ,
2026-07-07 21:10:29 +00:00
) ( ResolvedArtifactLane , [ ] ResolvedValidatorChain , capabilitySet , error ) {
2026-08-09 00:29:04 +00:00
extract , err := resolveBinding ( profile . Extract , "" , fmt . Sprintf ( "pipeline %q step %q lane %q extract reference slot" , pipelineID , stepID , laneID ) )
if err != nil {
return ResolvedArtifactLane { } , nil , nil , err
}
merge , err := resolveBinding ( profile . Merge , DefaultMergeModule , fmt . Sprintf ( "pipeline %q step %q lane %q merge reference slot" , pipelineID , stepID , laneID ) )
if err != nil {
return ResolvedArtifactLane { } , nil , nil , err
}
normalize , err := resolveBinding ( profile . Normalize , DefaultNormalizeModule , fmt . Sprintf ( "pipeline %q step %q lane %q normalize reference slot" , pipelineID , stepID , laneID ) )
if err != nil {
return ResolvedArtifactLane { } , nil , nil , err
}
validators , err := resolveBindings ( profile . Validators , "" , fmt . Sprintf ( "pipeline %q step %q lane %q validator reference slot" , pipelineID , stepID , laneID ) )
if err != nil {
return ResolvedArtifactLane { } , nil , nil , err
}
2026-07-03 15:37:07 +00:00
lane := ResolvedArtifactLane {
2026-07-21 21:05:28 +00:00
StepID : strings . TrimSpace ( stepID ) ,
2026-07-03 15:37:07 +00:00
ID : laneID ,
2026-08-09 00:29:04 +00:00
Extract : extract ,
Merge : merge ,
Normalize : normalize ,
Validators : validators ,
2026-07-03 15:37:07 +00:00
}
if lane . Extract . Module == "" {
2026-07-07 21:10:29 +00:00
return ResolvedArtifactLane { } , nil , nil , fmt . Errorf ( "pipeline %q lane %q extract module must not be empty" , pipelineID , laneID )
2026-07-03 15:37:07 +00:00
}
capabilities := inherited . clone ( )
extractSpec , err := extractorSpec ( catalog , lane . Extract . Module )
if err != nil {
2026-07-07 21:10:29 +00:00
return ResolvedArtifactLane { } , nil , nil , moduleLookupError ( pipelineID , laneID , StageExtract , lane . Extract . Module , err )
2026-07-03 15:37:07 +00:00
}
if missing , ok := capabilities . missing ( extractSpec . Requires ) ; ok {
2026-07-07 21:10:29 +00:00
return ResolvedArtifactLane { } , nil , nil , capabilityError ( pipelineID , laneID , StageExtract , lane . Extract . Module , missing )
2026-07-03 15:37:07 +00:00
}
2026-07-17 05:57:31 +00:00
artifactType , err := resolveArtifactIdentity ( pipelineID , laneID , & lane , extractSpec , catalog )
if err != nil {
return ResolvedArtifactLane { } , nil , nil , err
}
2026-07-21 21:05:28 +00:00
if referenceSourceMapsConflict ( profile . References , lane . Extract . References ) {
return ResolvedArtifactLane { } , nil , nil , fmt . Errorf ( "pipeline %q step %q lane %q extract has a generated and external reference collision" , pipelineID , stepID , laneID )
}
2026-07-05 16:18:00 +00:00
extractReferences := mergeReferenceMaps ( profile . References , lane . Extract . References )
2026-07-05 16:24:49 +00:00
references , err := resolveReferenceTargetBindings ( referenceResolutionTarget {
PipelineID : pipelineID ,
2026-08-09 00:29:04 +00:00
StepID : strings . TrimSpace ( stepID ) ,
2026-07-05 16:24:49 +00:00
LaneID : laneID ,
Stage : StageExtract ,
Module : lane . Extract . Module ,
Slots : extractSpec . ReferenceSlots ,
PipelineReferences : pipelineReferences ,
LocalReferences : extractReferences ,
Options : options ,
} )
2026-07-05 14:21:36 +00:00
if err != nil {
2026-07-07 21:10:29 +00:00
return ResolvedArtifactLane { } , nil , nil , err
2026-07-05 14:21:36 +00:00
}
2026-07-05 16:12:38 +00:00
lane . ExtractReferences = referenceTarget ( StageExtract , laneID , lane . Extract . Module , references )
2026-07-21 21:05:28 +00:00
lane . ExtractReferences . StepID = strings . TrimSpace ( stepID )
2026-08-03 16:49:51 +00:00
lane . ExtractExecutionClass = extractSpec . ExecutionClass
2026-08-26 23:42:11 +00:00
lane . ExtractCorrectionProtocol = extractSpec . CorrectionProtocol
2026-07-03 15:37:07 +00:00
capabilities . add ( extractSpec . Provides ... )
2026-07-17 05:57:31 +00:00
mergeSpec , err := mergerSpecForArtifact ( catalog , lane . Merge . Module , lane . ArtifactKind , artifactType )
2026-07-03 15:37:07 +00:00
if err != nil {
2026-07-07 21:10:29 +00:00
return ResolvedArtifactLane { } , nil , nil , moduleLookupError ( pipelineID , laneID , StageMerge , lane . Merge . Module , err )
2026-07-03 15:37:07 +00:00
}
if missing , ok := capabilities . missing ( mergeSpec . Requires ) ; ok {
2026-07-07 21:10:29 +00:00
return ResolvedArtifactLane { } , nil , nil , capabilityError ( pipelineID , laneID , StageMerge , lane . Merge . Module , missing )
2026-07-03 15:37:07 +00:00
}
2026-07-07 19:10:55 +00:00
mergeReferences , err := resolveReferenceTargetBindings ( referenceResolutionTarget {
PipelineID : pipelineID ,
2026-08-09 00:29:04 +00:00
StepID : strings . TrimSpace ( stepID ) ,
2026-07-07 19:10:55 +00:00
LaneID : laneID ,
Stage : StageMerge ,
Module : lane . Merge . Module ,
Slots : mergeSpec . ReferenceSlots ,
PipelineReferences : pipelineReferences ,
LocalReferences : lane . Merge . References ,
Options : options ,
} )
if err != nil {
2026-07-07 21:10:29 +00:00
return ResolvedArtifactLane { } , nil , nil , err
2026-07-07 19:10:55 +00:00
}
lane . MergeReferences = referenceTarget ( StageMerge , laneID , lane . Merge . Module , mergeReferences )
2026-07-21 21:05:28 +00:00
lane . MergeReferences . StepID = strings . TrimSpace ( stepID )
2026-08-03 16:49:51 +00:00
lane . MergeExecutionClass = mergeSpec . ExecutionClass
2026-08-26 23:42:11 +00:00
lane . MergeCorrectionProtocol = mergeSpec . CorrectionProtocol
2026-07-03 15:37:07 +00:00
capabilities . add ( mergeSpec . Provides ... )
2026-07-17 05:57:31 +00:00
normalizeSpec , err := normalizerSpecForArtifact ( catalog , lane . Normalize . Module , lane . ArtifactKind , artifactType )
2026-07-03 15:37:07 +00:00
if err != nil {
2026-07-07 21:10:29 +00:00
return ResolvedArtifactLane { } , nil , nil , moduleLookupError ( pipelineID , laneID , StageNormalize , lane . Normalize . Module , err )
2026-07-03 15:37:07 +00:00
}
if missing , ok := capabilities . missing ( normalizeSpec . Requires ) ; ok {
2026-07-07 21:10:29 +00:00
return ResolvedArtifactLane { } , nil , nil , capabilityError ( pipelineID , laneID , StageNormalize , lane . Normalize . Module , missing )
2026-07-03 15:37:07 +00:00
}
2026-07-05 16:24:49 +00:00
normalizeReferences , err := resolveReferenceTargetBindings ( referenceResolutionTarget {
PipelineID : pipelineID ,
2026-08-09 00:29:04 +00:00
StepID : strings . TrimSpace ( stepID ) ,
2026-07-05 16:24:49 +00:00
LaneID : laneID ,
Stage : StageNormalize ,
Module : lane . Normalize . Module ,
Slots : normalizeSpec . ReferenceSlots ,
PipelineReferences : pipelineReferences ,
LocalReferences : lane . Normalize . References ,
Options : options ,
} )
if err != nil {
2026-07-07 21:10:29 +00:00
return ResolvedArtifactLane { } , nil , nil , err
2026-07-05 16:24:49 +00:00
}
lane . NormalizeReferences = referenceTarget ( StageNormalize , laneID , lane . Normalize . Module , normalizeReferences )
2026-07-21 21:05:28 +00:00
lane . NormalizeReferences . StepID = strings . TrimSpace ( stepID )
2026-08-03 16:49:51 +00:00
lane . NormalizeExecutionClass = normalizeSpec . ExecutionClass
2026-08-26 23:42:11 +00:00
lane . NormalizeCorrectionProtocol = normalizeSpec . CorrectionProtocol
2026-07-03 15:37:07 +00:00
capabilities . add ( normalizeSpec . Provides ... )
2026-07-07 15:32:35 -05:00
if len ( lane . Validators ) > 0 {
2026-07-07 21:10:29 +00:00
return ResolvedArtifactLane { } , nil , nil , configuredValidatorsError ( pipelineID , laneID )
}
2026-07-17 05:57:31 +00:00
extractValidatorChain , err := resolveValidatorChain ( pipelineID , laneID , StageExtract , lane . Extract . Module , lane . Extract . Validators , lane . ArtifactKind , artifactType , catalog )
2026-07-07 21:10:29 +00:00
if err != nil {
return ResolvedArtifactLane { } , nil , nil , err
2026-07-03 15:37:07 +00:00
}
2026-07-17 05:57:31 +00:00
mergeValidatorChain , err := resolveValidatorChain ( pipelineID , laneID , StageMerge , lane . Merge . Module , lane . Merge . Validators , lane . ArtifactKind , artifactType , catalog )
2026-07-07 21:10:29 +00:00
if err != nil {
return ResolvedArtifactLane { } , nil , nil , err
}
2026-07-17 05:57:31 +00:00
normalizeValidatorChain , err := resolveValidatorChain ( pipelineID , laneID , StageNormalize , lane . Normalize . Module , lane . Normalize . Validators , lane . ArtifactKind , artifactType , catalog )
2026-07-07 21:10:29 +00:00
if err != nil {
return ResolvedArtifactLane { } , nil , nil , err
}
validatorChains := [ ] ResolvedValidatorChain { extractValidatorChain , mergeValidatorChain , normalizeValidatorChain }
2026-07-03 15:37:07 +00:00
2026-07-07 21:10:29 +00:00
return lane , validatorChains , capabilities , nil
2026-07-03 15:37:07 +00:00
}
2026-07-07 15:32:35 -05:00
func configuredValidatorsError ( pipelineID string , laneID string ) error {
2026-07-07 21:21:12 +00:00
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 )
2026-07-07 15:32:35 -05:00
}
2026-07-21 21:05:28 +00:00
func validateGeneratedBindings ( pipelineID string , resolved ResolvedPipeline , catalog ModuleCatalog ) error {
stepIndex := make ( map [ string ] int , len ( resolved . Steps ) )
laneIndex := make ( map [ string ] ResolvedArtifactLane )
for index , step := range resolved . Steps {
if _ , exists := stepIndex [ step . ID ] ; exists {
return fmt . Errorf ( "pipeline %q step id %q is ambiguous" , pipelineID , step . ID )
}
stepIndex [ step . ID ] = index
for _ , lane := range step . ArtifactLanes {
laneIndex [ step . ID + "\x00" + lane . ID ] = lane
}
}
for consumerIndex , step := range resolved . Steps {
for _ , lane := range step . ArtifactLanes {
for _ , target := range [ ] ResolvedReferenceTarget { lane . ExtractReferences , lane . MergeReferences , lane . NormalizeReferences } {
slots , err := resolvedReferenceSlots ( target , lane . ArtifactKind , catalog )
if err != nil {
return fmt . Errorf ( "pipeline %q step %q lane %q: %w" , pipelineID , step . ID , lane . ID , err )
}
slotByName := make ( map [ string ] contracts . ReferenceSlot , len ( slots ) )
for _ , slot := range slots {
slotByName [ slot . Name ] = slot
}
for _ , binding := range target . Bindings {
if binding . Artifact == nil {
continue
}
producerStep , ok := stepIndex [ strings . TrimSpace ( binding . Artifact . Step ) ]
if ! ok {
return fmt . Errorf ( "pipeline %q step %q lane %q %s reference slot %q producer step %q is not declared" , pipelineID , step . ID , lane . ID , target . Stage , binding . SlotName , binding . Artifact . Step )
}
if producerStep >= consumerIndex {
return fmt . Errorf ( "pipeline %q step %q lane %q reference slot %q producer %q must be an earlier step" , pipelineID , step . ID , lane . ID , binding . SlotName , binding . Artifact . Step )
}
producer , ok := laneIndex [ strings . TrimSpace ( binding . Artifact . Step ) + "\x00" + strings . TrimSpace ( binding . Artifact . Lane ) ]
if ! ok {
return fmt . Errorf ( "pipeline %q step %q lane %q reference slot %q producer lane %q is not selected" , pipelineID , step . ID , lane . ID , binding . SlotName , binding . Artifact . Lane )
}
slot , ok := slotByName [ strings . TrimSpace ( binding . SlotName ) ]
if ! ok {
return fmt . Errorf ( "pipeline %q step %q lane %q %s reference slot %q is not declared" , pipelineID , step . ID , lane . ID , target . Stage , binding . SlotName )
}
if len ( slot . AcceptedArtifactKinds ) == 0 {
return fmt . Errorf ( "pipeline %q step %q lane %q reference slot %q does not accept generated artifacts" , pipelineID , step . ID , lane . ID , binding . SlotName )
}
accepted := false
for _ , kind := range slot . AcceptedArtifactKinds {
if kind == producer . ArtifactKind {
accepted = true
break
}
}
if ! accepted {
return fmt . Errorf ( "pipeline %q step %q lane %q reference slot %q does not accept artifact kind %q" , pipelineID , step . ID , lane . ID , binding . SlotName , producer . ArtifactKind )
}
codecSpec , ok := catalog . ArtifactCodecs . Spec ( producer . ArtifactKind )
if ! ok {
return fmt . Errorf ( "pipeline %q step %q lane %q producer %q artifact codec %q is not registered" , pipelineID , step . ID , lane . ID , producer . ID , producer . ArtifactKind )
}
if ! referenceMediaTypeAccepted ( codecSpec . MediaType , slot . AcceptedMediaTypes ) {
return fmt . Errorf ( "pipeline %q step %q lane %q reference slot %q does not accept producer media type %q" , pipelineID , step . ID , lane . ID , binding . SlotName , codecSpec . MediaType )
}
}
}
}
}
return nil
}
func resolvedReferenceSlots ( target ResolvedReferenceTarget , artifactKind contracts . ArtifactKind , catalog ModuleCatalog ) ( [ ] contracts . ReferenceSlot , error ) {
spec , err := referenceTargetSpec ( target , artifactKind , catalog )
if err != nil {
return nil , err
}
return contracts . CloneReferenceSlots ( spec . ReferenceSlots ) , nil
}
2026-07-17 05:57:31 +00:00
func resolveArtifactIdentity ( pipelineID , laneID string , lane * ResolvedArtifactLane , extractSpec ModuleSpec , catalog ModuleCatalog ) ( reflect . Type , error ) {
if extractSpec . ArtifactKind == "" {
2026-07-17 08:13:08 +00:00
return nil , fmt . Errorf ( "pipeline %q lane %q extract module %q does not declare an artifact kind" , pipelineID , laneID , lane . Extract . Module )
2026-07-17 05:57:31 +00:00
}
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 == "" {
2026-07-17 13:53:55 +00:00
return ModuleSpec { } , fmt . Errorf ( "merger %q cannot be resolved without an artifact kind" , key )
2026-07-17 05:57:31 +00:00
}
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 == "" {
2026-07-17 13:53:55 +00:00
return ModuleSpec { } , fmt . Errorf ( "normalizer %q cannot be resolved without an artifact kind" , key )
2026-07-17 05:57:31 +00:00
}
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
}
2026-07-17 08:13:08 +00:00
return ValidatorSpec { } , "" , fmt . Errorf ( "references unknown validator %q without an artifact kind" , key )
2026-07-17 05:57:31 +00:00
}
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
}
2026-07-17 07:19:58 +00:00
if _ , ok := registry . Spec ( key ) ; ! ok {
return ValidatorSpec { } , "" , fmt . Errorf ( "references unknown validator %q for artifact kind %q" , key , kind )
}
2026-07-17 05:57:31 +00:00
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 ) {
2026-07-07 21:10:29 +00:00
chain := ResolvedValidatorChain {
Stage : stage ,
LaneID : strings . TrimSpace ( laneID ) ,
ModuleKey : strings . TrimSpace ( module ) ,
}
if chain . ModuleKey == "" {
return ResolvedValidatorChain { } , fmt . Errorf ( "pipeline %q validator chain %q module key must not be empty" , pipelineID , stage )
}
switch stage {
case StageChunk , StageExtract , StageMerge , StageNormalize :
default :
return ResolvedValidatorChain { } , fmt . Errorf ( "pipeline %q validator chain stage %q is not supported" , pipelineID , stage )
}
2026-07-07 21:21:12 +00:00
var bindings [ ] ModuleBinding
if override . Set {
bindings = cloneModuleBindings ( override . Validators )
} else if catalog . ValidatorChains != nil {
bindings = catalog . ValidatorChains . Validators ( stage , chain . ModuleKey )
2026-07-07 21:10:29 +00:00
}
if len ( bindings ) == 0 {
return chain , nil
}
if catalog . Validators == nil {
return ResolvedValidatorChain { } , fmt . Errorf ( "pipeline %q validator registry must not be nil for %s validator chain on module %q" , pipelineID , stage , chain . ModuleKey )
}
chain . Validators = make ( [ ] ResolvedValidator , 0 , len ( bindings ) )
for _ , validator := range bindings {
2026-07-17 05:57:31 +00:00
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 )
2026-07-07 21:10:29 +00:00
}
2026-08-26 23:42:11 +00:00
if validator . Retries > 0 && spec . ExecutionClass != contracts . ExecutionClassLLMBacked {
return ResolvedValidatorChain { } , fmt . Errorf ( "pipeline %q %s validator %q retries require an LLM-backed execution class" , pipelineID , stage , validator . Module )
}
2026-07-07 21:10:29 +00:00
chain . Validators = append ( chain . Validators , ResolvedValidator {
Binding : cloneModuleBinding ( validator ) ,
ExecutionClass : spec . ExecutionClass ,
2026-07-17 05:57:31 +00:00
Target : target ,
ArtifactKind : artifactKind ,
2026-07-07 21:10:29 +00:00
} )
}
return chain , nil
}
func cloneResolvedValidatorChains ( chains [ ] ResolvedValidatorChain ) [ ] ResolvedValidatorChain {
if len ( chains ) == 0 {
return nil
}
out := make ( [ ] ResolvedValidatorChain , len ( chains ) )
for i , chain := range chains {
out [ i ] = ResolvedValidatorChain {
Stage : chain . Stage ,
LaneID : strings . TrimSpace ( chain . LaneID ) ,
ModuleKey : strings . TrimSpace ( chain . ModuleKey ) ,
Validators : cloneResolvedValidators ( chain . Validators ) ,
}
}
return out
}
func cloneResolvedValidators ( validators [ ] ResolvedValidator ) [ ] ResolvedValidator {
if len ( validators ) == 0 {
return nil
}
out := make ( [ ] ResolvedValidator , len ( validators ) )
for i , validator := range validators {
out [ i ] = ResolvedValidator {
Binding : cloneModuleBinding ( validator . Binding ) ,
ExecutionClass : validator . ExecutionClass ,
2026-07-17 05:57:31 +00:00
Target : validator . Target ,
ArtifactKind : validator . ArtifactKind ,
2026-07-07 21:10:29 +00:00
}
}
return out
}
2026-07-05 16:12:38 +00:00
func referenceTarget ( stage ModuleStage , laneID string , module string , bindings [ ] ReferenceBinding ) ResolvedReferenceTarget {
return ResolvedReferenceTarget {
Stage : stage ,
LaneID : strings . TrimSpace ( laneID ) ,
Module : strings . TrimSpace ( module ) ,
Bindings : append ( [ ] ReferenceBinding ( nil ) , bindings ... ) ,
}
}
2026-07-21 21:05:28 +00:00
func mergeReferenceMaps ( base map [ string ] ReferenceSource , override map [ string ] ReferenceSource ) map [ string ] ReferenceSource {
2026-07-05 16:18:00 +00:00
if len ( base ) == 0 && len ( override ) == 0 {
return nil
}
2026-07-21 21:05:28 +00:00
out := make ( map [ string ] ReferenceSource , len ( base ) + len ( override ) )
2026-07-05 16:18:00 +00:00
for key , value := range base {
2026-07-21 21:05:28 +00:00
out [ key ] = cloneReferenceSource ( value )
2026-07-05 16:18:00 +00:00
}
for key , value := range override {
2026-07-21 21:05:28 +00:00
out [ key ] = cloneReferenceSource ( value )
2026-07-05 16:18:00 +00:00
}
return out
}
2026-07-21 21:05:28 +00:00
func referenceSourceMapsConflict ( base , override map [ string ] ReferenceSource ) bool {
for rawKey , left := range base {
key := strings . TrimSpace ( rawKey )
for rawOverrideKey , right := range override {
if key == strings . TrimSpace ( rawOverrideKey ) && left . IsGenerated ( ) != right . IsGenerated ( ) {
return true
}
}
}
return false
}
2026-07-05 10:53:16 -05:00
func validatePipelineReferenceDefaults (
pipelineID string ,
2026-07-21 21:05:28 +00:00
pipelineReferences map [ string ] ReferenceSource ,
2026-07-05 16:24:49 +00:00
chunkSpec ModuleSpec ,
2026-07-05 10:53:16 -05:00
lanesByID map [ string ] ArtifactLaneProfile ,
catalog ModuleCatalog ,
) error {
2026-08-09 00:29:04 +00:00
normalizedPipelineReferences , err := normalizeReferenceMap ( pipelineReferences , fmt . Sprintf ( "pipeline %q reference slot" , pipelineID ) )
2026-07-05 10:53:16 -05:00
if err != nil {
return err
}
if len ( normalizedPipelineReferences ) == 0 {
return nil
}
2026-07-05 16:24:49 +00:00
declaredByAnyTarget := make ( map [ string ] struct { } , len ( normalizedPipelineReferences ) )
for _ , slot := range chunkSpec . ReferenceSlots {
declaredByAnyTarget [ slot . Name ] = struct { } { }
}
2026-07-05 10:53:16 -05:00
for _ , laneID := range sortedArtifactLaneProfileKeys ( lanesByID ) {
laneProfile := lanesByID [ laneID ]
2026-08-09 00:29:04 +00:00
extractModule := resolveModuleKey ( laneProfile . Extract . Module , "" )
if extractModule == "" {
2026-07-05 10:53:16 -05:00
return fmt . Errorf ( "pipeline %q lane %q extract module must not be empty" , pipelineID , laneID )
}
2026-08-09 00:29:04 +00:00
extractSpec , err := extractorSpec ( catalog , extractModule )
2026-07-05 10:53:16 -05:00
if err != nil {
2026-08-09 00:29:04 +00:00
return moduleLookupError ( pipelineID , laneID , StageExtract , extractModule , err )
2026-07-05 10:53:16 -05:00
}
for _ , slot := range extractSpec . ReferenceSlots {
2026-07-05 16:24:49 +00:00
declaredByAnyTarget [ slot . Name ] = struct { } { }
}
2026-08-09 00:29:04 +00:00
mergeModule := resolveModuleKey ( laneProfile . Merge . Module , DefaultMergeModule )
2026-07-17 05:57:31 +00:00
var artifactType reflect . Type
2026-07-17 06:50:08 +00:00
artifactKind := extractSpec . ArtifactKind
if artifactKind != "" && catalog . Extractors != nil {
2026-08-09 00:29:04 +00:00
if entry , ok := catalog . Extractors . typedEntry ( extractModule ) ; ok {
2026-07-17 05:57:31 +00:00
artifactType = entry . valueType
}
}
2026-08-09 00:29:04 +00:00
mergeSpec , err := mergerSpecForArtifact ( catalog , mergeModule , artifactKind , artifactType )
2026-07-07 19:10:55 +00:00
if err != nil {
2026-08-09 00:29:04 +00:00
return moduleLookupError ( pipelineID , laneID , StageMerge , mergeModule , err )
2026-07-07 19:10:55 +00:00
}
for _ , slot := range mergeSpec . ReferenceSlots {
declaredByAnyTarget [ slot . Name ] = struct { } { }
}
2026-08-09 00:29:04 +00:00
normalizeModule := resolveModuleKey ( laneProfile . Normalize . Module , DefaultNormalizeModule )
normalizeSpec , err := normalizerSpecForArtifact ( catalog , normalizeModule , artifactKind , artifactType )
2026-07-05 16:24:49 +00:00
if err != nil {
2026-08-09 00:29:04 +00:00
return moduleLookupError ( pipelineID , laneID , StageNormalize , normalizeModule , err )
2026-07-05 16:24:49 +00:00
}
for _ , slot := range normalizeSpec . ReferenceSlots {
declaredByAnyTarget [ slot . Name ] = struct { } { }
2026-07-05 10:53:16 -05:00
}
}
for _ , slotName := range sortedStringMapKeys ( normalizedPipelineReferences ) {
2026-07-05 16:24:49 +00:00
if _ , ok := declaredByAnyTarget [ slotName ] ; ! ok {
return fmt . Errorf ( "pipeline %q reference slot %q is not declared by any eligible reference target" , pipelineID , slotName )
2026-07-05 10:53:16 -05:00
}
}
return nil
}
2026-07-05 16:24:49 +00:00
type referenceResolutionTarget struct {
PipelineID string
2026-08-09 00:29:04 +00:00
StepID string
2026-07-05 16:24:49 +00:00
LaneID string
Stage ModuleStage
Module string
Slots [ ] contracts . ReferenceSlot
2026-07-21 21:05:28 +00:00
PipelineReferences map [ string ] ReferenceSource
LocalReferences map [ string ] ReferenceSource
2026-07-05 16:24:49 +00:00
Options ResolveOptions
}
func resolveReferenceTargetBindings ( target referenceResolutionTarget ) ( [ ] ReferenceBinding , error ) {
slotByName := make ( map [ string ] contracts . ReferenceSlot , len ( target . Slots ) )
for _ , slot := range target . Slots {
2026-07-05 14:21:36 +00:00
slotByName [ slot . Name ] = slot
}
bindings := make ( map [ string ] ReferenceBinding )
2026-07-21 21:05:28 +00:00
addBinding := func ( slotName string , source ReferenceSource , bindingSource string ) error {
2026-07-05 14:21:36 +00:00
slotName = strings . TrimSpace ( slotName )
if slotName == "" {
2026-07-05 16:24:49 +00:00
return fmt . Errorf ( "%s reference slot name must not be empty" , referenceTargetErrorContext ( target ) )
2026-07-05 14:21:36 +00:00
}
2026-07-21 21:05:28 +00:00
if source . Artifact == nil {
source . Path = strings . TrimSpace ( source . Path )
}
if source . Artifact == nil && source . Path == "" {
2026-07-05 16:24:49 +00:00
return fmt . Errorf ( "%s reference slot %q source must not be empty" , referenceTargetErrorContext ( target ) , slotName )
2026-07-05 14:21:36 +00:00
}
2026-07-21 21:05:28 +00:00
if source . Artifact != nil {
if strings . TrimSpace ( source . Path ) != "" {
return fmt . Errorf ( "%s reference slot %q must contain exactly one source form" , referenceTargetErrorContext ( target ) , slotName )
}
artifact := * source . Artifact
artifact . Step = strings . TrimSpace ( artifact . Step )
artifact . Lane = strings . TrimSpace ( artifact . Lane )
if artifact . Step == "" || artifact . Lane == "" {
return fmt . Errorf ( "%s reference slot %q artifact selector step and lane must not be empty" , referenceTargetErrorContext ( target ) , slotName )
}
source . Artifact = & artifact
}
2026-07-05 14:21:36 +00:00
if _ , ok := slotByName [ slotName ] ; ! ok {
2026-07-05 16:24:49 +00:00
return fmt . Errorf ( "%s reference slot %q is not declared by %s module %q" , referenceTargetErrorContext ( target ) , slotName , target . Stage , target . Module )
2026-07-05 14:21:36 +00:00
}
2026-07-21 21:05:28 +00:00
if previous , exists := bindings [ slotName ] ; exists && ( previous . Artifact != nil || source . Artifact != nil ) {
return fmt . Errorf ( "%s reference slot %q has conflicting generated and external bindings" , referenceTargetErrorContext ( target ) , slotName )
}
binding := ReferenceBinding {
2026-07-05 16:24:49 +00:00
LaneID : target . LaneID ,
2026-07-05 14:21:36 +00:00
SlotName : slotName ,
BindingSource : bindingSource ,
}
2026-07-21 21:05:28 +00:00
if source . Artifact != nil {
binding . Artifact = source . Artifact
} else {
binding . Source = source . Path
}
bindings [ slotName ] = binding
2026-07-05 14:21:36 +00:00
return nil
}
2026-08-09 00:29:04 +00:00
normalizedPipelineReferences , err := normalizeReferenceMap ( target . PipelineReferences , fmt . Sprintf ( "pipeline %q reference slot" , target . PipelineID ) )
2026-07-05 14:21:36 +00:00
if err != nil {
return nil , err
}
for _ , slotName := range sortedStringMapKeys ( normalizedPipelineReferences ) {
if _ , ok := slotByName [ slotName ] ; ! ok {
continue
}
if err := addBinding ( slotName , normalizedPipelineReferences [ slotName ] , contracts . ReferenceBindingSourceConfig ) ; err != nil {
return nil , err
}
}
2026-08-09 00:29:04 +00:00
normalizedLocalReferences , err := normalizeReferenceMap ( target . LocalReferences , referenceTargetSlotLabel ( target ) )
2026-07-05 14:21:36 +00:00
if err != nil {
return nil , err
}
2026-07-05 16:24:49 +00:00
for _ , slotName := range sortedStringMapKeys ( normalizedLocalReferences ) {
if err := addBinding ( slotName , normalizedLocalReferences [ slotName ] , contracts . ReferenceBindingSourceConfig ) ; err != nil {
2026-07-05 14:21:36 +00:00
return nil , err
}
}
2026-07-05 16:24:49 +00:00
for _ , override := range target . Options . ReferenceOverrides {
2026-07-05 16:31:23 +00:00
match , err := referenceOverrideMatchesTarget ( target , override )
if err != nil {
return nil , err
2026-07-05 14:21:36 +00:00
}
2026-07-05 16:31:23 +00:00
if ! match {
2026-07-05 14:21:36 +00:00
continue
}
source := override . BindingSource
if strings . TrimSpace ( source ) == "" {
source = contracts . ReferenceBindingSourceCLI
}
2026-07-21 21:05:28 +00:00
if err := addBinding ( override . SlotName , ExternalReference ( override . Source ) , strings . TrimSpace ( source ) ) ; err != nil {
2026-07-05 14:21:36 +00:00
return nil , err
}
}
2026-07-05 16:24:49 +00:00
for _ , unbind := range target . Options . ReferenceUnbinds {
2026-07-05 16:31:23 +00:00
match , err := referenceUnbindMatchesTarget ( target , unbind )
if err != nil {
return nil , err
2026-07-05 14:21:36 +00:00
}
2026-07-05 16:31:23 +00:00
if ! match {
2026-07-05 14:21:36 +00:00
continue
}
slotName := strings . TrimSpace ( unbind . SlotName )
if slotName == "" {
2026-07-05 16:24:49 +00:00
return nil , fmt . Errorf ( "%s reference unbind slot name must not be empty" , referenceTargetErrorContext ( target ) )
2026-07-05 14:21:36 +00:00
}
if _ , ok := slotByName [ slotName ] ; ! ok {
2026-07-05 16:24:49 +00:00
return nil , fmt . Errorf ( "%s reference slot %q is not declared" , referenceTargetErrorContext ( target ) , slotName )
2026-07-05 14:21:36 +00:00
}
2026-07-21 21:05:28 +00:00
if binding , ok := bindings [ slotName ] ; ok && binding . Artifact == nil {
delete ( bindings , slotName )
}
2026-07-05 14:21:36 +00:00
}
2026-07-05 16:24:49 +00:00
for _ , slot := range target . Slots {
2026-07-05 14:21:36 +00:00
if slot . Required {
if _ , ok := bindings [ slot . Name ] ; ! ok {
2026-07-05 16:24:49 +00:00
return nil , fmt . Errorf ( "%s required reference slot %q is not bound" , referenceTargetErrorContext ( target ) , slot . Name )
2026-07-05 14:21:36 +00:00
}
}
}
2026-07-05 16:24:49 +00:00
return sortedReferenceBindings ( bindings ) , nil
}
2026-07-05 16:31:23 +00:00
func referenceOverrideMatchesTarget ( target referenceResolutionTarget , override ReferenceBinding ) ( bool , error ) {
stage , laneID , err := normalizeReferenceOptionTarget ( target . PipelineID , "override" , override . Stage , override . LaneID )
if err != nil {
return false , err
}
return stage == target . Stage && laneID == target . LaneID , nil
}
func referenceUnbindMatchesTarget ( target referenceResolutionTarget , unbind ReferenceUnbind ) ( bool , error ) {
stage , laneID , err := normalizeReferenceOptionTarget ( target . PipelineID , "unbind" , unbind . Stage , unbind . LaneID )
if err != nil {
return false , err
}
return stage == target . Stage && laneID == target . LaneID , nil
}
func normalizeReferenceOptionTarget ( pipelineID string , operation string , stage ModuleStage , laneID string ) ( ModuleStage , string , error ) {
stage = ModuleStage ( strings . TrimSpace ( string ( stage ) ) )
if stage == "" {
stage = StageExtract
}
laneID = strings . TrimSpace ( laneID )
switch stage {
case StageChunk :
if laneID != "" {
return "" , "" , fmt . Errorf ( "pipeline %q reference %s for chunk must not include a lane id" , pipelineID , operation )
}
2026-07-07 19:10:55 +00:00
case StageExtract , StageMerge , StageNormalize :
2026-07-05 16:31:23 +00:00
if laneID == "" {
return "" , "" , fmt . Errorf ( "pipeline %q reference %s lane id must not be empty" , pipelineID , operation )
}
default :
return "" , "" , fmt . Errorf ( "pipeline %q reference %s stage %q is not supported" , pipelineID , operation , stage )
}
return stage , laneID , nil
}
2026-07-05 16:24:49 +00:00
func sortedReferenceBindings ( bindings map [ string ] ReferenceBinding ) [ ] ReferenceBinding {
2026-07-05 14:21:36 +00:00
keys := sortedReferenceBindingKeys ( bindings )
resolved := make ( [ ] ReferenceBinding , 0 , len ( keys ) )
for _ , slotName := range keys {
resolved = append ( resolved , bindings [ slotName ] )
}
2026-07-05 16:24:49 +00:00
return resolved
}
func referenceTargetErrorContext ( target referenceResolutionTarget ) string {
if target . LaneID != "" {
2026-08-09 00:29:04 +00:00
if target . StepID != "" {
return fmt . Sprintf ( "pipeline %q step %q lane %q %s module %q" , target . PipelineID , target . StepID , target . LaneID , target . Stage , target . Module )
}
2026-07-05 16:24:49 +00:00
return fmt . Sprintf ( "pipeline %q lane %q %s module %q" , target . PipelineID , target . LaneID , target . Stage , target . Module )
}
return fmt . Sprintf ( "pipeline %q %s module %q" , target . PipelineID , target . Stage , target . Module )
}
func referenceTargetSlotLabel ( target referenceResolutionTarget ) string {
if target . LaneID != "" {
2026-08-09 00:29:04 +00:00
if target . StepID != "" {
return fmt . Sprintf ( "pipeline %q step %q lane %q %s reference slot" , target . PipelineID , target . StepID , target . LaneID , target . Stage )
}
2026-07-05 16:24:49 +00:00
return fmt . Sprintf ( "pipeline %q lane %q %s reference slot" , target . PipelineID , target . LaneID , target . Stage )
}
return fmt . Sprintf ( "pipeline %q %s reference slot" , target . PipelineID , target . Stage )
2026-07-05 14:21:36 +00:00
}
2026-08-09 00:29:04 +00:00
func normalizeReferenceMap ( values map [ string ] ReferenceSource , keyName string ) ( map [ string ] ReferenceSource , error ) {
2026-07-05 14:21:36 +00:00
if len ( values ) == 0 {
return nil , nil
}
2026-07-21 21:05:28 +00:00
out := make ( map [ string ] ReferenceSource , len ( values ) )
2026-08-09 00:29:04 +00:00
rawSlotNames := sortedStringMapKeys ( values )
for _ , rawSlotName := range rawSlotNames {
rawSource := values [ rawSlotName ]
2026-07-05 14:21:36 +00:00
slotName := strings . TrimSpace ( rawSlotName )
if slotName == "" {
return nil , fmt . Errorf ( "%s must not be empty" , keyName )
}
if _ , ok := out [ slotName ] ; ok {
return nil , fmt . Errorf ( "%s %q is duplicated after trimming" , keyName , slotName )
}
2026-07-21 21:05:28 +00:00
source := cloneReferenceSource ( rawSource )
if source . Artifact == nil && strings . TrimSpace ( source . Path ) == "" {
2026-07-05 14:21:36 +00:00
return nil , fmt . Errorf ( "%s %q source must not be empty" , keyName , slotName )
}
out [ slotName ] = source
}
return out , nil
}
2026-07-05 10:53:16 -05:00
func sortedArtifactLaneProfileKeys ( values map [ string ] ArtifactLaneProfile ) [ ] string {
if len ( values ) == 0 {
return nil
}
keys := make ( [ ] string , 0 , len ( values ) )
for key := range values {
keys = append ( keys , key )
}
sort . Strings ( keys )
return keys
}
2026-07-21 21:05:28 +00:00
func sortedStringMapKeys ( values map [ string ] ReferenceSource ) [ ] string {
2026-07-05 14:21:36 +00:00
if len ( values ) == 0 {
return nil
}
keys := make ( [ ] string , 0 , len ( values ) )
for key := range values {
keys = append ( keys , key )
}
sort . Strings ( keys )
return keys
}
func sortedReferenceBindingKeys ( values map [ string ] ReferenceBinding ) [ ] string {
if len ( values ) == 0 {
return nil
}
keys := make ( [ ] string , 0 , len ( values ) )
for key := range values {
keys = append ( keys , key )
}
sort . Strings ( keys )
return keys
}
2026-08-09 00:29:04 +00:00
func resolveBinding ( binding ModuleBinding , defaultModule string , referenceSlotLabel string ) ( ModuleBinding , error ) {
module := resolveModuleKey ( binding . Module , defaultModule )
2026-07-03 15:37:07 +00:00
llmProfile := strings . TrimSpace ( binding . LLMProfile )
2026-08-09 00:29:04 +00:00
references , err := normalizeReferenceMap ( binding . References , referenceSlotLabel )
if err != nil {
return ModuleBinding { } , err
}
2026-07-03 15:37:07 +00:00
return ModuleBinding {
2026-08-25 19:55:39 +00:00
Module : module ,
LLMProfile : llmProfile ,
StructuredOutputRepairAttempts : cloneStructuredOutputRepairAttempts ( binding . StructuredOutputRepairAttempts ) ,
2026-08-26 23:34:06 +00:00
ValidationPolicy : cloneValidationPolicyOverride ( binding . ValidationPolicy ) ,
2026-08-25 19:55:39 +00:00
Retries : binding . Retries ,
Options : cloneOptions ( binding . Options ) ,
References : references ,
Validators : cloneValidatorOverride ( binding . Validators ) ,
2026-08-09 00:29:04 +00:00
} , nil
}
func resolveModuleKey ( module string , defaultModule string ) string {
module = strings . TrimSpace ( module )
if module == "" {
return defaultModule
2026-07-03 15:37:07 +00:00
}
2026-08-09 00:29:04 +00:00
return module
2026-07-03 15:37:07 +00:00
}
2026-08-03 17:09:16 +00:00
func applyEffectiveLLMProfiles ( resolved * ResolvedPipeline , pipelineProfile , overrideProfile string ) error {
pipelineProfile = strings . TrimSpace ( pipelineProfile )
overrideProfile = strings . TrimSpace ( overrideProfile )
apply := func ( stage ModuleStage , laneID , module string , binding * ModuleBinding , executionClass contracts . ExecutionClass , kind string ) error {
binding . LLMProfile = strings . TrimSpace ( binding . LLMProfile )
if executionClass != contracts . ExecutionClassLLMBacked {
if binding . LLMProfile != "" {
if laneID == "" {
return fmt . Errorf ( "pipeline %q %s %q assigns llm_profile to deterministic %s %q" , resolved . ID , stage , module , kind , binding . Module )
}
return fmt . Errorf ( "pipeline %q lane %q %s %q assigns llm_profile to deterministic %s %q" , resolved . ID , laneID , stage , module , kind , binding . Module )
}
return nil
}
if overrideProfile != "" {
binding . LLMProfile = overrideProfile
return nil
}
if binding . LLMProfile == "" {
binding . LLMProfile = pipelineProfile
}
return nil
}
if err := apply ( StageInput , "" , resolved . Input . Module , & resolved . Input , resolved . InputExecutionClass , "module" ) ; err != nil {
return err
}
if err := apply ( StageChunk , "" , resolved . Chunk . Module , & resolved . Chunk , resolved . ChunkExecutionClass , "module" ) ; err != nil {
return err
}
for stepIndex := range resolved . Steps {
2026-08-25 19:55:39 +00:00
for laneIndex := range resolved . Steps [ stepIndex ] . ArtifactLanes {
lane := & resolved . Steps [ stepIndex ] . ArtifactLanes [ laneIndex ]
if err := apply ( StageExtract , lane . ID , lane . Extract . Module , & lane . Extract , lane . ExtractExecutionClass , "module" ) ; err != nil {
return err
}
if err := apply ( StageMerge , lane . ID , lane . Merge . Module , & lane . Merge , lane . MergeExecutionClass , "module" ) ; err != nil {
return err
}
if err := apply ( StageNormalize , lane . ID , lane . Normalize . Module , & lane . Normalize , lane . NormalizeExecutionClass , "module" ) ; err != nil {
return err
}
}
}
if err := apply ( StageOutput , "" , resolved . Output . Module , & resolved . Output , resolved . OutputExecutionClass , "module" ) ; err != nil {
return err
}
for chainIndex := range resolved . ValidatorChains {
chain := & resolved . ValidatorChains [ chainIndex ]
for validatorIndex := range chain . Validators {
validator := & chain . Validators [ validatorIndex ]
if err := apply ( chain . Stage , chain . LaneID , chain . ModuleKey , & validator . Binding , validator . ExecutionClass , "validator" ) ; err != nil {
return err
}
}
}
return nil
}
func applyEffectiveStructuredOutputRepairAttempts ( resolved * ResolvedPipeline , pipelineAttempts * int ) error {
apply := func ( stage ModuleStage , laneID , module string , binding * ModuleBinding , executionClass contracts . ExecutionClass , kind string ) error {
binding . StructuredOutputRepairAttempts = cloneStructuredOutputRepairAttempts ( binding . StructuredOutputRepairAttempts )
if executionClass != contracts . ExecutionClassLLMBacked {
if binding . StructuredOutputRepairAttempts != nil {
if laneID == "" {
return fmt . Errorf ( "pipeline %q %s %q assigns structured_output_repair_attempts to deterministic %s %q" , resolved . ID , stage , module , kind , binding . Module )
}
return fmt . Errorf ( "pipeline %q lane %q %s %q assigns structured_output_repair_attempts to deterministic %s %q" , resolved . ID , laneID , stage , module , kind , binding . Module )
}
return nil
}
if binding . StructuredOutputRepairAttempts == nil {
binding . StructuredOutputRepairAttempts = cloneStructuredOutputRepairAttempts ( pipelineAttempts )
}
return nil
}
if err := apply ( StageInput , "" , resolved . Input . Module , & resolved . Input , resolved . InputExecutionClass , "module" ) ; err != nil {
return err
}
if err := apply ( StageChunk , "" , resolved . Chunk . Module , & resolved . Chunk , resolved . ChunkExecutionClass , "module" ) ; err != nil {
return err
}
for stepIndex := range resolved . Steps {
2026-08-03 17:09:16 +00:00
for laneIndex := range resolved . Steps [ stepIndex ] . ArtifactLanes {
lane := & resolved . Steps [ stepIndex ] . ArtifactLanes [ laneIndex ]
if err := apply ( StageExtract , lane . ID , lane . Extract . Module , & lane . Extract , lane . ExtractExecutionClass , "module" ) ; err != nil {
return err
}
if err := apply ( StageMerge , lane . ID , lane . Merge . Module , & lane . Merge , lane . MergeExecutionClass , "module" ) ; err != nil {
return err
}
if err := apply ( StageNormalize , lane . ID , lane . Normalize . Module , & lane . Normalize , lane . NormalizeExecutionClass , "module" ) ; err != nil {
return err
}
}
}
if err := apply ( StageOutput , "" , resolved . Output . Module , & resolved . Output , resolved . OutputExecutionClass , "module" ) ; err != nil {
return err
}
for chainIndex := range resolved . ValidatorChains {
chain := & resolved . ValidatorChains [ chainIndex ]
for validatorIndex := range chain . Validators {
validator := & chain . Validators [ validatorIndex ]
if err := apply ( chain . Stage , chain . LaneID , chain . ModuleKey , & validator . Binding , validator . ExecutionClass , "validator" ) ; err != nil {
return err
}
}
}
return nil
}
2026-08-26 23:34:06 +00:00
func applyEffectiveValidationPolicies ( resolved * ResolvedPipeline , pipelinePolicy * ValidationPolicyOverride ) error {
apply := func ( stage ModuleStage , laneID string , binding ModuleBinding , executionClass contracts . ExecutionClass ) ( ValidationPolicy , error ) {
binding . ValidationPolicy = cloneValidationPolicyOverride ( binding . ValidationPolicy )
if binding . ValidationPolicy != nil {
if err := binding . ValidationPolicy . Validate ( ) ; err != nil {
return ValidationPolicy { } , fmt . Errorf ( "pipeline %q %s validation_policy: %w" , resolved . ID , stage , err )
}
if executionClass == contracts . ExecutionClassDeterministic && binding . ValidationPolicy . ProducerStructuralFailure != nil {
if laneID == "" {
return ValidationPolicy { } , fmt . Errorf ( "pipeline %q %s %q assigns producer_structural_failure to deterministic module" , resolved . ID , stage , binding . Module )
}
return ValidationPolicy { } , fmt . Errorf ( "pipeline %q lane %q %s %q assigns producer_structural_failure to deterministic module" , resolved . ID , laneID , stage , binding . Module )
}
}
return ResolveValidationPolicy ( binding . ValidationPolicy , pipelinePolicy ) , nil
}
if pipelinePolicy != nil {
if err := pipelinePolicy . Validate ( ) ; err != nil {
return fmt . Errorf ( "pipeline %q validation_policy: %w" , resolved . ID , err )
}
}
if resolved . Input . ValidationPolicy != nil {
return fmt . Errorf ( "pipeline %q input validation_policy is not supported" , resolved . ID )
}
if resolved . Output . ValidationPolicy != nil {
return fmt . Errorf ( "pipeline %q output validation_policy is not supported" , resolved . ID )
}
for _ , chain := range resolved . ValidatorChains {
for _ , validator := range chain . Validators {
if validator . Binding . ValidationPolicy != nil {
if chain . LaneID == "" {
return fmt . Errorf ( "pipeline %q %s validator %q validation_policy is not supported" , resolved . ID , chain . Stage , validator . Binding . Module )
}
return fmt . Errorf ( "pipeline %q lane %q %s validator %q validation_policy is not supported" , resolved . ID , chain . LaneID , chain . Stage , validator . Binding . Module )
}
}
}
policy , err := apply ( StageChunk , "" , resolved . Chunk , resolved . ChunkExecutionClass )
if err != nil {
return err
}
resolved . ChunkValidationPolicy = policy
for stepIndex := range resolved . Steps {
for laneIndex := range resolved . Steps [ stepIndex ] . ArtifactLanes {
lane := & resolved . Steps [ stepIndex ] . ArtifactLanes [ laneIndex ]
policy , err = apply ( StageExtract , lane . ID , lane . Extract , lane . ExtractExecutionClass )
if err != nil {
return err
}
lane . ExtractValidationPolicy = policy
policy , err = apply ( StageMerge , lane . ID , lane . Merge , lane . MergeExecutionClass )
if err != nil {
return err
}
lane . MergeValidationPolicy = policy
policy , err = apply ( StageNormalize , lane . ID , lane . Normalize , lane . NormalizeExecutionClass )
if err != nil {
return err
}
lane . NormalizeValidationPolicy = policy
}
}
return nil
}
2026-08-09 00:29:04 +00:00
func resolveBindings ( bindings [ ] ModuleBinding , defaultModule string , referenceSlotLabel string ) ( [ ] ModuleBinding , error ) {
2026-07-03 15:37:07 +00:00
if len ( bindings ) == 0 {
2026-08-09 00:29:04 +00:00
return nil , nil
2026-07-03 15:37:07 +00:00
}
resolved := make ( [ ] ModuleBinding , 0 , len ( bindings ) )
2026-08-09 00:29:04 +00:00
for index , binding := range bindings {
resolvedBinding , err := resolveBinding ( binding , defaultModule , fmt . Sprintf ( "%s %d" , referenceSlotLabel , index ) )
if err != nil {
return nil , err
}
2026-07-03 15:37:07 +00:00
resolved = append ( resolved , resolvedBinding )
}
2026-08-09 00:29:04 +00:00
return resolved , nil
2026-07-03 15:37:07 +00:00
}
func cloneOptions ( options map [ string ] any ) map [ string ] any {
if len ( options ) == 0 {
return nil
}
copied := make ( map [ string ] any , len ( options ) )
for key , value := range options {
2026-07-27 18:04:12 +00:00
copied [ key ] = cloneOptionValue ( value )
2026-07-03 15:37:07 +00:00
}
return copied
}
2026-07-27 18:04:12 +00:00
func cloneOptionValue ( value any ) any {
switch typed := value . ( type ) {
case map [ string ] any :
return cloneOptions ( typed )
case [ ] any :
out := make ( [ ] any , len ( typed ) )
for i := range typed {
out [ i ] = cloneOptionValue ( typed [ i ] )
}
return out
case [ ] string :
return append ( [ ] string ( nil ) , typed ... )
case [ ] byte :
return append ( [ ] byte ( nil ) , typed ... )
case json . RawMessage :
return append ( json . RawMessage ( nil ) , typed ... )
default :
return value
}
}
2026-07-03 15:37:07 +00:00
func selectedArtifactLanes ( pipelineID string , artifacts map [ string ] ArtifactLaneProfile , options ResolveOptions ) ( map [ string ] ArtifactLaneProfile , [ ] string , error ) {
lanesByID := make ( map [ string ] ArtifactLaneProfile , len ( artifacts ) )
for rawLaneID , lane := range artifacts {
laneID := strings . TrimSpace ( rawLaneID )
if laneID == "" {
return nil , nil , fmt . Errorf ( "pipeline %q artifact lane id must not be empty" , pipelineID )
}
if _ , ok := lanesByID [ laneID ] ; ok {
return nil , nil , fmt . Errorf ( "pipeline %q artifact lane %q is duplicated after trimming" , pipelineID , laneID )
}
lanesByID [ laneID ] = lane
}
if len ( options . Only ) == 0 {
keys := make ( [ ] string , 0 , len ( lanesByID ) )
for laneID := range lanesByID {
keys = append ( keys , laneID )
}
sort . Strings ( keys )
return lanesByID , keys , nil
}
selected := make ( map [ string ] struct { } , len ( options . Only ) )
for _ , rawLaneID := range options . Only {
laneID := strings . TrimSpace ( rawLaneID )
if laneID == "" {
return nil , nil , fmt . Errorf ( "pipeline %q selected artifact lane id must not be empty" , pipelineID )
}
if _ , ok := lanesByID [ laneID ] ; ! ok {
return nil , nil , fmt . Errorf ( "pipeline %q selected artifact lane %q is not declared" , pipelineID , laneID )
}
selected [ laneID ] = struct { } { }
}
keys := make ( [ ] string , 0 , len ( selected ) )
for laneID := range selected {
keys = append ( keys , laneID )
}
sort . Strings ( keys )
return lanesByID , keys , nil
}
func resolvedPipelineDigest ( resolved ResolvedPipeline ) ( string , error ) {
withoutDigest := struct {
2026-08-26 23:34:06 +00:00
ID string
ConfiguredValidationPolicy * ValidationPolicyOverride
Input ModuleBinding
InputExecutionClass contracts . ExecutionClass
Chunk ModuleBinding
ChunkExecutionClass contracts . ExecutionClass
2026-08-26 23:42:11 +00:00
ChunkCorrectionProtocol contracts . CorrectionProtocol
2026-08-26 23:34:06 +00:00
ChunkValidationPolicy ValidationPolicy
ChunkReferences ResolvedReferenceTarget
Steps [ ] ResolvedPipelineStep
ValidatorChains [ ] ResolvedValidatorChain
Output ModuleBinding
OutputExecutionClass contracts . ExecutionClass
2026-07-03 15:37:07 +00:00
} {
2026-08-26 23:34:06 +00:00
ID : resolved . ID ,
ConfiguredValidationPolicy : cloneValidationPolicyOverride ( resolved . ConfiguredValidationPolicy ) ,
ChunkValidationPolicy : resolved . ChunkValidationPolicy ,
Input : resolved . Input ,
InputExecutionClass : resolved . InputExecutionClass ,
Chunk : resolved . Chunk ,
ChunkExecutionClass : resolved . ChunkExecutionClass ,
2026-08-26 23:42:11 +00:00
ChunkCorrectionProtocol : resolved . ChunkCorrectionProtocol ,
2026-08-26 23:34:06 +00:00
ChunkReferences : resolved . ChunkReferences ,
Steps : resolved . Steps ,
ValidatorChains : resolved . ValidatorChains ,
Output : resolved . Output ,
OutputExecutionClass : resolved . OutputExecutionClass ,
2026-07-03 15:37:07 +00:00
}
encoded , err := json . Marshal ( withoutDigest )
if err != nil {
return "" , err
}
sum := sha256 . Sum256 ( encoded )
return "sha256:" + hex . EncodeToString ( sum [ : ] ) , nil
}
func inputSpec ( catalog ModuleCatalog , key string ) ( ModuleSpec , error ) {
return registrySpec ( catalog . Inputs , key )
}
func chunkerSpec ( catalog ModuleCatalog , key string ) ( ModuleSpec , error ) {
return registrySpec ( catalog . Chunkers , key )
}
func extractorSpec ( catalog ModuleCatalog , key string ) ( ModuleSpec , error ) {
return registrySpec ( catalog . Extractors , key )
}
func outputSpec ( catalog ModuleCatalog , key string ) ( ModuleSpec , error ) {
return registrySpec ( catalog . Outputs , key )
}
type specRegistry interface {
Spec ( key string ) ( ModuleSpec , bool )
}
func registrySpec ( registry specRegistry , key string ) ( ModuleSpec , error ) {
if registry == nil {
return ModuleSpec { } , fmt . Errorf ( "module %q is not registered" , key )
}
spec , ok := registry . Spec ( key )
if ! ok {
return ModuleSpec { } , fmt . Errorf ( "module %q is not registered" , key )
}
return spec , nil
}
func moduleLookupError ( pipelineID , laneID string , stage ModuleStage , module string , err error ) error {
if laneID != "" {
return fmt . Errorf ( "pipeline %q lane %q %s module %q: %w" , pipelineID , laneID , stage , module , err )
}
return fmt . Errorf ( "pipeline %q %s module %q: %w" , pipelineID , stage , module , err )
}
func capabilityError ( pipelineID , laneID string , stage ModuleStage , module , capability string ) error {
if laneID != "" {
return fmt . Errorf ( "pipeline %q lane %q %s module %q requires missing capability %q" , pipelineID , laneID , stage , module , capability )
}
return fmt . Errorf ( "pipeline %q %s module %q requires missing capability %q" , pipelineID , stage , module , capability )
}
type capabilitySet map [ string ] struct { }
func newCapabilitySet ( ) capabilitySet {
return make ( capabilitySet )
}
func ( set capabilitySet ) clone ( ) capabilitySet {
copied := make ( capabilitySet , len ( set ) )
for capability := range set {
copied [ capability ] = struct { } { }
}
return copied
}
func ( set capabilitySet ) add ( values ... string ) {
for _ , value := range values {
set [ value ] = struct { } { }
}
}
func ( set capabilitySet ) addSet ( other capabilitySet ) {
for value := range other {
set [ value ] = struct { } { }
}
}
func ( set capabilitySet ) missing ( required [ ] string ) ( string , bool ) {
for _ , capability := range required {
if _ , ok := set [ capability ] ; ! ok {
return capability , true
}
}
return "" , false
}