diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 1ec2926d..ba7072c0 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -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 diff --git a/internal/cli/assembled_spell_pipeline_contract_test.go b/internal/cli/assembled_spell_pipeline_contract_test.go index f56589dc..fca9d23b 100644 --- a/internal/cli/assembled_spell_pipeline_contract_test.go +++ b/internal/cli/assembled_spell_pipeline_contract_test.go @@ -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) diff --git a/internal/cli/run_result_command_test.go b/internal/cli/run_result_command_test.go index 5f2e8148..9eef0cea 100644 --- a/internal/cli/run_result_command_test.go +++ b/internal/cli/run_result_command_test.go @@ -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) } diff --git a/internal/framework/pipeline/evidence_output_test.go b/internal/framework/pipeline/evidence_output_test.go index 22b3f067..caacbd13 100644 --- a/internal/framework/pipeline/evidence_output_test.go +++ b/internal/framework/pipeline/evidence_output_test.go @@ -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 diff --git a/internal/framework/pipeline/runner_attempt_debug_test.go b/internal/framework/pipeline/runner_attempt_debug_test.go index 4c310c95..777f3f86 100644 --- a/internal/framework/pipeline/runner_attempt_debug_test.go +++ b/internal/framework/pipeline/runner_attempt_debug_test.go @@ -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) { diff --git a/internal/framework/pipeline/runner_candidate_encoding_test.go b/internal/framework/pipeline/runner_candidate_encoding_test.go index 69ab123a..47ae57da 100644 --- a/internal/framework/pipeline/runner_candidate_encoding_test.go +++ b/internal/framework/pipeline/runner_candidate_encoding_test.go @@ -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"}}) diff --git a/internal/framework/pipeline/runner_normalize_retry_test.go b/internal/framework/pipeline/runner_normalize_retry_test.go index 5bb7b090..c334ab78 100644 --- a/internal/framework/pipeline/runner_normalize_retry_test.go +++ b/internal/framework/pipeline/runner_normalize_retry_test.go @@ -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} } diff --git a/internal/framework/pipeline/runner_rejection_warnings_test.go b/internal/framework/pipeline/runner_rejection_warnings_test.go index 2e93ef73..46d40d28 100644 --- a/internal/framework/pipeline/runner_rejection_warnings_test.go +++ b/internal/framework/pipeline/runner_rejection_warnings_test.go @@ -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) } diff --git a/internal/framework/pipeline/runner_semantic_correction_test.go b/internal/framework/pipeline/runner_semantic_correction_test.go new file mode 100644 index 00000000..f46cf38a --- /dev/null +++ b/internal/framework/pipeline/runner_semantic_correction_test.go @@ -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) + } +} diff --git a/internal/framework/pipeline/runner_terminal_debug_test.go b/internal/framework/pipeline/runner_terminal_debug_test.go index 6edf6fec..f754f1d8 100644 --- a/internal/framework/pipeline/runner_terminal_debug_test.go +++ b/internal/framework/pipeline/runner_terminal_debug_test.go @@ -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) diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index de0bae8d..9deaf711 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -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 {