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

Declare producer correction capability

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-26 23:42:11 +00:00
parent 9d0faabf61
commit 85a5b52be7
12 changed files with 297 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
docs/config.md
View File

@@ -317,10 +317,12 @@ extract:
Omitting **validators** uses the registered chain. **validators: []** selects Omitting **validators** uses the registered chain. **validators: []** selects
an empty chain; a non-empty list replaces the chain in the listed order. an empty chain; a non-empty list replaces the chain in the listed order.
Validator bindings accept only **module**, **llm_profile**, Validator bindings accept only **module**, **llm_profile**,
**structured_output_repair_attempts**, and **options**. They reject **structured_output_repair_attempts**, **retries**, and **options**. Their
**validation_policy**, **retries** value is a non-negative validator-execution budget and is valid
**references**, **retries**, and nested **validators**. Deterministic validators only when the selected validator is LLM-backed. They reject
reject explicit **llm_profile** and **structured_output_repair_attempts**. **validation_policy**, **references**, and nested **validators**. Deterministic
validators reject explicit **llm_profile** and
**structured_output_repair_attempts**.
Deterministic module bindings also reject those explicit fields. Deterministic module bindings also reject those explicit fields.
The **json** output module accepts optional **include_chunk_map** and The **json** output module accepts optional **include_chunk_map** and

11
docs/internal/configuration.md
View File

@@ -82,8 +82,11 @@ and checkpoint identity; execution does not interpret configuration defaults.
The framework resolver supplies defaults, selects lanes, resolves validator The framework resolver supplies defaults, selects lanes, resolves validator
chains, checks registered module and artifact compatibility, validates module chains, checks registered module and artifact compatibility, validates module
options, and returns the fixed ordered pipeline shape. The resulting options, and returns the fixed ordered pipeline shape. Positive validator retry
**EffectiveConfig** retains the selected ID, requested selection and reference budgets require an LLM-backed selected validator; deterministic validators are
rejected during resolution. Eligible LLM-backed producer specifications also
contribute their declared correction protocol to the resolved metadata. The
resulting **EffectiveConfig** retains the selected ID, requested selection and reference
changes, a clone of the input configuration, and the resolved pipeline. changes, a clone of the input configuration, and the resolved pipeline.
Callers may therefore retain or modify their input slices and maps without Callers may therefore retain or modify their input slices and maps without
changing the resolved result, and later consumers cannot mutate the original changing the resolved result, and later consumers cannot mutate the original
@@ -100,8 +103,8 @@ runtime error class described in the [CLI reference](../cli.md#output-streams-an
The framework assigns the resolved pipeline a deterministic SHA-256 digest The framework assigns the resolved pipeline a deterministic SHA-256 digest
after defaults, lane selection, module bindings, reference bindings, validator after defaults, lane selection, module bindings, reference bindings, validator
chains, effective LLM profiles, and artifact schema identity have been chains, selected correction protocols, effective LLM profiles, and artifact
resolved. The digest excludes schema identity have been resolved. The digest excludes
its own stored value. It identifies resolved composition rather than raw YAML its own stored value. It identifies resolved composition rather than raw YAML
bytes, a debug payload, or all runtime state. The CLI records it as invocation bytes, a debug payload, or all runtime state. The CLI records it as invocation
provenance before execution; cache and checkpoint identity have additional provenance before execution; cache and checkpoint identity have additional

15
docs/internal/pipeline.md
View File

@@ -32,8 +32,11 @@ Resolution turns a configured pipeline profile into a **ResolvedPipeline**.
It normalizes the pipeline and lane identities, applies stage defaults, selects It normalizes the pipeline and lane identities, applies stage defaults, selects
requested lanes where that is supported, resolves validator chains, checks requested lanes where that is supported, resolves validator chains, checks
module capabilities and typed artifact compatibility, validates options, and module capabilities and typed artifact compatibility, validates options, and
assigns a deterministic resolved-composition digest. The resolved pipeline assigns a deterministic resolved-composition digest. A correction protocol is
contains bindings and declared reference targets, not external reference bytes. selected from each eligible LLM-backed producer specification and becomes part
of that resolved identity; only `single_response_v1` is currently supported.
The resolved pipeline contains bindings and declared reference targets, not
external reference bytes.
After selection, the resolver applies command, binding, and pipeline profile After selection, the resolver applies command, binding, and pipeline profile
precedence to LLM-backed bindings and validators only; prompt defaults remain precedence to LLM-backed bindings and validators only; prompt defaults remain
an empty resolved binding profile. It resolves structural output repair an empty resolved binding profile. It resolves structural output repair
@@ -59,9 +62,11 @@ remains declared but has no bytes until its producing step completes.
Preparation is the construction boundary. It validates the resolved shape and Preparation is the construction boundary. It validates the resolved shape and
registry set, clones the resolved data, then constructs the input adapter, registry set, clones the resolved data, then constructs the input adapter,
chunker, stage-local validators, every typed lane, and output encoder. Each chunker, stage-local validators, every typed lane, and output encoder. The
registered builder receives its own cloned build request immediately before its prepared producer metadata preserves each selected correction protocol, and
module-owned code runs. Preparation also collects stable checkpoint the resolved digest carrying that metadata participates in checkpoint identity.
Each registered builder receives its own cloned build request immediately
before its module-owned code runs. Preparation also collects stable checkpoint
fingerprints. Missing registrations, incompatible typed entries, nil fingerprints. Missing registrations, incompatible typed entries, nil
implementations, and constructor failures are reported before source parsing implementations, and constructor failures are reported before source parsing
or any stage operation begins. or any stage operation begins.

2
docs/roadmap/implementation.md
View File

@@ -236,7 +236,7 @@ configuration fails before execution, and policy changes invalidate resolved
identity. The runner still follows its old behavior until later stages. This identity. The runner still follows its old behavior until later stages. This
stage is one Terra prompt. stage is one Terra prompt.
## Stage 4 — Declare And Validate Producer Correction Capability ## Stage 4 ✅ — Declare And Validate Producer Correction Capability
### Goal ### Goal

4
internal/core/config/validation.go
View File

@@ -273,8 +273,8 @@ func validateValidatorOverride(pipelineID string, laneID string, slot string, ov
if validator.ValidationPolicy != nil { if validator.ValidationPolicy != nil {
return fmt.Errorf("%s validation_policy is not supported", context) return fmt.Errorf("%s validation_policy is not supported", context)
} }
if validator.Retries != 0 { if validator.Retries < 0 {
return fmt.Errorf("%s retries are not supported", context) return fmt.Errorf("%s retries must be greater than or equal to zero", context)
} }
if validator.LLMProfile != "" && strings.TrimSpace(validator.LLMProfile) == "" { if validator.LLMProfile != "" && strings.TrimSpace(validator.LLMProfile) == "" {
return fmt.Errorf("%s llm_profile must not be empty when set", context) return fmt.Errorf("%s llm_profile must not be empty when set", context)

6
internal/core/config/validation_contract_test.go
View File

@@ -374,17 +374,17 @@ func TestValidateValidatorBindingRules(t *testing.T) {
want: "chunk validators[0] module must not be empty", want: "chunk validators[0] module must not be empty",
}, },
{ {
name: "validator retries", name: "negative validator retries",
setup: func(profile *pipeline.PipelineProfile) { setup: func(profile *pipeline.PipelineProfile) {
profile.Chunk.Validators = pipeline.ValidatorOverride{ profile.Chunk.Validators = pipeline.ValidatorOverride{
Set: true, Set: true,
Validators: []pipeline.ModuleBinding{{ Validators: []pipeline.ModuleBinding{{
Module: "validator", Module: "validator",
Retries: 1, Retries: -1,
}}, }},
} }
}, },
want: "chunk validators[0] retries are not supported", want: "chunk validators[0] retries must be greater than or equal to zero",
}, },
{ {
name: "validator references", name: "validator references",

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

@@ -21,25 +21,28 @@ const (
) )
type ModuleSpec struct { type ModuleSpec struct {
Key string Key string
Stage ModuleStage Stage ModuleStage
ExecutionClass contracts.ExecutionClass ExecutionClass contracts.ExecutionClass
ArtifactKind contracts.ArtifactKind CorrectionProtocol contracts.CorrectionProtocol
Provides []string ArtifactKind contracts.ArtifactKind
Requires []string Provides []string
ReferenceSlots []contracts.ReferenceSlot Requires []string
ReferenceSlots []contracts.ReferenceSlot
} }
func normalizeModuleSpec(spec ModuleSpec) ModuleSpec { func normalizeModuleSpec(spec ModuleSpec) ModuleSpec {
executionClass := contracts.ExecutionClass(strings.TrimSpace(string(spec.ExecutionClass))) executionClass := contracts.ExecutionClass(strings.TrimSpace(string(spec.ExecutionClass)))
correctionProtocol := contracts.CorrectionProtocol(strings.TrimSpace(string(spec.CorrectionProtocol)))
return ModuleSpec{ return ModuleSpec{
Key: strings.TrimSpace(spec.Key), Key: strings.TrimSpace(spec.Key),
Stage: spec.Stage, Stage: spec.Stage,
ExecutionClass: executionClass, ExecutionClass: executionClass,
ArtifactKind: normalizeArtifactKind(spec.ArtifactKind), CorrectionProtocol: correctionProtocol,
Provides: normalizeCapabilities(spec.Provides), ArtifactKind: normalizeArtifactKind(spec.ArtifactKind),
Requires: normalizeCapabilities(spec.Requires), Provides: normalizeCapabilities(spec.Provides),
ReferenceSlots: normalizeReferenceSlots(spec.ReferenceSlots), Requires: normalizeCapabilities(spec.Requires),
ReferenceSlots: normalizeReferenceSlots(spec.ReferenceSlots),
} }
} }
@@ -70,13 +73,14 @@ func normalizeCapabilities(values []string) []string {
func cloneModuleSpec(spec ModuleSpec) ModuleSpec { func cloneModuleSpec(spec ModuleSpec) ModuleSpec {
return ModuleSpec{ return ModuleSpec{
Key: spec.Key, Key: spec.Key,
Stage: spec.Stage, Stage: spec.Stage,
ExecutionClass: spec.ExecutionClass, ExecutionClass: spec.ExecutionClass,
ArtifactKind: spec.ArtifactKind, CorrectionProtocol: spec.CorrectionProtocol,
Provides: append([]string(nil), spec.Provides...), ArtifactKind: spec.ArtifactKind,
Requires: append([]string(nil), spec.Requires...), Provides: append([]string(nil), spec.Provides...),
ReferenceSlots: contracts.CloneReferenceSlots(spec.ReferenceSlots), Requires: append([]string(nil), spec.Requires...),
ReferenceSlots: contracts.CloneReferenceSlots(spec.ReferenceSlots),
} }
} }
@@ -93,6 +97,19 @@ func validateModuleSpec(kind string, expectedStage ModuleStage, spec ModuleSpec)
if spec.ExecutionClass != contracts.ExecutionClassDeterministic && spec.ExecutionClass != contracts.ExecutionClassLLMBacked { if spec.ExecutionClass != contracts.ExecutionClassDeterministic && spec.ExecutionClass != contracts.ExecutionClassLLMBacked {
return fmt.Errorf("%s %q has unsupported execution class %q", kind, spec.Key, spec.ExecutionClass) return fmt.Errorf("%s %q has unsupported execution class %q", kind, spec.Key, spec.ExecutionClass)
} }
if spec.CorrectionProtocol != "" {
if err := spec.CorrectionProtocol.Validate(); err != nil {
return fmt.Errorf("%s %q correction protocol: %w", kind, spec.Key, err)
}
if spec.ExecutionClass != contracts.ExecutionClassLLMBacked {
return fmt.Errorf("%s %q correction protocol requires an LLM-backed execution class", kind, spec.Key)
}
switch spec.Stage {
case StageChunk, StageExtract, StageMerge, StageNormalize:
default:
return fmt.Errorf("%s %q correction protocol is not supported for %q stage", kind, spec.Key, spec.Stage)
}
}
if spec.ArtifactKind != "" && spec.Stage != StageExtract && spec.Stage != StageMerge && spec.Stage != StageNormalize { if spec.ArtifactKind != "" && spec.Stage != StageExtract && spec.Stage != StageMerge && spec.Stage != StageNormalize {
return fmt.Errorf("%s %q must not declare an artifact kind", kind, spec.Key) return fmt.Errorf("%s %q must not declare an artifact kind", kind, spec.Key)
} }

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

@@ -32,14 +32,70 @@ func TestValidateModuleSpecRequiresSupportedExecutionClass(t *testing.T) {
} }
} }
func TestCloneModuleSpecPreservesExecutionClass(t *testing.T) { func TestCloneModuleSpecPreservesCorrectionProtocol(t *testing.T) {
spec := normalizeModuleSpec(ModuleSpec{Key: " module ", Stage: StageChunk, ExecutionClass: contracts.ExecutionClassLLMBacked}) spec := normalizeModuleSpec(ModuleSpec{
Key: " module ",
Stage: StageChunk,
ExecutionClass: contracts.ExecutionClassLLMBacked,
CorrectionProtocol: " single_response_v1 ",
})
if spec.CorrectionProtocol != contracts.CorrectionProtocolSingleResponseV1 {
t.Fatalf("normalized CorrectionProtocol = %q, want %q", spec.CorrectionProtocol, contracts.CorrectionProtocolSingleResponseV1)
}
cloned := cloneModuleSpec(spec) cloned := cloneModuleSpec(spec)
if !reflect.DeepEqual(cloned, spec) { if !reflect.DeepEqual(cloned, spec) {
t.Fatalf("cloneModuleSpec() = %#v, want %#v", cloned, spec) t.Fatalf("cloneModuleSpec() = %#v, want %#v", cloned, spec)
} }
} }
func TestValidateModuleSpecCorrectionProtocolEligibility(t *testing.T) {
tests := []struct {
name string
stage ModuleStage
class contracts.ExecutionClass
protocol contracts.CorrectionProtocol
want string
}{
{name: "LLM chunk", stage: StageChunk, class: contracts.ExecutionClassLLMBacked, protocol: contracts.CorrectionProtocolSingleResponseV1},
{name: "LLM extract", stage: StageExtract, class: contracts.ExecutionClassLLMBacked, protocol: contracts.CorrectionProtocolSingleResponseV1},
{name: "LLM merge", stage: StageMerge, class: contracts.ExecutionClassLLMBacked, protocol: contracts.CorrectionProtocolSingleResponseV1},
{name: "LLM normalize", stage: StageNormalize, class: contracts.ExecutionClassLLMBacked, protocol: contracts.CorrectionProtocolSingleResponseV1},
{name: "deterministic chunk", stage: StageChunk, class: contracts.ExecutionClassDeterministic, protocol: contracts.CorrectionProtocolSingleResponseV1, want: "LLM-backed"},
{name: "input", stage: StageInput, class: contracts.ExecutionClassLLMBacked, protocol: contracts.CorrectionProtocolSingleResponseV1, want: "not supported"},
{name: "output", stage: StageOutput, class: contracts.ExecutionClassLLMBacked, protocol: contracts.CorrectionProtocolSingleResponseV1, want: "not supported"},
{name: "unknown protocol", stage: StageChunk, class: contracts.ExecutionClassLLMBacked, protocol: "unsupported", want: "unsupported correction protocol"},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
spec := normalizeModuleSpec(ModuleSpec{
Key: "module",
Stage: test.stage,
ExecutionClass: test.class,
CorrectionProtocol: test.protocol,
})
err := validateModuleSpec("module", test.stage, spec)
if test.want == "" && err != nil {
t.Fatalf("validateModuleSpec() error = %v, want nil", err)
}
if test.want != "" && (err == nil || !strings.Contains(err.Error(), test.want)) {
t.Fatalf("validateModuleSpec() error = %v, want %q", err, test.want)
}
})
}
}
func TestNormalizeValidatorSpecRejectsCorrectionProtocol(t *testing.T) {
_, err := normalizeValidatorSpec(ValidatorSpec{
Key: "validator",
ExecutionClass: contracts.ExecutionClassLLMBacked,
CorrectionProtocol: contracts.CorrectionProtocolSingleResponseV1,
})
if err == nil || !strings.Contains(err.Error(), "correction protocol") {
t.Fatalf("normalizeValidatorSpec() error = %v, want correction protocol error", err)
}
}
func TestValidateModuleSpecAllowsReferenceSlotsForEligibleStages(t *testing.T) { func TestValidateModuleSpecAllowsReferenceSlotsForEligibleStages(t *testing.T) {
tests := []struct { tests := []struct {
name string name string

34
internal/framework/pipeline/prepare.go
View File

@@ -13,10 +13,11 @@ import (
// resolved pipeline. Its implementation values are private so execution cannot // resolved pipeline. Its implementation values are private so execution cannot
// replace or reconfigure them after preparation. // replace or reconfigure them after preparation.
type PreparedPipeline struct { type PreparedPipeline struct {
Input ModuleBinding Input ModuleBinding
Chunk ModuleBinding Chunk ModuleBinding
Steps []PreparedPipelineStep ChunkCorrectionProtocol contracts.CorrectionProtocol
Output ModuleBinding Steps []PreparedPipelineStep
Output ModuleBinding
resolved ResolvedPipeline resolved ResolvedPipeline
dependencies ModuleDependencies dependencies ModuleDependencies
@@ -36,7 +37,10 @@ type PreparedPipelineStep struct {
} }
type PreparedArtifactLane struct { type PreparedArtifactLane struct {
Resolved ResolvedArtifactLane Resolved ResolvedArtifactLane
ExtractCorrectionProtocol contracts.CorrectionProtocol
MergeCorrectionProtocol contracts.CorrectionProtocol
NormalizeCorrectionProtocol contracts.CorrectionProtocol
} }
type preparedLaneExecutor struct { type preparedLaneExecutor struct {
@@ -92,12 +96,13 @@ func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDepend
} }
stable := cloneResolvedPipeline(resolved) stable := cloneResolvedPipeline(resolved)
prepared := &PreparedPipeline{ prepared := &PreparedPipeline{
Input: cloneModuleBinding(stable.Input), Input: cloneModuleBinding(stable.Input),
Chunk: cloneModuleBinding(stable.Chunk), Chunk: cloneModuleBinding(stable.Chunk),
Output: cloneModuleBinding(stable.Output), ChunkCorrectionProtocol: stable.ChunkCorrectionProtocol,
resolved: stable, Output: cloneModuleBinding(stable.Output),
dependencies: deps, resolved: stable,
artifactCodecs: registries.ArtifactCodecs, dependencies: deps,
artifactCodecs: registries.ArtifactCodecs,
} }
request := func(binding ModuleBinding, references contracts.ReferenceSet) BuildRequest { request := func(binding ModuleBinding, references contracts.ReferenceSet) BuildRequest {
return BuildRequest{Dependencies: deps, Options: binding.Options, References: references} return BuildRequest{Dependencies: deps, Options: binding.Options, References: references}
@@ -131,7 +136,12 @@ func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDepend
if err != nil { if err != nil {
return nil, err return nil, err
} }
preparedStep.ArtifactLanes = append(preparedStep.ArtifactLanes, PreparedArtifactLane{Resolved: cloneResolvedArtifactLane(lane)}) preparedStep.ArtifactLanes = append(preparedStep.ArtifactLanes, PreparedArtifactLane{
Resolved: cloneResolvedArtifactLane(lane),
ExtractCorrectionProtocol: lane.ExtractCorrectionProtocol,
MergeCorrectionProtocol: lane.MergeCorrectionProtocol,
NormalizeCorrectionProtocol: lane.NormalizeCorrectionProtocol,
})
preparedStep.lanes = append(preparedStep.lanes, executor) preparedStep.lanes = append(preparedStep.lanes, executor)
} }
prepared.Steps[stepIndex] = preparedStep prepared.Steps[stepIndex] = preparedStep

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

@@ -204,26 +204,29 @@ type ResolvedReferenceTarget struct {
} }
type ResolvedArtifactLane struct { type ResolvedArtifactLane struct {
StepID string StepID string
ID string ID string
ArtifactKind contracts.ArtifactKind `json:"artifact_kind,omitempty"` ArtifactKind contracts.ArtifactKind `json:"artifact_kind,omitempty"`
ArtifactSchemaID string `json:"artifact_schema_id,omitempty"` ArtifactSchemaID string `json:"artifact_schema_id,omitempty"`
ArtifactSchemaName string `json:"artifact_schema_name,omitempty"` ArtifactSchemaName string `json:"artifact_schema_name,omitempty"`
ArtifactSchemaVersion string `json:"artifact_schema_version,omitempty"` ArtifactSchemaVersion string `json:"artifact_schema_version,omitempty"`
ArtifactSchemaDigest string `json:"artifact_schema_digest,omitempty"` ArtifactSchemaDigest string `json:"artifact_schema_digest,omitempty"`
Extract ModuleBinding Extract ModuleBinding
ExtractExecutionClass contracts.ExecutionClass `json:"extract_execution_class"` ExtractExecutionClass contracts.ExecutionClass `json:"extract_execution_class"`
ExtractValidationPolicy ValidationPolicy `json:"extract_validation_policy"` ExtractCorrectionProtocol contracts.CorrectionProtocol `json:"extract_correction_protocol,omitempty"`
Merge ModuleBinding ExtractValidationPolicy ValidationPolicy `json:"extract_validation_policy"`
MergeExecutionClass contracts.ExecutionClass `json:"merge_execution_class"` Merge ModuleBinding
MergeValidationPolicy ValidationPolicy `json:"merge_validation_policy"` MergeExecutionClass contracts.ExecutionClass `json:"merge_execution_class"`
Normalize ModuleBinding MergeCorrectionProtocol contracts.CorrectionProtocol `json:"merge_correction_protocol,omitempty"`
NormalizeExecutionClass contracts.ExecutionClass `json:"normalize_execution_class"` MergeValidationPolicy ValidationPolicy `json:"merge_validation_policy"`
NormalizeValidationPolicy ValidationPolicy `json:"normalize_validation_policy"` Normalize ModuleBinding
Validators []ModuleBinding NormalizeExecutionClass contracts.ExecutionClass `json:"normalize_execution_class"`
ExtractReferences ResolvedReferenceTarget `json:"extract_references"` NormalizeCorrectionProtocol contracts.CorrectionProtocol `json:"normalize_correction_protocol,omitempty"`
MergeReferences ResolvedReferenceTarget `json:"merge_references"` NormalizeValidationPolicy ValidationPolicy `json:"normalize_validation_policy"`
NormalizeReferences ResolvedReferenceTarget `json:"normalize_references"` Validators []ModuleBinding
ExtractReferences ResolvedReferenceTarget `json:"extract_references"`
MergeReferences ResolvedReferenceTarget `json:"merge_references"`
NormalizeReferences ResolvedReferenceTarget `json:"normalize_references"`
} }
type ResolvedPipelineStep struct { type ResolvedPipelineStep struct {
@@ -252,9 +255,10 @@ type ResolvedPipeline struct {
Input ModuleBinding Input ModuleBinding
InputExecutionClass contracts.ExecutionClass `json:"input_execution_class"` InputExecutionClass contracts.ExecutionClass `json:"input_execution_class"`
Chunk ModuleBinding Chunk ModuleBinding
ChunkExecutionClass contracts.ExecutionClass `json:"chunk_execution_class"` ChunkExecutionClass contracts.ExecutionClass `json:"chunk_execution_class"`
ChunkValidationPolicy ValidationPolicy `json:"chunk_validation_policy"` ChunkCorrectionProtocol contracts.CorrectionProtocol `json:"chunk_correction_protocol,omitempty"`
ChunkReferences ResolvedReferenceTarget `json:"chunk_references"` ChunkValidationPolicy ValidationPolicy `json:"chunk_validation_policy"`
ChunkReferences ResolvedReferenceTarget `json:"chunk_references"`
Steps []ResolvedPipelineStep Steps []ResolvedPipelineStep
ValidatorChains []ResolvedValidatorChain `json:"validator_chains"` ValidatorChains []ResolvedValidatorChain `json:"validator_chains"`
Output ModuleBinding Output ModuleBinding
@@ -431,6 +435,7 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo
InputExecutionClass: inputModuleSpec.ExecutionClass, InputExecutionClass: inputModuleSpec.ExecutionClass,
Chunk: chunk, Chunk: chunk,
ChunkExecutionClass: chunkSpec.ExecutionClass, ChunkExecutionClass: chunkSpec.ExecutionClass,
ChunkCorrectionProtocol: chunkSpec.CorrectionProtocol,
ChunkReferences: referenceTarget(StageChunk, "", chunk.Module, chunkReferences), ChunkReferences: referenceTarget(StageChunk, "", chunk.Module, chunkReferences),
Output: output, Output: output,
} }
@@ -615,6 +620,7 @@ func resolveArtifactLane(
lane.ExtractReferences = referenceTarget(StageExtract, laneID, lane.Extract.Module, references) lane.ExtractReferences = referenceTarget(StageExtract, laneID, lane.Extract.Module, references)
lane.ExtractReferences.StepID = strings.TrimSpace(stepID) lane.ExtractReferences.StepID = strings.TrimSpace(stepID)
lane.ExtractExecutionClass = extractSpec.ExecutionClass lane.ExtractExecutionClass = extractSpec.ExecutionClass
lane.ExtractCorrectionProtocol = extractSpec.CorrectionProtocol
capabilities.add(extractSpec.Provides...) capabilities.add(extractSpec.Provides...)
mergeSpec, err := mergerSpecForArtifact(catalog, lane.Merge.Module, lane.ArtifactKind, artifactType) mergeSpec, err := mergerSpecForArtifact(catalog, lane.Merge.Module, lane.ArtifactKind, artifactType)
@@ -641,6 +647,7 @@ func resolveArtifactLane(
lane.MergeReferences = referenceTarget(StageMerge, laneID, lane.Merge.Module, mergeReferences) lane.MergeReferences = referenceTarget(StageMerge, laneID, lane.Merge.Module, mergeReferences)
lane.MergeReferences.StepID = strings.TrimSpace(stepID) lane.MergeReferences.StepID = strings.TrimSpace(stepID)
lane.MergeExecutionClass = mergeSpec.ExecutionClass lane.MergeExecutionClass = mergeSpec.ExecutionClass
lane.MergeCorrectionProtocol = mergeSpec.CorrectionProtocol
capabilities.add(mergeSpec.Provides...) capabilities.add(mergeSpec.Provides...)
normalizeSpec, err := normalizerSpecForArtifact(catalog, lane.Normalize.Module, lane.ArtifactKind, artifactType) normalizeSpec, err := normalizerSpecForArtifact(catalog, lane.Normalize.Module, lane.ArtifactKind, artifactType)
@@ -667,6 +674,7 @@ func resolveArtifactLane(
lane.NormalizeReferences = referenceTarget(StageNormalize, laneID, lane.Normalize.Module, normalizeReferences) lane.NormalizeReferences = referenceTarget(StageNormalize, laneID, lane.Normalize.Module, normalizeReferences)
lane.NormalizeReferences.StepID = strings.TrimSpace(stepID) lane.NormalizeReferences.StepID = strings.TrimSpace(stepID)
lane.NormalizeExecutionClass = normalizeSpec.ExecutionClass lane.NormalizeExecutionClass = normalizeSpec.ExecutionClass
lane.NormalizeCorrectionProtocol = normalizeSpec.CorrectionProtocol
capabilities.add(normalizeSpec.Provides...) capabilities.add(normalizeSpec.Provides...)
if len(lane.Validators) > 0 { if len(lane.Validators) > 0 {
@@ -928,6 +936,9 @@ func resolveValidatorChain(pipelineID string, laneID string, stage ModuleStage,
if err != nil { if err != nil {
return ResolvedValidatorChain{}, fmt.Errorf("pipeline %q %s validator chain for module %q: %w", pipelineID, stage, chain.ModuleKey, err) return ResolvedValidatorChain{}, fmt.Errorf("pipeline %q %s validator chain for module %q: %w", pipelineID, stage, chain.ModuleKey, err)
} }
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)
}
chain.Validators = append(chain.Validators, ResolvedValidator{ chain.Validators = append(chain.Validators, ResolvedValidator{
Binding: cloneModuleBinding(validator), Binding: cloneModuleBinding(validator),
ExecutionClass: spec.ExecutionClass, ExecutionClass: spec.ExecutionClass,
@@ -1637,6 +1648,7 @@ func resolvedPipelineDigest(resolved ResolvedPipeline) (string, error) {
InputExecutionClass contracts.ExecutionClass InputExecutionClass contracts.ExecutionClass
Chunk ModuleBinding Chunk ModuleBinding
ChunkExecutionClass contracts.ExecutionClass ChunkExecutionClass contracts.ExecutionClass
ChunkCorrectionProtocol contracts.CorrectionProtocol
ChunkValidationPolicy ValidationPolicy ChunkValidationPolicy ValidationPolicy
ChunkReferences ResolvedReferenceTarget ChunkReferences ResolvedReferenceTarget
Steps []ResolvedPipelineStep Steps []ResolvedPipelineStep
@@ -1651,6 +1663,7 @@ func resolvedPipelineDigest(resolved ResolvedPipeline) (string, error) {
InputExecutionClass: resolved.InputExecutionClass, InputExecutionClass: resolved.InputExecutionClass,
Chunk: resolved.Chunk, Chunk: resolved.Chunk,
ChunkExecutionClass: resolved.ChunkExecutionClass, ChunkExecutionClass: resolved.ChunkExecutionClass,
ChunkCorrectionProtocol: resolved.ChunkCorrectionProtocol,
ChunkReferences: resolved.ChunkReferences, ChunkReferences: resolved.ChunkReferences,
Steps: resolved.Steps, Steps: resolved.Steps,
ValidatorChains: resolved.ValidatorChains, ValidatorChains: resolved.ValidatorChains,

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

@@ -126,6 +126,109 @@ func TestModuleCatalogExecutionClassLooksUpRegisteredMetadata(t *testing.T) {
} }
} }
func TestResolvePipelineCarriesCorrectionProtocolsIntoPreparedMetadata(t *testing.T) {
correction := contracts.CorrectionProtocolSingleResponseV1
catalog := newProfileCatalogWithOverrides(t,
ModuleSpec{Key: "llm-input", Stage: StageInput, ExecutionClass: contracts.ExecutionClassLLMBacked, Provides: []string{"source"}},
ModuleSpec{Key: "llm-chunk", Stage: StageChunk, ExecutionClass: contracts.ExecutionClassLLMBacked, CorrectionProtocol: correction, Requires: []string{"source"}, Provides: []string{"chunk"}},
ModuleSpec{Key: "llm-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassLLMBacked, CorrectionProtocol: correction, ArtifactKind: "test/notes", Requires: []string{"chunk"}, Provides: []string{"candidate"}},
ModuleSpec{Key: "llm-merge", Stage: StageMerge, ExecutionClass: contracts.ExecutionClassLLMBacked, CorrectionProtocol: correction, ArtifactKind: "test/notes", Requires: []string{"candidate"}, Provides: []string{"merged"}},
ModuleSpec{Key: "llm-normalize", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, CorrectionProtocol: correction, ArtifactKind: "test/notes", Requires: []string{"merged"}, Provides: []string{"normalized"}},
ModuleSpec{Key: "llm-output", Stage: StageOutput, ExecutionClass: contracts.ExecutionClassLLMBacked, Requires: []string{"normalized"}, Provides: []string{"encoded"}},
)
resolved, err := ResolvePipeline(llmProfilePipeline(), ResolveOptions{}, catalog)
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
if resolved.ChunkCorrectionProtocol != correction {
t.Fatalf("ChunkCorrectionProtocol = %q, want %q", resolved.ChunkCorrectionProtocol, correction)
}
lane := resolved.Steps[0].ArtifactLanes[0]
if lane.ExtractCorrectionProtocol != correction || lane.MergeCorrectionProtocol != correction || lane.NormalizeCorrectionProtocol != correction {
t.Fatalf("lane correction protocols = %q/%q/%q, want %q", lane.ExtractCorrectionProtocol, lane.MergeCorrectionProtocol, lane.NormalizeCorrectionProtocol, correction)
}
withoutCapability := cloneResolvedPipeline(resolved)
withoutCapability.ChunkCorrectionProtocol = ""
withoutCapability.Steps[0].ArtifactLanes[0].ExtractCorrectionProtocol = ""
withoutCapability.Steps[0].ArtifactLanes[0].MergeCorrectionProtocol = ""
withoutCapability.Steps[0].ArtifactLanes[0].NormalizeCorrectionProtocol = ""
withoutCapabilityDigest, err := resolvedPipelineDigest(withoutCapability)
if err != nil {
t.Fatalf("resolvedPipelineDigest() error = %v", err)
}
if withoutCapabilityDigest == resolved.Digest {
t.Fatalf("resolved digest = %q with and without correction capability, want changed", resolved.Digest)
}
prepared, err := Prepare(resolved, registriesFromModuleCatalog(catalog), ModuleDependencies{})
if err != nil {
t.Fatalf("Prepare() error = %v, want nil", err)
}
if prepared.ChunkCorrectionProtocol != correction {
t.Fatalf("prepared ChunkCorrectionProtocol = %q, want %q", prepared.ChunkCorrectionProtocol, correction)
}
preparedLane := prepared.Steps[0].ArtifactLanes[0]
if preparedLane.ExtractCorrectionProtocol != correction || preparedLane.MergeCorrectionProtocol != correction || preparedLane.NormalizeCorrectionProtocol != correction {
t.Fatalf("prepared lane correction protocols = %q/%q/%q, want %q", preparedLane.ExtractCorrectionProtocol, preparedLane.MergeCorrectionProtocol, preparedLane.NormalizeCorrectionProtocol, correction)
}
}
func TestPrepareAllowsLLMProducerWithoutCorrectionCapability(t *testing.T) {
catalog := newProfileCatalogWithOverrides(t,
ModuleSpec{Key: "llm-input", Stage: StageInput, ExecutionClass: contracts.ExecutionClassLLMBacked, Provides: []string{"source"}},
ModuleSpec{Key: "llm-chunk", Stage: StageChunk, ExecutionClass: contracts.ExecutionClassLLMBacked, Requires: []string{"source"}, Provides: []string{"chunk"}},
ModuleSpec{Key: "llm-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassLLMBacked, ArtifactKind: "test/notes", Requires: []string{"chunk"}, Provides: []string{"candidate"}},
ModuleSpec{Key: "llm-merge", Stage: StageMerge, ExecutionClass: contracts.ExecutionClassLLMBacked, ArtifactKind: "test/notes", Requires: []string{"candidate"}, Provides: []string{"merged"}},
ModuleSpec{Key: "llm-normalize", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, ArtifactKind: "test/notes", Requires: []string{"merged"}, Provides: []string{"normalized"}},
ModuleSpec{Key: "llm-output", Stage: StageOutput, ExecutionClass: contracts.ExecutionClassLLMBacked, Requires: []string{"normalized"}, Provides: []string{"encoded"}},
)
profile := llmProfilePipeline()
profile.Chunk.Retries = 1
resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog)
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
if resolved.ChunkCorrectionProtocol != "" {
t.Fatalf("ChunkCorrectionProtocol = %q, want empty unsupported value", resolved.ChunkCorrectionProtocol)
}
if _, err := Prepare(resolved, registriesFromModuleCatalog(catalog), ModuleDependencies{}); err != nil {
t.Fatalf("Prepare() error = %v, want nil", err)
}
}
func TestResolvePipelineValidatorRetriesRequireLLMBackedValidator(t *testing.T) {
for _, test := range []struct {
name string
class contracts.ExecutionClass
want string
}{
{name: "deterministic validator", class: contracts.ExecutionClassDeterministic, want: "retries require an LLM-backed"},
{name: "LLM validator", class: contracts.ExecutionClassLLMBacked},
} {
t.Run(test.name, func(t *testing.T) {
catalog := newProfileCatalog(t)
registerProfileValidatorSpec(t, catalog, ValidatorSpec{Key: "retrying-validator", ExecutionClass: test.class})
profile := baselineProfile()
lane := profile.Artifacts["events"]
lane.Extract.Validators = ValidatorOverride{Set: true, Validators: []ModuleBinding{{Module: "retrying-validator", Retries: 1}}}
profile.Artifacts["events"] = lane
resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog)
if test.want != "" {
if err == nil || !strings.Contains(err.Error(), test.want) {
t.Fatalf("ResolvePipeline() error = %v, want %q", err, test.want)
}
return
}
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
if got := resolved.ValidatorChains[1].Validators[0].Binding.Retries; got != 1 {
t.Fatalf("validator retries = %d, want 1", got)
}
})
}
}
func TestResolvePipelineAppliesDefaults(t *testing.T) { func TestResolvePipelineAppliesDefaults(t *testing.T) {
resolved, err := ResolvePipeline(PipelineProfile{ resolved, err := ResolvePipeline(PipelineProfile{
ID: "defaulted", ID: "defaulted",

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

@@ -11,8 +11,9 @@ import (
) )
type ValidatorSpec struct { type ValidatorSpec struct {
Key string `json:"key"` Key string `json:"key"`
ExecutionClass contracts.ExecutionClass `json:"execution_class"` ExecutionClass contracts.ExecutionClass `json:"execution_class"`
CorrectionProtocol contracts.CorrectionProtocol `json:"correction_protocol,omitempty"`
} }
type SerializedValidatorSpec struct { type SerializedValidatorSpec struct {
@@ -321,7 +322,11 @@ func (r *ValidatorRegistry) RegisteredKeys() []string {
} }
func normalizeValidatorSpec(spec ValidatorSpec) (ValidatorSpec, error) { func normalizeValidatorSpec(spec ValidatorSpec) (ValidatorSpec, error) {
normalized := ValidatorSpec{Key: strings.TrimSpace(spec.Key), ExecutionClass: spec.ExecutionClass} normalized := ValidatorSpec{
Key: strings.TrimSpace(spec.Key),
ExecutionClass: spec.ExecutionClass,
CorrectionProtocol: contracts.CorrectionProtocol(strings.TrimSpace(string(spec.CorrectionProtocol))),
}
if normalized.Key == "" { if normalized.Key == "" {
return ValidatorSpec{}, fmt.Errorf("validator key must not be empty") return ValidatorSpec{}, fmt.Errorf("validator key must not be empty")
} }
@@ -330,6 +335,9 @@ func normalizeValidatorSpec(spec ValidatorSpec) (ValidatorSpec, error) {
default: default:
return ValidatorSpec{}, fmt.Errorf("validator %q execution class %q is not supported", normalized.Key, normalized.ExecutionClass) return ValidatorSpec{}, fmt.Errorf("validator %q execution class %q is not supported", normalized.Key, normalized.ExecutionClass)
} }
if normalized.CorrectionProtocol != "" {
return ValidatorSpec{}, fmt.Errorf("validator %q correction protocol is not supported", normalized.Key)
}
return normalized, nil return normalized, nil
} }