Isolate typed validator candidates

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
+their target: chunks, codec-decoded typed candidates, or serialized codec
-approve, approve with warnings, reject, or fail. A rejection is an ordinary
+bytes. Each typed validator receives a newly decoded value from the one
-pipeline result; a validator error is a framework error.
+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

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) {

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
+	candidateValues        []codecNotes
-	finalValues     []codecNotes
+	candidateDecodedValues []codecNotes
-	candidateError  string
+	finalValues            []codecNotes
-	finalError      string
+	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) {

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
+}