Validate artifact candidates before final encoding

internal/framework/pipeline/runner_candidate_encoding_test.go

@@ -0,0 +1,268 @@
+package pipeline
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"reflect"
+	"strings"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+)
+
+type observedNotesCodec struct {
+	candidateValues []codecNotes
+	finalValues     []codecNotes
+	candidateError  string
+	finalError      string
+}
+
+func (*observedNotesCodec) Kind() contracts.ArtifactKind { return "test/notes" }
+
+func (*observedNotesCodec) Schema() contracts.ArtifactSchema { return notesCodec().schema }
+
+func (*observedNotesCodec) MediaType() string { return "application/json" }
+
+func (c *observedNotesCodec) EncodeCandidate(value codecNotes) ([]byte, error) {
+	c.candidateValues = append(c.candidateValues, value)
+	if c.candidateError != "" && firstNote(value) == c.candidateError {
+		return nil, errors.New("candidate encoding failed")
+	}
+	return json.Marshal(value)
+}
+
+func (c *observedNotesCodec) Encode(value codecNotes) ([]byte, error) {
+	c.finalValues = append(c.finalValues, value)
+	if c.finalError != "" && firstNote(value) == c.finalError {
+		return nil, errors.New("final encoding failed")
+	}
+	if firstNote(value) == "invalid" {
+		return nil, errors.New("invalid note is not a final artifact")
+	}
+	return json.Marshal(value)
+}
+
+func (*observedNotesCodec) Decode(content []byte) (codecNotes, error) {
+	var value codecNotes
+	return value, json.Unmarshal(content, &value)
+}
+
+func firstNote(value codecNotes) string {
+	if len(value.Items) == 0 {
+		return ""
+	}
+	return value.Items[0]
+}
+
+func (c *observedNotesCodec) candidateCalls(value string) int {
+	return matchingNotes(c.candidateValues, value)
+}
+
+func (c *observedNotesCodec) finalCalls(value string) int {
+	return matchingNotes(c.finalValues, value)
+}
+
+func matchingNotes(values []codecNotes, value string) int {
+	count := 0
+	for _, candidate := range values {
+		if firstNote(candidate) == value {
+			count++
+		}
+	}
+	return count
+}
+
+type candidateCheckpointRecorder struct {
+	CheckpointRecorder
+	mergeSucceeded     int
+	normalizeSucceeded int
+	mergeOutput        CheckpointArtifact
+	normalizeOutput    CheckpointArtifact
+}
+
+func (r *candidateCheckpointRecorder) MergeSucceeded(_ string, _ string, _ []CheckpointFingerprint, output CheckpointArtifact, _ []contracts.Warning) error {
+	r.mergeSucceeded++
+	r.mergeOutput = cloneCheckpointArtifact(output)
+	return nil
+}
+
+func (r *candidateCheckpointRecorder) NormalizeSucceeded(_ string, _ string, _ []CheckpointFingerprint, output CheckpointArtifact, _ []contracts.Warning) error {
+	r.normalizeSucceeded++
+	r.normalizeOutput = cloneCheckpointArtifact(output)
+	return nil
+}
+
+func installObservedNotesCodec(t *testing.T, prepared *PreparedPipeline, codec *observedNotesCodec) {
+	t.Helper()
+	registry := NewArtifactCodecRegistry()
+	if err := RegisterArtifactCodec(registry, codec); err != nil {
+		t.Fatalf("RegisterArtifactCodec() error = %v", err)
+	}
+	entry, _, err := registry.entry("test/notes")
+	if err != nil {
+		t.Fatalf("codec entry error = %v", err)
+	}
+	prepared.lanes[0].typed.codec = entry
+}
+
+func configureCandidateOperation(prepared *PreparedPipeline, target ModuleStage, value codecNotes) {
+	lane := &prepared.lanes[0]
+	switch target {
+	case StageMerge:
+		lane.typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
+			return erasedTypedResult{Value: value}, nil
+		}
+		lane.typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
+			return erasedTypedResult{Value: codecNotes{Items: []string{"normalized-other"}}}, nil
+		}
+	case StageNormalize:
+		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: value}, nil
+		}
+	}
+}
+
+func setCandidateValidator(prepared *PreparedPipeline, target ModuleStage, approved bool) {
+	validator := preparedValidator{
+		resolved: ResolvedValidator{Binding: Binding("candidate-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"},
+		typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {
+			return contracts.ValidationResult{Approved: approved, ReasonCode: "candidate_rejected", Message: "candidate rejected by validator"}, nil
+		},
+	}
+	switch target {
+	case StageMerge:
+		prepared.lanes[0].mergeValidators.validators = []preparedValidator{validator}
+	case StageNormalize:
+		prepared.lanes[0].normalizeValidators.validators = []preparedValidator{validator}
+	}
+}
+
+func TestRunnerRejectsCandidatesBeforeFinalEncoding(t *testing.T) {
+	for _, target := range []ModuleStage{StageMerge, StageNormalize} {
+		t.Run(string(target), func(t *testing.T) {
+			prepared := preparedAttemptDebugPipeline(t)
+			codec := &observedNotesCodec{}
+			installObservedNotesCodec(t, prepared, codec)
+			configureCandidateOperation(prepared, target, codecNotes{Items: []string{"invalid"}})
+			setCandidateValidator(prepared, target, false)
+			recorder := &candidateCheckpointRecorder{CheckpointRecorder: NoopCheckpointRecorder()}
+
+			output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoints: recorder})
+			if err != nil {
+				t.Fatalf("Run() error = %v, want validator rejection", err)
+			}
+			if len(output.Rejected) != 1 || output.Rejected[0].Stage != string(target) || output.Rejected[0].ReasonCode != "candidate_rejected" {
+				t.Fatalf("rejected outputs = %#v, want %s validator rejection", output.Rejected, target)
+			}
+			if codec.candidateCalls("invalid") != 1 || codec.finalCalls("invalid") != 0 {
+				t.Fatalf("invalid candidate calls = candidate %d, final %d; want 1, 0", codec.candidateCalls("invalid"), codec.finalCalls("invalid"))
+			}
+			if target == StageMerge && recorder.mergeSucceeded != 0 {
+				t.Fatalf("merge checkpoints = %d, want none", recorder.mergeSucceeded)
+			}
+			if target == StageNormalize && recorder.normalizeSucceeded != 0 {
+				t.Fatalf("normalize checkpoints = %d, want none", recorder.normalizeSucceeded)
+			}
+		})
+	}
+}
+
+func TestRunnerFinalEncodesAcceptedCandidatesOnce(t *testing.T) {
+	for _, target := range []ModuleStage{StageMerge, StageNormalize} {
+		t.Run(string(target), func(t *testing.T) {
+			prepared := preparedAttemptDebugPipeline(t)
+			codec := &observedNotesCodec{}
+			installObservedNotesCodec(t, prepared, codec)
+			value := "accepted-" + string(target)
+			configureCandidateOperation(prepared, target, codecNotes{Items: []string{value}})
+			setCandidateValidator(prepared, target, true)
+			recorder := &candidateCheckpointRecorder{CheckpointRecorder: NoopCheckpointRecorder()}
+			loader := &extractResultLoader{CheckpointLoader: NoopCheckpointLoader(), decision: CheckpointDecision{Reason: "not found"}}
+
+			output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoints: recorder, Checkpoint: loader})
+			if err != nil {
+				t.Fatalf("Run() error = %v, want nil", err)
+			}
+			if codec.candidateCalls(value) != 1 || codec.finalCalls(value) != 1 {
+				t.Fatalf("accepted candidate calls = candidate %d, final %d; want 1, 1", codec.candidateCalls(value), codec.finalCalls(value))
+			}
+			if target == StageMerge && recorder.mergeSucceeded != 1 {
+				t.Fatalf("merge checkpoints = %d, want one", recorder.mergeSucceeded)
+			}
+			if target == StageNormalize && recorder.normalizeSucceeded != 1 {
+				t.Fatalf("normalize checkpoints = %d, want one", recorder.normalizeSucceeded)
+			}
+			checkpoint := recorder.mergeOutput
+			if target == StageNormalize {
+				checkpoint = recorder.normalizeOutput
+			}
+			var stored codecNotes
+			if err := json.Unmarshal(checkpoint.Artifact.Content, &stored); err != nil || !reflect.DeepEqual(stored, codecNotes{Items: []string{value}}) {
+				t.Fatalf("checkpoint content = %s, %v; want accepted value %q", checkpoint.Artifact.Content, err, value)
+			}
+			gotStages := make([]string, len(output.CheckpointEvents))
+			for i, event := range output.CheckpointEvents {
+				gotStages[i] = event.Stage
+			}
+			wantStages := []string{"source", string(StageChunk), string(StageExtract), string(StageMerge), string(StageNormalize)}
+			if !reflect.DeepEqual(gotStages, wantStages) {
+				t.Fatalf("checkpoint event stages = %#v, want %#v", gotStages, wantStages)
+			}
+		})
+	}
+}
+
+func TestRunnerRecordsCandidateAndFinalEncodingFailures(t *testing.T) {
+	tests := []struct {
+		name      string
+		target    ModuleStage
+		value     string
+		candidate bool
+		wantError string
+	}{
+		{name: "merge candidate", target: StageMerge, value: "merge-candidate-failure", candidate: true, wantError: "serialize merge candidate"},
+		{name: "normalize candidate", target: StageNormalize, value: "normalize-candidate-failure", candidate: true, wantError: "serialize normalize candidate"},
+		{name: "merge final", target: StageMerge, value: "merge-final-failure", wantError: "serialize accepted merge output"},
+		{name: "normalize final", target: StageNormalize, value: "normalize-final-failure", wantError: "serialize accepted normalize output"},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			prepared := preparedAttemptDebugPipeline(t)
+			codec := &observedNotesCodec{}
+			if tc.candidate {
+				codec.candidateError = tc.value
+			} else {
+				codec.finalError = tc.value
+			}
+			installObservedNotesCodec(t, prepared, codec)
+			configureCandidateOperation(prepared, tc.target, codecNotes{Items: []string{tc.value}})
+			setCandidateValidator(prepared, tc.target, true)
+			recorder := &candidateCheckpointRecorder{CheckpointRecorder: NoopCheckpointRecorder()}
+			debug := newCapturedDebugRecorder()
+
+			_, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoints: recorder, Debug: debug})
+			if err == nil || !strings.Contains(err.Error(), tc.wantError) {
+				t.Fatalf("Run() error = %v, want %q", err, tc.wantError)
+			}
+			attemptPath := fmt.Sprintf("%s/notes/attempt-01.json", tc.target)
+			envelope := debug.envelope(t, attemptPath)
+			if !strings.Contains(envelope.Error, tc.wantError) {
+				t.Fatalf("attempt error = %q, want %q", envelope.Error, tc.wantError)
+			}
+			if tc.candidate && codec.finalCalls(tc.value) != 0 {
+				t.Fatalf("final encode calls = %d, want none after candidate failure", codec.finalCalls(tc.value))
+			}
+			if tc.target == StageMerge && recorder.mergeSucceeded != 0 {
+				t.Fatalf("merge checkpoints = %d, want none", recorder.mergeSucceeded)
+			}
+			if tc.target == StageNormalize && recorder.normalizeSucceeded != 0 {
+				t.Fatalf("normalize checkpoints = %d, want none", recorder.normalizeSucceeded)
+			}
+		})
+	}
+}

internal/framework/pipeline/runner_typed.go

@@ -111,6 +111,14 @@ func checkpointArtifact(codec artifactCodecEntry, laneID, moduleKey, sourceID st
 	return CheckpointArtifact{LaneID: laneID, ModuleKey: moduleKey, SourceID: sourceID, Artifact: serialized, SchemaDigest: contracts.DigestArtifactSchema(serialized.Schema)}, nil
 }
 
+func serializeCandidateArtifact(codec artifactCodecEntry, laneID, moduleKey, sourceID string, value any) (CheckpointArtifact, error) {
+	serialized, err := serializeArtifact(codec, value, true)
+	if err != nil {
+		return CheckpointArtifact{}, err
+	}
+	return CheckpointArtifact{LaneID: laneID, ModuleKey: moduleKey, SourceID: sourceID, Artifact: serialized, SchemaDigest: contracts.DigestArtifactSchema(serialized.Schema)}, nil
+}
+
 type laneRunError struct {
 	stage ModuleStage
 	err   error
@@ -183,8 +191,8 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
 				return false, nil, attemptErr
 			}
 			candidate := erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: result.Value}
-			stored, encodeErr := checkpointArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value)
 			attemptWarnings := cloneWarnings(result.Warnings)
+			serializedCandidate, encodeErr := serializeCandidateArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value)
 			if encodeErr != nil {
 				attemptErr := fmt.Errorf("serialize merge candidate for lane %q: %w", lane.ID, encodeErr)
 				if debugErr := attemptEnvelope(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr); debugErr != nil {
@@ -192,15 +200,23 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
 				}
 				return false, nil, attemptErr
 			}
-			warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageMerge, laneID: lane.ID, moduleKey: lane.Merge.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.MergeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value}, prepared.mergeValidators, attempt, input.Debug)
+			warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageMerge, laneID: lane.ID, moduleKey: lane.Merge.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.MergeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value, candidate: &serializedCandidate}, prepared.mergeValidators, attempt, input.Debug)
 			attemptWarnings = append(attemptWarnings, warnings...)
-			payload := map[string]any{"output": debugCheckpointArtifact(stored), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)}
+			payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)}
 			if validateErr != nil || rejected != nil {
 				if debugErr := attemptEnvelope(payload, validateErr); debugErr != nil {
 					return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr)
 				}
 				return false, rejected, validateErr
 			}
+			stored, encodeErr := checkpointArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value)
+			if encodeErr != nil {
+				attemptErr := fmt.Errorf("serialize accepted merge output for lane %q: %w", lane.ID, encodeErr)
+				if debugErr := attemptEnvelope(payload, attemptErr); debugErr != nil {
+					return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr)
+				}
+				return false, nil, attemptErr
+			}
 			if debugErr := attemptEnvelope(payload, nil); debugErr != nil {
 				return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr)
 			}
@@ -272,8 +288,8 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
 				}
 				return false, nil, attemptErr
 			}
-			stored, encodeErr := checkpointArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value)
 			attemptWarnings := cloneWarnings(result.Warnings)
+			serializedCandidate, encodeErr := serializeCandidateArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value)
 			if encodeErr != nil {
 				attemptErr := fmt.Errorf("serialize normalize candidate for lane %q: %w", lane.ID, encodeErr)
 				if debugErr := attemptEnvelope(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr); debugErr != nil {
@@ -281,15 +297,23 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
 				}
 				return false, nil, attemptErr
 			}
-			warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageNormalize, laneID: lane.ID, moduleKey: lane.Normalize.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.NormalizeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value}, prepared.normalizeValidators, attempt, input.Debug)
+			warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageNormalize, laneID: lane.ID, moduleKey: lane.Normalize.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.NormalizeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value, candidate: &serializedCandidate}, prepared.normalizeValidators, attempt, input.Debug)
 			attemptWarnings = append(attemptWarnings, warnings...)
-			payload := map[string]any{"output": debugCheckpointArtifact(stored), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)}
+			payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)}
 			if validateErr != nil || rejected != nil {
 				if debugErr := attemptEnvelope(payload, validateErr); debugErr != nil {
 					return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr)
 				}
 				return false, rejected, validateErr
 			}
+			stored, encodeErr := checkpointArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value)
+			if encodeErr != nil {
+				attemptErr := fmt.Errorf("serialize accepted normalize output for lane %q: %w", lane.ID, encodeErr)
+				if debugErr := attemptEnvelope(payload, attemptErr); debugErr != nil {
+					return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr)
+				}
+				return false, nil, attemptErr
+			}
 			if debugErr := attemptEnvelope(payload, nil); debugErr != nil {
 				return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr)
 			}
@@ -358,17 +382,17 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE
 			target.llmProfile = binding.LLMProfile
 			result, err = item.typedValidate(validatorCtx, item.typed, target)
 		case ValidatorTargetSerialized:
-			artifact, encodeErr := serializeArtifact(codec, target.value, true)
+			artifact, encodeErr := validationCandidateArtifact(codec, target)
 			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: target.sourceInput.Clone(), SessionID: target.sessionID, References: CloneReferenceSet(target.references), LLMProfile: binding.LLMProfile, Metadata: cloneMetadata(target.metadata), Chunk: cloneSourceChunkPtr(target.chunk), Chunks: cloneSourceChunks(target.chunks), Schema: contracts.CloneArtifactSchema(artifact.Schema), MediaType: artifact.MediaType, Content: append([]byte(nil), 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: target.sourceInput.Clone(), SessionID: target.sessionID, References: CloneReferenceSet(target.references), LLMProfile: binding.LLMProfile, Metadata: cloneMetadata(target.metadata), Chunk: cloneSourceChunkPtr(target.chunk), Chunks: cloneSourceChunks(target.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)
 		}
-		artifact, _ := serializeArtifact(codec, target.value, true)
-		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(CheckpointArtifact{Artifact: artifact, SchemaDigest: contracts.DigestArtifactSchema(artifact.Schema)}), "metadata": redactSensitiveMap(target.metadata)}, Result: debugValidationResultEnvelope(result)}
+		artifact, _ := validationCandidateArtifact(codec, target)
+		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()
 		}
@@ -403,3 +427,10 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE
 	}
 	return warnings, nil, nil
 }
+
+func validationCandidateArtifact(codec artifactCodecEntry, target typedValidationTarget) (CheckpointArtifact, error) {
+	if target.candidate != nil {
+		return cloneCheckpointArtifact(*target.candidate), nil
+	}
+	return serializeCandidateArtifact(codec, target.laneID, target.moduleKey, target.sourceID, target.value)
+}

internal/framework/pipeline/typed_execution.go

@@ -41,6 +41,7 @@ type typedValidationTarget struct {
 	chunks      []source.Chunk
 	ref         source.SourceRef
 	value       any
+	candidate   *CheckpointArtifact
 }
 
 func exactTypedValue[T any](operation string, value any) (T, error) {