Add retryable normalization fallbacks

internal/framework/contracts/errors.go

@@ -0,0 +1,7 @@
+package contracts
+
+import "errors"
+
+// ErrInvalidStructuredOutput identifies a provider response that cannot satisfy
+// the caller's declared structured-output contract.
+var ErrInvalidStructuredOutput = errors.New("invalid structured output")

internal/framework/contracts/typed_pipeline.go

@@ -90,6 +90,15 @@ type TypedNormalizeRequest[T any] struct {
 type TypedNormalizeResult[T any] struct {
 	Value    T
 	Warnings []Warning
+	Retry    *NormalizeRetry
+}
+
+// NormalizeRetry asks the framework to retry normalization while retaining a
+// safe candidate for acceptance if the retry budget is exhausted.
+type NormalizeRetry struct {
+	ReasonCode       string
+	Message          string
+	FallbackWarnings []Warning
 }
 
 type Normalizer[T any] interface {

internal/framework/llm/scriptorium_client.go

@@ -113,17 +113,17 @@ func (c *ScriptoriumClient) CompleteStructured(ctx context.Context, req contract
 		return contracts.StructuredCompletionResponse{}, fmt.Errorf("run Scriptorium prompt %q: %w", promptID, redactScriptoriumError(err))
 	}
 	if result == nil {
-		return contracts.StructuredCompletionResponse{}, fmt.Errorf("run Scriptorium prompt %q: empty result", promptID)
+		return contracts.StructuredCompletionResponse{}, fmt.Errorf("run Scriptorium prompt %q: %w: empty result", promptID, contracts.ErrInvalidStructuredOutput)
 	}
 	response := c.responseFromResult(result, prepared)
 	if result.Validation.Status == scriptorium.ValidationFailed || !result.Validation.IsValid {
-		return response, fmt.Errorf("run Scriptorium prompt %q: validation failed: %s", promptID, strings.Join(result.Validation.Errors, "; "))
+		return response, fmt.Errorf("run Scriptorium prompt %q: %w: validation failed: %s", promptID, contracts.ErrInvalidStructuredOutput, strings.Join(result.Validation.Errors, "; "))
 	}
 	if len(strings.TrimSpace(string(response.Content))) == 0 {
-		return response, fmt.Errorf("run Scriptorium prompt %q: empty structured output", promptID)
+		return response, fmt.Errorf("run Scriptorium prompt %q: %w: empty structured output", promptID, contracts.ErrInvalidStructuredOutput)
 	}
 	if err := json.Unmarshal(response.Content, out); err != nil {
-		return response, fmt.Errorf("decode Scriptorium structured output for prompt %q: %w", promptID, err)
+		return response, fmt.Errorf("decode Scriptorium structured output for prompt %q: %w: %w", promptID, contracts.ErrInvalidStructuredOutput, err)
 	}
 	return response, nil
 }

internal/framework/llm/scriptorium_client_test.go

@@ -113,7 +113,7 @@ func TestScriptoriumClientValidationFailureReturnsError(t *testing.T) {
 			"transcript": contracts.NewLLMInputMaterial("transcript", "application/json", []byte(`{"source":true}`), "", ""),
 		},
 	}, &out)
-	if err == nil || !strings.Contains(err.Error(), "validation failed") {
+	if err == nil || !errors.Is(err, contracts.ErrInvalidStructuredOutput) || !strings.Contains(err.Error(), "validation failed") {
 		t.Fatalf("CompleteStructured() error = %v, want validation failure", err)
 	}
 	if got := string(resp.Content); got != `{"bad":true}` {
@@ -138,7 +138,7 @@ func TestScriptoriumClientDecodeFailureReturnsRawResponse(t *testing.T) {
 			"transcript": contracts.NewLLMInputMaterial("transcript", "application/json", []byte(`{"source":true}`), "", ""),
 		},
 	}, &out)
-	if err == nil || !strings.Contains(err.Error(), "decode Scriptorium structured output") {
+	if err == nil || !errors.Is(err, contracts.ErrInvalidStructuredOutput) || !strings.Contains(err.Error(), "decode Scriptorium structured output") {
 		t.Fatalf("CompleteStructured() error = %v, want decode failure", err)
 	}
 	if got := string(resp.Content); got != `{"ok":true}` {
@@ -163,6 +163,9 @@ func TestScriptoriumClientProviderFailureIncludesContextAndRedactsBearerToken(t
 	if err == nil {
 		t.Fatalf("CompleteStructured() error = nil, want provider error")
 	}
+	if errors.Is(err, contracts.ErrInvalidStructuredOutput) {
+		t.Fatalf("provider error = %v, must not be classified as invalid structured output", err)
+	}
 	if !strings.Contains(err.Error(), `run Scriptorium prompt "adapter.test"`) {
 		t.Fatalf("error = %q, want operation context", err.Error())
 	}
@@ -186,11 +189,27 @@ func TestScriptoriumClientContextCancellationIsRespected(t *testing.T) {
 			"transcript": contracts.NewLLMInputMaterial("transcript", "application/json", []byte(`{"source":true}`), "", ""),
 		},
 	}, &out)
-	if !errors.Is(err, context.Canceled) {
+	if !errors.Is(err, context.Canceled) || errors.Is(err, contracts.ErrInvalidStructuredOutput) {
 		t.Fatalf("CompleteStructured() error = %v, want context canceled", err)
 	}
 }
 
+func TestScriptoriumClientClassifiesEmptyStructuredCompletion(t *testing.T) {
+	client := newTestScriptoriumClient(t, &fakeScriptoriumLLM{allowEmpty: true})
+
+	var out map[string]any
+	_, err := client.CompleteStructured(context.Background(), contracts.StructuredCompletionRequest{
+		PromptID:  "adapter.test",
+		SessionID: "session-123",
+		Inputs: contracts.LLMInputSet{
+			"transcript": contracts.NewLLMInputMaterial("transcript", "application/json", []byte(`{"source":true}`), "", ""),
+		},
+	}, &out)
+	if !errors.Is(err, contracts.ErrInvalidStructuredOutput) {
+		t.Fatalf("CompleteStructured() error = %v, want invalid structured output", err)
+	}
+}
+
 func TestScheduledScriptoriumClientBoundsConcurrentCalls(t *testing.T) {
 	fake := &fakeScriptoriumLLM{
 		content: `{"ok":true}`,
@@ -296,6 +315,7 @@ output:
 
 type fakeScriptoriumLLM struct {
 	content     string
+	allowEmpty  bool
 	err         error
 	block       chan struct{}
 	mu          sync.Mutex
@@ -329,10 +349,10 @@ func (f *fakeScriptoriumLLM) Generate(ctx context.Context, req scriptorium.Gener
 		return nil, f.err
 	}
 	content := f.content
-	if content == "" {
+	if content == "" && !f.allowEmpty {
 		content = `{"ok":true}`
 	}
-	if !json.Valid([]byte(content)) {
+	if !f.allowEmpty && !json.Valid([]byte(content)) {
 		return nil, errors.New("test fake must return JSON content")
 	}
 	return &scriptorium.GenerateResponse{

internal/framework/pipeline/normalizer_registry.go

@@ -80,12 +80,23 @@ func RegisterNormalizerBuilder[T any](registry *NormalizerRegistry, spec ModuleS
 			if err != nil {
 				return erasedTypedResult{}, err
 			}
-			return erasedTypedResult{Value: result.Value, Warnings: result.Warnings}, nil
+			return erasedTypedResult{Value: result.Value, Warnings: cloneWarnings(result.Warnings), Retry: cloneNormalizeRetry(result.Retry)}, nil
 		},
 	}
 	return nil
 }
 
+func cloneNormalizeRetry(retry *contracts.NormalizeRetry) *contracts.NormalizeRetry {
+	if retry == nil {
+		return nil
+	}
+	return &contracts.NormalizeRetry{
+		ReasonCode:       retry.ReasonCode,
+		Message:          retry.Message,
+		FallbackWarnings: cloneWarnings(retry.FallbackWarnings),
+	}
+}
+
 func (r *NormalizerRegistry) validateOptions(key string, kind contracts.ArtifactKind, options map[string]any) error {
 	if r == nil {
 		return fmt.Errorf("normalizer registry must not be nil")

internal/framework/pipeline/normalizer_registry_test.go

@@ -0,0 +1,52 @@
+package pipeline
+
+import (
+	"context"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+)
+
+type retryingNotesNormalizer struct {
+	warnings []contracts.Warning
+	retry    *contracts.NormalizeRetry
+}
+
+func (retryingNotesNormalizer) Key() string                               { return "test/retry-normalize" }
+func (retryingNotesNormalizer) ReferenceSlots() []contracts.ReferenceSlot { return nil }
+func (n retryingNotesNormalizer) Normalize(_ context.Context, req contracts.TypedNormalizeRequest[codecNotes]) (contracts.TypedNormalizeResult[codecNotes], error) {
+	return contracts.TypedNormalizeResult[codecNotes]{Value: req.MergeOutput.Value, Warnings: n.warnings, Retry: n.retry}, nil
+}
+
+func TestNormalizerRegistryErasureClonesRetryDirective(t *testing.T) {
+	warnings := []contracts.Warning{{Scope: "attempt", ReasonCode: "ordinary", Message: "ordinary warning"}}
+	retry := &contracts.NormalizeRetry{
+		ReasonCode:       "retryable",
+		Message:          "safe fallback available",
+		FallbackWarnings: []contracts.Warning{{Scope: "fallback", ReasonCode: "omitted", Message: "fallback warning"}},
+	}
+	registry := NewNormalizerRegistry()
+	if err := RegisterNormalizer(registry, ModuleSpec{Key: "test/retry-normalize", Stage: StageNormalize, ArtifactKind: "test/notes"}, func() (contracts.Normalizer[codecNotes], error) {
+		return retryingNotesNormalizer{warnings: warnings, retry: retry}, nil
+	}); err != nil {
+		t.Fatalf("RegisterNormalizer() error = %v", err)
+	}
+	entry, ok := registry.typedEntry("test/retry-normalize", "test/notes")
+	if !ok {
+		t.Fatal("typed normalizer entry missing")
+	}
+	implementation, err := entry.builder(BuildRequest{})
+	if err != nil {
+		t.Fatalf("builder() error = %v", err)
+	}
+	result, err := entry.normalize(context.Background(), implementation, contracts.TypedNormalizeRequest[any]{MergeOutput: contracts.MergeArtifact[any]{Value: codecNotes{Items: []string{"safe"}}}})
+	if err != nil {
+		t.Fatalf("normalize() error = %v", err)
+	}
+	warnings[0].Message = "mutated"
+	retry.Message = "mutated"
+	retry.FallbackWarnings[0].Message = "mutated"
+	if result.Retry == nil || result.Warnings[0].Message != "ordinary warning" || result.Retry.Message != "safe fallback available" || result.Retry.FallbackWarnings[0].Message != "fallback warning" {
+		t.Fatalf("erased retry result = %#v, want independent warning data", result)
+	}
+}

internal/framework/pipeline/runner_normalize_retry_test.go

@@ -0,0 +1,174 @@
+package pipeline
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+)
+
+func TestRunnerHandlesRetryableNormalizeFallbacks(t *testing.T) {
+	tests := []struct {
+		name           string
+		retries        int
+		operation      func(int) erasedTypedResult
+		validator      *preparedValidator
+		wantCalls      int
+		wantItem       string
+		wantWarnings   []string
+		wantRejected   int
+		wantDebug      []string
+		wantCheckpoint int
+	}{
+		{
+			name:    "accepts zero-retry fallback",
+			retries: 0,
+			operation: func(int) erasedTypedResult {
+				return retryableNormalizeResult("fallback", "ordinary", "fallback-warning")
+			},
+			wantCalls:      1,
+			wantItem:       "fallback",
+			wantWarnings:   []string{"ordinary", "fallback-warning"},
+			wantDebug:      []string{`"another_attempt":false`, `"fallback_accepted":true`},
+			wantCheckpoint: 1,
+		},
+		{
+			name:    "retries before accepting ordinary result",
+			retries: 1,
+			operation: func(attempt int) erasedTypedResult {
+				if attempt == 1 {
+					return retryableNormalizeResult("discarded", "discarded-ordinary", "discarded-fallback")
+				}
+				return erasedTypedResult{Value: codecNotes{Items: []string{"accepted"}}, Warnings: []contracts.Warning{{Scope: "accepted", ReasonCode: "ordinary", Message: "accepted-warning"}}}
+			},
+			wantCalls:      2,
+			wantItem:       "accepted",
+			wantWarnings:   []string{"accepted-warning"},
+			wantDebug:      []string{`"another_attempt":true`, `"fallback_accepted":false`},
+			wantCheckpoint: 1,
+		},
+		{
+			name:    "accepts final fallback after exhaustion",
+			retries: 1,
+			operation: func(attempt int) erasedTypedResult {
+				return retryableNormalizeResult(fmt.Sprintf("fallback-%d", attempt), fmt.Sprintf("ordinary-%d", attempt), fmt.Sprintf("fallback-warning-%d", attempt))
+			},
+			wantCalls:      2,
+			wantItem:       "fallback-2",
+			wantWarnings:   []string{"ordinary-2", "fallback-warning-2"},
+			wantDebug:      []string{`"another_attempt":false`, `"fallback_accepted":true`},
+			wantCheckpoint: 1,
+		},
+		{
+			name:    "keeps final fallback rejection terminal",
+			retries: 1,
+			operation: func(int) erasedTypedResult {
+				return retryableNormalizeResult("rejected", "ordinary", "fallback-warning")
+			},
+			validator: &preparedValidator{
+				resolved: ResolvedValidator{Binding: Binding("reject-final-fallback"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"},
+				typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {
+					return contracts.ValidationResult{Approved: false, ReasonCode: "rejected", Message: "reject fallback"}, nil
+				},
+			},
+			wantCalls:      2,
+			wantRejected:   1,
+			wantDebug:      []string{`"fallback_accepted":true`, `"rejection"`},
+			wantCheckpoint: 0,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			prepared := preparedAttemptDebugPipeline(t)
+			lane := &prepared.Steps[0].lanes[0]
+			lane.resolved.Normalize.Retries = tc.retries
+			if tc.validator != nil {
+				lane.normalizeValidators.validators = []preparedValidator{*tc.validator}
+			}
+			calls := 0
+			lane.typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
+				calls++
+				return tc.operation(calls), nil
+			}
+			debug := newCapturedDebugRecorder()
+			checkpoints := &candidateCheckpointRecorder{CheckpointRecorder: NoopCheckpointRecorder()}
+			output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug, Checkpoints: checkpoints})
+			if err != nil {
+				t.Fatalf("Run() error = %v", err)
+			}
+			if calls != tc.wantCalls {
+				t.Fatalf("normalize calls = %d, want %d", calls, tc.wantCalls)
+			}
+			if checkpoints.normalizeSucceeded != tc.wantCheckpoint {
+				t.Fatalf("normalize checkpoints = %d, want %d", checkpoints.normalizeSucceeded, tc.wantCheckpoint)
+			}
+			if len(output.Rejected) != tc.wantRejected {
+				t.Fatalf("rejected outputs = %#v, want %d", output.Rejected, tc.wantRejected)
+			}
+			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])
+				}
+			}
+			for _, fragment := range tc.wantDebug {
+				if !strings.Contains(retryDebug.String(), fragment) {
+					t.Fatalf("retry debug = %s, want %q", retryDebug.String(), fragment)
+				}
+			}
+			if tc.wantRejected != 0 {
+				if output.Rejected[0].AttemptCount != tc.wantCalls {
+					t.Fatalf("rejection attempt count = %d, want %d", output.Rejected[0].AttemptCount, tc.wantCalls)
+				}
+				return
+			}
+			if len(output.NormalizeOutputs) != 1 {
+				t.Fatalf("normalize outputs = %#v, want one", output.NormalizeOutputs)
+			}
+			decoded, err := lane.typed.codec.decode(output.NormalizeOutputs[0].Artifact.Content)
+			if err != nil {
+				t.Fatalf("decode normalized output: %v", err)
+			}
+			normalized, ok := decoded.(codecNotes)
+			if !ok {
+				t.Fatalf("decoded normalized output = %T, want codecNotes", decoded)
+			}
+			if got := firstNote(normalized); got != tc.wantItem {
+				t.Fatalf("normalized item = %q, want %q", got, tc.wantItem)
+			}
+			gotWarnings := make([]string, len(output.Warnings))
+			for index, warning := range output.Warnings {
+				gotWarnings[index] = warning.Message
+			}
+			if strings.Join(gotWarnings, "|") != strings.Join(tc.wantWarnings, "|") {
+				t.Fatalf("durable warnings = %#v, want %#v", gotWarnings, tc.wantWarnings)
+			}
+		})
+	}
+}
+
+func TestRunnerRejectsBlankNormalizeRetryDiagnostic(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: &contracts.NormalizeRetry{ReasonCode: " ", Message: "missing reason"}}, nil
+	}
+	_, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: newCapturedDebugRecorder()})
+	if err == nil || !strings.Contains(err.Error(), "blank reason code or message") {
+		t.Fatalf("Run() error = %v, want retry diagnostic contract error", err)
+	}
+}
+
+func retryableNormalizeResult(item, ordinary, fallback string) erasedTypedResult {
+	return erasedTypedResult{
+		Value:    codecNotes{Items: []string{item}},
+		Warnings: []contracts.Warning{{Scope: "attempt", ReasonCode: "ordinary", Message: ordinary}},
+		Retry: &contracts.NormalizeRetry{
+			ReasonCode:       "retryable_normalization",
+			Message:          "safe fallback is available",
+			FallbackWarnings: []contracts.Warning{{Scope: "fallback", ReasonCode: "fallback", Message: fallback}},
+		},
+	}
+}

internal/framework/pipeline/runner_typed.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"fmt"
 	"path"
+	"strings"
 	"time"
 
 	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
@@ -362,9 +363,30 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi
 				attemptErr := fmt.Errorf("serialize normalize candidate for lane %q: %w", lane.ID, encodeErr)
 				return false, nil, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr)
 			}
+			var retryPayload map[string]any
+			if result.Retry != nil {
+				if strings.TrimSpace(result.Retry.ReasonCode) == "" || strings.TrimSpace(result.Retry.Message) == "" {
+					return false, nil, terminal.record(map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings)}, fmt.Errorf("normalize lane %q returned retry directive with blank reason code or message", lane.ID))
+				}
+				retryRemaining := attempt <= lane.Normalize.Retries
+				retryPayload = map[string]any{
+					"reason_code":       result.Retry.ReasonCode,
+					"message":           result.Retry.Message,
+					"another_attempt":   retryRemaining,
+					"fallback_accepted": !retryRemaining,
+				}
+				if retryRemaining {
+					payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "retry": retryPayload}
+					return false, nil, terminal.record(payload, nil)
+				}
+				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 || rejected != nil {
 				return false, rejected, terminal.record(payload, validateErr)
 			}

internal/framework/pipeline/typed_execution.go

@@ -24,6 +24,7 @@ type erasedMergeArtifact struct {
 type erasedTypedResult struct {
 	Value    any
 	Warnings []contracts.Warning
+	Retry    *contracts.NormalizeRetry
 }
 
 type typedValidationTarget struct {