Improve semantic reconciliation retries

internal/framework/pipeline/normalize_retry.go

@@ -30,5 +30,14 @@ func validateNormalizeRetry(retry *contracts.NormalizeRetry) error {
 	if len(retry.Message) > contracts.MaxNormalizeRetryMessageBytes {
 		return errors.New("normalize retry directive message exceeds maximum length")
 	}
+	if !utf8.ValidString(retry.CorrectionGuidance) {
+		return errors.New("normalize retry directive correction guidance has invalid UTF-8")
+	}
+	if retry.CorrectionGuidance != "" && strings.TrimSpace(retry.CorrectionGuidance) == "" {
+		return errors.New("normalize retry directive correction guidance is blank")
+	}
+	if len(retry.CorrectionGuidance) > contracts.MaxNormalizeRetryCorrectionGuidanceBytes {
+		return errors.New("normalize retry directive correction guidance exceeds maximum length")
+	}
 	return nil
 }

internal/framework/pipeline/normalizer_registry.go

@@ -101,6 +101,7 @@ func cloneNormalizeRetry(retry *contracts.NormalizeRetry) *contracts.NormalizeRe
 	return &contracts.NormalizeRetry{
 		ReasonCode:          retry.ReasonCode,
 		Message:             retry.Message,
+		CorrectionGuidance:  retry.CorrectionGuidance,
 		FallbackDiagnostics: contracts.CloneProducerDiagnostics(retry.FallbackDiagnostics),
 	}
 }

internal/framework/pipeline/normalizer_registry_test.go

@@ -19,8 +19,9 @@ func (n retryingNotesNormalizer) Normalize(_ context.Context, req contracts.Type
 
 func TestNormalizerRegistryErasureClonesRetryDirective(t *testing.T) {
 	retry := &contracts.NormalizeRetry{
-		ReasonCode: "retryable",
-		Message:    "safe fallback available",
+		ReasonCode:         "retryable",
+		Message:            "safe fallback available",
+		CorrectionGuidance: "return a complete corrected proposal",
 	}
 	registry := NewNormalizerRegistry()
 	if err := RegisterNormalizer(registry, ModuleSpec{Key: "test/retry-normalize", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassDeterministic, ArtifactKind: "test/notes"}, func() (contracts.Normalizer[codecNotes], error) {
@@ -41,7 +42,8 @@ func TestNormalizerRegistryErasureClonesRetryDirective(t *testing.T) {
 		t.Fatalf("normalize() error = %v", err)
 	}
 	retry.Message = "mutated"
-	if result.Retry == nil || result.Retry.Message != "safe fallback available" {
+	retry.CorrectionGuidance = "mutated guidance"
+	if result.Retry == nil || result.Retry.Message != "safe fallback available" || result.Retry.CorrectionGuidance != "return a complete corrected proposal" {
 		t.Fatalf("erased retry result = %#v, want independent retry data", result)
 	}
 }

internal/framework/pipeline/producer_attempts.go

@@ -52,6 +52,7 @@ type producerAttemptRequest struct {
 // the current value as a safe fallback if its shared budget is exhausted.
 // Artifact-specific adapters are responsible for validating and populating it.
 type producerRetryDirective struct {
+	CorrectionGuidance  string
 	FallbackDiagnostics []contracts.ProducerDiagnostic
 }
 
@@ -59,7 +60,10 @@ func (directive *producerRetryDirective) clone() *producerRetryDirective {
 	if directive == nil {
 		return nil
 	}
-	return &producerRetryDirective{FallbackDiagnostics: contracts.CloneProducerDiagnostics(directive.FallbackDiagnostics)}
+	return &producerRetryDirective{
+		CorrectionGuidance:  directive.CorrectionGuidance,
+		FallbackDiagnostics: contracts.CloneProducerDiagnostics(directive.FallbackDiagnostics),
+	}
 }
 
 // producerAttemptOutput is intentionally artifact-neutral. Value remains
@@ -204,10 +208,21 @@ func runProducerAttempts(ctx context.Context, config producerAttemptConfig, prod
 			return failedProducerAttempt(provenance), err
 		}
 		if output.Retry != nil && number < attemptLimit {
+			correction, err = moduleRetryCorrection(output)
+			if err != nil {
+				provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptFailed})
+				return failedProducerAttempt(provenance), err
+			}
 			provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptRetried})
-			kind, correction = producerAttemptModuleRetry, nil
+			kind = producerAttemptModuleRetry
 			continue
 		}
+		if output.Retry != nil && output.Retry.CorrectionGuidance != "" {
+			if _, err := moduleRetryCorrection(output); err != nil {
+				provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptFailed})
+				return failedProducerAttempt(provenance), err
+			}
+		}
 		if output.Retry != nil {
 			output.Diagnostics = append(output.Diagnostics, contracts.CloneProducerDiagnostics(output.Retry.FallbackDiagnostics)...)
 		}
@@ -269,6 +284,23 @@ func runProducerAttempts(ctx context.Context, config producerAttemptConfig, prod
 	return failedProducerAttempt(provenance), errors.New("producer attempt budget was not exhausted deterministically")
 }
 
+func moduleRetryCorrection(output producerAttemptOutput) (*contracts.SemanticCorrection, error) {
+	if output.Retry == nil || output.Retry.CorrectionGuidance == "" {
+		return nil, nil
+	}
+	if output.Candidate == nil {
+		return nil, errors.New("module retry correction guidance requires a model candidate")
+	}
+	if output.Candidate.Protocol != contracts.CorrectionProtocolSingleResponseV1 {
+		return nil, fmt.Errorf("module retry correction guidance requires protocol %q", contracts.CorrectionProtocolSingleResponseV1)
+	}
+	correction, err := contracts.NewSemanticCorrection(output.Candidate.Response, output.Retry.CorrectionGuidance)
+	if err != nil {
+		return nil, fmt.Errorf("construct module retry semantic correction: %w", err)
+	}
+	return correction, nil
+}
+
 func validateProducerAttemptDiagnostics(output producerAttemptOutput) error {
 	if err := contracts.ValidateProducerDiagnostics(output.Diagnostics); err != nil {
 		return fmt.Errorf("producer returned invalid diagnostics: %w", err)

internal/framework/pipeline/producer_attempts_test.go

@@ -218,6 +218,9 @@ func TestRunProducerAttemptsUsesModuleRetryBudgetAndFallback(t *testing.T) {
 		calls := 0
 		terminal, err := runProducerAttempts(context.Background(), producerAttemptConfig{Retries: 1, Policy: DefaultValidationPolicy()}, func(_ context.Context, request producerAttemptRequest) (producerAttemptOutput, error) {
 			calls++
+			if request.Correction != nil {
+				t.Fatalf("feedback-free module retry correction = %#v, want nil", request.Correction)
+			}
 			if request.Number == 1 {
 				return producerAttemptOutput{Value: "fallback", Retry: &producerRetryDirective{}}, nil
 			}
@@ -234,6 +237,58 @@ func TestRunProducerAttemptsUsesModuleRetryBudgetAndFallback(t *testing.T) {
 		}
 	})
 
+	t.Run("feedback retry", func(t *testing.T) {
+		const (
+			defective = `{"duplicate_groups":[{"candidate_ids":[1,99],"canonical_candidate_id":1}]}`
+			guidance  = "Use only candidate IDs from the supplied candidate list. Return one complete corrected response."
+		)
+		var observed *contracts.SemanticCorrection
+		terminal, err := runProducerAttempts(context.Background(), producerAttemptConfig{Retries: 1, Policy: DefaultValidationPolicy()}, func(_ context.Context, request producerAttemptRequest) (producerAttemptOutput, error) {
+			if request.Number == 1 {
+				return producerAttemptOutput{Value: "safe fallback", Candidate: attemptCandidate(t, defective), Retry: &producerRetryDirective{CorrectionGuidance: guidance}}, nil
+			}
+			observed = request.Correction
+			return producerAttemptOutput{Value: "corrected"}, nil
+		}, approveAttempt)
+		if err != nil {
+			t.Fatalf("runProducerAttempts() error = %v", err)
+		}
+		if terminal.Action != producerTerminalAccepted || terminal.Value != "corrected" {
+			t.Fatalf("terminal = %#v, want corrected accepted value", terminal)
+		}
+		if observed == nil || string(observed.AssistantResponse) != defective || observed.UserGuidance != guidance {
+			t.Fatalf("module retry correction = %#v, want exact latest response and guidance", observed)
+		}
+		if got := attemptKinds(terminal.Provenance); !reflect.DeepEqual(got, []producerAttemptKind{producerAttemptInitial, producerAttemptModuleRetry}) {
+			t.Fatalf("attempt kinds = %v", got)
+		}
+	})
+
+	t.Run("feedback-free retry clears prior correction", func(t *testing.T) {
+		terminal, err := runProducerAttempts(context.Background(), producerAttemptConfig{Retries: 2, Policy: DefaultValidationPolicy()}, func(_ context.Context, request producerAttemptRequest) (producerAttemptOutput, error) {
+			switch request.Number {
+			case 1:
+				return producerAttemptOutput{Value: "first fallback", Candidate: attemptCandidate(t, "first defective response"), Retry: &producerRetryDirective{CorrectionGuidance: "Correct the first response."}}, nil
+			case 2:
+				if request.Correction == nil || string(request.Correction.AssistantResponse) != "first defective response" {
+					t.Fatalf("second attempt correction = %#v", request.Correction)
+				}
+				return producerAttemptOutput{Value: "second fallback", Retry: &producerRetryDirective{}}, nil
+			case 3:
+				if request.Correction != nil {
+					t.Fatalf("third attempt retained stale correction %#v", request.Correction)
+				}
+				return producerAttemptOutput{Value: "accepted"}, nil
+			default:
+				t.Fatalf("unexpected producer attempt %d", request.Number)
+				return producerAttemptOutput{}, nil
+			}
+		}, approveAttempt)
+		if err != nil || terminal.Action != producerTerminalAccepted || terminal.Value != "accepted" {
+			t.Fatalf("terminal = %#v, error = %v", terminal, err)
+		}
+	})
+
 	t.Run("fallback", func(t *testing.T) {
 		fallbackDiagnostic := contracts.ProducerDiagnostic{Disposition: contracts.DiagnosticDispositionWarning, Category: contracts.DiagnosticCategoryFallback, ReasonCode: "fallback", OccurrenceCount: 1, Samples: []contracts.DiagnosticSample{{Scope: "fallback", Message: "fallback warning"}}}
 		terminal, err := runProducerAttempts(context.Background(), producerAttemptConfig{Policy: DefaultValidationPolicy()}, func(context.Context, producerAttemptRequest) (producerAttemptOutput, error) {
@@ -248,6 +303,24 @@ func TestRunProducerAttemptsUsesModuleRetryBudgetAndFallback(t *testing.T) {
 	})
 }
 
+func TestRunProducerAttemptsRejectsModuleCorrectionWithoutModelCandidate(t *testing.T) {
+	producerCalls := 0
+	validatorCalls := 0
+	terminal, err := runProducerAttempts(context.Background(), producerAttemptConfig{Retries: 1, Policy: DefaultValidationPolicy()}, func(context.Context, producerAttemptRequest) (producerAttemptOutput, error) {
+		producerCalls++
+		return producerAttemptOutput{Value: "safe", Retry: &producerRetryDirective{CorrectionGuidance: "Return a complete corrected response."}}, nil
+	}, func(context.Context, producerAttemptOutput) (validationReport, error) {
+		validatorCalls++
+		return validationReport{}, nil
+	})
+	if err == nil || !strings.Contains(err.Error(), "requires a model candidate") {
+		t.Fatalf("runProducerAttempts() error = %v, want model-candidate contract failure", err)
+	}
+	if terminal.Action != producerTerminalFailed || producerCalls != 1 || validatorCalls != 0 {
+		t.Fatalf("terminal = %#v, producer calls = %d, validator calls = %d", terminal, producerCalls, validatorCalls)
+	}
+}
+
 func TestRunProducerAttemptsRejectsInvalidDiagnosticsWithoutRetry(t *testing.T) {
 	tests := []struct {
 		name   string

internal/framework/pipeline/runner_normalize_retry_test.go

@@ -153,6 +153,60 @@ func TestRunnerHandlesRetryableNormalizeFallbacks(t *testing.T) {
 	}
 }
 
+func TestRunnerForwardsModuleRequestedNormalizeCorrection(t *testing.T) {
+	const (
+		defective = `{"duplicate_groups":[{"candidate_ids":[1,99],"canonical_candidate_id":1}]}`
+		guidance  = "Duplicate group 1 must use only supplied candidate IDs. Return one complete corrected response."
+	)
+	prepared := preparedAttemptDebugPipeline(t)
+	lane := &prepared.Steps[0].lanes[0]
+	lane.resolved.Normalize.Retries = 1
+	var observed *contracts.SemanticCorrection
+	calls := 0
+	lane.typed.normalize = func(_ context.Context, _ any, request contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
+		calls++
+		if request.Correction != nil {
+			clone, err := contracts.CloneSemanticCorrection(request.Correction)
+			if err != nil {
+				return erasedTypedResult{}, err
+			}
+			observed = clone
+		}
+		if calls == 1 {
+			return erasedTypedResult{
+				Value:          codecNotes{Items: []string{"safe fallback"}},
+				ModelCandidate: attemptCandidate(t, defective),
+				Retry: &contracts.NormalizeRetry{
+					ReasonCode:         "semantic_proposal_invalid",
+					Message:            "operator-facing proposal diagnostic",
+					CorrectionGuidance: guidance,
+				},
+			}, nil
+		}
+		return erasedTypedResult{Value: codecNotes{Items: []string{"corrected"}}}, nil
+	}
+	debug := newCapturedDebugRecorder()
+	output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug})
+	if err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	if calls != 2 || observed == nil || string(observed.AssistantResponse) != defective || observed.UserGuidance != guidance {
+		t.Fatalf("normalize calls = %d correction = %#v", calls, observed)
+	}
+	if len(output.NormalizeOutputs) != 1 || len(output.Rejected) != 0 {
+		t.Fatalf("run output = %#v, want corrected accepted output", output)
+	}
+	var retryDebug strings.Builder
+	for _, name := range debug.names() {
+		if strings.HasPrefix(name, "normalize/notes/attempt-") && strings.HasSuffix(name, ".json") {
+			retryDebug.Write(debug.json[name])
+		}
+	}
+	if !strings.Contains(retryDebug.String(), `"correction_available":true`) || strings.Contains(retryDebug.String(), guidance) || strings.Contains(retryDebug.String(), defective) {
+		t.Fatalf("retry debug = %s, want safe correction metadata without content", retryDebug.String())
+	}
+}
+
 func TestRunnerValidatesNormalizeRetryDiagnostics(t *testing.T) {
 	const (
 		reasonSentinel  = "reason-diagnostic-sentinel"
@@ -160,9 +214,11 @@ func TestRunnerValidatesNormalizeRetryDiagnostics(t *testing.T) {
 	)
 	reasonOverLimit := strings.Repeat("r", contracts.MaxNormalizeRetryReasonCodeBytes-len(reasonSentinel)) + reasonSentinel + "x"
 	messageOverLimit := strings.Repeat("m", contracts.MaxNormalizeRetryMessageBytes-len(messageSentinel)) + messageSentinel + "x"
+	guidanceOverLimit := strings.Repeat("g", contracts.MaxNormalizeRetryCorrectionGuidanceBytes-len(messageSentinel)) + messageSentinel + "x"
 	tests := []struct {
 		name         string
 		retry        contracts.NormalizeRetry
+		candidate    *contracts.ModelCandidate
 		wantError    string
 		hiddenValues []string
 	}{
@@ -173,6 +229,15 @@ func TestRunnerValidatesNormalizeRetryDiagnostics(t *testing.T) {
 				Message:    strings.Repeat("m", contracts.MaxNormalizeRetryMessageBytes),
 			},
 		},
+		{
+			name: "accepts correction guidance byte limit with candidate",
+			retry: contracts.NormalizeRetry{
+				ReasonCode:         reasonSentinel,
+				Message:            messageSentinel,
+				CorrectionGuidance: strings.Repeat("g", contracts.MaxNormalizeRetryCorrectionGuidanceBytes),
+			},
+			candidate: attemptCandidate(t, `{"duplicate_groups":[]}`),
+		},
 		{
 			name: "rejects oversized reason code",
 			retry: contracts.NormalizeRetry{
@@ -209,6 +274,46 @@ func TestRunnerValidatesNormalizeRetryDiagnostics(t *testing.T) {
 			wantError:    "message has invalid UTF-8",
 			hiddenValues: []string{reasonSentinel, messageSentinel},
 		},
+		{
+			name: "rejects oversized correction guidance",
+			retry: contracts.NormalizeRetry{
+				ReasonCode:         reasonSentinel,
+				Message:            messageSentinel,
+				CorrectionGuidance: guidanceOverLimit,
+			},
+			wantError:    "correction guidance exceeds maximum length",
+			hiddenValues: []string{reasonSentinel, messageSentinel},
+		},
+		{
+			name: "rejects invalid correction guidance UTF-8",
+			retry: contracts.NormalizeRetry{
+				ReasonCode:         reasonSentinel,
+				Message:            messageSentinel,
+				CorrectionGuidance: messageSentinel + string([]byte{0xff}),
+			},
+			wantError:    "correction guidance has invalid UTF-8",
+			hiddenValues: []string{reasonSentinel, messageSentinel},
+		},
+		{
+			name: "rejects blank correction guidance",
+			retry: contracts.NormalizeRetry{
+				ReasonCode:         reasonSentinel,
+				Message:            messageSentinel,
+				CorrectionGuidance: " \t\n ",
+			},
+			wantError:    "correction guidance is blank",
+			hiddenValues: []string{reasonSentinel, messageSentinel},
+		},
+		{
+			name: "rejects correction guidance without candidate",
+			retry: contracts.NormalizeRetry{
+				ReasonCode:         reasonSentinel,
+				Message:            messageSentinel,
+				CorrectionGuidance: "Return one complete corrected response.",
+			},
+			wantError:    "requires a model candidate",
+			hiddenValues: []string{reasonSentinel, messageSentinel},
+		},
 		{
 			name: "rejects blank reason code",
 			retry: contracts.NormalizeRetry{
@@ -233,7 +338,7 @@ func TestRunnerValidatesNormalizeRetryDiagnostics(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			prepared := preparedAttemptDebugPipeline(t)
 			prepared.Steps[0].lanes[0].typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
-				return erasedTypedResult{Value: codecNotes{Items: []string{"safe"}}, Retry: &tc.retry}, nil
+				return erasedTypedResult{Value: codecNotes{Items: []string{"safe"}}, Retry: &tc.retry, ModelCandidate: tc.candidate}, nil
 			}
 			debug := newCapturedDebugRecorder()
 			output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug})

internal/framework/pipeline/runner_typed.go

@@ -451,8 +451,8 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi
 					return producerAttemptOutput{}, terminal.record(map[string]any{"output": debugCheckpointArtifact(serializedCandidate)}, fmt.Errorf("normalize lane %q returned invalid retry directive: %w", lane.ID, err))
 				}
 				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{FallbackDiagnostics: contracts.CloneProducerDiagnostics(result.Retry.FallbackDiagnostics)}
+				attemptValue.retry = map[string]any{"reason_code": result.Retry.ReasonCode, "message": result.Retry.Message, "correction_available": result.Retry.CorrectionGuidance != "", "another_attempt": anotherAttempt, "fallback_accepted": !anotherAttempt}
+				directive = &producerRetryDirective{CorrectionGuidance: result.Retry.CorrectionGuidance, FallbackDiagnostics: contracts.CloneProducerDiagnostics(result.Retry.FallbackDiagnostics)}
 				if anotherAttempt {
 					payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "retry": attemptValue.retry}
 					if debugErr := terminal.record(payload, nil); debugErr != nil {