notarius/internal/framework/pipeline/runner_reuse_lineage_test.go

package pipeline

import (
	"context"
	"fmt"
	"strings"
	"testing"

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

type reuseLineageCheckpointSpy struct {
	CheckpointLoader
	CheckpointRecorder
	loads  map[string]int
	writes map[string]int
	forbid map[string]struct{}
}

func newReuseLineageCheckpointSpy() *reuseLineageCheckpointSpy {
	return &reuseLineageCheckpointSpy{
		CheckpointLoader:   NoopCheckpointLoader(),
		CheckpointRecorder: NoopCheckpointRecorder(),
		loads:              make(map[string]int),
		writes:             make(map[string]int),
		forbid:             make(map[string]struct{}),
	}
}

func (s *reuseLineageCheckpointSpy) Enabled() bool { return true }

func (s *reuseLineageCheckpointSpy) load(stage, laneID string) {
	key := stage + "/" + laneID
	if _, forbidden := s.forbid[key]; forbidden {
		panic("checkpoint lookup crossed validation-incomplete lineage: " + key)
	}
	s.loads[key]++
}

func (s *reuseLineageCheckpointSpy) write(stage, action, laneID string) {
	s.writes[stage+"/"+action+"/"+laneID]++
}

func (s *reuseLineageCheckpointSpy) Extract(laneID, _ string, _ []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) {
	s.load("extract", laneID)
	return ExtractCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing)
}

func (s *reuseLineageCheckpointSpy) Merge(laneID, _ string, _ []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) {
	s.load("merge", laneID)
	return MergeCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing)
}

func (s *reuseLineageCheckpointSpy) Normalize(laneID, _ string, _ []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) {
	s.load("normalize", laneID)
	return NormalizeCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing)
}

func (s *reuseLineageCheckpointSpy) ExtractRunning(laneID, _ string, _ []CheckpointFingerprint) error {
	s.write("extract", "running", laneID)
	return nil
}

func (s *reuseLineageCheckpointSpy) ExtractSucceeded(laneID, _ string, _ []CheckpointFingerprint, _ []CheckpointArtifact, _ []contracts.RejectedOutput, _ []contracts.Warning) error {
	s.write("extract", "succeeded", laneID)
	return nil
}

func (s *reuseLineageCheckpointSpy) ExtractFailed(laneID, _ string, _ []CheckpointFingerprint, _ error) error {
	s.write("extract", "failed", laneID)
	return nil
}

func (s *reuseLineageCheckpointSpy) MergeRunning(laneID, _ string, _ []CheckpointFingerprint) error {
	s.write("merge", "running", laneID)
	return nil
}

func (s *reuseLineageCheckpointSpy) MergeSucceeded(laneID, _ string, _ []CheckpointFingerprint, _ CheckpointArtifact, _ []contracts.Warning) error {
	s.write("merge", "succeeded", laneID)
	return nil
}

func (s *reuseLineageCheckpointSpy) MergeRejected(laneID, _ string, _ []CheckpointFingerprint, _ contracts.RejectedOutput) error {
	s.write("merge", "rejected", laneID)
	return nil
}

func (s *reuseLineageCheckpointSpy) MergeFailed(laneID, _ string, _ []CheckpointFingerprint, _ error) error {
	s.write("merge", "failed", laneID)
	return nil
}

func (s *reuseLineageCheckpointSpy) NormalizeRunning(laneID, _ string, _ []CheckpointFingerprint) error {
	s.write("normalize", "running", laneID)
	return nil
}

func (s *reuseLineageCheckpointSpy) NormalizeSucceeded(laneID, _ string, _ []CheckpointFingerprint, _ CheckpointArtifact, _ []contracts.Warning) error {
	s.write("normalize", "succeeded", laneID)
	return nil
}

func (s *reuseLineageCheckpointSpy) NormalizeRejected(laneID, _ string, _ []CheckpointFingerprint, _ contracts.RejectedOutput) error {
	s.write("normalize", "rejected", laneID)
	return nil
}

func (s *reuseLineageCheckpointSpy) NormalizeFailed(laneID, _ string, _ []CheckpointFingerprint, _ error) error {
	s.write("normalize", "failed", laneID)
	return nil
}

func (s *reuseLineageCheckpointSpy) writesFor(stage, laneID string) int {
	total := 0
	needle := stage + "/"
	suffix := "/" + laneID
	for key, count := range s.writes {
		if strings.HasPrefix(key, needle) && strings.HasSuffix(key, suffix) {
			total += count
		}
	}
	return total
}

func unavailableTypedValidator(kind contracts.ArtifactKind) preparedValidator {
	return preparedValidator{
		resolved: ResolvedValidator{Binding: Binding("typed/check"), Target: ValidatorTargetTyped, ArtifactKind: kind},
		typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {
			return contracts.ValidationResult{}, fmt.Errorf("validator unavailable")
		},
	}
}

func TestValidationIncompleteExtractDisablesDownstreamCheckpointIO(t *testing.T) {
	prepared := preparedAttemptDebugPipeline(t)
	lane := &prepared.Steps[0].lanes[0]
	lane.resolved.ExtractValidationPolicy.ValidatorFailure = ValidatorFailureWarnContinue
	installExtractOperation(prepared, 0, func(_ context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
		return erasedTypedResult{Value: typedValueForLane(0, request.Chunk.Index)}, nil
	})
	lane.extractValidators.validators[0].typedValidate = func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {
		return contracts.ValidationResult{}, fmt.Errorf("validator unavailable")
	}
	spy := newReuseLineageCheckpointSpy()

	output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoint: spy, Checkpoints: spy})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	if len(output.NormalizeOutputs) != 1 {
		t.Fatalf("normalize outputs = %d, want current-run handoff", len(output.NormalizeOutputs))
	}
	if spy.loads["merge/notes"] != 0 || spy.loads["normalize/notes"] != 0 {
		t.Fatalf("downstream loads = %#v, want none", spy.loads)
	}
	if spy.writesFor("merge", "notes") != 0 || spy.writesFor("normalize", "notes") != 0 {
		t.Fatalf("downstream writes = %#v, want none", spy.writes)
	}
	if spy.writes["extract/succeeded/notes"] != 0 {
		t.Fatalf("extract writes = %#v, want no reusable success", spy.writes)
	}
}

func TestValidationIncompleteMergeDisablesNormalizeCheckpointIO(t *testing.T) {
	prepared := preparedAttemptDebugPipeline(t)
	lane := &prepared.Steps[0].lanes[0]
	lane.resolved.MergeValidationPolicy.ValidatorFailure = ValidatorFailureWarnContinue
	installExtractOperation(prepared, 0, func(_ context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
		return erasedTypedResult{Value: typedValueForLane(0, request.Chunk.Index)}, nil
	})
	lane.mergeValidators.validators = []preparedValidator{unavailableTypedValidator(lane.resolved.ArtifactKind)}
	spy := newReuseLineageCheckpointSpy()

	output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoint: spy, Checkpoints: spy})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	if len(output.NormalizeOutputs) != 1 {
		t.Fatalf("normalize outputs = %d, want current-run handoff", len(output.NormalizeOutputs))
	}
	if spy.loads["merge/notes"] != 1 || spy.loads["normalize/notes"] != 0 {
		t.Fatalf("loads = %#v, want merge lookup only", spy.loads)
	}
	if spy.writes["merge/succeeded/notes"] != 0 || spy.writesFor("normalize", "notes") != 0 {
		t.Fatalf("writes = %#v, want no reusable merge or normalize state", spy.writes)
	}
}

func TestValidationIncompleteNormalizeIsNotPublished(t *testing.T) {
	prepared := preparedAttemptDebugPipeline(t)
	lane := &prepared.Steps[0].lanes[0]
	lane.resolved.NormalizeValidationPolicy.ValidatorFailure = ValidatorFailureWarnContinue
	installExtractOperation(prepared, 0, func(_ context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
		return erasedTypedResult{Value: typedValueForLane(0, request.Chunk.Index)}, nil
	})
	lane.normalizeValidators.validators = []preparedValidator{unavailableTypedValidator(lane.resolved.ArtifactKind)}
	spy := newReuseLineageCheckpointSpy()

	output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoint: spy, Checkpoints: spy})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	if len(output.NormalizeOutputs) != 1 || spy.writes["normalize/succeeded/notes"] != 0 {
		t.Fatalf("output/writes = %d / %#v, want in-memory output without normalize publication", len(output.NormalizeOutputs), spy.writes)
	}
}

func TestGeneratedReferenceFromIncompleteValidationDisablesDependentCheckpointIO(t *testing.T) {
	input, _, _ := handoffFixture(t, codecNotes{Items: []string{"producer"}})
	prepared := input.Prepared
	producer := &prepared.Steps[0].lanes[0]
	consumer := &prepared.Steps[1].lanes[0]
	producer.resolved.NormalizeValidationPolicy.ValidatorFailure = ValidatorFailureWarnContinue
	producer.normalizeValidators.validators = []preparedValidator{unavailableTypedValidator(producer.resolved.ArtifactKind)}
	referenceItems := 0
	consumer.typed.extract = func(_ context.Context, _ any, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
		referenceItems = len(request.References.Slots["producer-output"].Items)
		return erasedTypedResult{Value: codecScore{Value: 1}}, nil
	}
	spy := newReuseLineageCheckpointSpy()
	for _, stage := range []string{"extract", "merge", "normalize"} {
		spy.forbid[stage+"/score"] = struct{}{}
	}

	output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoint: spy, Checkpoints: spy})
	if err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	if referenceItems != 1 || len(output.NormalizeOutputs) != 2 {
		t.Fatalf("current-run handoff = items %d outputs %d, want one generated item and two outputs", referenceItems, len(output.NormalizeOutputs))
	}
	for _, stage := range []string{"extract", "merge", "normalize"} {
		if spy.writesFor(stage, "score") != 0 {
			t.Fatalf("dependent writes = %#v, want none for %s", spy.writes, stage)
		}
	}
}