Add producer attempt state machine
This commit is contained in:
310
internal/framework/pipeline/producer_attempts.go
Normal file
310
internal/framework/pipeline/producer_attempts.go
Normal file
@@ -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)}
|
||||
}
|
||||
358
internal/framework/pipeline/producer_attempts_test.go
Normal file
358
internal/framework/pipeline/producer_attempts_test.go
Normal file
@@ -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
|
||||
}
|
||||
@@ -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++ {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user