notarius/internal/framework/pipeline/runner_candidate_encoding_test.go

package pipeline

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"strings"
	"testing"

	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)

type observedNotesCodec struct {
	candidateValues        []codecNotes
	candidateDecodedValues []codecNotes
	finalValues            []codecNotes
	candidateError         string
	candidateDecodeError   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 (c *observedNotesCodec) DecodeCandidate(content []byte) (codecNotes, error) {
	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 {
	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 (c *observedNotesCodec) candidateDecodeCount() int {
	return len(c.candidateDecodedValues)
}

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) error {
	r.mergeSucceeded++
	r.mergeOutput = cloneCheckpointArtifact(output)
	return nil
}

func (r *candidateCheckpointRecorder) NormalizeSucceeded(_ string, _ string, _ []CheckpointFingerprint, output CheckpointArtifact) 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.Steps[0].lanes[0].typed.codec = entry
}

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
		}
		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", CorrectionGuidance: "return an acceptable candidate"}, nil
		},
	}
	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:
		prepared.Steps[0].lanes[0].normalizeValidators.validators = []preparedValidator{validator}
	}
}

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)
			if target == StageMerge {
				prepared.Steps[0].lanes[0].resolved.MergeValidationPolicy.SemanticRejection = SemanticRejectionRejectOutput
			} else {
				prepared.Steps[0].lanes[0].resolved.NormalizeValidationPolicy.SemanticRejection = SemanticRejectionRejectOutput
			}
			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) {
			prepared := preparedAttemptDebugPipeline(t)
			if target == StageMerge {
				prepared.Steps[0].lanes[0].resolved.MergeValidationPolicy.SemanticRejection = SemanticRejectionRejectOutput
			} else {
				prepared.Steps[0].lanes[0].resolved.NormalizeValidationPolicy.SemanticRejection = SemanticRejectionRejectOutput
			}
			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(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)
			}
		})
	}
}