From 8dd7a4324de71b1aa3e4880371056f38d5a3b381 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 27 Aug 2026 00:18:16 +0000 Subject: [PATCH] Execute complete validator chains --- docs/internal/pipeline.md | 10 +- docs/roadmap/implementation.md | 2 +- internal/framework/pipeline/prepare.go | 1 + internal/framework/pipeline/runner.go | 43 ++-- .../pipeline/runner_terminal_debug_test.go | 45 ++++ internal/framework/pipeline/runner_typed.go | 70 +++--- .../framework/pipeline/validation_executor.go | 214 ++++++++++++++++++ .../pipeline/validation_executor_test.go | 207 +++++++++++++++++ 8 files changed, 524 insertions(+), 68 deletions(-) create mode 100644 internal/framework/pipeline/validation_executor.go create mode 100644 internal/framework/pipeline/validation_executor_test.go diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 2664d29e..9e7f8ade 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -136,8 +136,14 @@ their target: chunks, codec-decoded typed candidates, or serialized codec bytes. Each typed validator receives a newly decoded value from the one candidate serialization for that attempt, while serialized validators receive separately owned representation bytes and schema metadata. They may approve, -approve with warnings, reject, or fail. A rejection is an ordinary pipeline -result; a validator error is a framework error. +approve with warnings, reject, fail, or be skipped when a runtime prerequisite +is unavailable. The shared executor settles every configured validator in +order. A failed LLM-backed validator retries only itself against the same +immutable candidate; it does not regenerate the producer or alter the +validator request. Rejections stop that validator, while other configured +validators still run. The executor retains ordered results, bounded +deduplicated correction guidance from rejections, and only the final exhausted +failure outcome for each validator. The runner applies the binding's retry policy around a stage operation and its complete validation chain. It preserves warnings only from the final accepted diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 05f2c5e5..50ee4d04 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -421,7 +421,7 @@ unsupported workflows fail before source parsing, and ordinary operational retries remain allowed when semantic correction cannot occur. This stage is one Terra prompt. -## Stage 9 — Build Complete Validator-Chain Execution +## Stage 9 — Build Complete Validator-Chain Execution ✅ ### Goal diff --git a/internal/framework/pipeline/prepare.go b/internal/framework/pipeline/prepare.go index dcd02bcd..e74ec33d 100644 --- a/internal/framework/pipeline/prepare.go +++ b/internal/framework/pipeline/prepare.go @@ -83,6 +83,7 @@ type preparedValidator struct { typedValidate typedValidateOperation chunk contracts.ChunkValidator serialized contracts.SerializedValidator + position int } // Prepare validates all configured options and constructs every selected diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 07e5a23b..2387122f 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -476,20 +476,19 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument, return nil, nil, fmt.Errorf("encode canonical chunks for validation: %w", err) } schema := contracts.ArtifactSchema{ID: "notarius.source.chunks", Name: "notarius_source_chunks", Version: "v1", JSONSchema: []byte(`{"$schema":"https://json-schema.org/draft/2020-12/schema","type":"array"}`)} - var warnings []contracts.Warning - for index, item := range prepared.validators { + report, err := executeValidationChain(ctx, prepared, func(validatorCtx context.Context, item preparedValidator, validatorAttempt int) (validationInvocation, error) { binding := item.resolved.Binding started := time.Now().UTC() - attemptPath := path.Join("validate", fileio.EncodePathComponent(string(StageChunk)), "", fileio.EncodePathComponent(moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d", index+1, fileio.EncodePathComponent(binding.Module), attempt)) - validatorCtx, llmScope := withIsolatedDebugLLMScope(ctx, attemptPath) + attemptPath := validatorAttemptPath(path.Join("validate", fileio.EncodePathComponent(string(StageChunk)), "", fileio.EncodePathComponent(moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d", item.position, fileio.EncodePathComponent(binding.Module), attempt)), validatorAttempt) + validatorCtx, llmScope := withIsolatedDebugLLMScope(validatorCtx, attemptPath) var result contracts.ValidationResult requestMetadata, cloneErr := cloneMetadata(metadata) if cloneErr != nil { - return nil, nil, fmt.Errorf("clone chunk validation metadata: %w", cloneErr) + return validationInvocation{}, fatalValidationError(fmt.Errorf("clone chunk validation metadata: %w", cloneErr)) } requestChunks, cloneErr := cloneSourceChunks(chunks) if cloneErr != nil { - return nil, nil, fmt.Errorf("clone chunks for validation: %w", cloneErr) + return validationInvocation{}, fatalValidationError(fmt.Errorf("clone chunks for validation: %w", cloneErr)) } switch item.resolved.Target { case ValidatorTargetChunk: @@ -497,7 +496,7 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument, case ValidatorTargetSerialized: result, err = item.serialized.Validate(validatorCtx, contracts.SerializedValidationRequest{Stage: string(StageChunk), ModuleKey: moduleKey, Source: doc, SourceID: doc.ID, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(references), LLMProfile: binding.LLMProfile, StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(binding.StructuredOutputRepairAttempts), Metadata: requestMetadata, Chunks: requestChunks, Schema: contracts.CloneArtifactSchema(schema), MediaType: "application/json", Content: append([]byte(nil), content...)}) default: - return nil, nil, fmt.Errorf("validator %q is incompatible with chunk validation", binding.Module) + return validationInvocation{}, fatalValidationError(fmt.Errorf("validator %q is incompatible with chunk validation", binding.Module)) } if err == nil { err = contracts.ValidateValidationResult(result) @@ -511,27 +510,25 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument, if err != nil { validationErr := fmt.Errorf("validate chunks with validator %q: %w", binding.Module, err) if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope); debugErr != nil { - return warnings, nil, errors.Join(validationErr, fmt.Errorf("write chunk validator attempt debug artifact: %w", debugErr)) + return validationInvocation{}, fatalValidationError(errors.Join(validationErr, fmt.Errorf("write chunk validator attempt debug artifact: %w", debugErr))) } - return warnings, nil, validationErr + return validationInvocation{}, validationErr } if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall}, llmScope); debugErr != nil { - return warnings, nil, fmt.Errorf("write chunk validator attempt debug artifact: %w", debugErr) + return validationInvocation{}, fatalValidationError(fmt.Errorf("write chunk validator attempt debug artifact: %w", debugErr)) } - if !result.Approved { - reason := result.ReasonCode - if reason == "" { - reason = "output_rejected" - } - message := result.Message - if message == "" { - message = "output rejected" - } - return warnings, &contracts.RejectedOutput{Stage: string(StageChunk), ModuleKey: moduleKey, ValidatorName: binding.Module, ReasonCode: reason, Message: message, AttemptCount: attempt, DiagnosticArtifactPath: result.DiagnosticArtifactPath}, nil - } - warnings = append(warnings, result.Warnings...) + return validationInvocation{result: result}, nil + }) + if err != nil { + return report.Warnings(), nil, err } - return warnings, nil, nil + if failure := report.FirstFailure(); failure != nil { + return report.Warnings(), nil, validatorFailureError(*failure) + } + if rejection := report.FirstRejection(); rejection != nil { + return report.Warnings(), &contracts.RejectedOutput{Stage: string(StageChunk), ModuleKey: moduleKey, ValidatorName: rejection.validatorName, ReasonCode: rejection.reasonCode, Message: rejection.message, AttemptCount: attempt, DiagnosticArtifactPath: rejection.diagnosticPath}, nil + } + return report.Warnings(), nil, nil } func resolvedValidatorChain(stage ModuleStage, laneID string, moduleKey string, chains []ResolvedValidatorChain) ResolvedValidatorChain { diff --git a/internal/framework/pipeline/runner_terminal_debug_test.go b/internal/framework/pipeline/runner_terminal_debug_test.go index ed3b47fe..0d9c560f 100644 --- a/internal/framework/pipeline/runner_terminal_debug_test.go +++ b/internal/framework/pipeline/runner_terminal_debug_test.go @@ -37,6 +37,11 @@ type terminalChunkValidator struct { err error } +type retryingLLMChunkValidator struct { + requests []contracts.ChunkValidationRequest + calls int +} + type observingChunkValidator struct { request contracts.ChunkValidationRequest } @@ -73,6 +78,24 @@ func (v terminalChunkValidator) Validate(context.Context, contracts.ChunkValidat return v.result, v.err } +func (*retryingLLMChunkValidator) Name() string { return "retrying/llm-chunk-validator" } + +func (*retryingLLMChunkValidator) ExecutionClass() contracts.ExecutionClass { + return contracts.ExecutionClassLLMBacked +} + +func (validator *retryingLLMChunkValidator) Validate(_ context.Context, request contracts.ChunkValidationRequest) (contracts.ValidationResult, error) { + validator.calls++ + validator.requests = append(validator.requests, request) + if validator.calls == 1 && len(request.Chunks) > 0 { + request.Chunks[0].ID = "mutated" + } + if validator.calls == 1 { + return contracts.ValidationResult{}, errors.New("temporary validator failure") + } + return contracts.ValidationResult{Approved: true}, nil +} + func assertAttemptEnvelopeSequence(t *testing.T, debug *capturedDebugRecorder, prefix string, attempts ...int) { t.Helper() marker := strings.TrimSuffix(prefix, "/") + "/attempt-" @@ -152,6 +175,28 @@ func TestRunnerRecordsChunkTerminalOutcomes(t *testing.T) { } } +func TestRunnerRetriesLLMValidatorsWithoutRegeneratingChunkCandidate(t *testing.T) { + prepared, plan := preparedTerminalDebugPipeline(t) + chunkCalls := 0 + prepared.resolved.Chunk.Retries = 0 + prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan, calls: &chunkCalls} + validator := &retryingLLMChunkValidator{} + prepared.chunkValidators.validators = []preparedValidator{{ + resolved: ResolvedValidator{Binding: ModuleBinding{Module: validator.Name(), Retries: 1}, ExecutionClass: contracts.ExecutionClassLLMBacked, Target: ValidatorTargetChunk}, + chunk: validator, + }} + + if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")}); err != nil { + t.Fatalf("Run() error = %v", err) + } + if chunkCalls != 1 || validator.calls != 2 { + t.Fatalf("producer calls = %d validator calls = %d, want 1 and 2", chunkCalls, validator.calls) + } + if len(validator.requests) != 2 || len(validator.requests[0].Chunks) == 0 || len(validator.requests[1].Chunks) == 0 || validator.requests[1].Chunks[0].ID == "mutated" { + t.Fatalf("validator requests = %#v, want independently owned immutable candidates", validator.requests) + } +} + func TestRunnerMaterializesAnnotatedPlanBeforeChunkValidation(t *testing.T) { prepared, plan := preparedTerminalDebugPipeline(t) plan.Annotations = source.ChunkAnnotations{"same": []byte(`{"plan":1}`)} diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 0565a22b..6581d52d 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -472,7 +472,6 @@ func setTypedLaneManifestMetadata(output *RunOutput, laneID string, extractor, m } func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecEntry, target typedValidationTarget, chain preparedValidatorChain, attempt int, debug DebugRecorder) ([]contracts.Warning, *contracts.RejectedOutput, error) { - var warnings []contracts.Warning if len(chain.validators) > 0 && target.candidate == nil { candidate, err := validationCandidateArtifact(codec, target) if err != nil { @@ -480,53 +479,48 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE } target.candidate = &candidate } - for index, item := range chain.validators { + report, err := executeValidationChain(ctx, chain, func(validatorCtx context.Context, item preparedValidator, validatorAttempt int) (validationInvocation, error) { binding := item.resolved.Binding var result contracts.ValidationResult var err error started := time.Now().UTC() - attemptPath := path.Join("validate", fileio.EncodePathComponent(string(target.stage)), fileio.EncodePathComponent(target.laneID), fileio.EncodePathComponent(target.moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d", index+1, fileio.EncodePathComponent(binding.Module), attempt)) - validatorCtx, llmScope := withIsolatedDebugLLMScope(ctx, attemptPath) + attemptPath := validatorAttemptPath(path.Join("validate", fileio.EncodePathComponent(string(target.stage)), fileio.EncodePathComponent(target.laneID), fileio.EncodePathComponent(target.moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d", item.position, fileio.EncodePathComponent(binding.Module), attempt)), validatorAttempt) + validatorCtx, llmScope := withIsolatedDebugLLMScope(validatorCtx, attemptPath) requestTarget := target requestTarget.sourceInput = target.sourceInput.Clone() requestTarget.references = CloneReferenceSet(target.references) requestTarget.metadata, err = cloneMetadata(target.metadata) if err != nil { - return nil, nil, fmt.Errorf("clone typed validation metadata: %w", err) + return validationInvocation{}, fatalValidationError(fmt.Errorf("clone typed validation metadata: %w", err)) } requestTarget.chunk, err = cloneSourceChunkPtr(target.chunk) if err != nil { - return nil, nil, fmt.Errorf("clone typed validation chunk: %w", err) + return validationInvocation{}, fatalValidationError(fmt.Errorf("clone typed validation chunk: %w", err)) } requestTarget.chunks, err = cloneSourceChunks(target.chunks) if err != nil { - return nil, nil, fmt.Errorf("clone typed validation chunks: %w", err) + return validationInvocation{}, fatalValidationError(fmt.Errorf("clone typed validation chunks: %w", err)) } switch item.resolved.Target { case ValidatorTargetTyped: candidateValue, decodeErr := decodeTypedValidationCandidate(codec, *target.candidate) if decodeErr != nil { - err = fmt.Errorf("decode %s candidate for typed validator %q: %w", target.stage, binding.Module, decodeErr) - break + return validationInvocation{}, fatalValidationError(fmt.Errorf("decode %s candidate for typed validator %q: %w", target.stage, binding.Module, decodeErr)) } requestTarget.value = candidateValue requestTarget.llmProfile = binding.LLMProfile requestTarget.structuredOutputRepairAttempts = cloneStructuredOutputRepairAttempts(binding.StructuredOutputRepairAttempts) result, err = item.typedValidate(validatorCtx, item.typed, requestTarget) case ValidatorTargetSerialized: - artifact, encodeErr := validationCandidateArtifact(codec, target) - if encodeErr != nil { - err = encodeErr - break - } + artifact := cloneCheckpointArtifact(*target.candidate) result, err = item.serialized.Validate(validatorCtx, contracts.SerializedValidationRequest{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Source: target.source, SourceID: target.sourceID, SourceInput: requestTarget.sourceInput, SessionID: target.sessionID, References: requestTarget.references, LLMProfile: binding.LLMProfile, StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(binding.StructuredOutputRepairAttempts), Metadata: requestTarget.metadata, Chunk: requestTarget.chunk, Chunks: requestTarget.chunks, Schema: contracts.CloneArtifactSchema(artifact.Artifact.Schema), MediaType: artifact.Artifact.MediaType, Content: append([]byte(nil), artifact.Artifact.Content...)}) default: - return nil, nil, fmt.Errorf("validator %q is incompatible with typed artifact validation", binding.Module) + return validationInvocation{}, fatalValidationError(fmt.Errorf("validator %q is incompatible with typed artifact validation", binding.Module)) } if err == nil { err = contracts.ValidateValidationResult(result) } - artifact, _ := validationCandidateArtifact(codec, target) + artifact := cloneCheckpointArtifact(*target.candidate) debugCall := debugValidationCall{ValidatorName: binding.Module, Request: map[string]any{"stage": string(target.stage), "lane_id": target.laneID, "module_key": target.moduleKey, "source_id": target.sourceID, "artifact": debugCheckpointArtifact(artifact), "metadata": redactSensitiveMap(target.metadata)}, Result: debugValidationResultEnvelope(result)} if err != nil { debugCall.Error = err.Error() @@ -534,37 +528,29 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE if err != nil { validationErr := fmt.Errorf("validate typed %s output with validator %q: %w", target.stage, binding.Module, err) if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(target.stage), StepID: target.stepID, LaneID: target.laneID, ModuleKey: target.moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope); debugErr != nil { - return warnings, nil, errors.Join(validationErr, fmt.Errorf("write typed validator attempt debug artifact: %w", debugErr)) + return validationInvocation{}, fatalValidationError(errors.Join(validationErr, fmt.Errorf("write typed validator attempt debug artifact: %w", debugErr))) } - return warnings, nil, validationErr + return validationInvocation{}, validationErr } if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(target.stage), StepID: target.stepID, LaneID: target.laneID, ModuleKey: target.moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall}, llmScope); debugErr != nil { - return warnings, nil, fmt.Errorf("write typed validator attempt debug artifact: %w", debugErr) + return validationInvocation{}, fatalValidationError(fmt.Errorf("write typed validator attempt debug artifact: %w", debugErr)) } - if !result.Approved { - reason := result.ReasonCode - if reason == "" { - reason = "artifact_rejected" - } - message := result.Message - if message == "" { - message = "artifact rejected" - } - return warnings, &contracts.RejectedOutput{Stage: string(target.stage), StepID: target.stepID, LaneID: target.laneID, ModuleKey: target.moduleKey, ChunkID: func() string { - if target.chunk != nil { - return target.chunk.ID - } - return "" - }(), ChunkIndex: func() int { - if target.chunk != nil { - return target.chunk.Index - } - return 0 - }(), ValidatorName: binding.Module, ReasonCode: reason, Message: message, AttemptCount: attempt, DiagnosticArtifactPath: result.DiagnosticArtifactPath}, nil - } - warnings = append(warnings, result.Warnings...) + return validationInvocation{result: result}, nil + }) + if err != nil { + return report.Warnings(), nil, err } - return warnings, nil, nil + if failure := report.FirstFailure(); failure != nil { + return report.Warnings(), nil, validatorFailureError(*failure) + } + if rejection := report.FirstRejection(); rejection != nil { + chunkID, chunkIndex := "", 0 + if target.chunk != nil { + chunkID, chunkIndex = target.chunk.ID, target.chunk.Index + } + return report.Warnings(), &contracts.RejectedOutput{Stage: string(target.stage), StepID: target.stepID, LaneID: target.laneID, ModuleKey: target.moduleKey, ChunkID: chunkID, ChunkIndex: chunkIndex, ValidatorName: rejection.validatorName, ReasonCode: rejection.reasonCode, Message: rejection.message, AttemptCount: attempt, DiagnosticArtifactPath: rejection.diagnosticPath}, nil + } + return report.Warnings(), nil, nil } func validationCandidateArtifact(codec artifactCodecEntry, target typedValidationTarget) (CheckpointArtifact, error) { diff --git a/internal/framework/pipeline/validation_executor.go b/internal/framework/pipeline/validation_executor.go new file mode 100644 index 00000000..1d570386 --- /dev/null +++ b/internal/framework/pipeline/validation_executor.go @@ -0,0 +1,214 @@ +package pipeline + +import ( + "context" + "errors" + "fmt" + "strings" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +type validationOutcome string + +const ( + validationApproved validationOutcome = "approved" + validationRejected validationOutcome = "rejected" + validationFailed validationOutcome = "failed" + validationSkipped validationOutcome = "skipped" +) + +const defaultCorrectionGuidance = "Correct the candidate to satisfy the validator requirements." + +// validationRecord captures the settled result of one configured validator. +// Its fields remain private so reports cannot expose mutable warning storage. +type validationRecord struct { + validatorName string + outcome validationOutcome + attemptCount int + reasonCode string + message string + diagnosticPath string + warnings []contracts.Warning + correctionGuidance string + failure error +} + +func (record validationRecord) clone() validationRecord { + record.warnings = cloneWarnings(record.warnings) + return record +} + +// validationReport is the immutable, ordered result of one validator chain. +type validationReport struct { + records []validationRecord +} + +func (report validationReport) Records() []validationRecord { + records := make([]validationRecord, len(report.records)) + for index, record := range report.records { + records[index] = record.clone() + } + return records +} + +func (report validationReport) Warnings() []contracts.Warning { + var warnings []contracts.Warning + for _, record := range report.records { + if record.outcome == validationApproved || record.outcome == validationRejected { + warnings = append(warnings, cloneWarnings(record.warnings)...) + } + } + return warnings +} + +func (report validationReport) FirstRejection() *validationRecord { + for _, record := range report.records { + if record.outcome == validationRejected { + clone := record.clone() + return &clone + } + } + return nil +} + +func (report validationReport) FirstFailure() *validationRecord { + for _, record := range report.records { + if record.outcome == validationFailed { + clone := record.clone() + return &clone + } + } + return nil +} + +func (report validationReport) CorrectionGuidance() string { + seen := make(map[string]struct{}) + parts := make([]string, 0, len(report.records)) + used := 0 + for _, record := range report.records { + if record.outcome != validationRejected { + continue + } + guidance := strings.TrimSpace(record.correctionGuidance) + if guidance == "" { + guidance = defaultCorrectionGuidance + } + if _, exists := seen[guidance]; exists { + continue + } + separator := 0 + if len(parts) > 0 { + separator = 1 + } + if used+separator+len(guidance) > contracts.MaxCorrectionGuidanceBytes { + break + } + seen[guidance] = struct{}{} + parts = append(parts, guidance) + used += separator + len(guidance) + } + return strings.Join(parts, "\n") +} + +type validationInvocation struct { + result contracts.ValidationResult + skipped bool + reason string + message string +} + +func skippedValidation(reason, message string) validationInvocation { + return validationInvocation{skipped: true, reason: strings.TrimSpace(reason), message: strings.TrimSpace(message)} +} + +type validationInvoker func(context.Context, preparedValidator, int) (validationInvocation, error) + +type validationFrameworkError struct{ err error } + +func (err validationFrameworkError) Error() string { return err.err.Error() } + +func (err validationFrameworkError) Unwrap() error { return err.err } + +func fatalValidationError(err error) error { + if err == nil { + return nil + } + return validationFrameworkError{err: err} +} + +func executeValidationChain(ctx context.Context, chain preparedValidatorChain, invoke validationInvoker) (validationReport, error) { + if ctx == nil { + return validationReport{}, errors.New("validator execution context must not be nil") + } + if invoke == nil { + return validationReport{}, errors.New("validator invocation must not be nil") + } + report := validationReport{records: make([]validationRecord, 0, len(chain.validators))} + for index, validator := range chain.validators { + validator.position = index + 1 + if err := ctx.Err(); err != nil { + return validationReport{}, err + } + binding := validator.resolved.Binding + attemptLimit := 1 + if validator.resolved.ExecutionClass == contracts.ExecutionClassLLMBacked { + attemptLimit += binding.Retries + } + for attempt := 1; attempt <= attemptLimit; attempt++ { + if err := ctx.Err(); err != nil { + return validationReport{}, err + } + invocation, err := invoke(ctx, validator, attempt) + if err != nil { + var frameworkErr validationFrameworkError + if errors.As(err, &frameworkErr) || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return validationReport{}, err + } + if attempt == attemptLimit { + report.records = append(report.records, validationRecord{validatorName: binding.Module, outcome: validationFailed, attemptCount: attempt, message: "validator failed after retry exhaustion", failure: err}) + } + continue + } + if invocation.skipped { + report.records = append(report.records, validationRecord{validatorName: binding.Module, outcome: validationSkipped, attemptCount: attempt, reasonCode: invocation.reason, message: invocation.message}) + break + } + if err := contracts.ValidateValidationResult(invocation.result); err != nil { + if attempt == attemptLimit { + report.records = append(report.records, validationRecord{validatorName: binding.Module, outcome: validationFailed, attemptCount: attempt, message: "validator returned an invalid result", failure: err}) + } + continue + } + if invocation.result.Approved { + report.records = append(report.records, validationRecord{validatorName: binding.Module, outcome: validationApproved, attemptCount: attempt, warnings: cloneWarnings(invocation.result.Warnings), diagnosticPath: invocation.result.DiagnosticArtifactPath}) + break + } + reason := invocation.result.ReasonCode + if reason == "" { + reason = "output_rejected" + } + message := invocation.result.Message + if message == "" { + message = "output rejected" + } + report.records = append(report.records, validationRecord{validatorName: binding.Module, outcome: validationRejected, attemptCount: attempt, reasonCode: reason, message: message, diagnosticPath: invocation.result.DiagnosticArtifactPath, warnings: cloneWarnings(invocation.result.Warnings), correctionGuidance: invocation.result.CorrectionGuidance}) + break + } + } + return report, nil +} + +func validatorFailureError(record validationRecord) error { + if record.failure != nil { + return record.failure + } + return fmt.Errorf("validator %q failed after %d attempt(s)", record.validatorName, record.attemptCount) +} + +func validatorAttemptPath(base string, attempt int) string { + if attempt == 1 { + return base + } + return fmt.Sprintf("%s-validator-%02d", base, attempt) +} diff --git a/internal/framework/pipeline/validation_executor_test.go b/internal/framework/pipeline/validation_executor_test.go new file mode 100644 index 00000000..c6cf7321 --- /dev/null +++ b/internal/framework/pipeline/validation_executor_test.go @@ -0,0 +1,207 @@ +package pipeline + +import ( + "context" + "errors" + "reflect" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +func TestExecuteValidationChainSettlesEveryValidatorInOrder(t *testing.T) { + tests := []struct { + name string + chain preparedValidatorChain + invoke validationInvoker + want []validationOutcome + wantCalls []string + wantGuidance string + wantWarnings []string + }{ + { + name: "all approved", + chain: validationChain(validationSpec("shape", contracts.ExecutionClassDeterministic, 0), validationSpec("refs", contracts.ExecutionClassDeterministic, 0)), + invoke: validationSequence(map[string][]validationStep{ + "shape": {{result: contracts.ValidationResult{Approved: true, Warnings: []contracts.Warning{{ReasonCode: "shape"}}}}}, + "refs": {{result: contracts.ValidationResult{Approved: true, Warnings: []contracts.Warning{{ReasonCode: "refs"}}}}}, + }), + want: []validationOutcome{validationApproved, validationApproved}, + wantCalls: []string{"shape:1", "refs:1"}, + wantWarnings: []string{"shape", "refs"}, + }, + { + name: "multiple rejections deduplicate guidance", + chain: validationChain(validationSpec("shape", contracts.ExecutionClassDeterministic, 0), validationSpec("refs", contracts.ExecutionClassDeterministic, 0), validationSpec("coverage", contracts.ExecutionClassDeterministic, 0)), + invoke: validationSequence(map[string][]validationStep{ + "shape": {{result: contracts.ValidationResult{ReasonCode: "shape", Message: "invalid", CorrectionGuidance: "repair shape"}}}, + "refs": {{result: contracts.ValidationResult{ReasonCode: "refs", Message: "missing", CorrectionGuidance: "repair references"}}}, + "coverage": {{result: contracts.ValidationResult{ReasonCode: "coverage", Message: "missing", CorrectionGuidance: "repair shape"}}}, + }), + want: []validationOutcome{validationRejected, validationRejected, validationRejected}, + wantCalls: []string{"shape:1", "refs:1", "coverage:1"}, + wantGuidance: "repair shape\nrepair references", + }, + { + name: "rejection and failure both settle", + chain: validationChain(validationSpec("shape", contracts.ExecutionClassDeterministic, 0), validationSpec("remote", contracts.ExecutionClassDeterministic, 0)), + invoke: validationSequence(map[string][]validationStep{ + "shape": {{result: contracts.ValidationResult{ReasonCode: "shape", CorrectionGuidance: "repair shape"}}}, + "remote": {{err: errors.New("offline")}}, + }), + want: []validationOutcome{validationRejected, validationFailed}, + wantCalls: []string{"shape:1", "remote:1"}, + wantGuidance: "repair shape", + }, + { + name: "failure only has no correction guidance", + chain: validationChain(validationSpec("remote", contracts.ExecutionClassDeterministic, 0)), + invoke: validationSequence(map[string][]validationStep{"remote": {{err: errors.New("offline")}}}), + want: []validationOutcome{validationFailed}, + wantCalls: []string{"remote:1"}, + }, + { + name: "skipped is retained without guidance", + chain: validationChain(validationSpec("optional", contracts.ExecutionClassDeterministic, 0)), + invoke: validationSequence(map[string][]validationStep{"optional": {{invocation: skippedValidation("runtime_unavailable", "optional dependency unavailable")}}}), + want: []validationOutcome{validationSkipped}, + wantCalls: []string{"optional:1"}, + }, + { + name: "LLM failure retries then succeeds", + chain: validationChain(validationSpec("remote", contracts.ExecutionClassLLMBacked, 2)), + invoke: validationSequence(map[string][]validationStep{"remote": {{err: errors.New("first")}, {err: errors.New("second")}, {result: contracts.ValidationResult{Approved: true}}}}), + want: []validationOutcome{validationApproved}, + wantCalls: []string{"remote:1", "remote:2", "remote:3"}, + }, + { + name: "LLM failure retry exhaustion records once", + chain: validationChain(validationSpec("remote", contracts.ExecutionClassLLMBacked, 1)), + invoke: validationSequence(map[string][]validationStep{"remote": {{err: errors.New("first")}, {err: errors.New("second")}}}), + want: []validationOutcome{validationFailed}, + wantCalls: []string{"remote:1", "remote:2"}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + calls := []string(nil) + invoke := func(ctx context.Context, validator preparedValidator, attempt int) (validationInvocation, error) { + calls = append(calls, validator.resolved.Binding.Module+":"+string(rune('0'+attempt))) + return test.invoke(ctx, validator, attempt) + } + report, err := executeValidationChain(context.Background(), test.chain, invoke) + if err != nil { + t.Fatalf("executeValidationChain() error = %v", err) + } + records := report.Records() + outcomes := make([]validationOutcome, len(records)) + for index, record := range records { + outcomes[index] = record.outcome + } + if !reflect.DeepEqual(outcomes, test.want) || !reflect.DeepEqual(calls, test.wantCalls) { + t.Fatalf("outcomes = %#v calls = %#v, want %#v %#v", outcomes, calls, test.want, test.wantCalls) + } + if guidance := report.CorrectionGuidance(); guidance != test.wantGuidance { + t.Fatalf("CorrectionGuidance() = %q, want %q", guidance, test.wantGuidance) + } + warnings := report.Warnings() + var warningCodes []string + for _, warning := range warnings { + warningCodes = append(warningCodes, warning.ReasonCode) + } + if !reflect.DeepEqual(warningCodes, test.wantWarnings) { + t.Fatalf("Warnings() = %#v, want codes %#v", warnings, test.wantWarnings) + } + }) + } +} + +func TestExecuteValidationChainKeepsInvocationInputsImmutable(t *testing.T) { + candidate := []byte(`{"events":["one"]}`) + chain := validationChain(validationSpec("remote", contracts.ExecutionClassLLMBacked, 1)) + var received [][]byte + report, err := executeValidationChain(context.Background(), chain, func(_ context.Context, _ preparedValidator, attempt int) (validationInvocation, error) { + requestCandidate := append([]byte(nil), candidate...) + received = append(received, requestCandidate) + requestCandidate[0] = 'x' + if attempt == 1 { + return validationInvocation{}, errors.New("retry") + } + return validationInvocation{result: contracts.ValidationResult{Approved: true}}, nil + }) + if err != nil || len(report.Records()) != 1 || report.Records()[0].attemptCount != 2 { + t.Fatalf("report = %#v, error = %v", report, err) + } + if string(candidate) != `{"events":["one"]}` || len(received) != 2 || string(received[0]) != string(received[1]) { + t.Fatalf("candidate = %q received = %#v, want immutable repeated request content", candidate, received) + } +} + +func TestValidationReportBoundsGuidanceAndOwnsRecords(t *testing.T) { + guidance := strings.Repeat("x", contracts.MaxValidationCorrectionGuidanceBytes) + report := validationReport{records: make([]validationRecord, 17)} + for index := range report.records { + unique := []byte(guidance) + unique[0] = byte('a' + index) + report.records[index] = validationRecord{validatorName: "validator", outcome: validationRejected, correctionGuidance: string(unique)} + } + report.records[0].warnings = []contracts.Warning{{ReasonCode: "warning"}} + if got := report.CorrectionGuidance(); len(got) > contracts.MaxCorrectionGuidanceBytes || !strings.Contains(got, string([]byte{byte('a')})) || strings.Contains(got, string([]byte{byte('q')})) { + t.Fatalf("CorrectionGuidance() length = %d, want bounded ordered guidance", len(got)) + } + records := report.Records() + records[0].warnings[0].ReasonCode = "changed" + if report.Records()[0].warnings[0].ReasonCode != "warning" { + t.Fatal("Records() exposed mutable warning storage") + } +} + +func TestExecuteValidationChainReturnsFrameworkAndCancellationErrors(t *testing.T) { + chain := validationChain(validationSpec("validator", contracts.ExecutionClassLLMBacked, 1)) + framework := errors.New("debug persistence failed") + if _, err := executeValidationChain(context.Background(), chain, func(context.Context, preparedValidator, int) (validationInvocation, error) { + return validationInvocation{}, fatalValidationError(framework) + }); !errors.Is(err, framework) { + t.Fatalf("framework error = %v, want %v", err, framework) + } + canceled, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := executeValidationChain(canceled, chain, func(context.Context, preparedValidator, int) (validationInvocation, error) { + return validationInvocation{result: contracts.ValidationResult{Approved: true}}, nil + }); !errors.Is(err, context.Canceled) { + t.Fatalf("cancellation error = %v, want context canceled", err) + } +} + +type validationStep struct { + result contracts.ValidationResult + invocation validationInvocation + err error +} + +func validationSequence(steps map[string][]validationStep) validationInvoker { + positions := make(map[string]int) + return func(_ context.Context, validator preparedValidator, _ int) (validationInvocation, error) { + name := validator.resolved.Binding.Module + index := positions[name] + positions[name]++ + if index >= len(steps[name]) { + return validationInvocation{}, errors.New("unexpected validator invocation") + } + step := steps[name][index] + if step.invocation.skipped { + return step.invocation, step.err + } + return validationInvocation{result: step.result}, step.err + } +} + +func validationChain(validators ...preparedValidator) preparedValidatorChain { + return preparedValidatorChain{validators: validators} +} + +func validationSpec(name string, class contracts.ExecutionClass, retries int) preparedValidator { + return preparedValidator{resolved: ResolvedValidator{Binding: ModuleBinding{Module: name, Retries: retries}, ExecutionClass: class}} +}