From 63c397d86ab2c70ccf795b27b168f24b885d7fb7 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Tue, 25 Aug 2026 19:55:39 +0000 Subject: [PATCH] Resolve structured output repair configuration --- docs/config.md | 20 ++- docs/internal/pipeline.md | 10 +- docs/operations.md | 10 +- docs/roadmap/implementation.md | 2 +- internal/framework/pipeline/profile.go | 69 ++++++++- internal/framework/pipeline/profile_test.go | 146 ++++++++++++++++++ internal/framework/pipeline/runner.go | 4 +- .../pipeline/runner_attempt_debug_test.go | 32 ++++ .../framework/pipeline/runner_chunk_plan.go | 2 +- .../framework/pipeline/runner_concurrent.go | 2 +- internal/framework/pipeline/runner_typed.go | 8 +- .../pipeline/validator_chain_registry.go | 13 +- 12 files changed, 289 insertions(+), 29 deletions(-) diff --git a/docs/config.md b/docs/config.md index 7e8b2068..166b98a9 100644 --- a/docs/config.md +++ b/docs/config.md @@ -222,6 +222,7 @@ pipelines: | Field | Type | Default | Rules | | --- | --- | --- | --- | | **llm_profile** | string | none | Optional non-empty default PromptKit profile ID for selected LLM-backed bindings and validators. An explicitly present blank value is invalid. | +| **structured_output_repair_attempts** | integer | prompt-owned | Optional structural-repair limit from 0 through 3 for selected LLM-backed bindings and validators. Omission leaves the prompt's declared policy in control; explicit 0 disables structural repair at that scope. | | **input** | module binding | none | Required. | | **chunk** | module binding | **generic** | Optional. | | **output** | module binding | **json** | Optional. | @@ -241,6 +242,15 @@ run-level **--llm-profile** value first, then the binding's **llm_profile**, then the pipeline's **llm_profile**, and finally the PromptKit default. Deterministic bindings do not receive these defaults or run overrides. +Structural output repair is resolved after module, validator, and `--only` lane +selection. An object's **structured_output_repair_attempts** value takes +precedence over the pipeline value; otherwise, an LLM-backed binding or +validator inherits the pipeline value. If both are omitted, PromptKit uses the +prompt's declared repair policy. The value must be an integer from 0 through 3; +explicit `null` and non-integer values are invalid. An explicit value on a +deterministic binding or validator is invalid, while a pipeline value simply +does not apply to deterministic selections. + A lane has these fields: | Field | Type | Default | Rules | @@ -278,6 +288,7 @@ extract: | --- | --- | --- | --- | | **module** | string | none | Required for an object binding. Must be a registered compatible key. | | **llm_profile** | string | none | Optional non-empty PromptKit profile ID for an LLM-backed binding. It overrides the pipeline default unless the run supplies **--llm-profile**. | +| **structured_output_repair_attempts** | integer | pipeline or prompt-owned | Optional structural-repair limit from 0 through 3 for an LLM-backed binding. It overrides the pipeline value; explicit 0 disables structural repair. | | **retries** | integer | 0 | Non-negative additional attempts for chunk, extract, merge, and normalize bindings. | | **options** | object | none | Must satisfy the selected module. | | **references** | map | none | Valid only on chunk, extract, merge, and normalize bindings. | @@ -285,10 +296,11 @@ extract: Omitting **validators** uses the registered chain. **validators: []** selects an empty chain; a non-empty list replaces the chain in the listed order. -Validator bindings accept only **module**, **llm_profile**, and **options**. -They reject **references**, **retries**, and nested **validators**. Deterministic -validators reject an explicit **llm_profile**. Deterministic module bindings -also reject an explicit **llm_profile**. +Validator bindings accept only **module**, **llm_profile**, +**structured_output_repair_attempts**, and **options**. They reject +**references**, **retries**, and nested **validators**. Deterministic validators +reject explicit **llm_profile** and **structured_output_repair_attempts**. +Deterministic module bindings also reject those explicit fields. The **json** output module accepts optional **include_chunk_map** and **evidence_context** settings: diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 3c559690..758e6bb7 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -36,9 +36,13 @@ assigns a deterministic resolved-composition digest. The resolved pipeline contains bindings and declared reference targets, not external reference bytes. After selection, the resolver applies command, binding, and pipeline profile precedence to LLM-backed bindings and validators only; prompt defaults remain -an empty resolved binding profile. Deterministic bindings remain profile-free. -These effective values are part of the digest, so execution and checkpoint -consumers do not repeat profile inheritance. +an empty resolved binding profile. It resolves structural output repair +separately: a binding's `structured_output_repair_attempts` value wins, then a +pipeline value applies to LLM-backed bindings and validators, and omission +leaves the prompt-owned policy intact. An explicit repair value on a +deterministic binding is rejected. Resolved bindings own copied repair values, +and these effective values are part of the digest, so execution and checkpoint +consumers do not repeat profile inheritance or configuration resolution. Configuration resolution supplies the selected profile and catalog; see [Configuration Internals](configuration.md). diff --git a/docs/operations.md b/docs/operations.md index 54e0c83b..3effa0f3 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -275,9 +275,13 @@ Provider execution settings and the generation timeout come from the selected PromptKit profile. The invocation-only **--reasoning-effort** and **--clear-reasoning-effort** controls may replace or clear that profile setting for all LLM-backed calls in one run without changing the profile. PromptKit -v0.5.0 does not add a provider retry loop. Notarius binding retries rerun the -complete module operation and validation chain as defined by -[module bindings](config.md#module-bindings-and-validators). +structural output repair happens within one structured-completion call. Its +effective `structured_output_repair_attempts` limit is resolved from the +selected binding, then the pipeline, then the prompt declaration; see +[module bindings](config.md#module-bindings-and-validators). This is distinct +from Notarius binding **retries**, which rerun the complete module operation +and validation chain and do not consume or replenish the structural-repair +limit. Timeouts are layered. Caller cancellation is the outer authority. A positive effective generation timeout adds an inner request deadline, while zero diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index e1f4084e..54f7728b 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -604,7 +604,7 @@ git diff --check - Both maintained configurations remain valid without requiring the new field. - No runtime or prompt default has changed prematurely. -## Stage 9: Resolve And Apply Repair Configuration +## Stage 9: Resolve And Apply Repair Configuration ✅ ### Goal diff --git a/internal/framework/pipeline/profile.go b/internal/framework/pipeline/profile.go index a24324c5..a7655877 100644 --- a/internal/framework/pipeline/profile.go +++ b/internal/framework/pipeline/profile.go @@ -490,6 +490,9 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo if err := applyEffectiveLLMProfiles(&resolved, profile.LLMProfile, options.LLMProfileOverride); err != nil { return ResolvedPipeline{}, err } + if err := applyEffectiveStructuredOutputRepairAttempts(&resolved, profile.StructuredOutputRepairAttempts); err != nil { + return ResolvedPipeline{}, err + } if err := validateResolvedOptions(resolved, catalog, configuredLaneIDs); err != nil { return ResolvedPipeline{}, err } @@ -1322,12 +1325,13 @@ func resolveBinding(binding ModuleBinding, defaultModule string, referenceSlotLa return ModuleBinding{}, err } return ModuleBinding{ - Module: module, - LLMProfile: llmProfile, - Retries: binding.Retries, - Options: cloneOptions(binding.Options), - References: references, - Validators: cloneValidatorOverride(binding.Validators), + Module: module, + LLMProfile: llmProfile, + StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(binding.StructuredOutputRepairAttempts), + Retries: binding.Retries, + Options: cloneOptions(binding.Options), + References: references, + Validators: cloneValidatorOverride(binding.Validators), }, nil } @@ -1399,6 +1403,59 @@ func applyEffectiveLLMProfiles(resolved *ResolvedPipeline, pipelineProfile, over 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 { + 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 resolveBindings(bindings []ModuleBinding, defaultModule string, referenceSlotLabel string) ([]ModuleBinding, error) { if len(bindings) == 0 { return nil, nil diff --git a/internal/framework/pipeline/profile_test.go b/internal/framework/pipeline/profile_test.go index 8b8f7ea2..3d1af19c 100644 --- a/internal/framework/pipeline/profile_test.go +++ b/internal/framework/pipeline/profile_test.go @@ -408,6 +408,148 @@ func TestResolvePipelineDigestUsesEffectiveLLMProfiles(t *testing.T) { } } +func TestResolvePipelineAppliesStructuredOutputRepairAttemptsToLLMBindings(t *testing.T) { + pipelineAttempts := 2 + chunkAttempts := 1 + extractAttempts := 0 + validatorAttempts := 3 + profile := llmProfilePipeline() + profile.StructuredOutputRepairAttempts = &pipelineAttempts + profile.Chunk.StructuredOutputRepairAttempts = &chunkAttempts + lane := profile.Artifacts["events"] + lane.Extract.StructuredOutputRepairAttempts = &extractAttempts + lane.Extract.Validators = ValidatorOverride{Set: true, Validators: []ModuleBinding{{Module: "llm-validator", StructuredOutputRepairAttempts: &validatorAttempts}}} + profile.Artifacts["events"] = lane + + resolved, err := ResolvePipeline(profile, ResolveOptions{}, llmProfileCatalog(t)) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + encoded, err := json.Marshal(resolved) + if err != nil { + t.Fatalf("json.Marshal(resolved) error = %v", err) + } + if !strings.Contains(string(encoded), `"structured_output_repair_attempts":2`) { + t.Fatalf("resolved JSON = %s, want effective repair policy", encoded) + } + resolvedLane := resolved.Steps[0].ArtifactLanes[0] + for _, test := range []struct { + name string + got *int + want int + }{ + {name: "input", got: resolved.Input.StructuredOutputRepairAttempts, want: pipelineAttempts}, + {name: "chunk binding", got: resolved.Chunk.StructuredOutputRepairAttempts, want: chunkAttempts}, + {name: "extract binding", got: resolvedLane.Extract.StructuredOutputRepairAttempts, want: extractAttempts}, + {name: "merge binding", got: resolvedLane.Merge.StructuredOutputRepairAttempts, want: pipelineAttempts}, + {name: "normalize binding", got: resolvedLane.Normalize.StructuredOutputRepairAttempts, want: pipelineAttempts}, + {name: "output", got: resolved.Output.StructuredOutputRepairAttempts, want: pipelineAttempts}, + } { + t.Run(test.name, func(t *testing.T) { + if test.got == nil || *test.got != test.want { + t.Fatalf("repair attempts = %v, want %d", test.got, test.want) + } + }) + } + + extractChain := findResolvedValidatorChain(resolved.ValidatorChains, StageExtract, "events", "llm-extractor") + if extractChain == nil || len(extractChain.Validators) != 1 { + t.Fatalf("resolved extract validator chain = %#v, want one validator", extractChain) + } + if got := extractChain.Validators[0].Binding.StructuredOutputRepairAttempts; got == nil || *got != validatorAttempts { + t.Fatalf("extract validator repair attempts = %v, want %d", got, validatorAttempts) + } + chunkChain := findResolvedValidatorChain(resolved.ValidatorChains, StageChunk, "", "llm-chunk") + if chunkChain == nil || len(chunkChain.Validators) != 1 { + t.Fatalf("resolved chunk validator chain = %#v, want one validator", chunkChain) + } + if got := chunkChain.Validators[0].Binding.StructuredOutputRepairAttempts; got == nil || *got != pipelineAttempts { + t.Fatalf("chunk validator repair attempts = %v, want %d", got, pipelineAttempts) + } + + pipelineAttempts = 1 + chunkAttempts = 2 + extractAttempts = 3 + validatorAttempts = 0 + if got := *resolved.Input.StructuredOutputRepairAttempts; got != 2 { + t.Fatalf("resolved input repair attempts aliased profile: got %d, want 2", got) + } + if got := *resolved.Chunk.StructuredOutputRepairAttempts; got != 1 { + t.Fatalf("resolved chunk repair attempts aliased profile: got %d, want 1", got) + } + if got := *resolvedLane.Extract.StructuredOutputRepairAttempts; got != 0 { + t.Fatalf("resolved extract repair attempts aliased profile: got %d, want 0", got) + } + if got := *extractChain.Validators[0].Binding.StructuredOutputRepairAttempts; got != 3 { + t.Fatalf("resolved validator repair attempts aliased profile: got %d, want 3", got) + } +} + +func TestResolvePipelineRejectsStructuredOutputRepairAttemptsOnDeterministicBindings(t *testing.T) { + for _, test := range []struct { + name string + mutate func(*PipelineProfile) + }{ + {name: "module", mutate: func(profile *PipelineProfile) { profile.Input.StructuredOutputRepairAttempts = repairAttempts(1) }}, + {name: "validator", mutate: func(profile *PipelineProfile) { + lane := profile.Artifacts["events"] + lane.Extract.Validators = ValidatorOverride{Set: true, Validators: []ModuleBinding{{Module: "grounded", StructuredOutputRepairAttempts: repairAttempts(1)}}} + profile.Artifacts["events"] = lane + }}, + } { + t.Run(test.name, func(t *testing.T) { + profile := baselineProfile() + test.mutate(&profile) + _, err := ResolvePipeline(profile, ResolveOptions{}, newProfileCatalog(t)) + if err == nil || !strings.Contains(err.Error(), "structured_output_repair_attempts") || !strings.Contains(err.Error(), "deterministic") { + t.Fatalf("ResolvePipeline() error = %v, want deterministic repair-attempt rejection", err) + } + }) + } +} + +func TestResolvePipelineLeavesPipelineRepairAttemptsOffDeterministicBindings(t *testing.T) { + profile := baselineProfile() + profile.StructuredOutputRepairAttempts = repairAttempts(2) + + resolved, err := ResolvePipeline(profile, ResolveOptions{}, newProfileCatalog(t)) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + if resolved.Input.StructuredOutputRepairAttempts != nil || resolved.Chunk.StructuredOutputRepairAttempts != nil || resolved.Output.StructuredOutputRepairAttempts != nil { + t.Fatalf("deterministic pipeline inherited repair attempts: %#v", resolved) + } + lane := resolved.Steps[0].ArtifactLanes[0] + if lane.Extract.StructuredOutputRepairAttempts != nil || lane.Merge.StructuredOutputRepairAttempts != nil || lane.Normalize.StructuredOutputRepairAttempts != nil { + t.Fatalf("deterministic lane inherited repair attempts: %#v", lane) + } +} + +func TestResolvePipelineDigestUsesEffectiveStructuredOutputRepairAttempts(t *testing.T) { + digests := make(map[string]string) + for _, test := range []struct { + name string + attempts *int + }{ + {name: "prompt owned", attempts: nil}, + {name: "disabled", attempts: repairAttempts(0)}, + {name: "configured", attempts: repairAttempts(1)}, + } { + t.Run(test.name, func(t *testing.T) { + profile := llmProfilePipeline() + profile.StructuredOutputRepairAttempts = test.attempts + resolved, err := ResolvePipeline(profile, ResolveOptions{}, llmProfileCatalogWithoutValidatorChains(t)) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + digests[test.name] = resolved.Digest + }) + } + if digests["prompt owned"] == digests["disabled"] || digests["disabled"] == digests["configured"] || digests["prompt owned"] == digests["configured"] { + t.Fatalf("digests = %#v, want distinct effective repair policies", digests) + } +} + func TestResolvePipelineRecordsValidatorChains(t *testing.T) { catalog := newProfileCatalog(t) if err := catalog.ValidatorChains.Register(ValidatorChainMapping{ @@ -1843,6 +1985,10 @@ func llmProfilePipeline() PipelineProfile { } } +func repairAttempts(value int) *int { + return &value +} + func llmProfileValues(profile string) map[string]string { return map[string]string{ "input": profile, diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 4ab03793..203cfccc 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -177,7 +177,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err Path: input.Path, Raw: input.RawInput, LLMProfile: input.pipeline.Input.LLMProfile, - StructuredOutputRepairAttempts: input.pipeline.Input.StructuredOutputRepairAttempts, + StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(input.pipeline.Input.StructuredOutputRepairAttempts), Metadata: requestMetadata, }) if ctxErr := ctx.Err(); ctxErr != nil { @@ -366,7 +366,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err Rejected: cloneRejectedOutputs(output.Rejected), Warnings: output.Warnings, LLMProfile: input.pipeline.Output.LLMProfile, - StructuredOutputRepairAttempts: input.pipeline.Output.StructuredOutputRepairAttempts, + StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(input.pipeline.Output.StructuredOutputRepairAttempts), Metadata: outputMetadata, ChunkMap: contracts.CloneSerializedArtifactPointer(acceptedChunkMap), EvidenceContext: contracts.CloneSerializedArtifactPointer(evidenceArtifact), diff --git a/internal/framework/pipeline/runner_attempt_debug_test.go b/internal/framework/pipeline/runner_attempt_debug_test.go index 018e7ff0..4c310c95 100644 --- a/internal/framework/pipeline/runner_attempt_debug_test.go +++ b/internal/framework/pipeline/runner_attempt_debug_test.go @@ -117,6 +117,38 @@ func preparedAttemptDebugPipeline(t *testing.T) *PreparedPipeline { return prepared } +func TestRunnerForwardsDetachedStructuredOutputRepairAttempts(t *testing.T) { + prepared := preparedAttemptDebugPipeline(t) + lane := &prepared.Steps[0].lanes[0] + producerAttempts := 2 + validatorAttempts := 0 + lane.resolved.Extract.StructuredOutputRepairAttempts = &producerAttempts + if len(lane.extractValidators.validators) == 0 { + t.Fatal("extract validators are empty") + } + lane.extractValidators.validators[0].resolved.Binding.StructuredOutputRepairAttempts = &validatorAttempts + + var observedProducer, observedValidator *int + installExtractOperation(prepared, 0, func(_ context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) { + observedProducer = request.StructuredOutputRepairAttempts + return erasedTypedResult{Value: codecNotes{Items: []string{"extract"}}}, nil + }) + lane.extractValidators.validators[0].typedValidate = func(_ context.Context, _ any, target typedValidationTarget) (contracts.ValidationResult, error) { + observedValidator = target.structuredOutputRepairAttempts + return contracts.ValidationResult{Approved: true}, nil + } + + if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")}); err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + if observedProducer == nil || *observedProducer != producerAttempts || observedProducer == lane.resolved.Extract.StructuredOutputRepairAttempts { + t.Fatalf("producer repair attempts = %v, want detached value %d", observedProducer, producerAttempts) + } + if observedValidator == nil || *observedValidator != validatorAttempts || observedValidator == lane.extractValidators.validators[0].resolved.Binding.StructuredOutputRepairAttempts { + t.Fatalf("validator repair attempts = %v, want detached value %d", observedValidator, validatorAttempts) + } +} + func callAttemptDebugLLM(ctx context.Context, client contracts.StructuredLLMClient, name string) error { _, err := client.CompleteStructured(ctx, contracts.StructuredCompletionRequest{StageName: "unscoped-" + name, PromptID: name, ProfileID: "test"}, nil) return err diff --git a/internal/framework/pipeline/runner_chunk_plan.go b/internal/framework/pipeline/runner_chunk_plan.go index 5e69dc87..b25d6467 100644 --- 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, StructuredOutputRepairAttempts: input.pipeline.Chunk.StructuredOutputRepairAttempts, Metadata: requestMetadata, + LLMProfile: input.pipeline.Chunk.LLMProfile, StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(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)) diff --git a/internal/framework/pipeline/runner_concurrent.go b/internal/framework/pipeline/runner_concurrent.go index aecd44b3..82606516 100644 --- 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, StructuredOutputRepairAttempts: lane.Extract.StructuredOutputRepairAttempts, 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: cloneStructuredOutputRepairAttempts(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) diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index a5e0d7c0..852cfacd 100644 --- 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, StructuredOutputRepairAttempts: lane.Merge.StructuredOutputRepairAttempts, 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: cloneStructuredOutputRepairAttempts(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, StructuredOutputRepairAttempts: lane.Normalize.StructuredOutputRepairAttempts, 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: cloneStructuredOutputRepairAttempts(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,7 +511,7 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE } requestTarget.value = candidateValue requestTarget.llmProfile = binding.LLMProfile - requestTarget.structuredOutputRepairAttempts = binding.StructuredOutputRepairAttempts + requestTarget.structuredOutputRepairAttempts = cloneStructuredOutputRepairAttempts(binding.StructuredOutputRepairAttempts) result, err = item.typedValidate(validatorCtx, item.typed, requestTarget) case ValidatorTargetSerialized: artifact, encodeErr := validationCandidateArtifact(codec, target) @@ -519,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, 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...)}) + 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: cloneStructuredOutputRepairAttempts(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) } diff --git a/internal/framework/pipeline/validator_chain_registry.go b/internal/framework/pipeline/validator_chain_registry.go index 4fd5e9ca..447cf71f 100644 --- a/internal/framework/pipeline/validator_chain_registry.go +++ b/internal/framework/pipeline/validator_chain_registry.go @@ -94,10 +94,7 @@ 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.StructuredOutputRepairAttempts = cloneStructuredOutputRepairAttempts(binding.StructuredOutputRepairAttempts) binding.Options = cloneOptions(binding.Options) if len(binding.References) > 0 { references := make(map[string]ReferenceSource, len(binding.References)) @@ -110,6 +107,14 @@ func cloneModuleBinding(binding ModuleBinding) ModuleBinding { return binding } +func cloneStructuredOutputRepairAttempts(attempts *int) *int { + if attempts == nil { + return nil + } + value := *attempts + return &value +} + func cloneReferenceSource(source ReferenceSource) ReferenceSource { out := source if source.Artifact != nil {