From 14bfae216dbeb960cca920f8e40fa4a9eddedd58 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sun, 9 Aug 2026 01:11:03 +0000 Subject: [PATCH] Isolate typed validator candidates --- docs/internal/pipeline.md | 9 +- .../pipeline/artifact_codec_registry_test.go | 12 +- .../runner_candidate_encoding_test.go | 166 +++++++++++++++++- internal/framework/pipeline/runner_typed.go | 19 ++ 4 files changed, 195 insertions(+), 11 deletions(-) diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 0589337..6e0ab95 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -115,9 +115,12 @@ for started workers, and prevents output encoding. Every chunk, extract, merge, and normalize candidate passes its resolved validator chain. Validators receive immutable canonical input appropriate to -their target: chunks, typed values, or serialized codec bytes. They may -approve, approve with warnings, reject, or fail. A rejection is an ordinary -pipeline result; a validator error is a framework error. +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. 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/internal/framework/pipeline/artifact_codec_registry_test.go b/internal/framework/pipeline/artifact_codec_registry_test.go index f3594b9..55e981a 100644 --- a/internal/framework/pipeline/artifact_codec_registry_test.go +++ b/internal/framework/pipeline/artifact_codec_registry_test.go @@ -14,7 +14,13 @@ import ( ) type codecNotes struct { - Items []string `json:"items"` + Items []string `json:"items"` + Labels map[string]string `json:"labels,omitempty"` + Details *codecNoteDetails `json:"details,omitempty"` +} + +type codecNoteDetails struct { + Name string `json:"name"` } type codecScore struct { @@ -210,7 +216,7 @@ func TestArtifactCodecRegistryStoresValidatedSchemaMetadata(t *testing.T) { spec.Schema.JSONSchema[0] = '[' again, _ := registry.Spec("test/notes") - if string(again.Schema.JSONSchema) != `{"additionalProperties":false,"properties":{"items":{"items":{"type":"string"},"type":"array"}},"required":["items"],"type":"object"}` { + if string(again.Schema.JSONSchema) != `{"additionalProperties":false,"properties":{"details":{"additionalProperties":false,"properties":{"name":{"type":"string"}},"required":["name"],"type":"object"},"items":{"items":{"type":"string"},"type":"array"},"labels":{"additionalProperties":{"type":"string"},"type":"object"}},"required":["items"],"type":"object"}` { t.Fatalf("stored JSON Schema changed through Spec result: %q", again.Schema.JSONSchema) } } @@ -388,7 +394,7 @@ func notesCodec() testArtifactCodec[codecNotes] { ID: "notes.v1", Name: "notes", Version: "v1", - JSONSchema: []byte(`{"additionalProperties":false,"properties":{"items":{"items":{"type":"string"},"type":"array"}},"required":["items"],"type":"object"}`), + JSONSchema: []byte(`{"additionalProperties":false,"properties":{"details":{"additionalProperties":false,"properties":{"name":{"type":"string"}},"required":["name"],"type":"object"},"items":{"items":{"type":"string"},"type":"array"},"labels":{"additionalProperties":{"type":"string"},"type":"object"}},"required":["items"],"type":"object"}`), }, mediaType: "application/json", encodeFunc: func(value codecNotes) ([]byte, error) { diff --git a/internal/framework/pipeline/runner_candidate_encoding_test.go b/internal/framework/pipeline/runner_candidate_encoding_test.go index 803fda6..69ab123 100644 --- a/internal/framework/pipeline/runner_candidate_encoding_test.go +++ b/internal/framework/pipeline/runner_candidate_encoding_test.go @@ -2,6 +2,7 @@ package pipeline import ( "context" + "encoding/base64" "encoding/json" "errors" "fmt" @@ -13,10 +14,12 @@ import ( ) type observedNotesCodec struct { - candidateValues []codecNotes - finalValues []codecNotes - candidateError string - finalError string + candidateValues []codecNotes + candidateDecodedValues []codecNotes + finalValues []codecNotes + candidateError string + candidateDecodeError string + finalError string } func (*observedNotesCodec) Kind() contracts.ArtifactKind { return "test/notes" } @@ -50,7 +53,15 @@ func (*observedNotesCodec) Decode(content []byte) (codecNotes, error) { } func (c *observedNotesCodec) DecodeCandidate(content []byte) (codecNotes, error) { - return c.Decode(content) + value, err := c.Decode(content) + if err != nil { + return codecNotes{}, err + } + if c.candidateDecodeError != "" && firstNote(value) == c.candidateDecodeError { + return codecNotes{}, errors.New("candidate decoding failed") + } + c.candidateDecodedValues = append(c.candidateDecodedValues, value) + return value, nil } func firstNote(value codecNotes) string { @@ -68,6 +79,10 @@ func (c *observedNotesCodec) finalCalls(value string) int { return matchingNotes(c.finalValues, value) } +func (c *observedNotesCodec) candidateDecodeCount() int { + return len(c.candidateDecodedValues) +} + func matchingNotes(values []codecNotes, value string) int { count := 0 for _, candidate := range values { @@ -114,6 +129,16 @@ func installObservedNotesCodec(t *testing.T, prepared *PreparedPipeline, codec * func configureCandidateOperation(prepared *PreparedPipeline, target ModuleStage, value codecNotes) { lane := &prepared.Steps[0].lanes[0] switch target { + case StageExtract: + lane.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + return erasedTypedResult{Value: value}, nil + } + lane.typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { + return erasedTypedResult{Value: codecNotes{Items: []string{"merged-other"}}}, nil + } + lane.typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { + return erasedTypedResult{Value: codecNotes{Items: []string{"normalized-other"}}}, nil + } case StageMerge: lane.typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { return erasedTypedResult{Value: value}, nil @@ -139,6 +164,8 @@ func setCandidateValidator(prepared *PreparedPipeline, target ModuleStage, appro }, } switch target { + case StageExtract: + prepared.Steps[0].lanes[0].extractValidators.validators = []preparedValidator{validator} case StageMerge: prepared.Steps[0].lanes[0].mergeValidators.validators = []preparedValidator{validator} case StageNormalize: @@ -146,6 +173,135 @@ func setCandidateValidator(prepared *PreparedPipeline, target ModuleStage, appro } } +type serializedValidationFunc func(context.Context, contracts.SerializedValidationRequest) (contracts.ValidationResult, error) + +func (serializedValidationFunc) Name() string { return "candidate-serialized" } +func (serializedValidationFunc) ExecutionClass() contracts.ExecutionClass { + return contracts.ExecutionClassDeterministic +} +func (validate serializedValidationFunc) Validate(ctx context.Context, request contracts.SerializedValidationRequest) (contracts.ValidationResult, error) { + return validate(ctx, request) +} + +func TestRunnerIsolatesTypedValidatorCandidates(t *testing.T) { + for _, target := range []ModuleStage{StageExtract, StageMerge, StageNormalize} { + t.Run(string(target), func(t *testing.T) { + prepared := preparedAttemptDebugPipeline(t) + prepared.Steps[0].lanes[0].extractValidators = preparedValidatorChain{} + prepared.Steps[0].lanes[0].mergeValidators = preparedValidatorChain{} + prepared.Steps[0].lanes[0].normalizeValidators = preparedValidatorChain{} + codec := &observedNotesCodec{} + installObservedNotesCodec(t, prepared, codec) + candidate := codecNotes{Items: []string{"candidate-" + string(target)}, Labels: map[string]string{"label": "original"}, Details: &codecNoteDetails{Name: "original"}} + configureCandidateOperation(prepared, target, candidate) + + var laterValue, serializedValue, downstreamValue codecNotes + firstValidator := preparedValidator{ + resolved: ResolvedValidator{Binding: Binding("candidate-mutator"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"}, + typedValidate: func(_ context.Context, _ any, request typedValidationTarget) (contracts.ValidationResult, error) { + value := request.value.(codecNotes) + value.Items[0] = "mutated" + value.Labels["label"] = "mutated" + value.Details.Name = "mutated" + return contracts.ValidationResult{Approved: true}, nil + }, + } + secondValidator := preparedValidator{ + resolved: ResolvedValidator{Binding: Binding("candidate-observer"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"}, + typedValidate: func(_ context.Context, _ any, request typedValidationTarget) (contracts.ValidationResult, error) { + laterValue = request.value.(codecNotes) + return contracts.ValidationResult{Approved: true}, nil + }, + } + serializedValidator := preparedValidator{ + resolved: ResolvedValidator{Binding: Binding("candidate-serialized"), Target: ValidatorTargetSerialized, ArtifactKind: "test/notes"}, + serialized: serializedValidationFunc(func(_ context.Context, request contracts.SerializedValidationRequest) (contracts.ValidationResult, error) { + if err := json.Unmarshal(request.Content, &serializedValue); err != nil { + return contracts.ValidationResult{}, err + } + return contracts.ValidationResult{Approved: true}, nil + }), + } + setCandidateValidators(prepared, target, []preparedValidator{firstValidator, secondValidator, serializedValidator}) + + lane := &prepared.Steps[0].lanes[0] + switch target { + case StageExtract: + lane.typed.merge = func(_ context.Context, _ any, request contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { + downstreamValue = request.ExtractOutputs[0].Value.(codecNotes) + return erasedTypedResult{Value: codecNotes{Items: []string{"merged-other"}}}, nil + } + case StageMerge: + lane.typed.normalize = func(_ context.Context, _ any, request contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { + downstreamValue = request.MergeOutput.Value.(codecNotes) + return erasedTypedResult{Value: codecNotes{Items: []string{"normalized-other"}}}, nil + } + } + + debug := newCapturedDebugRecorder() + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug}) + if err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + if !reflect.DeepEqual(laterValue, candidate) || !reflect.DeepEqual(serializedValue, candidate) { + t.Fatalf("validator values = typed %#v serialized %#v, want %#v", laterValue, serializedValue, candidate) + } + if codec.candidateCalls(candidate.Items[0]) != 1 || codec.candidateDecodeCount() != 2 || codec.finalCalls(candidate.Items[0]) != 1 { + t.Fatalf("candidate calls = encode %d decode %d final %d, want 1, 2, 1", codec.candidateCalls(candidate.Items[0]), codec.candidateDecodeCount(), codec.finalCalls(candidate.Items[0])) + } + switch target { + case StageExtract, StageMerge: + if !reflect.DeepEqual(downstreamValue, candidate) { + t.Fatalf("downstream value = %#v, want %#v", downstreamValue, candidate) + } + case StageNormalize: + var normalized codecNotes + if err := json.Unmarshal(output.NormalizeOutputs[0].Artifact.Content, &normalized); err != nil || !reflect.DeepEqual(normalized, candidate) { + t.Fatalf("normalized output = %#v, %v; want %#v", normalized, err, candidate) + } + } + attemptPath := fmt.Sprintf("%s/notes/attempt-01.json", target) + if target == StageExtract { + attemptPath = "extract/notes/chunk-000001/attempt-01.json" + } + payload := debug.envelope(t, attemptPath).Payload.(map[string]any) + content, err := base64.StdEncoding.DecodeString(payload["output"].(map[string]any)["content"].(map[string]any)["content_base64"].(string)) + var debugValue codecNotes + if err == nil { + err = json.Unmarshal(content, &debugValue) + } + if err != nil || !reflect.DeepEqual(debugValue, candidate) { + t.Fatalf("debug candidate = %#v, %v; want %#v", debugValue, err, candidate) + } + }) + } +} + +func setCandidateValidators(prepared *PreparedPipeline, target ModuleStage, validators []preparedValidator) { + switch target { + case StageExtract: + prepared.Steps[0].lanes[0].extractValidators.validators = validators + case StageMerge: + prepared.Steps[0].lanes[0].mergeValidators.validators = validators + case StageNormalize: + prepared.Steps[0].lanes[0].normalizeValidators.validators = validators + } +} + +func TestRunnerReportsCandidateDecodeFailure(t *testing.T) { + prepared := preparedAttemptDebugPipeline(t) + codec := &observedNotesCodec{candidateDecodeError: "decode-failure"} + installObservedNotesCodec(t, prepared, codec) + configureCandidateOperation(prepared, StageMerge, codecNotes{Items: []string{"decode-failure"}}) + setCandidateValidator(prepared, StageMerge, true) + + _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")}) + var codecErr *ArtifactCodecOperationError + if err == nil || !errors.As(err, &codecErr) || codecErr.Operation != "decode candidate" || codec.finalCalls("decode-failure") != 0 { + t.Fatalf("Run() error = %v, codec error = %#v, final calls = %d; want contextual candidate decode failure", err, codecErr, codec.finalCalls("decode-failure")) + } +} + func TestRunnerRejectsCandidatesBeforeFinalEncoding(t *testing.T) { for _, target := range []ModuleStage{StageMerge, StageNormalize} { t.Run(string(target), func(t *testing.T) { diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 1258bb0..7eed857 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "path" + "reflect" "time" "gitea.maximumdirect.net/eric/notarius/internal/core/fileio" @@ -489,6 +490,12 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE } 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 + } + requestTarget.value = candidateValue requestTarget.llmProfile = binding.LLMProfile result, err = item.typedValidate(validatorCtx, item.typed, requestTarget) case ValidatorTargetSerialized: @@ -548,3 +555,15 @@ func validationCandidateArtifact(codec artifactCodecEntry, target typedValidatio } return serializeCandidateArtifact(codec, target.laneID, target.moduleKey, target.sourceID, target.value) } + +func decodeTypedValidationCandidate(codec artifactCodecEntry, candidate CheckpointArtifact) (any, error) { + value, err := codec.decodeCandidate(candidate.Artifact.Content) + if err != nil { + return nil, err + } + actualType := reflect.TypeOf(value) + if actualType != codec.valueType { + return nil, newArtifactCodecTypeError("decode candidate", codec.spec.Kind, codec.valueType, actualType) + } + return value, nil +}