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

Apply validation retries to merge and normalize

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-27 01:13:32 +00:00
parent 4bb4582695
commit 7b5f4ebd42
11 changed files with 302 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/roadmap/implementation.md
View File

@@ -585,7 +585,7 @@ Every extraction job has an isolated bounded correction conversation and
continues to obey existing concurrency and ordering contracts. This stage is
one Terra prompt.
## Stage 13 — Integrate Merge And Normalize
## Stage 13 ✅ — Integrate Merge And Normalize
### Goal

1
internal/cli/assembled_spell_pipeline_contract_test.go
View File

@@ -164,6 +164,7 @@ func TestAssembledSpellPipelineHonorsNormalizeValidatorOverride(t *testing.T) {
func TestAssembledSpellPipelinePromotesTerminalUnknownSpellWarning(t *testing.T) {
registries, resolved, _ := assembledSpellPipeline(t, assembledSpellPipelineOptions{unknownSpell: true})
resolved.Steps[0].ArtifactLanes[0].NormalizeValidationPolicy.SemanticRejection = pipeline.SemanticRejectionRejectOutput
prepared, err := pipeline.Prepare(resolved, registries, pipeline.ModuleDependencies{})
if err != nil {
t.Fatalf("Prepare() error = %v, want nil", err)

2
internal/cli/run_result_command_test.go
View File

@@ -92,7 +92,7 @@ func TestRunResultReportsSuccessfulRejection(t *testing.T) {
if err != nil {
t.Fatal(err)
}
configBytes = []byte(replaceRequiredOnce(t, string(configBytes), " normalize: test/normalize\n", " normalize:\n module: test/normalize\n validators:\n - generic/always_reject\n"))
configBytes = []byte(replaceRequiredOnce(t, string(configBytes), " normalize: test/normalize\n", " normalize:\n module: test/normalize\n validators:\n - generic/always_reject\n validation_policy:\n semantic_rejection: reject_output\n"))
if err := os.WriteFile(roots.config, configBytes, 0o600); err != nil {
t.Fatal(err)
}

1
internal/framework/pipeline/evidence_output_test.go
View File

@@ -117,6 +117,7 @@ func TestRunnerEvidenceContextOmitsAbsentAndRejectedLanes(t *testing.T) {
return contracts.ValidationResult{Approved: false, ReasonCode: "rejected", Message: "not accepted"}, nil
},
}}
prepared.Steps[1].lanes[0].resolved.MergeValidationPolicy.SemanticRejection = SemanticRejectionRejectOutput
installEvidencePlan(prepared, 0, []string{"absent", "present", "rejected"}, func(notes codecNotes) ([]source.SourceRef, error) {
if len(notes.Items) > 0 && notes.Items[0] == "present" {
return []source.SourceRef{{SourceID: "source", StartUnitID: 1, EndUnitID: 1}}, nil

4
internal/framework/pipeline/runner_attempt_debug_test.go
View File

@@ -230,7 +230,7 @@ func TestRunnerRecordsDistinctRetryAttemptsAndPromotesAcceptedWarningsOnly(t *te
if attempts == 1 {
scope = "discarded"
}
return erasedTypedResult{Value: codecNotes{Items: []string{scope}}, Warnings: []contracts.Warning{{Scope: scope, ReasonCode: "observed", Message: scope}}}, nil
return erasedTypedResult{Value: codecNotes{Items: []string{scope}}, Warnings: []contracts.Warning{{Scope: scope, ReasonCode: "observed", Message: scope}}, ModelCandidate: attemptCandidate(t, fmt.Sprintf(`{"items":["%s"]}`, scope))}, nil
}
validatorCalls := 0
validator := preparedValidator{
@@ -299,6 +299,7 @@ func TestRunnerAttemptDebugRepresentsFailuresAndFinalRejection(t *testing.T) {
{
name: "normalize validator error",
configure: func(prepared *PreparedPipeline) {
prepared.Steps[0].lanes[0].resolved.NormalizeValidationPolicy.ValidatorFailure = ValidatorFailureFailRun
prepared.Steps[0].lanes[0].normalizeValidators.validators = []preparedValidator{{
resolved: ResolvedValidator{Binding: Binding("error-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"},
typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {
@@ -313,6 +314,7 @@ func TestRunnerAttemptDebugRepresentsFailuresAndFinalRejection(t *testing.T) {
{
name: "merge final rejection",
configure: func(prepared *PreparedPipeline) {
prepared.Steps[0].lanes[0].resolved.MergeValidationPolicy.SemanticRejection = SemanticRejectionRejectOutput
prepared.Steps[0].lanes[0].mergeValidators.validators = []preparedValidator{{
resolved: ResolvedValidator{Binding: Binding("reject-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"},
typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {

10
internal/framework/pipeline/runner_candidate_encoding_test.go
View File

@@ -187,6 +187,11 @@ func TestRunnerIsolatesTypedValidatorCandidates(t *testing.T) {
for _, target := range []ModuleStage{StageExtract, StageMerge, StageNormalize} {
t.Run(string(target), func(t *testing.T) {
prepared := preparedAttemptDebugPipeline(t)
if target == StageMerge {
prepared.Steps[0].lanes[0].resolved.MergeValidationPolicy.SemanticRejection = SemanticRejectionRejectOutput
} else {
prepared.Steps[0].lanes[0].resolved.NormalizeValidationPolicy.SemanticRejection = SemanticRejectionRejectOutput
}
prepared.Steps[0].lanes[0].extractValidators = preparedValidatorChain{}
prepared.Steps[0].lanes[0].mergeValidators = preparedValidatorChain{}
prepared.Steps[0].lanes[0].normalizeValidators = preparedValidatorChain{}
@@ -306,6 +311,11 @@ func TestRunnerRejectsCandidatesBeforeFinalEncoding(t *testing.T) {
for _, target := range []ModuleStage{StageMerge, StageNormalize} {
t.Run(string(target), func(t *testing.T) {
prepared := preparedAttemptDebugPipeline(t)
if target == StageMerge {
prepared.Steps[0].lanes[0].resolved.MergeValidationPolicy.SemanticRejection = SemanticRejectionRejectOutput
} else {
prepared.Steps[0].lanes[0].resolved.NormalizeValidationPolicy.SemanticRejection = SemanticRejectionRejectOutput
}
codec := &observedNotesCodec{}
installObservedNotesCodec(t, prepared, codec)
configureCandidateOperation(prepared, target, codecNotes{Items: []string{"invalid"}})

1
internal/framework/pipeline/runner_normalize_retry_test.go
View File

@@ -85,6 +85,7 @@ func TestRunnerHandlesRetryableNormalizeFallbacks(t *testing.T) {
prepared := preparedAttemptDebugPipeline(t)
lane := &prepared.Steps[0].lanes[0]
lane.resolved.Normalize.Retries = tc.retries
lane.resolved.NormalizeValidationPolicy.SemanticRejection = SemanticRejectionRejectOutput
if tc.validator != nil {
lane.normalizeValidators.validators = []preparedValidator{*tc.validator}
}

6
internal/framework/pipeline/runner_rejection_warnings_test.go
View File

@@ -83,16 +83,18 @@ func TestRunnerPromotesOnlyTerminalRejectionWarnings(t *testing.T) {
lane.extractValidators.validators = rejectionWarningTypedValidators(first, reject)
case StageMerge:
lane.resolved.Merge.Retries = 1
lane.resolved.MergeValidationPolicy.SemanticRejection = SemanticRejectionRejectOutput
lane.typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
attempts++
return erasedTypedResult{Value: codecNotes{Items: []string{"merge"}}, Warnings: []contracts.Warning{{Scope: fmt.Sprintf("operation-%d", attempts), ReasonCode: "operation", Message: "operation warning"}}}, nil
return erasedTypedResult{Value: codecNotes{Items: []string{"merge"}}, Warnings: []contracts.Warning{{Scope: fmt.Sprintf("operation-%d", attempts), ReasonCode: "operation", Message: "operation warning"}}, ModelCandidate: attemptCandidate(t, fmt.Sprintf(`{"items":["%d"]}`, attempts))}, nil
}
lane.mergeValidators.validators = rejectionWarningTypedValidators(first, reject)
case StageNormalize:
lane.resolved.Normalize.Retries = 1
lane.resolved.NormalizeValidationPolicy.SemanticRejection = SemanticRejectionRejectOutput
lane.typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
attempts++
return erasedTypedResult{Value: codecNotes{Items: []string{"normalize"}}, Warnings: []contracts.Warning{{Scope: fmt.Sprintf("operation-%d", attempts), ReasonCode: "operation", Message: "operation warning"}}}, nil
return erasedTypedResult{Value: codecNotes{Items: []string{"normalize"}}, Warnings: []contracts.Warning{{Scope: fmt.Sprintf("operation-%d", attempts), ReasonCode: "operation", Message: "operation warning"}}, ModelCandidate: attemptCandidate(t, fmt.Sprintf(`{"items":["%d"]}`, attempts))}, nil
}
lane.normalizeValidators.validators = rejectionWarningTypedValidators(first, reject)
}

119
internal/framework/pipeline/runner_semantic_correction_test.go Normal file
View File

@@ -0,0 +1,119 @@
package pipeline
import (
"context"
"fmt"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
func TestRunnerCorrectsRejectedMergeAndNormalizeCandidates(t *testing.T) {
for _, target := range []ModuleStage{StageMerge, StageNormalize} {
t.Run(string(target), func(t *testing.T) {
prepared := preparedAttemptDebugPipeline(t)
lane := &prepared.Steps[0].lanes[0]
var correction *contracts.SemanticCorrection
validator := preparedValidator{
resolved: ResolvedValidator{Binding: Binding("correction-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"},
typedValidate: func(_ context.Context, _ any, target typedValidationTarget) (contracts.ValidationResult, error) {
value := target.value.(codecNotes)
if firstNote(value) == "invalid" {
return contracts.ValidationResult{Approved: false, ReasonCode: "invalid_candidate", Message: "candidate needs correction", CorrectionGuidance: "produce the accepted value"}, nil
}
return contracts.ValidationResult{Approved: true}, nil
},
}
operation := func(observed *contracts.SemanticCorrection) (erasedTypedResult, error) {
if observed != nil {
cloned, err := contracts.CloneSemanticCorrection(observed)
if err != nil {
return erasedTypedResult{}, err
}
correction = cloned
}
item, response := "invalid", `{"items":["invalid"]}`
if observed != nil {
item, response = "accepted", `{"items":["accepted"]}`
}
candidate, err := contracts.NewModelCandidate([]byte(response), contracts.CorrectionProtocolSingleResponseV1)
if err != nil {
return erasedTypedResult{}, err
}
return erasedTypedResult{Value: codecNotes{Items: []string{item}}, ModelCandidate: candidate}, nil
}
switch target {
case StageMerge:
lane.resolved.Merge.Retries = 1
lane.mergeValidators.validators = []preparedValidator{validator}
lane.typed.merge = func(_ context.Context, _ any, request contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
return operation(request.Correction)
}
case StageNormalize:
lane.resolved.Normalize.Retries = 1
lane.normalizeValidators.validators = []preparedValidator{validator}
lane.typed.normalize = func(_ context.Context, _ any, request contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
return operation(request.Correction)
}
}
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")})
if err != nil {
t.Fatalf("Run() error = %v, want nil", err)
}
if correction == nil || string(correction.AssistantResponse) != `{"items":["invalid"]}` || correction.UserGuidance != "produce the accepted value" {
t.Fatalf("%s correction = %#v, want exact rejected response and guidance", target, correction)
}
if len(output.Rejected) != 0 || len(output.NormalizeOutputs) != 1 {
t.Fatalf("run output = %#v, want corrected accepted output", output)
}
})
}
}
func TestRunnerRejectsDeterministicMergeCandidateWithoutCorrection(t *testing.T) {
prepared := preparedAttemptDebugPipeline(t)
lane := &prepared.Steps[0].lanes[0]
lane.resolved.Merge.Retries = 1
lane.resolved.MergeValidationPolicy.SemanticRejection = SemanticRejectionRejectOutput
calls := 0
lane.typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
calls++
return erasedTypedResult{Value: codecNotes{Items: []string{"invalid"}}}, nil
}
lane.mergeValidators.validators = []preparedValidator{{
resolved: ResolvedValidator{Binding: Binding("reject"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"},
typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {
return contracts.ValidationResult{Approved: false, ReasonCode: "invalid_candidate", Message: "not accepted"}, nil
},
}}
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")})
if err != nil {
t.Fatalf("Run() error = %v, want non-fatal rejection", err)
}
if calls != 1 || len(output.Rejected) != 1 || output.Rejected[0].AttemptCount != 1 {
t.Fatalf("merge calls = %d output = %#v, want one deterministic terminal rejection", calls, output)
}
}
func TestRunnerContinuesNormalizeAfterValidatorFailure(t *testing.T) {
prepared := preparedAttemptDebugPipeline(t)
lane := &prepared.Steps[0].lanes[0]
lane.resolved.NormalizeValidationPolicy.ValidatorFailure = ValidatorFailureWarnContinue
lane.normalizeValidators.validators = []preparedValidator{{
resolved: ResolvedValidator{Binding: Binding("unavailable"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"},
typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {
return contracts.ValidationResult{}, fmt.Errorf("validator unavailable")
},
}}
checkpoints := &candidateCheckpointRecorder{CheckpointRecorder: NoopCheckpointRecorder()}
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoints: checkpoints})
if err != nil {
t.Fatalf("Run() error = %v, want incomplete accepted output", err)
}
if len(output.NormalizeOutputs) != 1 || len(output.Rejected) != 0 || checkpoints.normalizeSucceeded != 0 {
t.Fatalf("run output = %#v checkpoints = %#v, want non-checkpointed incomplete normalized output", output, checkpoints)
}
}

10
internal/framework/pipeline/runner_terminal_debug_test.go
View File

@@ -133,6 +133,16 @@ func preparedTerminalDebugPipeline(t *testing.T) (*PreparedPipeline, source.Chun
SemanticRejection: SemanticRejectionRejectOutput,
ValidatorFailure: ValidatorFailureFailRun,
}
prepared.Steps[0].lanes[0].resolved.MergeValidationPolicy = ValidationPolicy{
ProducerStructuralFailure: ProducerStructuralFailureFailRun,
SemanticRejection: SemanticRejectionRejectOutput,
ValidatorFailure: ValidatorFailureFailRun,
}
prepared.Steps[0].lanes[0].resolved.NormalizeValidationPolicy = ValidationPolicy{
ProducerStructuralFailure: ProducerStructuralFailureFailRun,
SemanticRejection: SemanticRejectionRejectOutput,
ValidatorFailure: ValidatorFailureFailRun,
}
chunker, ok := prepared.chunker.(*typedTestChunker)
if !ok {
t.Fatalf("prepared chunker = %T, want *typedTestChunker", prepared.chunker)

246
internal/framework/pipeline/runner_typed.go
View File

@@ -188,9 +188,23 @@ func (e *laneRunError) Error() string { return e.err.Error() }
func (e *laneRunError) Unwrap() error { return e.err }
type mergeStageResult struct {
artifact erasedMergeArtifact
serialized CheckpointArtifact
terminal bool
artifact erasedMergeArtifact
serialized CheckpointArtifact
terminal bool
validationIncomplete bool
}
type mergeAttemptValue struct {
artifact erasedMergeArtifact
candidate CheckpointArtifact
terminal *attemptTerminalRecorder
}
type normalizeAttemptValue struct {
value any
candidate CheckpointArtifact
terminal *attemptTerminalRecorder
retry map[string]any
}
func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, prepared preparedLaneExecutor, extracts finalizedExtractResults, output *RunOutput) error {
@@ -249,67 +263,88 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints
if err := checkpointMergeRunning(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps); err != nil {
return stageResult, err
}
retryResult, runErr := runSimpleRetry(ctx, lane.Merge.Retries, func(attempt int) (retryAttemptResult, error) {
terminalResult, runErr := runProducerAttempts(ctx, producerAttemptConfig{Retries: lane.Merge.Retries, Policy: lane.MergeValidationPolicy, AllowStructuralRetry: lane.MergeExecutionClass == contracts.ExecutionClassLLMBacked}, func(attemptCtx context.Context, request producerAttemptRequest) (producerAttemptOutput, error) {
started := time.Now().UTC()
attemptPath := path.Join("merge", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "merge", llmScope, debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: attempt, StartedAt: started})
attemptPath := path.Join("merge", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("attempt-%02d", request.Number))
attemptCtx, llmScope := withDebugLLMScope(attemptCtx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "merge", llmScope, debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: request.Number, StartedAt: started})
requestMetadata, metadataErr := cloneMetadata(input.Metadata)
if metadataErr != nil {
return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone merge request metadata: %w", metadataErr))
return producerAttemptOutput{}, 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: cloneStructuredOutputRepairAttempts(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), Correction: request.Correction, 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)
return producerAttemptOutput{}, terminal.record(nil, fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, lane.Merge.Module, callErr))
}
candidate := erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: result.Value}
attemptWarnings := cloneWarnings(result.Warnings)
warnings := cloneWarnings(result.Warnings)
serializedCandidate, encodeErr := serializeCandidateArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value)
if encodeErr != nil {
attemptErr := fmt.Errorf("serialize merge candidate for lane %q: %w", lane.ID, encodeErr)
return retryAttemptResult{}, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr)
return producerAttemptOutput{}, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(warnings)}, fmt.Errorf("serialize merge candidate for lane %q: %w", lane.ID, encodeErr))
}
warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageMerge, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Merge.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: mergeReferences, metadata: input.Metadata, value: result.Value, candidate: &serializedCandidate}, prepared.mergeValidators, attempt, input.Debug)
attemptWarnings = append(attemptWarnings, warnings...)
payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)}
if validateErr != nil {
return retryAttemptResult{}, terminal.record(payload, validateErr)
return producerAttemptOutput{Value: mergeAttemptValue{artifact: candidate, candidate: serializedCandidate, terminal: &terminal}, Candidate: result.ModelCandidate, Warnings: warnings}, nil
}, func(validationCtx context.Context, output producerAttemptOutput) (validationReport, error) {
candidate, ok := output.Value.(mergeAttemptValue)
if !ok {
return validationReport{}, fmt.Errorf("merge attempt has incompatible value")
}
if rejected != nil {
if debugErr := terminal.record(payload, nil); debugErr != nil {
return retryAttemptResult{}, debugErr
target := typedValidationTarget{stage: StageMerge, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Merge.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: mergeReferences, metadata: input.Metadata, value: candidate.artifact.Value, candidate: &candidate.candidate}
report, validationErr := r.validateTypedReport(validationCtx, typed.codec, target, prepared.mergeValidators, candidate.terminal.envelope.Attempt, input.Debug)
payload := map[string]any{"output": debugCheckpointArtifact(candidate.candidate), "warnings": debugWarningEnvelopes(append(cloneWarnings(output.Warnings), report.Warnings()...)), "rejection": debugRejectedOutputPtr(typedRejection(report, target, candidate.terminal.envelope.Attempt))}
if validationErr != nil {
return report, candidate.terminal.record(payload, validationErr)
}
if report.FirstRejection() != nil {
return report, candidate.terminal.record(payload, nil)
}
if lane.MergeValidationPolicy.ValidatorFailure == ValidatorFailureFailRun {
if failure := firstIncompleteValidation(report); failure != nil {
return report, candidate.terminal.record(payload, validatorFailureError(*failure))
}
return retryAttemptResult{rejection: rejected, warnings: attemptWarnings}, nil
}
stored, encodeErr := checkpointArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value)
if encodeErr != nil {
attemptErr := fmt.Errorf("serialize accepted merge output for lane %q: %w", lane.ID, encodeErr)
return retryAttemptResult{}, terminal.record(payload, attemptErr)
}
if debugErr := terminal.record(payload, nil); debugErr != nil {
return retryAttemptResult{}, debugErr
}
merged, serializedMerge = candidate, stored
mergeWarnings = attemptWarnings
return retryAttemptResult{accepted: true}, nil
return report, nil
})
if runErr != nil {
_ = checkpointMergeFailed(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, runErr)
return stageResult, runErr
}
if !retryResult.accepted {
output.Warnings = append(output.Warnings, retryResult.warnings...)
output.Rejected = append(output.Rejected, *retryResult.rejection)
if err := checkpointMergeRejected(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, *retryResult.rejection); err != nil {
if terminalResult.Action == producerTerminalRejected {
rejected := terminalResult.Rejection
if rejected == nil {
return stageResult, fmt.Errorf("merge attempt terminal is missing rejection")
}
rejected.Stage, rejected.StepID, rejected.LaneID, rejected.ModuleKey = string(StageMerge), input.stepID, lane.ID, lane.Merge.Module
output.Warnings = append(output.Warnings, terminalResult.Warnings...)
output.Rejected = append(output.Rejected, *rejected)
if err := checkpointMergeRejected(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, *rejected); err != nil {
return stageResult, err
}
stageResult.terminal = true
return stageResult, nil
}
candidate, ok := terminalResult.Value.(mergeAttemptValue)
if !ok {
return stageResult, fmt.Errorf("merge attempt terminal has incompatible value")
}
stored, encodeErr := checkpointArtifact(typed.codec, candidate.artifact.LaneID, candidate.artifact.MergerKey, candidate.artifact.SourceID, candidate.artifact.Value)
payload := map[string]any{"output": debugCheckpointArtifact(candidate.candidate), "warnings": debugWarningEnvelopes(terminalResult.Warnings), "rejection": debugRejectedOutputPtr(nil)}
if encodeErr != nil {
attemptErr := fmt.Errorf("serialize accepted merge output for lane %q: %w", lane.ID, encodeErr)
_ = checkpointMergeFailed(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, attemptErr)
return stageResult, candidate.terminal.record(payload, attemptErr)
}
if debugErr := candidate.terminal.record(payload, nil); debugErr != nil {
_ = checkpointMergeFailed(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, debugErr)
return stageResult, debugErr
}
merged, serializedMerge = candidate.artifact, stored
mergeWarnings = cloneWarnings(terminalResult.Warnings)
stageResult.validationIncomplete = terminalResult.ValidationIncomplete
output.Warnings = append(output.Warnings, mergeWarnings...)
if err := recordMerge(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, serializedMerge, mergeWarnings); err != nil {
return stageResult, err
if !stageResult.validationIncomplete {
if err := recordMerge(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, serializedMerge, mergeWarnings); err != nil {
return stageResult, err
}
}
}
if err := writeDebugTimed(input.Debug, path.Join("merge", fileio.EncodePathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "accepted": true, "output": debugCheckpointArtifact(serializedMerge), "warnings": debugWarningEnvelopes(mergeWarnings)}}); err != nil {
@@ -351,86 +386,107 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi
if err := checkpointNormalizeRunning(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps); err != nil {
return stageResult, err
}
retryResult, runErr := runSimpleRetry(ctx, lane.Normalize.Retries, func(attempt int) (retryAttemptResult, error) {
terminalResult, runErr := runProducerAttempts(ctx, producerAttemptConfig{Retries: lane.Normalize.Retries, Policy: lane.NormalizeValidationPolicy, AllowStructuralRetry: lane.NormalizeExecutionClass == contracts.ExecutionClassLLMBacked}, func(attemptCtx context.Context, request producerAttemptRequest) (producerAttemptOutput, error) {
started := time.Now().UTC()
attemptPath := path.Join("normalize", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "normalize", llmScope, debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: attempt, StartedAt: started})
attemptPath := path.Join("normalize", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("attempt-%02d", request.Number))
attemptCtx, llmScope := withDebugLLMScope(attemptCtx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "normalize", llmScope, debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: request.Number, StartedAt: started})
requestMetadata, metadataErr := cloneMetadata(input.Metadata)
if metadataErr != nil {
return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone normalize request metadata: %w", metadataErr))
return producerAttemptOutput{}, 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: cloneStructuredOutputRepairAttempts(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), Correction: request.Correction, 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)
return producerAttemptOutput{}, terminal.record(nil, fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, lane.Normalize.Module, callErr))
}
attemptWarnings := cloneWarnings(result.Warnings)
warnings := cloneWarnings(result.Warnings)
serializedCandidate, encodeErr := serializeCandidateArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value)
if encodeErr != nil {
attemptErr := fmt.Errorf("serialize normalize candidate for lane %q: %w", lane.ID, encodeErr)
return retryAttemptResult{}, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr)
return producerAttemptOutput{}, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(warnings)}, fmt.Errorf("serialize normalize candidate for lane %q: %w", lane.ID, encodeErr))
}
var retryPayload map[string]any
attemptValue := normalizeAttemptValue{value: result.Value, candidate: serializedCandidate, terminal: &terminal}
var directive *producerRetryDirective
if result.Retry != nil {
if err := validateNormalizeRetry(result.Retry); err != nil {
return retryAttemptResult{}, terminal.record(map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings)}, fmt.Errorf("normalize lane %q returned invalid retry directive: %w", lane.ID, err))
return producerAttemptOutput{}, terminal.record(map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(warnings)}, fmt.Errorf("normalize lane %q returned invalid retry directive: %w", lane.ID, err))
}
retryRemaining := attempt <= lane.Normalize.Retries
retryPayload = map[string]any{
"reason_code": result.Retry.ReasonCode,
"message": result.Retry.Message,
"another_attempt": retryRemaining,
"fallback_accepted": !retryRemaining,
anotherAttempt := request.Number <= lane.Normalize.Retries
attemptValue.retry = map[string]any{"reason_code": result.Retry.ReasonCode, "message": result.Retry.Message, "another_attempt": anotherAttempt, "fallback_accepted": !anotherAttempt}
directive = &producerRetryDirective{FallbackWarnings: cloneWarnings(result.Retry.FallbackWarnings)}
if anotherAttempt {
payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(warnings), "retry": attemptValue.retry}
if debugErr := terminal.record(payload, nil); debugErr != nil {
return producerAttemptOutput{}, debugErr
}
}
if retryRemaining {
payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "retry": retryPayload}
return retryAttemptResult{}, terminal.record(payload, nil)
}
return producerAttemptOutput{Value: attemptValue, Candidate: result.ModelCandidate, Warnings: warnings, Retry: directive}, nil
}, func(validationCtx context.Context, output producerAttemptOutput) (validationReport, error) {
candidate, ok := output.Value.(normalizeAttemptValue)
if !ok {
return validationReport{}, fmt.Errorf("normalize attempt has incompatible value")
}
target := typedValidationTarget{stage: StageNormalize, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Normalize.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: normalizeReferences, metadata: input.Metadata, value: candidate.value, candidate: &candidate.candidate}
report, validationErr := r.validateTypedReport(validationCtx, typed.codec, target, prepared.normalizeValidators, candidate.terminal.envelope.Attempt, input.Debug)
payload := map[string]any{"output": debugCheckpointArtifact(candidate.candidate), "warnings": debugWarningEnvelopes(append(cloneWarnings(output.Warnings), report.Warnings()...)), "rejection": debugRejectedOutputPtr(typedRejection(report, target, candidate.terminal.envelope.Attempt))}
if candidate.retry != nil {
payload["retry"] = candidate.retry
}
if validationErr != nil {
return report, candidate.terminal.record(payload, validationErr)
}
if report.FirstRejection() != nil {
return report, candidate.terminal.record(payload, nil)
}
if lane.NormalizeValidationPolicy.ValidatorFailure == ValidatorFailureFailRun {
if failure := firstIncompleteValidation(report); failure != nil {
return report, candidate.terminal.record(payload, validatorFailureError(*failure))
}
attemptWarnings = append(attemptWarnings, cloneWarnings(result.Retry.FallbackWarnings)...)
}
warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageNormalize, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Normalize.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: normalizeReferences, metadata: input.Metadata, value: result.Value, candidate: &serializedCandidate}, prepared.normalizeValidators, attempt, input.Debug)
attemptWarnings = append(attemptWarnings, warnings...)
payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)}
if retryPayload != nil {
payload["retry"] = retryPayload
}
if validateErr != nil {
return retryAttemptResult{}, terminal.record(payload, validateErr)
}
if rejected != nil {
if debugErr := terminal.record(payload, nil); debugErr != nil {
return retryAttemptResult{}, debugErr
}
return retryAttemptResult{rejection: rejected, warnings: attemptWarnings}, nil
}
stored, encodeErr := checkpointArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value)
if encodeErr != nil {
attemptErr := fmt.Errorf("serialize accepted normalize output for lane %q: %w", lane.ID, encodeErr)
return retryAttemptResult{}, terminal.record(payload, attemptErr)
}
if debugErr := terminal.record(payload, nil); debugErr != nil {
return retryAttemptResult{}, debugErr
}
serializedNormalize = stored
normalizeWarnings = attemptWarnings
return retryAttemptResult{accepted: true}, nil
return report, nil
})
if runErr != nil {
_ = checkpointNormalizeFailed(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, runErr)
return stageResult, runErr
}
if !retryResult.accepted {
output.Warnings = append(output.Warnings, retryResult.warnings...)
output.Rejected = append(output.Rejected, *retryResult.rejection)
if err := checkpointNormalizeRejected(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, *retryResult.rejection); err != nil {
if terminalResult.Action == producerTerminalRejected {
rejected := terminalResult.Rejection
if rejected == nil {
return stageResult, fmt.Errorf("normalize attempt terminal is missing rejection")
}
rejected.Stage, rejected.StepID, rejected.LaneID, rejected.ModuleKey = string(StageNormalize), input.stepID, lane.ID, lane.Normalize.Module
output.Warnings = append(output.Warnings, terminalResult.Warnings...)
output.Rejected = append(output.Rejected, *rejected)
if err := checkpointNormalizeRejected(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, *rejected); err != nil {
return stageResult, err
}
return stageResult, nil
}
candidate, ok := terminalResult.Value.(normalizeAttemptValue)
if !ok {
return stageResult, fmt.Errorf("normalize attempt terminal has incompatible value")
}
stored, encodeErr := checkpointArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, candidate.value)
payload := map[string]any{"output": debugCheckpointArtifact(candidate.candidate), "warnings": debugWarningEnvelopes(terminalResult.Warnings), "rejection": debugRejectedOutputPtr(nil)}
if candidate.retry != nil {
payload["retry"] = candidate.retry
}
if encodeErr != nil {
attemptErr := fmt.Errorf("serialize accepted normalize output for lane %q: %w", lane.ID, encodeErr)
_ = checkpointNormalizeFailed(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, attemptErr)
return stageResult, candidate.terminal.record(payload, attemptErr)
}
if debugErr := candidate.terminal.record(payload, nil); debugErr != nil {
_ = checkpointNormalizeFailed(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, debugErr)
return stageResult, debugErr
}
serializedNormalize = stored
normalizeWarnings = cloneWarnings(terminalResult.Warnings)
output.Warnings = append(output.Warnings, normalizeWarnings...)
if err := recordNormalize(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, serializedNormalize, normalizeWarnings); err != nil {
return stageResult, err
if !terminalResult.ValidationIncomplete {
if err := recordNormalize(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, serializedNormalize, normalizeWarnings); err != nil {
return stageResult, err
}
}
}
if err := writeDebugTimed(input.Debug, path.Join("normalize", fileio.EncodePathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "accepted": true, "output": debugCheckpointArtifact(serializedNormalize), "warnings": debugWarningEnvelopes(normalizeWarnings)}}); err != nil {