Files
notarius/internal/framework/pipeline/producer_attempts_test.go

458 lines
20 KiB
Go

package pipeline
import (
"context"
"errors"
"reflect"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
func TestRunProducerAttemptsUsesOneSharedBudget(t *testing.T) {
candidate := attemptCandidate(t, "first response")
producerCalls := 0
terminal, err := runProducerAttempts(context.Background(), producerAttemptConfig{Retries: 2, Policy: DefaultValidationPolicy(), AllowStructuralRetry: true}, func(_ context.Context, request producerAttemptRequest) (producerAttemptOutput, error) {
producerCalls++
switch request.Number {
case 1:
return producerAttemptOutput{}, errors.New("temporary producer failure")
case 2:
return producerAttemptOutput{}, contracts.ErrInvalidStructuredOutput
case 3:
return producerAttemptOutput{Value: "accepted", Candidate: candidate}, nil
default:
t.Fatalf("unexpected producer attempt %d", request.Number)
return producerAttemptOutput{}, nil
}
}, approveAttempt)
if err != nil {
t.Fatalf("runProducerAttempts() error = %v", err)
}
if terminal.Action != producerTerminalAccepted || terminal.Value != "accepted" {
t.Fatalf("terminal = %#v, want accepted value", terminal)
}
if producerCalls != 3 {
t.Fatalf("producer calls = %d, want 3", producerCalls)
}
if got := attemptKinds(terminal.Provenance); !reflect.DeepEqual(got, []producerAttemptKind{producerAttemptInitial, producerAttemptOperationalRetry, producerAttemptStructuralRetry}) {
t.Fatalf("attempt kinds = %v", got)
}
}
func TestRunProducerAttemptsReplacesSemanticCorrectionWithLatestResponse(t *testing.T) {
first := attemptCandidate(t, "first response")
second := attemptCandidate(t, "second response")
var corrections []*contracts.SemanticCorrection
terminal, err := runProducerAttempts(context.Background(), producerAttemptConfig{Retries: 2, Policy: DefaultValidationPolicy()}, func(_ context.Context, request producerAttemptRequest) (producerAttemptOutput, error) {
if request.Correction != nil {
corrections = append(corrections, request.Correction)
}
switch request.Number {
case 1:
return producerAttemptOutput{Value: "one", Candidate: first}, nil
case 2:
return producerAttemptOutput{Value: "two", Candidate: second}, nil
case 3:
return producerAttemptOutput{Value: "three", Candidate: attemptCandidate(t, "third response")}, nil
default:
t.Fatalf("unexpected producer attempt %d", request.Number)
return producerAttemptOutput{}, nil
}
}, func(_ context.Context, output producerAttemptOutput) (validationReport, error) {
if output.Value == "three" {
return validationReport{}, nil
}
return rejectedAttemptReport("semantic_defect"), nil
})
if err != nil {
t.Fatalf("runProducerAttempts() error = %v", err)
}
if terminal.Action != producerTerminalAccepted {
t.Fatalf("terminal action = %q, want accepted", terminal.Action)
}
if len(corrections) != 2 {
t.Fatalf("correction count = %d, want 2", len(corrections))
}
if corrections[0] == corrections[1] {
t.Fatal("semantic corrections reused the same pointer")
}
if got := string(corrections[0].AssistantResponse); got != "first response" {
t.Fatalf("first correction response = %q", got)
}
if got := string(corrections[1].AssistantResponse); got != "second response" {
t.Fatalf("second correction response = %q", got)
}
if got := attemptKinds(terminal.Provenance); !reflect.DeepEqual(got, []producerAttemptKind{producerAttemptInitial, producerAttemptSemanticRetry, producerAttemptSemanticRetry}) {
t.Fatalf("attempt kinds = %v", got)
}
}
func TestRunProducerAttemptsAppliesTerminalPolicies(t *testing.T) {
tests := []struct {
name string
config producerAttemptConfig
produce producerAttemptProducer
validate producerAttemptValidator
wantAction producerTerminalAction
wantError bool
wantValue any
}{
{
name: "structural reject output",
config: producerAttemptConfig{Policy: ValidationPolicy{ProducerStructuralFailure: ProducerStructuralFailureRejectOutput, SemanticRejection: SemanticRejectionFailRun, ValidatorFailure: ValidatorFailureWarnContinue}, AllowStructuralRetry: true},
produce: func(context.Context, producerAttemptRequest) (producerAttemptOutput, error) {
return producerAttemptOutput{}, contracts.ErrInvalidStructuredOutput
},
validate: approveAttempt,
wantAction: producerTerminalRejected,
},
{
name: "structural fail run",
config: producerAttemptConfig{Policy: DefaultValidationPolicy(), AllowStructuralRetry: true},
produce: func(context.Context, producerAttemptRequest) (producerAttemptOutput, error) {
return producerAttemptOutput{}, contracts.ErrInvalidStructuredOutput
},
validate: approveAttempt,
wantAction: producerTerminalFailed,
wantError: true,
},
{
name: "operational failure after retry exhaustion",
config: producerAttemptConfig{Policy: DefaultValidationPolicy()},
produce: func(context.Context, producerAttemptRequest) (producerAttemptOutput, error) {
return producerAttemptOutput{}, errors.New("producer unavailable")
},
validate: approveAttempt,
wantAction: producerTerminalFailed,
wantError: true,
},
{
name: "semantic reject output",
config: producerAttemptConfig{Policy: ValidationPolicy{ProducerStructuralFailure: ProducerStructuralFailureFailRun, SemanticRejection: SemanticRejectionRejectOutput, ValidatorFailure: ValidatorFailureWarnContinue}},
produce: func(context.Context, producerAttemptRequest) (producerAttemptOutput, error) {
return producerAttemptOutput{Value: "discard", Candidate: attemptCandidate(t, "response")}, nil
},
validate: func(context.Context, producerAttemptOutput) (validationReport, error) {
return rejectedAttemptReport("bad"), nil
},
wantAction: producerTerminalRejected,
},
{
name: "semantic fail run",
config: producerAttemptConfig{Policy: DefaultValidationPolicy()},
produce: func(context.Context, producerAttemptRequest) (producerAttemptOutput, error) {
return producerAttemptOutput{Candidate: attemptCandidate(t, "response")}, nil
},
validate: func(context.Context, producerAttemptOutput) (validationReport, error) {
return rejectedAttemptReport("bad"), nil
},
wantAction: producerTerminalFailed,
wantError: true,
},
{
name: "validator failure warns and continues",
config: producerAttemptConfig{Policy: DefaultValidationPolicy()},
produce: func(context.Context, producerAttemptRequest) (producerAttemptOutput, error) {
return producerAttemptOutput{Value: "kept"}, nil
},
validate: func(context.Context, producerAttemptOutput) (validationReport, error) {
return failedAttemptReport(), nil
},
wantAction: producerTerminalIncompleteAccepted,
wantValue: "kept",
},
{
name: "validator failure fails run",
config: producerAttemptConfig{Policy: ValidationPolicy{ProducerStructuralFailure: ProducerStructuralFailureFailRun, SemanticRejection: SemanticRejectionFailRun, ValidatorFailure: ValidatorFailureFailRun}},
produce: func(context.Context, producerAttemptRequest) (producerAttemptOutput, error) {
return producerAttemptOutput{}, nil
},
validate: func(context.Context, producerAttemptOutput) (validationReport, error) {
return failedAttemptReport(), nil
},
wantAction: producerTerminalFailed,
wantError: true,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
terminal, err := runProducerAttempts(context.Background(), test.config, test.produce, test.validate)
if (err != nil) != test.wantError {
t.Fatalf("error = %v, want error %t", err, test.wantError)
}
if terminal.Action != test.wantAction {
t.Fatalf("terminal action = %q, want %q", terminal.Action, test.wantAction)
}
if terminal.Value != test.wantValue {
t.Fatalf("terminal value = %#v, want %#v", terminal.Value, test.wantValue)
}
if test.wantAction == producerTerminalRejected && (terminal.Rejection == nil || terminal.Value != nil) {
t.Fatalf("rejected terminal = %#v, want rejection without value", terminal)
}
})
}
}
func TestRunProducerAttemptsDoesNotRetryUncorrectableRejection(t *testing.T) {
calls := 0
policy := DefaultValidationPolicy()
policy.SemanticRejection = SemanticRejectionRejectOutput
terminal, err := runProducerAttempts(context.Background(), producerAttemptConfig{Retries: 3, Policy: policy}, func(context.Context, producerAttemptRequest) (producerAttemptOutput, error) {
calls++
return producerAttemptOutput{Value: "deterministic"}, nil
}, func(context.Context, producerAttemptOutput) (validationReport, error) {
return rejectedAttemptReport("deterministic_rejection"), nil
})
if err != nil {
t.Fatalf("runProducerAttempts() error = %v", err)
}
if calls != 1 || terminal.Action != producerTerminalRejected || len(terminal.Provenance) != 1 {
t.Fatalf("terminal = %#v, calls = %d; want immediate rejection", terminal, calls)
}
}
func TestRunProducerAttemptsUsesModuleRetryBudgetAndFallback(t *testing.T) {
t.Run("retry", func(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.Number == 1 {
return producerAttemptOutput{Value: "fallback", Retry: &producerRetryDirective{}}, nil
}
return producerAttemptOutput{Value: "replacement"}, nil
}, approveAttempt)
if err != nil {
t.Fatalf("runProducerAttempts() error = %v", err)
}
if terminal.Action != producerTerminalAccepted || terminal.Value != "replacement" || calls != 2 {
t.Fatalf("terminal = %#v, calls = %d", terminal, calls)
}
if got := attemptKinds(terminal.Provenance); !reflect.DeepEqual(got, []producerAttemptKind{producerAttemptInitial, producerAttemptModuleRetry}) {
t.Fatalf("attempt kinds = %v", got)
}
})
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) {
return producerAttemptOutput{Value: "fallback", Retry: &producerRetryDirective{FallbackDiagnostics: []contracts.ProducerDiagnostic{fallbackDiagnostic}}}, nil
}, approveAttempt)
if err != nil {
t.Fatalf("runProducerAttempts() error = %v", err)
}
if terminal.Action != producerTerminalAccepted || terminal.Value != "fallback" || !reflect.DeepEqual(terminal.Diagnostics, []contracts.ProducerDiagnostic{fallbackDiagnostic}) {
t.Fatalf("terminal = %#v", terminal)
}
})
}
func TestRunProducerAttemptsRejectsInvalidDiagnosticsWithoutRetry(t *testing.T) {
tests := []struct {
name string
output producerAttemptOutput
want string
}{
{
name: "producer diagnostics",
output: producerAttemptOutput{Diagnostics: []contracts.ProducerDiagnostic{{}}},
want: "producer returned invalid diagnostics",
},
{
name: "retry fallback diagnostics",
output: producerAttemptOutput{Retry: &producerRetryDirective{FallbackDiagnostics: []contracts.ProducerDiagnostic{{}}}},
want: "producer returned invalid retry fallback diagnostics",
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
producerCalls := 0
validatorCalls := 0
terminal, err := runProducerAttempts(context.Background(), producerAttemptConfig{Retries: 3, Policy: DefaultValidationPolicy()}, func(context.Context, producerAttemptRequest) (producerAttemptOutput, error) {
producerCalls++
return test.output, nil
}, func(context.Context, producerAttemptOutput) (validationReport, error) {
validatorCalls++
return validationReport{}, nil
})
if err == nil || !strings.Contains(err.Error(), test.want) {
t.Fatalf("runProducerAttempts() error = %v, want %q", err, test.want)
}
if terminal.Action != producerTerminalFailed || producerCalls != 1 || validatorCalls != 0 {
t.Fatalf("terminal = %#v, producer calls = %d, validator calls = %d; want immediate framework failure", terminal, producerCalls, validatorCalls)
}
})
}
}
func TestRunProducerAttemptsRejectionWinsOverValidatorFailure(t *testing.T) {
firstDiagnostics := []contracts.ProducerDiagnostic{producerDiagnostic("discarded", "discarded warning")}
secondDiagnostics := []contracts.ProducerDiagnostic{producerDiagnostic("accepted", "accepted warning")}
terminal, err := runProducerAttempts(context.Background(), producerAttemptConfig{Retries: 1, Policy: DefaultValidationPolicy()}, func(_ context.Context, request producerAttemptRequest) (producerAttemptOutput, error) {
if request.Number == 1 {
return producerAttemptOutput{Value: "first", Candidate: attemptCandidate(t, "defective"), Diagnostics: firstDiagnostics}, nil
}
return producerAttemptOutput{Value: "second", Candidate: attemptCandidate(t, "corrected"), Diagnostics: secondDiagnostics}, nil
}, func(_ context.Context, output producerAttemptOutput) (validationReport, error) {
if output.Value == "first" {
report := rejectedAttemptReport("defect")
report.records = append(report.records, validationRecord{validatorName: "unavailable", outcome: validationFailed, failure: errors.New("validator unavailable")})
return report, nil
}
return validationReport{}, nil
})
if err != nil {
t.Fatalf("runProducerAttempts() error = %v", err)
}
if terminal.Action != producerTerminalAccepted {
t.Fatalf("terminal action = %q, want accepted", terminal.Action)
}
if got := attemptKinds(terminal.Provenance); !reflect.DeepEqual(got, []producerAttemptKind{producerAttemptInitial, producerAttemptSemanticRetry}) {
t.Fatalf("attempt kinds = %v", got)
}
if got := terminal.Diagnostics; !reflect.DeepEqual(got, secondDiagnostics) {
t.Fatalf("terminal diagnostics = %#v, want %#v", got, secondDiagnostics)
}
if len(terminal.Provenance[0].Validation.records) != 2 {
t.Fatalf("first validation records = %#v, want rejection and failure", terminal.Provenance[0].Validation.records)
}
}
func TestValidationSummaryIsBoundedAndCorrectedSuccessIsQuiet(t *testing.T) {
terminal, err := runProducerAttempts(context.Background(), producerAttemptConfig{Retries: 1, Policy: DefaultValidationPolicy()}, func(_ context.Context, request producerAttemptRequest) (producerAttemptOutput, error) {
if request.Number == 1 {
return producerAttemptOutput{Value: "first", Candidate: attemptCandidate(t, "sensitive defective response")}, nil
}
return producerAttemptOutput{Value: "corrected", Candidate: attemptCandidate(t, "sensitive corrected response")}, nil
}, func(_ context.Context, output producerAttemptOutput) (validationReport, error) {
if output.Value == "first" {
return validationReport{records: []validationRecord{{validatorName: "first", outcome: validationRejected, reasonCode: "needs_fix", message: "sensitive diagnostic", correctionGuidance: "sensitive guidance"}}}, nil
}
return validationReport{}, nil
})
if err != nil {
t.Fatalf("runProducerAttempts() error = %v", err)
}
summary := validationSummary(terminal, StageExtract, "step", "lane", "module", "chunk", 3)
if summary.Status != "complete" || summary.ProducerAttemptCount != 2 || summary.TerminalAction != string(producerTerminalAccepted) {
t.Fatalf("summary = %#v", summary)
}
if len(summary.ReasonCodes) != 0 || len(summary.RejectingValidators) != 0 || len(summary.IncompleteValidators) != 0 {
t.Fatalf("corrected success summary retained prior findings: %#v", summary)
}
}
func TestWarnContinueRecordsOneWarningForEachExhaustedValidator(t *testing.T) {
report := validationReport{records: []validationRecord{
{validatorName: "first", outcome: validationFailed, attemptCount: 2, failure: errors.New("provider error with sensitive details")},
{validatorName: "second", outcome: validationSkipped, attemptCount: 1, reasonCode: "missing_prerequisite", message: "sensitive skipped detail"},
{validatorName: "third", outcome: validationFailed, attemptCount: 1, failure: errors.New("other provider error")},
}}
terminal, err := runProducerAttempts(context.Background(), producerAttemptConfig{Policy: DefaultValidationPolicy()}, func(context.Context, producerAttemptRequest) (producerAttemptOutput, error) {
return producerAttemptOutput{Value: "candidate"}, nil
}, func(context.Context, producerAttemptOutput) (validationReport, error) {
return report, nil
})
if err != nil {
t.Fatalf("runProducerAttempts() error = %v", err)
}
if terminal.Action != producerTerminalIncompleteAccepted {
t.Fatalf("terminal action = %q", terminal.Action)
}
groups, groupErr := terminalDiagnosticGroups(terminal, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageNormalize, StepID: "step", LaneID: "lane", ModuleKey: "module"}, nil)
if groupErr != nil {
t.Fatalf("terminalDiagnosticGroups() error = %v", groupErr)
}
if len(groups) != 3 {
t.Fatalf("diagnostic groups = %#v, want every failed and skipped validator", groups)
}
for index, validator := range []string{"first", "second", "third"} {
if group := groups[index]; group.Origin.ValidatorKey != validator || group.Disposition != contracts.DiagnosticDispositionWarning || group.Category != contracts.DiagnosticCategoryValidationIncomplete || group.ReasonCode != "validator_execution_incomplete" {
t.Fatalf("diagnostic group %d = %#v, want incomplete warning for %q", index, group, validator)
}
}
summary := validationSummary(terminal, StageNormalize, "step", "lane", "module", "", 0)
if summary.Status != "incomplete" || !reflect.DeepEqual(summary.IncompleteValidators, []string{"first", "second", "third"}) || !reflect.DeepEqual(summary.ReasonCodes, []string{"missing_prerequisite"}) {
t.Fatalf("summary = %#v", summary)
}
}
func TestRunProducerAttemptsStopsForCancellationAndDebugFailure(t *testing.T) {
tests := []struct {
name string
context context.Context
produce producerAttemptProducer
}{
{
name: "cancelled context",
context: cancelledAttemptContext(),
produce: func(context.Context, producerAttemptRequest) (producerAttemptOutput, error) {
return producerAttemptOutput{}, nil
},
},
{
name: "debug persistence failure",
context: context.Background(),
produce: func(context.Context, producerAttemptRequest) (producerAttemptOutput, error) {
return producerAttemptOutput{}, &attemptDebugPersistenceError{label: "attempt", err: errors.New("write debug")}
},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
calls := 0
producer := func(ctx context.Context, request producerAttemptRequest) (producerAttemptOutput, error) {
calls++
return test.produce(ctx, request)
}
terminal, err := runProducerAttempts(test.context, producerAttemptConfig{Retries: 3, Policy: DefaultValidationPolicy()}, producer, approveAttempt)
if err == nil || terminal.Action != producerTerminalFailed {
t.Fatalf("terminal = %#v, error = %v", terminal, err)
}
if test.name == "cancelled context" && calls != 0 {
t.Fatalf("cancelled producer calls = %d, want 0", calls)
}
if test.name == "debug persistence failure" && calls != 1 {
t.Fatalf("debug failure producer calls = %d, want 1", calls)
}
})
}
}
func attemptCandidate(t *testing.T, response string) *contracts.ModelCandidate {
t.Helper()
candidate, err := contracts.NewModelCandidate([]byte(response), contracts.CorrectionProtocolSingleResponseV1)
if err != nil {
t.Fatalf("NewModelCandidate() error = %v", err)
}
return candidate
}
func approveAttempt(context.Context, producerAttemptOutput) (validationReport, error) {
return validationReport{}, nil
}
func rejectedAttemptReport(reason string) validationReport {
return validationReport{records: []validationRecord{{validatorName: "validator", outcome: validationRejected, reasonCode: reason, message: "candidate rejected", correctionGuidance: "fix the defect"}}}
}
func failedAttemptReport() validationReport {
return validationReport{records: []validationRecord{{validatorName: "validator", outcome: validationFailed, attemptCount: 1, failure: errors.New("validator failure")}}}
}
func attemptKinds(provenance []producerAttemptProvenance) []producerAttemptKind {
kinds := make([]producerAttemptKind, len(provenance))
for index, item := range provenance {
kinds[index] = item.Kind
}
return kinds
}
func cancelledAttemptContext() context.Context {
ctx, cancel := context.WithCancel(context.Background())
cancel()
return ctx
}