diff --git a/internal/framework/contracts/errors.go b/internal/framework/contracts/errors.go new file mode 100644 index 0000000..218641a --- /dev/null +++ b/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") diff --git a/internal/framework/contracts/typed_pipeline.go b/internal/framework/contracts/typed_pipeline.go index 4670e03..7c049e1 100644 --- a/internal/framework/contracts/typed_pipeline.go +++ b/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 { diff --git a/internal/framework/llm/scriptorium_client.go b/internal/framework/llm/scriptorium_client.go index c5566e1..0c0fd65 100644 --- a/internal/framework/llm/scriptorium_client.go +++ b/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 } diff --git a/internal/framework/llm/scriptorium_client_test.go b/internal/framework/llm/scriptorium_client_test.go index 4993eaa..2dfd65d 100644 --- a/internal/framework/llm/scriptorium_client_test.go +++ b/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{ diff --git a/internal/framework/pipeline/normalizer_registry.go b/internal/framework/pipeline/normalizer_registry.go index afd7ff3..cd403b2 100644 --- a/internal/framework/pipeline/normalizer_registry.go +++ b/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") diff --git a/internal/framework/pipeline/normalizer_registry_test.go b/internal/framework/pipeline/normalizer_registry_test.go new file mode 100644 index 0000000..6bad130 --- /dev/null +++ b/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) + } +} diff --git a/internal/framework/pipeline/runner_normalize_retry_test.go b/internal/framework/pipeline/runner_normalize_retry_test.go new file mode 100644 index 0000000..c1f4807 --- /dev/null +++ b/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}}, + }, + } +} diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 7187a33..ea0f559 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/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) } diff --git a/internal/framework/pipeline/typed_execution.go b/internal/framework/pipeline/typed_execution.go index 99f41f5..05e47f6 100644 --- a/internal/framework/pipeline/typed_execution.go +++ b/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 {