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

Enable correction-aware semantic reconciliation

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-27 00:08:05 +00:00
parent a26d6ed042
commit 04ba87e174
14 changed files with 195 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -91,6 +91,9 @@ func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDepend
if err := validateResolvedPipeline(resolved); err != nil {
return nil, err
}
if err := validateCorrectionRetryCapabilities(resolved); err != nil {
return nil, err
}
if err := validateRegistrySet(resolved, registries); err != nil {
return nil, err
}
@@ -209,6 +212,40 @@ func prepareEvidencePlan(resolved ResolvedPipeline, registries Registries, outpu
return plan, nil
}
func validateCorrectionRetryCapabilities(pipeline ResolvedPipeline) error {
validate := func(stage ModuleStage, laneID string, binding ModuleBinding, executionClass contracts.ExecutionClass, protocol contracts.CorrectionProtocol) error {
if executionClass != contracts.ExecutionClassLLMBacked || binding.Retries == 0 {
return nil
}
chain := resolvedValidatorChain(stage, laneID, binding.Module, pipeline.ValidatorChains)
if len(chain.Validators) == 0 || protocol == contracts.CorrectionProtocolSingleResponseV1 {
return nil
}
if laneID == "" {
return fmt.Errorf("pipeline %q %s module %q configures validators and retries but does not declare correction protocol %q", pipeline.ID, stage, binding.Module, contracts.CorrectionProtocolSingleResponseV1)
}
return fmt.Errorf("pipeline %q lane %q %s module %q configures validators and retries but does not declare correction protocol %q", pipeline.ID, laneID, stage, binding.Module, contracts.CorrectionProtocolSingleResponseV1)
}
if err := validate(StageChunk, "", pipeline.Chunk, pipeline.ChunkExecutionClass, pipeline.ChunkCorrectionProtocol); err != nil {
return err
}
for _, step := range pipeline.Steps {
for _, lane := range step.ArtifactLanes {
if err := validate(StageExtract, lane.ID, lane.Extract, lane.ExtractExecutionClass, lane.ExtractCorrectionProtocol); err != nil {
return err
}
if err := validate(StageMerge, lane.ID, lane.Merge, lane.MergeExecutionClass, lane.MergeCorrectionProtocol); err != nil {
return err
}
if err := validate(StageNormalize, lane.ID, lane.Normalize, lane.NormalizeExecutionClass, lane.NormalizeCorrectionProtocol); err != nil {
return err
}
}
}
return nil
}
func prepareLane(pipeline ResolvedPipeline, lane ResolvedArtifactLane, registries Registries, deps ModuleDependencies) (preparedLaneExecutor, error) {
executor := preparedLaneExecutor{resolved: cloneResolvedArtifactLane(lane)}
request := func(binding ModuleBinding, references contracts.ReferenceSet) BuildRequest {

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

@@ -173,29 +173,68 @@ func TestResolvePipelineCarriesCorrectionProtocolsIntoPreparedMetadata(t *testin
}
}
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 TestPrepareRequiresCorrectionCapabilityForValidatorRetries(t *testing.T) {
for _, test := range []struct {
name string
protocol contracts.CorrectionProtocol
retries int
validators bool
want string
}{
{name: "supported", protocol: contracts.CorrectionProtocolSingleResponseV1, retries: 1, validators: true},
{name: "unsupported", retries: 1, validators: true, want: "does not declare correction protocol"},
{name: "no validators", retries: 1},
{name: "no retries", validators: true},
} {
t.Run(test.name, func(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, CorrectionProtocol: test.protocol, 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"}},
)
if err := RegisterChunkValidator(catalog.Validators, ValidatorSpec{Key: "chunk-validator", ExecutionClass: contracts.ExecutionClassLLMBacked}, func() (contracts.ChunkValidator, error) {
return llmProfileTestChunkValidator{key: "chunk-validator"}, nil
}); err != nil {
t.Fatal(err)
}
profile := llmProfilePipeline()
profile.Chunk.Retries = test.retries
if test.validators {
profile.Chunk.Validators = ValidatorOverride{Set: true, Validators: []ModuleBinding{{Module: "chunk-validator"}}}
}
resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog)
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
_, err = Prepare(resolved, registriesFromModuleCatalog(catalog), ModuleDependencies{})
if test.want != "" {
if err == nil || !strings.Contains(err.Error(), test.want) {
t.Fatalf("Prepare() error = %v, want %q", err, test.want)
}
return
}
if err != nil {
t.Fatalf("Prepare() error = %v, want nil", err)
}
})
}
}
type llmProfileTestChunkValidator struct{ key string }
func (validator llmProfileTestChunkValidator) Name() string { return validator.key }
func (llmProfileTestChunkValidator) ExecutionClass() contracts.ExecutionClass {
return contracts.ExecutionClassLLMBacked
}
func (llmProfileTestChunkValidator) Validate(context.Context, contracts.ChunkValidationRequest) (contracts.ValidationResult, error) {
return contracts.ValidationResult{Approved: true}, nil
}
func TestResolvePipelineValidatorRetriesRequireLLMBackedValidator(t *testing.T) {
for _, test := range []struct {
name string