Propagate structured repair requests

2026-08-25 19:39:04 +00:00
parent ab9b743df6
commit e00cc45c6b
12 changed files with 154 additions and 134 deletions
--- a/internal/framework/pipeline/profile.go
+++ b/internal/framework/pipeline/profile.go
@@ -21,12 +21,13 @@ const (
 )

 type ModuleBinding struct {
-	Module     string                     `json:"module"`
-	LLMProfile string                     `json:"llm_profile,omitempty"`
-	Retries    int                        `json:"retries,omitempty"`
-	Options    map[string]any             `json:"options,omitempty"`
-	References map[string]ReferenceSource `json:"references,omitempty"`
-	Validators ValidatorOverride          `json:"validators,omitempty"`
+	Module                         string                     `json:"module"`
+	LLMProfile                     string                     `json:"llm_profile,omitempty"`
+	StructuredOutputRepairAttempts *int                       `json:"structured_output_repair_attempts,omitempty"`
+	Retries                        int                        `json:"retries,omitempty"`
+	Options                        map[string]any             `json:"options,omitempty"`
+	References                     map[string]ReferenceSource `json:"references,omitempty"`
+	Validators                     ValidatorOverride          `json:"validators,omitempty"`
 }

 // ArtifactReference identifies a normalized artifact produced by an earlier
--- a/internal/framework/pipeline/runner.go
+++ b/internal/framework/pipeline/runner.go
@@ -173,11 +173,12 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
 			return failOutput(output), err
 		}
 		doc, err = adapter.Parse(ctx, contracts.ParseRequest{
-			SourceID:   input.SourceID,
-			Path:       input.Path,
-			Raw:        input.RawInput,
-			LLMProfile: input.pipeline.Input.LLMProfile,
-			Metadata:   requestMetadata,
+			SourceID:                       input.SourceID,
+			Path:                           input.Path,
+			Raw:                            input.RawInput,
+			LLMProfile:                     input.pipeline.Input.LLMProfile,
+			StructuredOutputRepairAttempts: input.pipeline.Input.StructuredOutputRepairAttempts,
+			Metadata:                       requestMetadata,
 		})
 		if ctxErr := ctx.Err(); ctxErr != nil {
 			_ = checkpoints.SourceFailed(adapter.Key(), ctxErr)
@@ -360,14 +361,15 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
 		return failOutput(output), err
 	}
 	encoded, err := encoder.Encode(ctx, contracts.OutputRequest{
-		Manifest:         output.Manifest,
-		NormalizeOutputs: cloneSerializedOutputs(output.NormalizeOutputs),
-		Rejected:         cloneRejectedOutputs(output.Rejected),
-		Warnings:         output.Warnings,
-		LLMProfile:       input.pipeline.Output.LLMProfile,
-		Metadata:         outputMetadata,
-		ChunkMap:         contracts.CloneSerializedArtifactPointer(acceptedChunkMap),
-		EvidenceContext:  contracts.CloneSerializedArtifactPointer(evidenceArtifact),
+		Manifest:                       output.Manifest,
+		NormalizeOutputs:               cloneSerializedOutputs(output.NormalizeOutputs),
+		Rejected:                       cloneRejectedOutputs(output.Rejected),
+		Warnings:                       output.Warnings,
+		LLMProfile:                     input.pipeline.Output.LLMProfile,
+		StructuredOutputRepairAttempts: input.pipeline.Output.StructuredOutputRepairAttempts,
+		Metadata:                       outputMetadata,
+		ChunkMap:                       contracts.CloneSerializedArtifactPointer(acceptedChunkMap),
+		EvidenceContext:                contracts.CloneSerializedArtifactPointer(evidenceArtifact),
 	})
 	if ctxErr := ctx.Err(); ctxErr != nil {
 		return failOutput(output), ctxErr
--- a/internal/framework/pipeline/runner_chunk_plan.go
+++ b/internal/framework/pipeline/runner_chunk_plan.go
@@ -96,7 +96,7 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
 		chunkResult, callErr := chunker.Plan(attemptCtx, contracts.ChunkRequest{
 			Source: doc, SourceInput: sourceInput.Clone(), SessionID: sessionID,
 			References: CloneReferenceSet(input.pipeline.ChunkReferences.ReferenceSet),
-			LLMProfile: input.pipeline.Chunk.LLMProfile, Metadata: requestMetadata,
+			LLMProfile: input.pipeline.Chunk.LLMProfile, StructuredOutputRepairAttempts: input.pipeline.Chunk.StructuredOutputRepairAttempts, Metadata: requestMetadata,
 		})
 		if callErr != nil {
 			return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), callErr))
--- a/internal/framework/pipeline/runner_concurrent.go
+++ b/internal/framework/pipeline/runner_concurrent.go
@@ -420,7 +420,7 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source.
 			return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone extract request metadata: %w", metadataErr))
 		}
 		extractReferences := operationReferenceSet(input, lane.ExtractReferences)
-		extracted, callErr := typed.extract(attemptCtx, typed.extractor, contracts.TypedExtractionRequest{Source: doc, Chunk: &chunk, SourceInput: chunkInputMaterial(sourceInput, chunk), SessionID: sessionID, References: CloneReferenceSet(extractReferences), LLMProfile: lane.Extract.LLMProfile, Metadata: requestMetadata})
+		extracted, callErr := typed.extract(attemptCtx, typed.extractor, contracts.TypedExtractionRequest{Source: doc, Chunk: &chunk, SourceInput: chunkInputMaterial(sourceInput, chunk), SessionID: sessionID, References: CloneReferenceSet(extractReferences), LLMProfile: lane.Extract.LLMProfile, StructuredOutputRepairAttempts: lane.Extract.StructuredOutputRepairAttempts, Metadata: requestMetadata})
 		if callErr != nil {
 			attemptErr := fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, lane.Extract.Module, callErr)
 			return retryAttemptResult{}, terminal.record(nil, attemptErr)
--- a/internal/framework/pipeline/runner_typed.go
+++ b/internal/framework/pipeline/runner_typed.go
@@ -258,7 +258,7 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints
 			if metadataErr != nil {
 				return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone merge request metadata: %w", metadataErr))
 			}
-			result, callErr := typed.merge(attemptCtx, typed.merger, contracts.TypedMergeRequest[any]{Source: doc, LaneID: lane.ID, ExtractOutputs: mergeInputs, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(mergeReferences), LLMProfile: lane.Merge.LLMProfile, Metadata: requestMetadata})
+			result, callErr := typed.merge(attemptCtx, typed.merger, contracts.TypedMergeRequest[any]{Source: doc, LaneID: lane.ID, ExtractOutputs: mergeInputs, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(mergeReferences), LLMProfile: lane.Merge.LLMProfile, StructuredOutputRepairAttempts: lane.Merge.StructuredOutputRepairAttempts, Metadata: requestMetadata})
 			if callErr != nil {
 				attemptErr := fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, lane.Merge.Module, callErr)
 				return retryAttemptResult{}, terminal.record(nil, attemptErr)
@@ -360,7 +360,7 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi
 			if metadataErr != nil {
 				return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone normalize request metadata: %w", metadataErr))
 			}
-			result, callErr := typed.normalize(attemptCtx, typed.normalizer, contracts.TypedNormalizeRequest[any]{Source: doc, LaneID: lane.ID, MergeOutput: contracts.MergeArtifact[any]{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: merged.Value}, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(normalizeReferences), LLMProfile: lane.Normalize.LLMProfile, Metadata: requestMetadata})
+			result, callErr := typed.normalize(attemptCtx, typed.normalizer, contracts.TypedNormalizeRequest[any]{Source: doc, LaneID: lane.ID, MergeOutput: contracts.MergeArtifact[any]{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: merged.Value}, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(normalizeReferences), LLMProfile: lane.Normalize.LLMProfile, StructuredOutputRepairAttempts: lane.Normalize.StructuredOutputRepairAttempts, Metadata: requestMetadata})
 			if callErr != nil {
 				attemptErr := fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, lane.Normalize.Module, callErr)
 				return retryAttemptResult{}, terminal.record(nil, attemptErr)
@@ -511,6 +511,7 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE
 			}
 			requestTarget.value = candidateValue
 			requestTarget.llmProfile = binding.LLMProfile
+			requestTarget.structuredOutputRepairAttempts = binding.StructuredOutputRepairAttempts
 			result, err = item.typedValidate(validatorCtx, item.typed, requestTarget)
 		case ValidatorTargetSerialized:
 			artifact, encodeErr := validationCandidateArtifact(codec, target)
@@ -518,7 +519,7 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE
 				err = encodeErr
 				break
 			}
-			result, err = item.serialized.Validate(validatorCtx, contracts.SerializedValidationRequest{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Source: target.source, SourceID: target.sourceID, SourceInput: requestTarget.sourceInput, SessionID: target.sessionID, References: requestTarget.references, LLMProfile: binding.LLMProfile, Metadata: requestTarget.metadata, Chunk: requestTarget.chunk, Chunks: requestTarget.chunks, Schema: contracts.CloneArtifactSchema(artifact.Artifact.Schema), MediaType: artifact.Artifact.MediaType, Content: append([]byte(nil), artifact.Artifact.Content...)})
+			result, err = item.serialized.Validate(validatorCtx, contracts.SerializedValidationRequest{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Source: target.source, SourceID: target.sourceID, SourceInput: requestTarget.sourceInput, SessionID: target.sessionID, References: requestTarget.references, LLMProfile: binding.LLMProfile, StructuredOutputRepairAttempts: binding.StructuredOutputRepairAttempts, Metadata: requestTarget.metadata, Chunk: requestTarget.chunk, Chunks: requestTarget.chunks, Schema: contracts.CloneArtifactSchema(artifact.Artifact.Schema), MediaType: artifact.Artifact.MediaType, Content: append([]byte(nil), artifact.Artifact.Content...)})
 		default:
 			return nil, nil, fmt.Errorf("validator %q is incompatible with typed artifact validation", binding.Module)
 		}
--- a/internal/framework/pipeline/typed_execution.go
+++ b/internal/framework/pipeline/typed_execution.go
@@ -28,22 +28,23 @@ type erasedTypedResult struct {
 }

 type typedValidationTarget struct {
-	stage       ModuleStage
-	stepID      string
-	laneID      string
-	moduleKey   string
-	source      *source.SourceDocument
-	sourceID    string
-	sourceInput contracts.LLMInputMaterial
-	sessionID   string
-	references  contracts.ReferenceSet
-	llmProfile  string
-	metadata    map[string]any
-	chunk       *source.Chunk
-	chunks      []source.Chunk
-	ref         source.SourceRef
-	value       any
-	candidate   *CheckpointArtifact
+	stage                          ModuleStage
+	stepID                         string
+	laneID                         string
+	moduleKey                      string
+	source                         *source.SourceDocument
+	sourceID                       string
+	sourceInput                    contracts.LLMInputMaterial
+	sessionID                      string
+	references                     contracts.ReferenceSet
+	llmProfile                     string
+	structuredOutputRepairAttempts *int
+	metadata                       map[string]any
+	chunk                          *source.Chunk
+	chunks                         []source.Chunk
+	ref                            source.SourceRef
+	value                          any
+	candidate                      *CheckpointArtifact
 }

 func exactTypedValue[T any](operation string, value any) (T, error) {
--- a/internal/framework/pipeline/validator_chain_registry.go
+++ b/internal/framework/pipeline/validator_chain_registry.go
@@ -94,6 +94,10 @@ func cloneModuleBindings(bindings []ModuleBinding) []ModuleBinding {
 func cloneModuleBinding(binding ModuleBinding) ModuleBinding {
 	binding.Module = strings.TrimSpace(binding.Module)
 	binding.LLMProfile = strings.TrimSpace(binding.LLMProfile)
+	if binding.StructuredOutputRepairAttempts != nil {
+		value := *binding.StructuredOutputRepairAttempts
+		binding.StructuredOutputRepairAttempts = &value
+	}
 	binding.Options = cloneOptions(binding.Options)
 	if len(binding.References) > 0 {
 		references := make(map[string]ReferenceSource, len(binding.References))
--- a/internal/framework/pipeline/validator_registry.go
+++ b/internal/framework/pipeline/validator_registry.go
@@ -115,7 +115,7 @@ func RegisterTypedValidatorBuilder[T any](registry *ValidatorRegistry, kind cont
 			if err != nil {
 				return contracts.ValidationResult{}, err
 			}
-			return validator.Validate(ctx, contracts.TypedValidationRequest[T]{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Source: target.source, SourceID: target.sourceID, SourceInput: target.sourceInput, SessionID: target.sessionID, References: target.references, LLMProfile: target.llmProfile, Metadata: target.metadata, Chunk: target.chunk, Chunks: target.chunks, Ref: target.ref, Value: value})
+			return validator.Validate(ctx, contracts.TypedValidationRequest[T]{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Source: target.source, SourceID: target.sourceID, SourceInput: target.sourceInput, SessionID: target.sessionID, References: target.references, LLMProfile: target.llmProfile, StructuredOutputRepairAttempts: target.structuredOutputRepairAttempts, Metadata: target.metadata, Chunk: target.chunk, Chunks: target.chunks, Ref: target.ref, Value: value})
 		},
 	}
 	return nil