From b487d93186d824b5273e1798e47c1738b3b58770 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 27 Aug 2026 00:27:27 +0000 Subject: [PATCH] Add producer attempt state machine --- docs/internal/pipeline.md | 11 + docs/roadmap/implementation.md | 2 +- .../framework/pipeline/producer_attempts.go | 310 +++++++++++++++ .../pipeline/producer_attempts_test.go | 358 ++++++++++++++++++ internal/framework/pipeline/runner.go | 2 +- .../framework/pipeline/runner_chunk_plan.go | 2 +- .../framework/pipeline/runner_concurrent.go | 2 +- internal/framework/pipeline/runner_typed.go | 4 +- 8 files changed, 685 insertions(+), 6 deletions(-) create mode 100644 internal/framework/pipeline/producer_attempts.go create mode 100644 internal/framework/pipeline/producer_attempts_test.go diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 9e7f8ade..a92f1719 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -151,6 +151,17 @@ or rejected attempt. Cancellation stops retries. Normalizer-specific retry directives consume this same budget and validate any final safe fallback through the normalizer chain. +The artifact-neutral producer-attempt state machine owns that shared budget, +attempt provenance, semantic-correction material, and terminal-policy +selection. It accepts producer and complete-validation closures, so artifact +materialization, cache handling, checkpoints, and debug output stay at the +operation boundary. It distinguishes operational, structural, module-requested, +and semantic retries. A semantic retry is available only for a valid latest +`single_response_v1` candidate; a deterministic or no-model rejection instead +settles the semantic policy immediately. Structural-output errors alone use the +structural policy, and validation failure without rejection settles the +validator-failure policy without regenerating the producer. + After terminal lane work, the runner assembles manifest provenance, normalized artifacts, rejections, warnings, and an optional accepted chunk map. When an output policy selected evidence lanes, it decodes accepted serialized normalize diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 50ee4d04..2a5c59a0 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -471,7 +471,7 @@ the producer candidate advances. Existing stage callers may still adapt the report through their old disposition path until the next stages. This stage is one Terra prompt. -## Stage 10 — Implement The Generic Producer Attempt State Machine +## Stage 10 — Implement The Generic Producer Attempt State Machine ✅ ### Goal diff --git a/internal/framework/pipeline/producer_attempts.go b/internal/framework/pipeline/producer_attempts.go new file mode 100644 index 00000000..bb0fef81 --- /dev/null +++ b/internal/framework/pipeline/producer_attempts.go @@ -0,0 +1,310 @@ +package pipeline + +import ( + "context" + "errors" + "fmt" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +// producerAttemptKind records why a producer invocation followed the prior +// one. It is deliberately independent of any artifact family. +type producerAttemptKind string + +const ( + producerAttemptInitial producerAttemptKind = "initial" + producerAttemptOperationalRetry producerAttemptKind = "operational_error_retry" + producerAttemptStructuralRetry producerAttemptKind = "structural_retry" + producerAttemptModuleRetry producerAttemptKind = "module_requested_retry" + producerAttemptSemanticRetry producerAttemptKind = "semantic_correction" +) + +type producerAttemptOutcome string + +const ( + producerAttemptAccepted producerAttemptOutcome = "accepted" + producerAttemptRejected producerAttemptOutcome = "rejected" + producerAttemptIncompleteAccepted producerAttemptOutcome = "incomplete_accepted" + producerAttemptRetried producerAttemptOutcome = "retried" + producerAttemptFailed producerAttemptOutcome = "failed" +) + +type producerTerminalAction string + +const ( + producerTerminalAccepted producerTerminalAction = "accepted" + producerTerminalRejected producerTerminalAction = "reject_output" + producerTerminalIncompleteAccepted producerTerminalAction = "warn_continue" + producerTerminalFailed producerTerminalAction = "fail_run" +) + +// producerAttemptRequest contains only attempt-local control material. The +// producer reconstructs its ordinary request from its own durable inputs. +type producerAttemptRequest struct { + Number int + Kind producerAttemptKind + Correction *contracts.SemanticCorrection +} + +// producerRetryDirective asks for another producer invocation while retaining +// 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 { + FallbackWarnings []contracts.Warning +} + +func (directive *producerRetryDirective) clone() *producerRetryDirective { + if directive == nil { + return nil + } + return &producerRetryDirective{FallbackWarnings: cloneWarnings(directive.FallbackWarnings)} +} + +// producerAttemptOutput is intentionally artifact-neutral. Value remains +// opaque to the state machine; Candidate is the attempt-local response that +// may support semantic correction. +type producerAttemptOutput struct { + Value any + Candidate *contracts.ModelCandidate + Warnings []contracts.Warning + Retry *producerRetryDirective +} + +func (output producerAttemptOutput) clone() (producerAttemptOutput, error) { + candidate, err := contracts.CloneModelCandidate(output.Candidate) + if err != nil { + return producerAttemptOutput{}, fmt.Errorf("clone model candidate: %w", err) + } + return producerAttemptOutput{ + Value: output.Value, + Candidate: candidate, + Warnings: cloneWarnings(output.Warnings), + Retry: output.Retry.clone(), + }, nil +} + +type producerAttemptProducer func(context.Context, producerAttemptRequest) (producerAttemptOutput, error) +type producerAttemptValidator func(context.Context, producerAttemptOutput) (validationReport, error) + +type producerAttemptConfig struct { + Retries int + Policy ValidationPolicy + AllowStructuralRetry bool +} + +// producerAttemptProvenance is ordered by producer invocation. It retains the +// settled validation report for later debug and durable-provenance adapters. +type producerAttemptProvenance struct { + Number int + Kind producerAttemptKind + Outcome producerAttemptOutcome + Validation validationReport +} + +func (provenance producerAttemptProvenance) clone() producerAttemptProvenance { + provenance.Validation = cloneValidationReport(provenance.Validation) + return provenance +} + +// producerAttemptTerminal describes the state-machine decision without +// materializing an artifact or persisting stage-specific diagnostics. +type producerAttemptTerminal struct { + Action producerTerminalAction + Value any + Warnings []contracts.Warning + Rejection *contracts.RejectedOutput + Validation validationReport + ValidationIncomplete bool + Provenance []producerAttemptProvenance +} + +func (terminal producerAttemptTerminal) clone() producerAttemptTerminal { + terminal.Warnings = cloneWarnings(terminal.Warnings) + if terminal.Rejection != nil { + rejection := *terminal.Rejection + terminal.Rejection = &rejection + } + terminal.Validation = cloneValidationReport(terminal.Validation) + terminal.Provenance = cloneProducerAttemptProvenance(terminal.Provenance) + return terminal +} + +func cloneProducerAttemptProvenance(provenance []producerAttemptProvenance) []producerAttemptProvenance { + if len(provenance) == 0 { + return nil + } + cloned := make([]producerAttemptProvenance, len(provenance)) + for index, item := range provenance { + cloned[index] = item.clone() + } + return cloned +} + +func cloneValidationReport(report validationReport) validationReport { + return validationReport{records: report.Records()} +} + +func runProducerAttempts(ctx context.Context, config producerAttemptConfig, produce producerAttemptProducer, validate producerAttemptValidator) (producerAttemptTerminal, error) { + if ctx == nil { + return producerAttemptTerminal{}, errors.New("producer attempt context must not be nil") + } + if config.Retries < 0 { + return producerAttemptTerminal{}, errors.New("producer retries must not be negative") + } + if produce == nil { + return producerAttemptTerminal{}, errors.New("producer attempt closure must not be nil") + } + if validate == nil { + return producerAttemptTerminal{}, errors.New("producer validation closure must not be nil") + } + + attemptLimit := config.Retries + 1 + provenance := make([]producerAttemptProvenance, 0, attemptLimit) + kind := producerAttemptInitial + var correction *contracts.SemanticCorrection + + for number := 1; number <= attemptLimit; number++ { + if err := ctx.Err(); err != nil { + return failedProducerAttempt(provenance), err + } + requestCorrection, err := contracts.CloneSemanticCorrection(correction) + if err != nil { + return failedProducerAttempt(provenance), fmt.Errorf("clone semantic correction: %w", err) + } + output, err := produce(ctx, producerAttemptRequest{Number: number, Kind: kind, Correction: requestCorrection}) + if err != nil { + provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptFailed}) + if isImmediateProducerFailure(err) { + return failedProducerAttempt(provenance), err + } + if errors.Is(err, contracts.ErrInvalidStructuredOutput) && config.AllowStructuralRetry { + if number < attemptLimit { + kind, correction = producerAttemptStructuralRetry, nil + continue + } + return applyStructuralTerminalPolicy(config.Policy, provenance, number, err) + } + if number < attemptLimit { + kind, correction = producerAttemptOperationalRetry, nil + continue + } + return failedProducerAttempt(provenance), fmt.Errorf("producer failed after %d attempt(s): %w", number, err) + } + + output, err = output.clone() + if err != nil { + provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptFailed}) + return failedProducerAttempt(provenance), err + } + if output.Retry != nil && number < attemptLimit { + provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptRetried}) + kind, correction = producerAttemptModuleRetry, nil + continue + } + if output.Retry != nil { + output.Warnings = append(output.Warnings, cloneWarnings(output.Retry.FallbackWarnings)...) + } + correctionCandidate, err := contracts.CloneModelCandidate(output.Candidate) + if err != nil { + provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptFailed}) + return failedProducerAttempt(provenance), fmt.Errorf("clone correction candidate: %w", err) + } + + report, err := validate(ctx, output) + if err != nil { + provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptFailed}) + return failedProducerAttempt(provenance), err + } + report = cloneValidationReport(report) + if err := ctx.Err(); err != nil { + provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptFailed, Validation: report}) + return failedProducerAttempt(provenance), err + } + + if rejection := report.FirstRejection(); rejection != nil { + if correctionCandidate != nil { + if err := correctionCandidate.Validate(); err != nil { + provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptFailed, Validation: report}) + return failedProducerAttempt(provenance), fmt.Errorf("validate producer model candidate: %w", err) + } + } + if number < attemptLimit && correctionCandidate != nil && correctionCandidate.Protocol == contracts.CorrectionProtocolSingleResponseV1 { + correction, err = contracts.NewSemanticCorrection(correctionCandidate.Response, report.CorrectionGuidance()) + if err != nil { + provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptFailed, Validation: report}) + return failedProducerAttempt(provenance), fmt.Errorf("construct semantic correction: %w", err) + } + provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptRetried, Validation: report}) + kind = producerAttemptSemanticRetry + continue + } + provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptRejected, Validation: report}) + return applySemanticTerminalPolicy(config.Policy, provenance, number, output, report, *rejection) + } + + if incomplete := firstIncompleteValidation(report); incomplete != nil { + provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptIncompleteAccepted, Validation: report}) + if config.Policy.ValidatorFailure == ValidatorFailureWarnContinue { + return producerAttemptTerminal{Action: producerTerminalIncompleteAccepted, Value: output.Value, Warnings: terminalWarnings(output, report), Validation: report, ValidationIncomplete: true, Provenance: cloneProducerAttemptProvenance(provenance)}, nil + } + return failedProducerAttempt(provenance), validatorFailureError(*incomplete) + } + + provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptAccepted, Validation: report}) + return producerAttemptTerminal{Action: producerTerminalAccepted, Value: output.Value, Warnings: terminalWarnings(output, report), Validation: report, Provenance: cloneProducerAttemptProvenance(provenance)}, nil + } + + return failedProducerAttempt(provenance), errors.New("producer attempt budget was not exhausted deterministically") +} + +func isImmediateProducerFailure(err error) bool { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return true + } + var debugErr *attemptDebugPersistenceError + return errors.As(err, &debugErr) +} + +func applyStructuralTerminalPolicy(policy ValidationPolicy, provenance []producerAttemptProvenance, number int, err error) (producerAttemptTerminal, error) { + switch policy.ProducerStructuralFailure { + case ProducerStructuralFailureRejectOutput: + return producerAttemptTerminal{Action: producerTerminalRejected, Rejection: &contracts.RejectedOutput{ReasonCode: "invalid_structured_output", Message: "producer returned invalid structured output", AttemptCount: number}, Provenance: cloneProducerAttemptProvenance(provenance)}, nil + case ProducerStructuralFailureFailRun: + return failedProducerAttempt(provenance), fmt.Errorf("producer returned invalid structured output after %d attempt(s): %w", number, err) + default: + return failedProducerAttempt(provenance), fmt.Errorf("unknown producer structural-failure action %q", policy.ProducerStructuralFailure) + } +} + +func applySemanticTerminalPolicy(policy ValidationPolicy, provenance []producerAttemptProvenance, number int, output producerAttemptOutput, report validationReport, rejection validationRecord) (producerAttemptTerminal, error) { + rejected := contracts.RejectedOutput{ReasonCode: rejection.reasonCode, Message: rejection.message, AttemptCount: number} + switch policy.SemanticRejection { + case SemanticRejectionRejectOutput: + return producerAttemptTerminal{Action: producerTerminalRejected, Warnings: terminalWarnings(output, report), Rejection: &rejected, Validation: report, ValidationIncomplete: firstIncompleteValidation(report) != nil, Provenance: cloneProducerAttemptProvenance(provenance)}, nil + case SemanticRejectionFailRun: + return failedProducerAttempt(provenance), fmt.Errorf("producer candidate rejected after %d attempt(s): %s", number, rejection.message) + default: + return failedProducerAttempt(provenance), fmt.Errorf("unknown semantic-rejection action %q", policy.SemanticRejection) + } +} + +func firstIncompleteValidation(report validationReport) *validationRecord { + for _, record := range report.records { + if record.outcome == validationFailed || record.outcome == validationSkipped { + clone := record.clone() + return &clone + } + } + return nil +} + +func terminalWarnings(output producerAttemptOutput, report validationReport) []contracts.Warning { + warnings := cloneWarnings(output.Warnings) + warnings = append(warnings, report.Warnings()...) + return warnings +} + +func failedProducerAttempt(provenance []producerAttemptProvenance) producerAttemptTerminal { + return producerAttemptTerminal{Action: producerTerminalFailed, Provenance: cloneProducerAttemptProvenance(provenance)} +} diff --git a/internal/framework/pipeline/producer_attempts_test.go b/internal/framework/pipeline/producer_attempts_test.go new file mode 100644 index 00000000..b9fb1afe --- /dev/null +++ b/internal/framework/pipeline/producer_attempts_test.go @@ -0,0 +1,358 @@ +package pipeline + +import ( + "context" + "errors" + "reflect" + "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) { + fallbackWarning := contracts.Warning{ReasonCode: "fallback", Message: "fallback warning"} + terminal, err := runProducerAttempts(context.Background(), producerAttemptConfig{Policy: DefaultValidationPolicy()}, func(context.Context, producerAttemptRequest) (producerAttemptOutput, error) { + return producerAttemptOutput{Value: "fallback", Retry: &producerRetryDirective{FallbackWarnings: []contracts.Warning{fallbackWarning}}}, nil + }, approveAttempt) + if err != nil { + t.Fatalf("runProducerAttempts() error = %v", err) + } + if terminal.Action != producerTerminalAccepted || terminal.Value != "fallback" || !reflect.DeepEqual(terminal.Warnings, []contracts.Warning{fallbackWarning}) { + t.Fatalf("terminal = %#v", terminal) + } + }) +} + +func TestRunProducerAttemptsRejectionWinsOverValidatorFailure(t *testing.T) { + firstWarnings := []contracts.Warning{{ReasonCode: "discarded", Message: "discarded warning"}} + secondWarnings := []contracts.Warning{{ReasonCode: "accepted", Message: "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"), Warnings: firstWarnings}, nil + } + return producerAttemptOutput{Value: "second", Candidate: attemptCandidate(t, "corrected"), Warnings: secondWarnings}, 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.Warnings; !reflect.DeepEqual(got, secondWarnings) { + t.Fatalf("terminal warnings = %#v, want %#v", got, secondWarnings) + } + if len(terminal.Provenance[0].Validation.records) != 2 { + t.Fatalf("first validation records = %#v, want rejection and failure", terminal.Provenance[0].Validation.records) + } +} + +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 +} diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 2387122f..289a9e5c 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -428,7 +428,7 @@ type retryAttemptResult struct { warnings []contracts.Warning } -func runWithRetry(ctx context.Context, retries int, run func(attempt int) (retryAttemptResult, error)) (retryAttemptResult, error) { +func runSimpleRetry(ctx context.Context, retries int, run func(attempt int) (retryAttemptResult, error)) (retryAttemptResult, error) { attempts := retries + 1 var last retryAttemptResult for attempt := 1; attempt <= attempts; attempt++ { diff --git a/internal/framework/pipeline/runner_chunk_plan.go b/internal/framework/pipeline/runner_chunk_plan.go index b25d6467..e9107a62 100644 --- a/internal/framework/pipeline/runner_chunk_plan.go +++ b/internal/framework/pipeline/runner_chunk_plan.go @@ -84,7 +84,7 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S } var producerWarnings []contracts.Warning - retryResult, err := runWithRetry(ctx, input.pipeline.Chunk.Retries, func(attempt int) (retryAttemptResult, error) { + retryResult, err := runSimpleRetry(ctx, input.pipeline.Chunk.Retries, func(attempt int) (retryAttemptResult, error) { attemptStarted := time.Now().UTC() attemptPath := path.Join("chunk", fmt.Sprintf("attempt-%02d", attempt)) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) diff --git a/internal/framework/pipeline/runner_concurrent.go b/internal/framework/pipeline/runner_concurrent.go index 82606516..e63f34f6 100644 --- a/internal/framework/pipeline/runner_concurrent.go +++ b/internal/framework/pipeline/runner_concurrent.go @@ -410,7 +410,7 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source. var accepted erasedExtractArtifact var serialized CheckpointArtifact var acceptedWarnings []contracts.Warning - retryResult, err := runWithRetry(ctx, lane.Extract.Retries, func(attempt int) (retryAttemptResult, error) { + retryResult, err := runSimpleRetry(ctx, lane.Extract.Retries, func(attempt int) (retryAttemptResult, error) { started := time.Now().UTC() attemptPath := path.Join("extract", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("chunk-%06d", chunk.Index+1), fmt.Sprintf("attempt-%02d", attempt)) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 6581d52d..0a8fd6f9 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -249,7 +249,7 @@ 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 := runWithRetry(ctx, lane.Merge.Retries, func(attempt int) (retryAttemptResult, error) { + retryResult, runErr := runSimpleRetry(ctx, lane.Merge.Retries, func(attempt int) (retryAttemptResult, error) { started := time.Now().UTC() attemptPath := path.Join("merge", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt)) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) @@ -351,7 +351,7 @@ 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 := runWithRetry(ctx, lane.Normalize.Retries, func(attempt int) (retryAttemptResult, error) { + retryResult, runErr := runSimpleRetry(ctx, lane.Normalize.Retries, func(attempt int) (retryAttemptResult, error) { started := time.Now().UTC() attemptPath := path.Join("normalize", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt)) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)