Execute complete validator chains
This commit is contained in:
@@ -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
|
bytes. Each typed validator receives a newly decoded value from the one
|
||||||
candidate serialization for that attempt, while serialized validators receive
|
candidate serialization for that attempt, while serialized validators receive
|
||||||
separately owned representation bytes and schema metadata. They may approve,
|
separately owned representation bytes and schema metadata. They may approve,
|
||||||
approve with warnings, reject, or fail. A rejection is an ordinary pipeline
|
approve with warnings, reject, fail, or be skipped when a runtime prerequisite
|
||||||
result; a validator error is a framework error.
|
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
|
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
|
complete validation chain. It preserves warnings only from the final accepted
|
||||||
|
|||||||
@@ -421,7 +421,7 @@ unsupported workflows fail before source parsing, and ordinary operational
|
|||||||
retries remain allowed when semantic correction cannot occur. This stage is
|
retries remain allowed when semantic correction cannot occur. This stage is
|
||||||
one Terra prompt.
|
one Terra prompt.
|
||||||
|
|
||||||
## Stage 9 — Build Complete Validator-Chain Execution
|
## Stage 9 — Build Complete Validator-Chain Execution ✅
|
||||||
|
|
||||||
### Goal
|
### Goal
|
||||||
|
|
||||||
|
|||||||
@@ -83,6 +83,7 @@ type preparedValidator struct {
|
|||||||
typedValidate typedValidateOperation
|
typedValidate typedValidateOperation
|
||||||
chunk contracts.ChunkValidator
|
chunk contracts.ChunkValidator
|
||||||
serialized contracts.SerializedValidator
|
serialized contracts.SerializedValidator
|
||||||
|
position int
|
||||||
}
|
}
|
||||||
|
|
||||||
// Prepare validates all configured options and constructs every selected
|
// Prepare validates all configured options and constructs every selected
|
||||||
|
|||||||
@@ -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)
|
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"}`)}
|
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
|
report, err := executeValidationChain(ctx, prepared, func(validatorCtx context.Context, item preparedValidator, validatorAttempt int) (validationInvocation, error) {
|
||||||
for index, item := range prepared.validators {
|
|
||||||
binding := item.resolved.Binding
|
binding := item.resolved.Binding
|
||||||
started := time.Now().UTC()
|
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))
|
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(ctx, attemptPath)
|
validatorCtx, llmScope := withIsolatedDebugLLMScope(validatorCtx, attemptPath)
|
||||||
var result contracts.ValidationResult
|
var result contracts.ValidationResult
|
||||||
requestMetadata, cloneErr := cloneMetadata(metadata)
|
requestMetadata, cloneErr := cloneMetadata(metadata)
|
||||||
if cloneErr != nil {
|
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)
|
requestChunks, cloneErr := cloneSourceChunks(chunks)
|
||||||
if cloneErr != nil {
|
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 {
|
switch item.resolved.Target {
|
||||||
case ValidatorTargetChunk:
|
case ValidatorTargetChunk:
|
||||||
@@ -497,7 +496,7 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument,
|
|||||||
case ValidatorTargetSerialized:
|
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...)})
|
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:
|
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 {
|
if err == nil {
|
||||||
err = contracts.ValidateValidationResult(result)
|
err = contracts.ValidateValidationResult(result)
|
||||||
@@ -511,27 +510,25 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument,
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
validationErr := fmt.Errorf("validate chunks with validator %q: %w", binding.Module, err)
|
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 {
|
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 {
|
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 {
|
return validationInvocation{result: result}, nil
|
||||||
reason := result.ReasonCode
|
})
|
||||||
if reason == "" {
|
if err != nil {
|
||||||
reason = "output_rejected"
|
return report.Warnings(), nil, err
|
||||||
}
|
}
|
||||||
message := result.Message
|
if failure := report.FirstFailure(); failure != nil {
|
||||||
if message == "" {
|
return report.Warnings(), nil, validatorFailureError(*failure)
|
||||||
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
|
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
|
||||||
}
|
}
|
||||||
warnings = append(warnings, result.Warnings...)
|
return report.Warnings(), nil, nil
|
||||||
}
|
|
||||||
return warnings, nil, nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func resolvedValidatorChain(stage ModuleStage, laneID string, moduleKey string, chains []ResolvedValidatorChain) ResolvedValidatorChain {
|
func resolvedValidatorChain(stage ModuleStage, laneID string, moduleKey string, chains []ResolvedValidatorChain) ResolvedValidatorChain {
|
||||||
|
|||||||
@@ -37,6 +37,11 @@ type terminalChunkValidator struct {
|
|||||||
err error
|
err error
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type retryingLLMChunkValidator struct {
|
||||||
|
requests []contracts.ChunkValidationRequest
|
||||||
|
calls int
|
||||||
|
}
|
||||||
|
|
||||||
type observingChunkValidator struct {
|
type observingChunkValidator struct {
|
||||||
request contracts.ChunkValidationRequest
|
request contracts.ChunkValidationRequest
|
||||||
}
|
}
|
||||||
@@ -73,6 +78,24 @@ func (v terminalChunkValidator) Validate(context.Context, contracts.ChunkValidat
|
|||||||
return v.result, v.err
|
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) {
|
func assertAttemptEnvelopeSequence(t *testing.T, debug *capturedDebugRecorder, prefix string, attempts ...int) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
marker := strings.TrimSuffix(prefix, "/") + "/attempt-"
|
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) {
|
func TestRunnerMaterializesAnnotatedPlanBeforeChunkValidation(t *testing.T) {
|
||||||
prepared, plan := preparedTerminalDebugPipeline(t)
|
prepared, plan := preparedTerminalDebugPipeline(t)
|
||||||
plan.Annotations = source.ChunkAnnotations{"same": []byte(`{"plan":1}`)}
|
plan.Annotations = source.ChunkAnnotations{"same": []byte(`{"plan":1}`)}
|
||||||
|
|||||||
@@ -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) {
|
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 {
|
if len(chain.validators) > 0 && target.candidate == nil {
|
||||||
candidate, err := validationCandidateArtifact(codec, target)
|
candidate, err := validationCandidateArtifact(codec, target)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -480,53 +479,48 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE
|
|||||||
}
|
}
|
||||||
target.candidate = &candidate
|
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
|
binding := item.resolved.Binding
|
||||||
var result contracts.ValidationResult
|
var result contracts.ValidationResult
|
||||||
var err error
|
var err error
|
||||||
started := time.Now().UTC()
|
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))
|
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(ctx, attemptPath)
|
validatorCtx, llmScope := withIsolatedDebugLLMScope(validatorCtx, attemptPath)
|
||||||
requestTarget := target
|
requestTarget := target
|
||||||
requestTarget.sourceInput = target.sourceInput.Clone()
|
requestTarget.sourceInput = target.sourceInput.Clone()
|
||||||
requestTarget.references = CloneReferenceSet(target.references)
|
requestTarget.references = CloneReferenceSet(target.references)
|
||||||
requestTarget.metadata, err = cloneMetadata(target.metadata)
|
requestTarget.metadata, err = cloneMetadata(target.metadata)
|
||||||
if err != nil {
|
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)
|
requestTarget.chunk, err = cloneSourceChunkPtr(target.chunk)
|
||||||
if err != nil {
|
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)
|
requestTarget.chunks, err = cloneSourceChunks(target.chunks)
|
||||||
if err != nil {
|
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 {
|
switch item.resolved.Target {
|
||||||
case ValidatorTargetTyped:
|
case ValidatorTargetTyped:
|
||||||
candidateValue, decodeErr := decodeTypedValidationCandidate(codec, *target.candidate)
|
candidateValue, decodeErr := decodeTypedValidationCandidate(codec, *target.candidate)
|
||||||
if decodeErr != nil {
|
if decodeErr != nil {
|
||||||
err = fmt.Errorf("decode %s candidate for typed validator %q: %w", target.stage, binding.Module, decodeErr)
|
return validationInvocation{}, fatalValidationError(fmt.Errorf("decode %s candidate for typed validator %q: %w", target.stage, binding.Module, decodeErr))
|
||||||
break
|
|
||||||
}
|
}
|
||||||
requestTarget.value = candidateValue
|
requestTarget.value = candidateValue
|
||||||
requestTarget.llmProfile = binding.LLMProfile
|
requestTarget.llmProfile = binding.LLMProfile
|
||||||
requestTarget.structuredOutputRepairAttempts = cloneStructuredOutputRepairAttempts(binding.StructuredOutputRepairAttempts)
|
requestTarget.structuredOutputRepairAttempts = cloneStructuredOutputRepairAttempts(binding.StructuredOutputRepairAttempts)
|
||||||
result, err = item.typedValidate(validatorCtx, item.typed, requestTarget)
|
result, err = item.typedValidate(validatorCtx, item.typed, requestTarget)
|
||||||
case ValidatorTargetSerialized:
|
case ValidatorTargetSerialized:
|
||||||
artifact, encodeErr := validationCandidateArtifact(codec, target)
|
artifact := cloneCheckpointArtifact(*target.candidate)
|
||||||
if encodeErr != nil {
|
|
||||||
err = encodeErr
|
|
||||||
break
|
|
||||||
}
|
|
||||||
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...)})
|
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:
|
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 {
|
if err == nil {
|
||||||
err = contracts.ValidateValidationResult(result)
|
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)}
|
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 {
|
if err != nil {
|
||||||
debugCall.Error = err.Error()
|
debugCall.Error = err.Error()
|
||||||
@@ -534,37 +528,29 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
validationErr := fmt.Errorf("validate typed %s output with validator %q: %w", target.stage, binding.Module, err)
|
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 {
|
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 {
|
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 {
|
return validationInvocation{result: result}, nil
|
||||||
reason := result.ReasonCode
|
})
|
||||||
if reason == "" {
|
if err != nil {
|
||||||
reason = "artifact_rejected"
|
return report.Warnings(), nil, err
|
||||||
}
|
}
|
||||||
message := result.Message
|
if failure := report.FirstFailure(); failure != nil {
|
||||||
if message == "" {
|
return report.Warnings(), nil, validatorFailureError(*failure)
|
||||||
message = "artifact rejected"
|
|
||||||
}
|
}
|
||||||
return warnings, &contracts.RejectedOutput{Stage: string(target.stage), StepID: target.stepID, LaneID: target.laneID, ModuleKey: target.moduleKey, ChunkID: func() string {
|
if rejection := report.FirstRejection(); rejection != nil {
|
||||||
|
chunkID, chunkIndex := "", 0
|
||||||
if target.chunk != nil {
|
if target.chunk != nil {
|
||||||
return target.chunk.ID
|
chunkID, chunkIndex = target.chunk.ID, target.chunk.Index
|
||||||
}
|
}
|
||||||
return ""
|
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
|
||||||
}(), ChunkIndex: func() int {
|
|
||||||
if target.chunk != nil {
|
|
||||||
return target.chunk.Index
|
|
||||||
}
|
}
|
||||||
return 0
|
return report.Warnings(), nil, nil
|
||||||
}(), ValidatorName: binding.Module, ReasonCode: reason, Message: message, AttemptCount: attempt, DiagnosticArtifactPath: result.DiagnosticArtifactPath}, nil
|
|
||||||
}
|
|
||||||
warnings = append(warnings, result.Warnings...)
|
|
||||||
}
|
|
||||||
return warnings, nil, nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func validationCandidateArtifact(codec artifactCodecEntry, target typedValidationTarget) (CheckpointArtifact, error) {
|
func validationCandidateArtifact(codec artifactCodecEntry, target typedValidationTarget) (CheckpointArtifact, error) {
|
||||||
|
|||||||
214
internal/framework/pipeline/validation_executor.go
Normal file
214
internal/framework/pipeline/validation_executor.go
Normal file
@@ -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)
|
||||||
|
}
|
||||||
207
internal/framework/pipeline/validation_executor_test.go
Normal file
207
internal/framework/pipeline/validation_executor_test.go
Normal file
@@ -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}}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user