package pipeline

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"path"
	"reflect"
	"time"

	"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
	"gitea.maximumdirect.net/eric/notarius/internal/core/fileio"
	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)

func loadMerge(loader CheckpointLoader, stepID, laneID, moduleKey string, deps []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) {
	if stepAware, ok := loader.(StepCheckpointLoader); ok {
		return stepAware.MergeForStep(stepID, laneID, moduleKey, deps)
	}
	return loader.Merge(laneID, moduleKey, deps)
}
func loadNormalize(loader CheckpointLoader, stepID, laneID, moduleKey string, deps []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) {
	if stepAware, ok := loader.(StepCheckpointLoader); ok {
		return stepAware.NormalizeForStep(stepID, laneID, moduleKey, deps)
	}
	return loader.Normalize(laneID, moduleKey, deps)
}
func recordMerge(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint, output CheckpointArtifact) error {
	return checkpointMergeSucceeded(recorder, stepID, laneID, moduleKey, deps, output)
}
func recordNormalize(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint, output CheckpointArtifact) error {
	return checkpointNormalizeSucceeded(recorder, stepID, laneID, moduleKey, deps, output)
}
func cloneCheckpointArtifact(output CheckpointArtifact) CheckpointArtifact {
	output.Artifact = contracts.CloneSerializedArtifact(output.Artifact)
	output.Diagnostics = cloneCheckpointDiagnostics(output.Diagnostics)
	return output
}

func hydrateCheckpointArtifact(codec artifactCodecEntry, output CheckpointArtifact, value any) (CheckpointArtifact, error) {
	output.Artifact.Schema = contracts.CloneArtifactSchema(codec.spec.Schema)
	if codec.metadata != nil {
		metadata, err := codec.metadata(value)
		if err != nil {
			return CheckpointArtifact{}, fmt.Errorf("clone artifact metadata: %w", err)
		}
		output.Artifact.Metadata = metadata
	} else {
		output.Artifact.Metadata = nil
	}
	return output, nil
}
func artifactCheckpointDigests(outputs []CheckpointArtifact) []CheckpointFingerprint {
	values := make([]CheckpointFingerprint, 0, len(outputs))
	for i, output := range outputs {
		sum := sha256.Sum256(output.Artifact.Content)
		values = append(values, CheckpointFingerprint{Name: fmt.Sprintf("artifact[%d]", i), Value: "sha256:" + hex.EncodeToString(sum[:])})
	}
	return normalizeCheckpointFingerprints(values)
}
func debugCheckpointArtifact(output CheckpointArtifact) map[string]any {
	artifact := output.Artifact
	schema := contracts.CloneArtifactSchema(artifact.Schema)
	digest := output.SchemaDigest
	if digest == "" {
		digest = contracts.DigestArtifactSchema(schema)
	}
	schema.JSONSchema = nil
	content := debugContentEnvelope(artifact.Content, artifact.MediaType, artifact.Metadata, nil)
	content.ContentDigest = debugContentDigest(artifact.Content)
	return map[string]any{"lane_id": output.LaneID, "module_key": output.ModuleKey, "source_id": output.SourceID, "chunk_id": output.ChunkID, "chunk_index": output.ChunkIndex, "chunk_ref": output.ChunkRef, "artifact_kind": artifact.Kind, "schema": schema, "schema_digest": digest, "content": content}
}
func debugCheckpointArtifacts(outputs []CheckpointArtifact) []map[string]any {
	if len(outputs) == 0 {
		return nil
	}
	out := make([]map[string]any, 0, len(outputs))
	for _, output := range outputs {
		out = append(out, debugCheckpointArtifact(output))
	}
	return out
}

func serializeArtifact(codec artifactCodecEntry, value any, candidate bool) (contracts.SerializedArtifact, error) {
	encode := codec.encode
	if candidate {
		encode = codec.encodeCandidate
	}
	content, err := encode(value)
	if err != nil {
		return contracts.SerializedArtifact{}, err
	}
	schema := codec.spec.Schema
	metadata := map[string]any(nil)
	if codec.metadata != nil {
		metadata, err = codec.metadata(value)
		if err != nil {
			return contracts.SerializedArtifact{}, fmt.Errorf("clone artifact metadata: %w", err)
		}
	}
	return contracts.SerializedArtifact{Kind: codec.spec.Kind, Schema: contracts.CloneArtifactSchema(schema), MediaType: codec.spec.MediaType, Content: append([]byte(nil), content...), Metadata: metadata}, nil
}

type checkpointArtifactValidationError struct {
	code CheckpointReasonCode
	err  error
}

func (e *checkpointArtifactValidationError) Error() string { return e.err.Error() }
func (e *checkpointArtifactValidationError) Unwrap() error { return e.err }

func checkpointArtifactValidationFailure(code CheckpointReasonCode, format string, args ...any) error {
	return &checkpointArtifactValidationError{code: code, err: fmt.Errorf(format, args...)}
}

func checkpointArtifactReasonCode(err error) CheckpointReasonCode {
	var validationErr *checkpointArtifactValidationError
	if errors.As(err, &validationErr) {
		return validationErr.code
	}
	return CheckpointReasonArtifactPayloadInvalid
}

func decodeCheckpointArtifact(codec artifactCodecEntry, artifact CheckpointArtifact) (any, error) {
	expectedDigest := contracts.DigestArtifactSchema(codec.spec.Schema)
	if artifact.Artifact.Kind != codec.spec.Kind {
		return nil, checkpointArtifactValidationFailure(CheckpointReasonArtifactCodecIncompatible, "artifact kind %q does not match codec %q", artifact.Artifact.Kind, codec.spec.Kind)
	}
	if artifact.Artifact.Schema.ID != codec.spec.Schema.ID || artifact.Artifact.Schema.Name != codec.spec.Schema.Name || artifact.Artifact.Schema.Version != codec.spec.Schema.Version {
		return nil, checkpointArtifactValidationFailure(CheckpointReasonArtifactCodecIncompatible, "artifact schema %q version %q does not match codec schema %q version %q", artifact.Artifact.Schema.ID, artifact.Artifact.Schema.Version, codec.spec.Schema.ID, codec.spec.Schema.Version)
	}
	if artifact.SchemaDigest != expectedDigest {
		return nil, checkpointArtifactValidationFailure(CheckpointReasonArtifactCodecIncompatible, "artifact schema digest %q does not match codec schema digest %q", artifact.SchemaDigest, expectedDigest)
	}
	if artifact.Artifact.MediaType != codec.spec.MediaType {
		return nil, checkpointArtifactValidationFailure(CheckpointReasonArtifactCodecIncompatible, "artifact media type %q does not match codec media type %q", artifact.Artifact.MediaType, codec.spec.MediaType)
	}
	value, err := codec.decode(append([]byte(nil), artifact.Artifact.Content...))
	if err != nil {
		return nil, checkpointArtifactValidationFailure(CheckpointReasonArtifactPayloadInvalid, "decode artifact payload: %w", err)
	}
	return value, nil
}

func decodeCanonicalCheckpointArtifact(codec artifactCodecEntry, artifact CheckpointArtifact) (any, CheckpointArtifact, error) {
	value, err := decodeCheckpointArtifact(codec, artifact)
	if err != nil {
		return nil, CheckpointArtifact{}, err
	}
	canonical, err := serializeArtifact(codec, value, false)
	if err != nil {
		return nil, CheckpointArtifact{}, checkpointArtifactValidationFailure(CheckpointReasonArtifactPayloadInvalid, "encode canonical artifact: %w", err)
	}
	if canonical.Kind != artifact.Artifact.Kind || canonical.Schema.ID != artifact.Artifact.Schema.ID || canonical.Schema.Name != artifact.Artifact.Schema.Name || canonical.Schema.Version != artifact.Artifact.Schema.Version || canonical.MediaType != artifact.Artifact.MediaType || !bytes.Equal(canonical.Content, artifact.Artifact.Content) || checkpointContentDigest(canonical.Content) != checkpointContentDigest(artifact.Artifact.Content) {
		return nil, CheckpointArtifact{}, checkpointArtifactValidationFailure(CheckpointReasonArtifactNotCanonical, "stored artifact is not canonical")
	}
	hydrated, err := hydrateCheckpointArtifact(codec, cloneCheckpointArtifact(artifact), value)
	if err != nil {
		return nil, CheckpointArtifact{}, checkpointArtifactValidationFailure(CheckpointReasonArtifactPayloadInvalid, "hydrate checkpoint artifact: %w", err)
	}
	return value, hydrated, nil
}

func checkpointArtifact(codec artifactCodecEntry, laneID, moduleKey, sourceID string, value any) (CheckpointArtifact, error) {
	serialized, err := serializeArtifact(codec, value, false)
	if err != nil {
		return CheckpointArtifact{}, err
	}
	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
}

func (e *laneRunError) Error() string { return e.err.Error() }
func (e *laneRunError) Unwrap() error { return e.err }

type mergeStageResult struct {
	artifact             erasedMergeArtifact
	serialized           CheckpointArtifact
	terminal             bool
	validationIncomplete bool
	reuseEligible        bool
}

type mergeAttemptValue struct {
	artifact  erasedMergeArtifact
	candidate CheckpointArtifact
	terminal  *attemptTerminalRecorder
}

type normalizeAttemptValue struct {
	value     any
	candidate CheckpointArtifact
	terminal  *attemptTerminalRecorder
	retry     map[string]any
}

func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, prepared preparedLaneExecutor, extracts finalizedExtractResults, output *RunOutput) error {
	lane, typed := prepared.resolved, prepared.typed
	if typed == nil {
		return fmt.Errorf("typed lane %q executor is not prepared", lane.ID)
	}
	if err := setTypedLaneManifestMetadata(output, lane.ID, typed.extractor, typed.merger, typed.normalizer); err != nil {
		return err
	}
	merged, err := r.runMergeStage(ctx, input, checkpoints, loader, doc, sourceInput, sessionID, prepared, extracts, output)
	if err != nil {
		return &laneRunError{stage: StageMerge, err: err}
	}
	if merged.terminal {
		return nil
	}
	normalized, err := r.runNormalizeStage(ctx, input, checkpoints, loader, doc, sourceInput, sessionID, prepared, merged.artifact, merged.serialized, merged.reuseEligible, output)
	if err != nil {
		return &laneRunError{stage: StageNormalize, err: err}
	}
	if normalized.accepted {
		output.NormalizeOutputs = append(output.NormalizeOutputs, contracts.SerializedOutput{StepID: input.stepID, LaneID: lane.ID, NormalizerKey: lane.Normalize.Module, SourceID: doc.ID, Artifact: contracts.CloneSerializedArtifact(normalized.serialized.Artifact)})
		if output.normalizeReuseEligibility == nil {
			output.normalizeReuseEligibility = make(map[generatedOutputKey]bool)
		}
		output.normalizeReuseEligibility[generatedOutputKeyFor(input.stepID, lane.ID)] = normalized.reuseEligible
	}
	return nil
}

func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, prepared preparedLaneExecutor, extracts finalizedExtractResults, output *RunOutput) (mergeStageResult, error) {
	var stageResult mergeStageResult
	lane, typed := prepared.resolved, prepared.typed
	mergeInputs := make([]contracts.ExtractArtifact[any], len(extracts.accepted))
	for i, value := range extracts.accepted {
		mergeInputs[i] = contracts.ExtractArtifact[any]{LaneID: value.LaneID, ExtractorKey: value.ExtractorKey, SourceID: value.SourceID, ChunkID: value.ChunkID, ChunkIndex: value.ChunkIndex, ChunkRef: value.ChunkRef, Value: value.Value}
	}
	mergeReferences := operationReferenceSet(input, lane.MergeReferences)
	stageResult.reuseEligible = extracts.reuseEligible && referenceTargetReuseEligible(input, lane.MergeReferences)
	mergeDeps := append(artifactCheckpointDigests(extracts.serialized), generatedReferenceDependencies(mergeReferences)...)
	mergeDeps = normalizeCheckpointFingerprints(mergeDeps)
	var mergeCP MergeCheckpoint
	mergeDecision := checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonValidationIncompleteLineage)
	if stageResult.reuseEligible {
		mergeCP, mergeDecision = loadMerge(loader, input.stepID, lane.ID, lane.Merge.Module, mergeDeps)
	}
	mergeResolution, err := resolveCheckpointDecision(output, loader, input.CheckpointPolicy, StageMerge, input.stepID, lane.ID, lane.Merge.Module, mergeDecision, typed.codec, []CheckpointArtifact{mergeCP.Output})
	if err != nil {
		return stageResult, err
	}
	mergeDecision = mergeResolution.decision
	if err := writeDebugTimed(input.Debug, path.Join("merge", fileio.EncodePathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "decision": mergeDecision, "source": debugSourceDocumentEnvelope(doc), "extract_outputs": debugCheckpointArtifacts(extracts.serialized), "options": redactSensitiveMap(lane.Merge.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
		return stageResult, err
	}
	var merged erasedMergeArtifact
	var serializedMerge CheckpointArtifact
	if mergeDecision.Reused {
		merged = erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: mergeResolution.values[0]}
		serializedMerge = mergeResolution.artifacts[0]
		diagnostics, diagnosticErr := promoteCheckpointDiagnostics(serializedMerge.Diagnostics, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageMerge, StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module}, nil)
		if diagnosticErr != nil {
			return stageResult, fmt.Errorf("promote reused merge diagnostics: %w", diagnosticErr)
		}
		appendDiagnosticGroups(output, diagnostics)
	} else {
		if stageResult.reuseEligible {
			if err := checkpointMergeRunning(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps); err != nil {
				return stageResult, err
			}
		}
		terminalResult, runErr := runProducerAttempts(ctx, producerAttemptConfig{Retries: lane.Merge.Retries, Policy: lane.MergeValidationPolicy, AllowStructuralRetry: lane.MergeExecutionClass == contracts.ExecutionClassLLMBacked}, func(attemptCtx context.Context, request producerAttemptRequest) (producerAttemptOutput, error) {
			started := time.Now().UTC()
			attemptPath := path.Join("merge", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("attempt-%02d", request.Number))
			attemptCtx, llmScope := withDebugLLMScope(attemptCtx, attemptPath)
			terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "merge", llmScope, debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: request.Number, AttemptKind: string(request.Kind), StartedAt: started})
			requestMetadata, metadataErr := cloneMetadata(input.Metadata)
			if metadataErr != nil {
				return producerAttemptOutput{}, terminal.record(nil, fmt.Errorf("clone merge request metadata: %w", metadataErr))
			}
			result, callErr := typed.merge(attemptCtx, typed.merger, contracts.TypedMergeRequest[any]{Source: doc, LaneID: lane.ID, ExtractOutputs: mergeInputs, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(mergeReferences), LLMProfile: lane.Merge.LLMProfile, StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(lane.Merge.StructuredOutputRepairAttempts), Correction: request.Correction, Metadata: requestMetadata})
			if callErr != nil {
				return producerAttemptOutput{}, terminal.record(nil, fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, lane.Merge.Module, callErr))
			}
			candidate := erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: result.Value}
			serializedCandidate, encodeErr := serializeCandidateArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value)
			if encodeErr != nil {
				return producerAttemptOutput{}, terminal.record(nil, fmt.Errorf("serialize merge candidate for lane %q: %w", lane.ID, encodeErr))
			}
			return producerAttemptOutput{Value: mergeAttemptValue{artifact: candidate, candidate: serializedCandidate, terminal: &terminal}, Candidate: result.ModelCandidate, Diagnostics: contracts.CloneProducerDiagnostics(result.Diagnostics)}, nil
		}, func(validationCtx context.Context, output producerAttemptOutput) (validationReport, error) {
			candidate, ok := output.Value.(mergeAttemptValue)
			if !ok {
				return validationReport{}, fmt.Errorf("merge attempt has incompatible value")
			}
			target := typedValidationTarget{stage: StageMerge, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Merge.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: mergeReferences, metadata: input.Metadata, value: candidate.artifact.Value, candidate: &candidate.candidate}
			report, validationErr := r.validateTypedReport(validationCtx, typed.codec, target, prepared.mergeValidators, candidate.terminal.envelope.Attempt, input.Debug)
			payload := map[string]any{"output": debugCheckpointArtifact(candidate.candidate), "rejection": debugRejectedOutputPtr(typedRejection(report, target, candidate.terminal.envelope.Attempt))}
			if validationErr != nil {
				return report, candidate.terminal.record(payload, validationErr)
			}
			return report, candidate.terminal.record(payload, nil)
		})
		terminalSummary := validationSummary(terminalResult, StageMerge, input.stepID, lane.ID, lane.Merge.Module, "", 0)
		if debugErr := writeProducerTerminalDebug(input.Debug, path.Join("merge", fileio.EncodePathComponent(lane.ID), "terminal.json"), terminalResult, lane.MergeValidationPolicy, terminalSummary); debugErr != nil {
			if stageResult.reuseEligible {
				_ = checkpointMergeFailed(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, debugErr)
			}
			return stageResult, debugErr
		}
		if runErr != nil {
			if stageResult.reuseEligible {
				_ = checkpointMergeFailed(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, runErr)
			}
			return stageResult, runErr
		}
		terminalDiagnostics, diagnosticErr := terminalDiagnosticGroups(terminalResult, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageMerge, StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module}, nil)
		if diagnosticErr != nil {
			return stageResult, fmt.Errorf("promote merge diagnostics: %w", diagnosticErr)
		}
		if terminalResult.Action == producerTerminalRejected {
			rejected := terminalResult.Rejection
			if rejected == nil {
				return stageResult, fmt.Errorf("merge attempt terminal is missing rejection")
			}
			rejected.Stage, rejected.StepID, rejected.LaneID, rejected.ModuleKey = string(StageMerge), input.stepID, lane.ID, lane.Merge.Module
			rejected.Validation = &terminalSummary
			output.ValidationSummaries = append(output.ValidationSummaries, artifacts.CloneValidationSummary(terminalSummary))
			appendDiagnosticGroups(output, terminalDiagnostics)
			output.Rejected = append(output.Rejected, *rejected)
			if stageResult.reuseEligible {
				if err := checkpointMergeRejected(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, *rejected); err != nil {
					return stageResult, err
				}
			}
			stageResult.terminal = true
			return stageResult, nil
		}
		candidate, ok := terminalResult.Value.(mergeAttemptValue)
		if !ok {
			return stageResult, fmt.Errorf("merge attempt terminal has incompatible value")
		}
		stored, encodeErr := checkpointArtifact(typed.codec, candidate.artifact.LaneID, candidate.artifact.MergerKey, candidate.artifact.SourceID, candidate.artifact.Value)
		payload := map[string]any{"output": debugCheckpointArtifact(candidate.candidate), "rejection": debugRejectedOutputPtr(nil)}
		if encodeErr != nil {
			attemptErr := fmt.Errorf("serialize accepted merge output for lane %q: %w", lane.ID, encodeErr)
			if stageResult.reuseEligible {
				_ = checkpointMergeFailed(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, attemptErr)
			}
			return stageResult, candidate.terminal.record(payload, attemptErr)
		}
		stored.Diagnostics = terminalCheckpointDiagnostics(terminalResult)
		if debugErr := candidate.terminal.record(payload, nil); debugErr != nil {
			if stageResult.reuseEligible {
				_ = checkpointMergeFailed(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, debugErr)
			}
			return stageResult, debugErr
		}
		merged, serializedMerge = candidate.artifact, stored
		output.ValidationSummaries = append(output.ValidationSummaries, terminalSummary)
		stageResult.validationIncomplete = terminalResult.ValidationIncomplete
		if stageResult.validationIncomplete {
			stageResult.reuseEligible = false
		}
		appendDiagnosticGroups(output, terminalDiagnostics)
		if stageResult.reuseEligible {
			if err := recordMerge(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, serializedMerge); err != nil {
				return stageResult, err
			}
		}
	}
	if err := writeDebugTimed(input.Debug, path.Join("merge", fileio.EncodePathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "accepted": true, "output": debugCheckpointArtifact(serializedMerge)}}); err != nil {
		return stageResult, err
	}
	stageResult.artifact = merged
	stageResult.serialized = serializedMerge
	return stageResult, nil
}

type normalizeStageResult struct {
	serialized    CheckpointArtifact
	accepted      bool
	reuseEligible bool
}

func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, prepared preparedLaneExecutor, merged erasedMergeArtifact, serializedMerge CheckpointArtifact, upstreamReuseEligible bool, output *RunOutput) (normalizeStageResult, error) {
	var stageResult normalizeStageResult
	lane, typed := prepared.resolved, prepared.typed
	normalizeReferences := operationReferenceSet(input, lane.NormalizeReferences)
	stageResult.reuseEligible = upstreamReuseEligible && referenceTargetReuseEligible(input, lane.NormalizeReferences)
	normalizeDeps := append(artifactCheckpointDigests([]CheckpointArtifact{serializedMerge}), generatedReferenceDependencies(normalizeReferences)...)
	normalizeDeps = normalizeCheckpointFingerprints(normalizeDeps)
	var normalizeCP NormalizeCheckpoint
	normalizeDecision := checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonValidationIncompleteLineage)
	if stageResult.reuseEligible {
		normalizeCP, normalizeDecision = loadNormalize(loader, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps)
	}
	normalizeResolution, err := resolveCheckpointDecision(output, loader, input.CheckpointPolicy, StageNormalize, input.stepID, lane.ID, lane.Normalize.Module, normalizeDecision, typed.codec, []CheckpointArtifact{normalizeCP.Output})
	if err != nil {
		return stageResult, err
	}
	normalizeDecision = normalizeResolution.decision
	if err := writeDebugTimed(input.Debug, path.Join("normalize", fileio.EncodePathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "decision": normalizeDecision, "source": debugSourceDocumentEnvelope(doc), "merge_output": debugCheckpointArtifact(serializedMerge), "options": redactSensitiveMap(lane.Normalize.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
		return stageResult, err
	}
	var serializedNormalize CheckpointArtifact
	if normalizeDecision.Reused {
		serializedNormalize = normalizeResolution.artifacts[0]
		diagnostics, diagnosticErr := promoteCheckpointDiagnostics(serializedNormalize.Diagnostics, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageNormalize, StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module}, nil)
		if diagnosticErr != nil {
			return stageResult, fmt.Errorf("promote reused normalize diagnostics: %w", diagnosticErr)
		}
		appendDiagnosticGroups(output, diagnostics)
	} else {
		if stageResult.reuseEligible {
			if err := checkpointNormalizeRunning(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps); err != nil {
				return stageResult, err
			}
		}
		terminalResult, runErr := runProducerAttempts(ctx, producerAttemptConfig{Retries: lane.Normalize.Retries, Policy: lane.NormalizeValidationPolicy, AllowStructuralRetry: lane.NormalizeExecutionClass == contracts.ExecutionClassLLMBacked}, func(attemptCtx context.Context, request producerAttemptRequest) (producerAttemptOutput, error) {
			started := time.Now().UTC()
			attemptPath := path.Join("normalize", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("attempt-%02d", request.Number))
			attemptCtx, llmScope := withDebugLLMScope(attemptCtx, attemptPath)
			terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "normalize", llmScope, debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: request.Number, AttemptKind: string(request.Kind), StartedAt: started})
			requestMetadata, metadataErr := cloneMetadata(input.Metadata)
			if metadataErr != nil {
				return producerAttemptOutput{}, terminal.record(nil, fmt.Errorf("clone normalize request metadata: %w", metadataErr))
			}
			result, callErr := typed.normalize(attemptCtx, typed.normalizer, contracts.TypedNormalizeRequest[any]{Source: doc, LaneID: lane.ID, MergeOutput: contracts.MergeArtifact[any]{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: merged.Value}, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(normalizeReferences), LLMProfile: lane.Normalize.LLMProfile, StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(lane.Normalize.StructuredOutputRepairAttempts), Correction: request.Correction, Metadata: requestMetadata})
			if callErr != nil {
				return producerAttemptOutput{}, terminal.record(nil, fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, lane.Normalize.Module, callErr))
			}
			serializedCandidate, encodeErr := serializeCandidateArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value)
			if encodeErr != nil {
				return producerAttemptOutput{}, terminal.record(nil, fmt.Errorf("serialize normalize candidate for lane %q: %w", lane.ID, encodeErr))
			}
			attemptValue := normalizeAttemptValue{value: result.Value, candidate: serializedCandidate, terminal: &terminal}
			var directive *producerRetryDirective
			if result.Retry != nil {
				if err := validateNormalizeRetry(result.Retry); err != nil {
					return producerAttemptOutput{}, terminal.record(map[string]any{"output": debugCheckpointArtifact(serializedCandidate)}, fmt.Errorf("normalize lane %q returned invalid retry directive: %w", lane.ID, err))
				}
				anotherAttempt := request.Number <= lane.Normalize.Retries
				attemptValue.retry = map[string]any{"reason_code": result.Retry.ReasonCode, "message": result.Retry.Message, "correction_available": result.Retry.CorrectionGuidance != "", "another_attempt": anotherAttempt, "fallback_accepted": !anotherAttempt}
				directive = &producerRetryDirective{CorrectionGuidance: result.Retry.CorrectionGuidance, FallbackDiagnostics: contracts.CloneProducerDiagnostics(result.Retry.FallbackDiagnostics)}
				if anotherAttempt {
					payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "retry": attemptValue.retry}
					if debugErr := terminal.record(payload, nil); debugErr != nil {
						return producerAttemptOutput{}, debugErr
					}
				}
			}
			return producerAttemptOutput{Value: attemptValue, Candidate: result.ModelCandidate, Diagnostics: contracts.CloneProducerDiagnostics(result.Diagnostics), Retry: directive}, nil
		}, func(validationCtx context.Context, output producerAttemptOutput) (validationReport, error) {
			candidate, ok := output.Value.(normalizeAttemptValue)
			if !ok {
				return validationReport{}, fmt.Errorf("normalize attempt has incompatible value")
			}
			target := typedValidationTarget{stage: StageNormalize, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Normalize.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: normalizeReferences, metadata: input.Metadata, value: candidate.value, candidate: &candidate.candidate}
			report, validationErr := r.validateTypedReport(validationCtx, typed.codec, target, prepared.normalizeValidators, candidate.terminal.envelope.Attempt, input.Debug)
			payload := map[string]any{"output": debugCheckpointArtifact(candidate.candidate), "rejection": debugRejectedOutputPtr(typedRejection(report, target, candidate.terminal.envelope.Attempt))}
			if candidate.retry != nil {
				payload["retry"] = candidate.retry
			}
			if validationErr != nil {
				return report, candidate.terminal.record(payload, validationErr)
			}
			return report, candidate.terminal.record(payload, nil)
		})
		terminalSummary := validationSummary(terminalResult, StageNormalize, input.stepID, lane.ID, lane.Normalize.Module, "", 0)
		if debugErr := writeProducerTerminalDebug(input.Debug, path.Join("normalize", fileio.EncodePathComponent(lane.ID), "terminal.json"), terminalResult, lane.NormalizeValidationPolicy, terminalSummary); debugErr != nil {
			if stageResult.reuseEligible {
				_ = checkpointNormalizeFailed(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, debugErr)
			}
			return stageResult, debugErr
		}
		if runErr != nil {
			if stageResult.reuseEligible {
				_ = checkpointNormalizeFailed(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, runErr)
			}
			return stageResult, runErr
		}
		terminalDiagnostics, diagnosticErr := terminalDiagnosticGroups(terminalResult, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageNormalize, StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module}, nil)
		if diagnosticErr != nil {
			return stageResult, fmt.Errorf("promote normalize diagnostics: %w", diagnosticErr)
		}
		if terminalResult.Action == producerTerminalRejected {
			rejected := terminalResult.Rejection
			if rejected == nil {
				return stageResult, fmt.Errorf("normalize attempt terminal is missing rejection")
			}
			rejected.Stage, rejected.StepID, rejected.LaneID, rejected.ModuleKey = string(StageNormalize), input.stepID, lane.ID, lane.Normalize.Module
			rejected.Validation = &terminalSummary
			output.ValidationSummaries = append(output.ValidationSummaries, artifacts.CloneValidationSummary(terminalSummary))
			appendDiagnosticGroups(output, terminalDiagnostics)
			output.Rejected = append(output.Rejected, *rejected)
			if stageResult.reuseEligible {
				if err := checkpointNormalizeRejected(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, *rejected); err != nil {
					return stageResult, err
				}
			}
			return stageResult, nil
		}
		candidate, ok := terminalResult.Value.(normalizeAttemptValue)
		if !ok {
			return stageResult, fmt.Errorf("normalize attempt terminal has incompatible value")
		}
		stored, encodeErr := checkpointArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, candidate.value)
		payload := map[string]any{"output": debugCheckpointArtifact(candidate.candidate), "rejection": debugRejectedOutputPtr(nil)}
		if candidate.retry != nil {
			payload["retry"] = candidate.retry
		}
		if encodeErr != nil {
			attemptErr := fmt.Errorf("serialize accepted normalize output for lane %q: %w", lane.ID, encodeErr)
			if stageResult.reuseEligible {
				_ = checkpointNormalizeFailed(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, attemptErr)
			}
			return stageResult, candidate.terminal.record(payload, attemptErr)
		}
		stored.Diagnostics = terminalCheckpointDiagnostics(terminalResult)
		if debugErr := candidate.terminal.record(payload, nil); debugErr != nil {
			if stageResult.reuseEligible {
				_ = checkpointNormalizeFailed(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, debugErr)
			}
			return stageResult, debugErr
		}
		serializedNormalize = stored
		output.ValidationSummaries = append(output.ValidationSummaries, terminalSummary)
		appendDiagnosticGroups(output, terminalDiagnostics)
		if terminalResult.ValidationIncomplete {
			stageResult.reuseEligible = false
		}
		if stageResult.reuseEligible {
			if err := recordNormalize(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, serializedNormalize); err != nil {
				return stageResult, err
			}
		}
	}
	if err := writeDebugTimed(input.Debug, path.Join("normalize", fileio.EncodePathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "accepted": true, "output": debugCheckpointArtifact(serializedNormalize)}}); err != nil {
		return stageResult, err
	}
	stageResult.serialized = serializedNormalize
	stageResult.accepted = true
	return stageResult, nil
}

func setTypedLaneManifestMetadata(output *RunOutput, laneID string, extractor, merger, normalizer any) error {
	if output == nil {
		return nil
	}
	for i := range output.Manifest.ArtifactLanes {
		if output.Manifest.ArtifactLanes[i].ID != laneID {
			continue
		}
		metadata := make(map[string]any)
		for _, item := range []struct {
			name   string
			module any
		}{{"extractor", extractor}, {"merger", merger}, {"normalizer", normalizer}} {
			value, ok, err := moduleManifestMetadata(item.module)
			if err != nil {
				return fmt.Errorf("clone %s manifest metadata for lane %q: %w", item.name, laneID, err)
			}
			if ok {
				metadata[item.name] = value
			}
		}
		if len(metadata) > 0 {
			output.Manifest.ArtifactLanes[i].Metadata = metadata
		}
		return nil
	}
	return nil
}

func (r *Runner) validateTypedReport(ctx context.Context, codec artifactCodecEntry, target typedValidationTarget, chain preparedValidatorChain, attempt int, debug DebugRecorder) (validationReport, error) {
	if len(chain.validators) > 0 && target.candidate == nil {
		candidate, err := validationCandidateArtifact(codec, target)
		if err != nil {
			return validationReport{}, fmt.Errorf("serialize %s candidate for validation: %w", target.stage, err)
		}
		target.candidate = &candidate
	}
	report, err := executeValidationChain(ctx, chain, func(validatorCtx context.Context, item preparedValidator, validatorAttempt int) (validationInvocation, error) {
		binding := item.resolved.Binding
		var result contracts.ValidationResult
		var err error
		started := time.Now().UTC()
		pathParts := []string{"validate", fileio.EncodePathComponent(string(target.stage)), fileio.EncodePathComponent(target.laneID), fileio.EncodePathComponent(target.moduleKey)}
		if target.chunk != nil {
			pathParts = append(pathParts, fmt.Sprintf("chunk-%06d", target.chunk.Index+1))
		}
		pathParts = append(pathParts, fmt.Sprintf("%02d-%s-attempt-%02d", item.position, fileio.EncodePathComponent(binding.Module), attempt))
		attemptPath := validatorAttemptPath(path.Join(pathParts...), validatorAttempt)
		validatorCtx, llmScope := withIsolatedDebugLLMScope(validatorCtx, attemptPath)
		requestTarget := target
		requestTarget.sourceInput = target.sourceInput.Clone()
		requestTarget.references = CloneReferenceSet(target.references)
		requestTarget.metadata, err = cloneMetadata(target.metadata)
		if err != nil {
			return validationInvocation{}, fatalValidationError(fmt.Errorf("clone typed validation metadata: %w", err))
		}
		requestTarget.chunk, err = cloneSourceChunkPtr(target.chunk)
		if err != nil {
			return validationInvocation{}, fatalValidationError(fmt.Errorf("clone typed validation chunk: %w", err))
		}
		requestTarget.chunks, err = cloneSourceChunks(target.chunks)
		if err != nil {
			return validationInvocation{}, fatalValidationError(fmt.Errorf("clone typed validation chunks: %w", err))
		}
		switch item.resolved.Target {
		case ValidatorTargetTyped:
			candidateValue, decodeErr := decodeTypedValidationCandidate(codec, *target.candidate)
			if decodeErr != nil {
				return validationInvocation{}, fatalValidationError(fmt.Errorf("decode %s candidate for typed validator %q: %w", target.stage, binding.Module, decodeErr))
			}
			requestTarget.value = candidateValue
			requestTarget.llmProfile = binding.LLMProfile
			requestTarget.structuredOutputRepairAttempts = cloneStructuredOutputRepairAttempts(binding.StructuredOutputRepairAttempts)
			result, err = item.typedValidate(validatorCtx, item.typed, requestTarget)
		case ValidatorTargetSerialized:
			artifact := cloneCheckpointArtifact(*target.candidate)
			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: requestTarget.sourceInput, SessionID: target.sessionID, References: requestTarget.references, LLMProfile: binding.LLMProfile, StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(binding.StructuredOutputRepairAttempts), Metadata: requestTarget.metadata, Chunk: requestTarget.chunk, Chunks: requestTarget.chunks, Schema: contracts.CloneArtifactSchema(artifact.Artifact.Schema), MediaType: artifact.Artifact.MediaType, Content: append([]byte(nil), artifact.Artifact.Content...)})
		default:
			return validationInvocation{}, fatalValidationError(fmt.Errorf("validator %q is incompatible with typed artifact validation", binding.Module))
		}
		if err == nil {
			err = contracts.ValidateValidationResult(result)
		}
		artifact := cloneCheckpointArtifact(*target.candidate)
		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()
		}
		if err != nil {
			validationErr := fmt.Errorf("validate typed %s output with validator %q: %w", target.stage, binding.Module, err)
			if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(target.stage), StepID: target.stepID, LaneID: target.laneID, ModuleKey: target.moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope); debugErr != nil {
				return validationInvocation{}, fatalValidationError(errors.Join(validationErr, fmt.Errorf("write typed validator attempt debug artifact: %w", debugErr)))
			}
			return validationInvocation{}, validationErr
		}
		if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(target.stage), StepID: target.stepID, LaneID: target.laneID, ModuleKey: target.moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall}, llmScope); debugErr != nil {
			return validationInvocation{}, fatalValidationError(fmt.Errorf("write typed validator attempt debug artifact: %w", debugErr))
		}
		return validationInvocation{result: result}, nil
	})
	if err != nil {
		return report, err
	}
	return report, nil
}

func typedRejection(report validationReport, target typedValidationTarget, attempt int) *contracts.RejectedOutput {
	rejection := report.FirstRejection()
	if rejection == nil {
		return nil
	}
	chunkID, chunkIndex := "", 0
	if target.chunk != nil {
		chunkID, chunkIndex = target.chunk.ID, target.chunk.Index
	}
	return &contracts.RejectedOutput{Stage: string(target.stage), StepID: target.stepID, LaneID: target.laneID, ModuleKey: target.moduleKey, ChunkID: chunkID, ChunkIndex: chunkIndex, ValidatorName: rejection.validatorName, ReasonCode: rejection.reasonCode, Message: rejection.message, AttemptCount: attempt, DiagnosticArtifactPath: rejection.diagnosticPath}
}

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

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
}
