notarius/internal/framework/pipeline/runner.go

package pipeline

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"mime"
	"path"
	"path/filepath"
	"sort"
	"strings"
	"time"

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

type Registries struct {
	Inputs          *InputAdapterRegistry
	Chunkers        *ChunkerRegistry
	Extractors      *ExtractorRegistry
	Mergers         *MergerRegistry
	Normalizers     *NormalizerRegistry
	Validators      *ValidatorRegistry
	ValidatorChains *ValidatorChainRegistry
	Outputs         *OutputEncoderRegistry
}

type Runner struct {
	registries Registries
}

func New(registries Registries) *Runner {
	return &Runner{registries: registries}
}

type RunInput struct {
	Pipeline    ResolvedPipeline
	SourceID    string
	Path        string
	RawInput    []byte
	LLMClient   contracts.StructuredLLMClient
	SessionID   string
	RunID       string
	StartedAt   time.Time
	LLMProfiles []artifacts.LLMProfileManifest
	Metadata    map[string]any
	Warnings    []contracts.Warning
	Checkpoints CheckpointRecorder
	Checkpoint  CheckpointLoader
	Debug       DebugRecorder
}

type RunOutput struct {
	Manifest         artifacts.RunManifest       `json:"manifest"`
	NormalizeOutputs []contracts.NormalizeOutput `json:"normalize_outputs,omitempty"`
	Rejected         []contracts.RejectedOutput  `json:"rejected,omitempty"`
	Warnings         []contracts.Warning         `json:"warnings,omitempty"`
	OutputFiles      []contracts.OutputFile      `json:"-"`
	CheckpointEvents []CheckpointEvent           `json:"checkpoint_events,omitempty"`
}

func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err error) {
	if r == nil {
		return output, fmt.Errorf("runner must not be nil")
	}
	if err := validateRunInput(input); err != nil {
		return output, err
	}
	if err := r.validateRegistries(input.Pipeline); err != nil {
		return output, err
	}

	output.Manifest = manifestFromPipeline(input)
	checkpoints := input.Checkpoints
	if checkpoints == nil {
		checkpoints = NoopCheckpointRecorder()
	}
	checkpointLoader := input.Checkpoint
	if checkpointLoader == nil {
		checkpointLoader = NoopCheckpointLoader()
	}
	debugRecorder := input.Debug
	if debugRecorder == nil {
		debugRecorder = NoopDebugRecorder()
	}
	input.Debug = debugRecorder
	input.LLMClient = wrapDebugLLMClient(input.LLMClient, debugRecorder)
	defer func() {
		output.Manifest.LLMProfiles = mergeLLMProfileManifests(input.LLMProfiles, llmProfileManifests(input.LLMClient))
	}()
	output.Warnings = append(output.Warnings, cloneWarnings(input.Warnings)...)
	if err := writeDebugTimed(debugRecorder, "run.json", debugTimedEnvelope{
		Stage:     "run",
		StartedAt: startedTime(input.StartedAt),
		Payload: map[string]any{
			"pipeline_id":     input.Pipeline.ID,
			"pipeline_digest": input.Pipeline.Digest,
			"run_id":          output.Manifest.RunID,
		},
	}); err != nil {
		return failOutput(output), fmt.Errorf("write debug run artifact: %w", err)
	}

	adapter, err := r.registries.Inputs.Build(input.Pipeline.Input.Module)
	if err != nil {
		return failOutput(output), fmt.Errorf("build input adapter %q: %w", input.Pipeline.Input.Module, err)
	}
	attachModuleManifestMetadata(&output, "input", adapter)
	sourceCheckpoint, sourceDecision := checkpointLoader.Source(adapter.Key())
	recordCheckpointEvent(&output, checkpointLoader, "source", "", adapter.Key(), sourceDecision)
	doc := sourceCheckpoint.Document
	sourceStarted := time.Now().UTC()
	if err := writeDebugTimed(debugRecorder, "source/input.json", debugTimedEnvelope{
		Stage:     "source",
		ModuleKey: adapter.Key(),
		StartedAt: sourceStarted,
		Payload: debugSourceInput{
			SourceID: input.SourceID,
			Path:     input.Path,
			Raw:      debugContentEnvelope(input.RawInput, sourceInputMediaType(input.Path), nil, nil),
			Options:  redactSensitiveMap(input.Pipeline.Input.Options),
			Metadata: redactSensitiveMap(input.Metadata),
		},
	}); err != nil {
		return failOutput(output), fmt.Errorf("write source debug artifact: %w", err)
	}
	if !sourceDecision.Reused {
		if err := checkpoints.SourceRunning(adapter.Key()); err != nil {
			return failOutput(output), fmt.Errorf("write source checkpoint: %w", err)
		}
		doc, err = adapter.Parse(ctx, contracts.ParseRequest{
			SourceID:   input.SourceID,
			Path:       input.Path,
			Raw:        input.RawInput,
			LLMProfile: input.Pipeline.Input.LLMProfile,
			Options:    cloneOptions(input.Pipeline.Input.Options),
			Metadata:   input.Metadata,
		})
		if err != nil {
			_ = checkpoints.SourceFailed(adapter.Key(), err)
			return failOutput(output), fmt.Errorf("parse input with adapter %q: %w", adapter.Key(), err)
		}
		if err := source.ValidateDocument(doc); err != nil {
			_ = checkpoints.SourceFailed(adapter.Key(), err)
			return failOutput(output), fmt.Errorf("validate source document: %w", err)
		}
		if err := checkpoints.SourceSucceeded(adapter.Key(), doc); err != nil {
			return failOutput(output), fmt.Errorf("write source checkpoint: %w", err)
		}
	}
	if err := writeDebugTimed(debugRecorder, "source/output.json", debugTimedEnvelope{
		Stage:     "source",
		ModuleKey: adapter.Key(),
		StartedAt: sourceStarted,
		Payload: map[string]any{
			"reused":   sourceDecision.Reused,
			"decision": sourceDecision,
			"document": debugSourceDocumentEnvelope(doc),
		},
	}); err != nil {
		return failOutput(output), fmt.Errorf("write source debug artifact: %w", err)
	}
	sourceInput := sourceInputMaterial(input.Path, input.RawInput)
	sessionID := resolvedSessionID(input.SessionID, doc.ID)
	output.Manifest.Metadata = manifestMetadataWithSessionID(output.Manifest.Metadata, sessionID)
	output.Manifest.SourceDigests = []string{doc.Digest}

	chunker, err := r.registries.Chunkers.Build(input.Pipeline.Chunk.Module)
	if err != nil {
		return failOutput(output), fmt.Errorf("build chunker %q: %w", input.Pipeline.Chunk.Module, err)
	}
	attachModuleManifestMetadata(&output, "chunker", chunker)
	var canonicalChunks []contracts.SourceChunk
	var chunkWarnings []contracts.Warning
	chunkCheckpoint, chunkDecision := checkpointLoader.Chunk(chunker.Key(), doc.Digest)
	recordCheckpointEvent(&output, checkpointLoader, string(StageChunk), "", chunker.Key(), chunkDecision)
	chunkStarted := time.Now().UTC()
	if err := writeDebugTimed(debugRecorder, "chunk/input.json", debugTimedEnvelope{
		Stage:     string(StageChunk),
		ModuleKey: chunker.Key(),
		StartedAt: chunkStarted,
		Payload: map[string]any{
			"reused":       chunkDecision.Reused,
			"decision":     chunkDecision,
			"source":       debugSourceDocumentEnvelope(doc),
			"source_input": debugContentEnvelope(sourceInput.Content, sourceInput.MediaType, nil, nil),
			"options":      redactSensitiveMap(input.Pipeline.Chunk.Options),
			"metadata":     redactSensitiveMap(input.Metadata),
		},
	}); err != nil {
		return failOutput(output), fmt.Errorf("write chunk debug artifact: %w", err)
	}
	chunksAccepted := chunkDecision.Reused
	var chunkRejection *contracts.RejectedOutput
	if chunkDecision.Reused {
		canonicalChunks = cloneSourceChunks(chunkCheckpoint.Chunks)
		chunkWarnings = cloneWarnings(chunkCheckpoint.Warnings)
		output.Warnings = append(output.Warnings, chunkWarnings...)
	} else {
		if err := checkpoints.ChunkRunning(chunker.Key(), doc.Digest); err != nil {
			return failOutput(output), fmt.Errorf("write chunk checkpoint: %w", err)
		}
		chunksAccepted, chunkRejection, err = runWithRetry(ctx, input.Pipeline.Chunk.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
			attemptStarted := time.Now().UTC()
			attemptPath := path.Join("chunk", fmt.Sprintf("attempt-%02d", attempt))
			attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
			chunkResult, err := chunker.Chunk(attemptCtx, contracts.ChunkRequest{
				Source:      doc,
				SourceInput: sourceInput.Clone(),
				SessionID:   sessionID,
				References:  CloneReferenceSet(input.Pipeline.ChunkReferences.ReferenceSet),
				LLMClient:   input.LLMClient,
				LLMProfile:  input.Pipeline.Chunk.LLMProfile,
				Options:     cloneOptions(input.Pipeline.Chunk.Options),
				Metadata:    input.Metadata,
			})
			if err != nil {
				_ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
					Stage:     string(StageChunk),
					ModuleKey: chunker.Key(),
					Attempt:   attempt,
					StartedAt: attemptStarted,
					Error:     err.Error(),
				}, llmScope))
				return false, nil, fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), err)
			}
			if len(chunkResult.Chunks) == 0 {
				err := fmt.Errorf("chunker %q returned no chunks", chunker.Key())
				_ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
					Stage:     string(StageChunk),
					ModuleKey: chunker.Key(),
					Attempt:   attempt,
					StartedAt: attemptStarted,
					Error:     err.Error(),
				}, llmScope))
				return false, nil, err
			}
			chunks, err := validateAndCanonicalizeChunkResult(doc, chunkResult.Chunks)
			if err != nil {
				err := fmt.Errorf("validate chunks from chunker %q: %w", chunker.Key(), err)
				_ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
					Stage:     string(StageChunk),
					ModuleKey: chunker.Key(),
					Attempt:   attempt,
					StartedAt: attemptStarted,
					Payload: map[string]any{
						"warnings": cloneWarnings(chunkResult.Warnings),
					},
					Error: err.Error(),
				}, llmScope))
				return false, nil, err
			}
			validationWarnings, rejection, err := r.validateChunksRaw(attemptCtx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.Pipeline.ChunkReferences.ReferenceSet, input.LLMClient, input.Metadata, input.Pipeline.ValidatorChains, attempt, input.Debug)
			if err != nil || rejection != nil {
				_ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
					Stage:     string(StageChunk),
					ModuleKey: chunker.Key(),
					Attempt:   attempt,
					StartedAt: attemptStarted,
					Payload: map[string]any{
						"chunks":    debugSourceChunkEnvelopes(chunks),
						"warnings":  append(cloneWarnings(chunkResult.Warnings), validationWarnings...),
						"rejection": debugRejectedOutputPtr(rejection),
					},
				}, llmScope))
				return false, rejection, err
			}
			canonicalChunks = chunks
			chunkWarnings = append(cloneWarnings(chunkResult.Warnings), validationWarnings...)
			if err := writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
				Stage:     string(StageChunk),
				ModuleKey: chunker.Key(),
				Attempt:   attempt,
				StartedAt: attemptStarted,
				Payload: map[string]any{
					"chunks":   debugSourceChunkEnvelopes(chunks),
					"warnings": chunkWarnings,
				},
			}, llmScope)); err != nil {
				return false, nil, err
			}
			return true, nil, nil
		})
		if err != nil {
			_ = checkpoints.ChunkFailed(chunker.Key(), doc.Digest, err)
			return failOutput(output), err
		}
		if !chunksAccepted {
			output.Rejected = append(output.Rejected, *chunkRejection)
			if err := checkpoints.ChunkRejected(chunker.Key(), doc.Digest, *chunkRejection); err != nil {
				return failOutput(output), fmt.Errorf("write chunk checkpoint: %w", err)
			}
		} else {
			output.Warnings = append(output.Warnings, chunkWarnings...)
			if err := checkpoints.ChunkSucceeded(chunker.Key(), doc.Digest, canonicalChunks, chunkWarnings); err != nil {
				return failOutput(output), fmt.Errorf("write chunk checkpoint: %w", err)
			}
		}
	}
	chunkDebugPayload := map[string]any{
		"reused":   chunkDecision.Reused,
		"accepted": chunksAccepted,
		"chunks":   debugSourceChunkEnvelopes(canonicalChunks),
		"warnings": chunkWarnings,
	}
	if chunkRejection != nil {
		chunkDebugPayload["rejection"] = debugRejectedOutputEnvelope(*chunkRejection)
	}
	if err := writeDebugTimed(debugRecorder, "chunk/output.json", debugTimedEnvelope{
		Stage:     string(StageChunk),
		ModuleKey: chunker.Key(),
		StartedAt: chunkStarted,
		Payload:   chunkDebugPayload,
	}); err != nil {
		return failOutput(output), fmt.Errorf("write chunk debug artifact: %w", err)
	}

	if chunksAccepted {
		for _, lane := range input.Pipeline.ArtifactLanes {
			if err := r.runLane(ctx, input, checkpoints, checkpointLoader, doc, sourceInput, sessionID, canonicalChunks, lane, &output); err != nil {
				return failOutput(output), err
			}
		}
	}

	if len(output.Rejected) > 0 {
		output.Manifest.ValidationStatus = "rejected"
	} else {
		output.Manifest.ValidationStatus = "approved"
	}
	populateRawOutputManifest(&output)
	output.Manifest.CompletedAt = timePtr(time.Now().UTC())

	encoder, err := r.registries.Outputs.Build(input.Pipeline.Output.Module)
	if err != nil {
		return failOutput(output), fmt.Errorf("build output encoder %q: %w", input.Pipeline.Output.Module, err)
	}
	attachModuleManifestMetadata(&output, "output", encoder)
	outputStarted := time.Now().UTC()
	if err := writeDebugTimed(debugRecorder, "output/input.json", debugTimedEnvelope{
		Stage:     string(StageOutput),
		ModuleKey: encoder.Key(),
		StartedAt: outputStarted,
		Payload: map[string]any{
			"manifest":          output.Manifest,
			"normalize_outputs": debugNormalizeOutputEnvelopes(output.NormalizeOutputs),
			"rejected":          debugRejectedOutputEnvelopes(output.Rejected),
			"warnings":          output.Warnings,
			"options":           redactSensitiveMap(input.Pipeline.Output.Options),
			"metadata":          redactSensitiveMap(input.Metadata),
		},
	}); err != nil {
		return failOutput(output), fmt.Errorf("write output debug artifact: %w", err)
	}
	encoded, err := encoder.Encode(ctx, contracts.OutputRequest{
		Manifest:         output.Manifest,
		NormalizeOutputs: cloneNormalizeOutputs(output.NormalizeOutputs),
		Rejected:         cloneRejectedOutputs(output.Rejected),
		Warnings:         output.Warnings,
		LLMProfile:       input.Pipeline.Output.LLMProfile,
		Options:          cloneOptions(input.Pipeline.Output.Options),
		Metadata:         input.Metadata,
	})
	output.Warnings = append(output.Warnings, encoded.Warnings...)
	if err != nil {
		return failOutput(output), fmt.Errorf("encode output with encoder %q: %w", encoder.Key(), err)
	}
	files, err := outputFilesFromResult(encoded)
	if err != nil {
		return failOutput(output), fmt.Errorf("validate output files from encoder %q: %w", encoder.Key(), err)
	}
	output.OutputFiles = files
	if err := writeDebugTimed(debugRecorder, "output/output.json", debugTimedEnvelope{
		Stage:     string(StageOutput),
		ModuleKey: encoder.Key(),
		StartedAt: outputStarted,
		Payload: map[string]any{
			"files":    debugOutputFiles(files),
			"warnings": encoded.Warnings,
		},
	}); err != nil {
		return failOutput(output), fmt.Errorf("write output debug artifact: %w", err)
	}

	return output, nil
}

func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, checkpointLoader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []contracts.SourceChunk, lane ResolvedArtifactLane, output *RunOutput) error {
	extractor, err := r.registries.Extractors.Build(lane.Extract.Module)
	if err != nil {
		return fmt.Errorf("build extractor %q for lane %q: %w", lane.Extract.Module, lane.ID, err)
	}
	merger, err := r.registries.Mergers.Build(lane.Merge.Module)
	if err != nil {
		return fmt.Errorf("build merger %q for lane %q: %w", lane.Merge.Module, lane.ID, err)
	}
	normalizer, err := r.registries.Normalizers.Build(lane.Normalize.Module)
	if err != nil {
		return fmt.Errorf("build normalizer %q for lane %q: %w", lane.Normalize.Module, lane.ID, err)
	}
	setLaneManifestMetadata(output, lane.ID, extractor, merger, normalizer)

	extractOutputs := make([]contracts.ExtractOutput, 0, len(chunks))
	extractWarnings := []contracts.Warning{}
	extractRejectedStart := len(output.Rejected)
	extractDependencies := digestFingerprints("chunks", joinedChunkDigest(chunks))
	extractCheckpoint, extractDecision := checkpointLoader.Extract(lane.ID, extractor.Key(), extractDependencies)
	recordCheckpointEvent(output, checkpointLoader, string(StageExtract), lane.ID, extractor.Key(), extractDecision)
	extractStarted := time.Now().UTC()
	if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{
		Stage:     string(StageExtract),
		LaneID:    lane.ID,
		ModuleKey: extractor.Key(),
		StartedAt: extractStarted,
		Payload: map[string]any{
			"reused":   extractDecision.Reused,
			"decision": extractDecision,
			"source":   debugSourceDocumentEnvelope(doc),
			"chunks":   debugSourceChunkEnvelopes(chunks),
			"options":  redactSensitiveMap(lane.Extract.Options),
			"metadata": redactSensitiveMap(input.Metadata),
		},
	}); err != nil {
		return fmt.Errorf("write extract debug artifact for lane %q: %w", lane.ID, err)
	}
	if extractDecision.Reused {
		extractOutputs = cloneExtractOutputs(extractCheckpoint.Outputs)
		extractWarnings = cloneWarnings(extractCheckpoint.Warnings)
		output.Rejected = append(output.Rejected, cloneRejectedOutputs(extractCheckpoint.Rejected)...)
		output.Warnings = append(output.Warnings, extractWarnings...)
	} else {
		if err := checkpoints.ExtractRunning(lane.ID, extractor.Key(), extractDependencies); err != nil {
			return fmt.Errorf("write extract checkpoint for lane %q: %w", lane.ID, err)
		}
		for index := range chunks {
			chunk := chunks[index]
			var acceptedOutput contracts.ExtractOutput
			var acceptedWarnings []contracts.Warning
			accepted, rejection, err := runWithRetry(ctx, lane.Extract.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
				attemptStarted := time.Now().UTC()
				attemptPath := path.Join("extract", debugPathComponent(lane.ID), fmt.Sprintf("chunk-%06d", chunk.Index+1), fmt.Sprintf("attempt-%02d", attempt))
				attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
				result, err := extractor.Extract(attemptCtx, contracts.ExtractionRequest{
					Source:      doc,
					Chunk:       &chunk,
					SourceInput: chunkInputMaterial(sourceInput, chunk),
					SessionID:   sessionID,
					References:  CloneReferenceSet(lane.ExtractReferences.ReferenceSet),
					LLMClient:   input.LLMClient,
					LLMProfile:  lane.Extract.LLMProfile,
					Options:     cloneOptions(lane.Extract.Options),
					Metadata:    input.Metadata,
				})
				if err != nil {
					_ = writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
						Stage:     string(StageExtract),
						LaneID:    lane.ID,
						ModuleKey: extractor.Key(),
						Attempt:   attempt,
						StartedAt: attemptStarted,
						Error:     err.Error(),
					}, llmScope))
					return false, nil, fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, extractor.Key(), err)
				}
				extractOutput := result.Output
				extractOutput.LaneID = lane.ID
				extractOutput.ExtractorKey = extractor.Key()
				extractOutput.SourceID = doc.ID
				extractOutput.ChunkID = chunk.ID
				extractOutput.ChunkIndex = chunk.Index
				extractOutput.Payload.Warnings = append(extractOutput.Payload.Warnings, result.Warnings...)
				validationWarnings, rejection, err := r.validateRaw(attemptCtx, rawValidationTarget{
					stage:       StageExtract,
					laneID:      lane.ID,
					moduleKey:   extractor.Key(),
					source:      doc,
					sourceID:    doc.ID,
					chunkID:     chunk.ID,
					chunkIndex:  chunk.Index,
					chunk:       &chunk,
					sourceInput: chunkInputMaterial(sourceInput, chunk),
					sessionID:   sessionID,
					references:  lane.ExtractReferences.ReferenceSet,
					llmClient:   input.LLMClient,
					schema:      extractOutput.Schema,
					payload:     extractOutput.Payload,
					metadata:    input.Metadata,
					chains:      input.Pipeline.ValidatorChains,
					attempt:     attempt,
					debug:       input.Debug,
				})
				if err != nil || rejection != nil {
					_ = writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
						Stage:     string(StageExtract),
						LaneID:    lane.ID,
						ModuleKey: extractor.Key(),
						Attempt:   attempt,
						StartedAt: attemptStarted,
						Payload: map[string]any{
							"output":    debugExtractOutputEnvelope(extractOutput),
							"warnings":  append(cloneWarnings(result.Warnings), validationWarnings...),
							"rejection": debugRejectedOutputPtr(rejection),
						},
					}, llmScope))
					return false, rejection, err
				}
				acceptedOutput = cloneExtractOutput(extractOutput)
				acceptedWarnings = append(cloneWarnings(result.Warnings), validationWarnings...)
				if err := writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
					Stage:     string(StageExtract),
					LaneID:    lane.ID,
					ModuleKey: extractor.Key(),
					Attempt:   attempt,
					StartedAt: attemptStarted,
					Payload: map[string]any{
						"output":   debugExtractOutputEnvelope(extractOutput),
						"warnings": acceptedWarnings,
					},
				}, llmScope)); err != nil {
					return false, nil, err
				}
				return true, nil, nil
			})
			if err != nil {
				_ = checkpoints.ExtractFailed(lane.ID, extractor.Key(), extractDependencies, err)
				return err
			}
			if !accepted {
				output.Rejected = append(output.Rejected, *rejection)
				continue
			}
			output.Warnings = append(output.Warnings, acceptedWarnings...)
			extractWarnings = append(extractWarnings, acceptedWarnings...)
			extractOutputs = append(extractOutputs, acceptedOutput)
		}
		extractRejected := cloneRejectedOutputs(output.Rejected[extractRejectedStart:])
		if err := checkpoints.ExtractSucceeded(lane.ID, extractor.Key(), extractDependencies, extractOutputs, extractRejected, extractWarnings); err != nil {
			return fmt.Errorf("write extract checkpoint for lane %q: %w", lane.ID, err)
		}
	}
	if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{
		Stage:     string(StageExtract),
		LaneID:    lane.ID,
		ModuleKey: extractor.Key(),
		StartedAt: extractStarted,
		Payload: map[string]any{
			"reused":   extractDecision.Reused,
			"outputs":  debugExtractOutputEnvelopes(extractOutputs),
			"rejected": debugRejectedOutputEnvelopes(output.Rejected[extractRejectedStart:]),
			"warnings": extractWarnings,
		},
	}); err != nil {
		return fmt.Errorf("write extract debug artifact for lane %q: %w", lane.ID, err)
	}

	if len(extractOutputs) == 0 {
		return nil
	}

	var acceptedMerge contracts.MergeOutput
	var mergeWarnings []contracts.Warning
	mergeDependencies := rawOutputDigests(extractPayloads(extractOutputs))
	mergeCheckpoint, mergeDecision := checkpointLoader.Merge(lane.ID, merger.Key(), mergeDependencies)
	recordCheckpointEvent(output, checkpointLoader, string(StageMerge), lane.ID, merger.Key(), mergeDecision)
	mergeStarted := time.Now().UTC()
	if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{
		Stage:     string(StageMerge),
		LaneID:    lane.ID,
		ModuleKey: merger.Key(),
		StartedAt: mergeStarted,
		Payload: map[string]any{
			"reused":          mergeDecision.Reused,
			"decision":        mergeDecision,
			"source":          debugSourceDocumentEnvelope(doc),
			"extract_outputs": debugExtractOutputEnvelopes(extractOutputs),
			"options":         redactSensitiveMap(lane.Merge.Options),
			"metadata":        redactSensitiveMap(input.Metadata),
		},
	}); err != nil {
		return fmt.Errorf("write merge debug artifact for lane %q: %w", lane.ID, err)
	}
	if mergeDecision.Reused {
		acceptedMerge = cloneMergeOutput(mergeCheckpoint.Output)
		mergeWarnings = cloneWarnings(mergeCheckpoint.Warnings)
		output.Warnings = append(output.Warnings, mergeWarnings...)
	} else {
		if err := checkpoints.MergeRunning(lane.ID, merger.Key(), mergeDependencies); err != nil {
			return fmt.Errorf("write merge checkpoint for lane %q: %w", lane.ID, err)
		}
		mergeAccepted, mergeRejection, err := runWithRetry(ctx, lane.Merge.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
			attemptStarted := time.Now().UTC()
			attemptPath := path.Join("merge", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
			attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
			mergeResult, err := merger.Merge(attemptCtx, contracts.MergeRequest{
				Source:         doc,
				LaneID:         lane.ID,
				ExtractOutputs: cloneExtractOutputs(extractOutputs),
				SourceInput:    sourceInput.Clone(),
				SessionID:      sessionID,
				References:     CloneReferenceSet(lane.MergeReferences.ReferenceSet),
				LLMClient:      input.LLMClient,
				LLMProfile:     lane.Merge.LLMProfile,
				Options:        cloneOptions(lane.Merge.Options),
				Metadata:       input.Metadata,
			})
			if err != nil {
				_ = writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
					Stage:     string(StageMerge),
					LaneID:    lane.ID,
					ModuleKey: merger.Key(),
					Attempt:   attempt,
					StartedAt: attemptStarted,
					Error:     err.Error(),
				}, llmScope))
				return false, nil, fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, merger.Key(), err)
			}
			mergeOutput := mergeResult.Output
			mergeOutput.LaneID = lane.ID
			mergeOutput.MergerKey = merger.Key()
			mergeOutput.SourceID = doc.ID
			mergeOutput.Payload.Warnings = append(mergeOutput.Payload.Warnings, mergeResult.Warnings...)
			validationWarnings, rejection, err := r.validateRaw(attemptCtx, rawValidationTarget{
				stage:          StageMerge,
				laneID:         lane.ID,
				moduleKey:      merger.Key(),
				source:         doc,
				sourceID:       doc.ID,
				sourceInput:    sourceInput.Clone(),
				sessionID:      sessionID,
				references:     lane.MergeReferences.ReferenceSet,
				llmClient:      input.LLMClient,
				schema:         mergeOutput.Schema,
				payload:        mergeOutput.Payload,
				extractOutputs: extractOutputs,
				metadata:       input.Metadata,
				chains:         input.Pipeline.ValidatorChains,
				attempt:        attempt,
				debug:          input.Debug,
			})
			if err != nil || rejection != nil {
				_ = writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
					Stage:     string(StageMerge),
					LaneID:    lane.ID,
					ModuleKey: merger.Key(),
					Attempt:   attempt,
					StartedAt: attemptStarted,
					Payload: map[string]any{
						"output":    debugMergeOutputEnvelope(mergeOutput),
						"warnings":  append(cloneWarnings(mergeResult.Warnings), validationWarnings...),
						"rejection": debugRejectedOutputPtr(rejection),
					},
				}, llmScope))
				return false, rejection, err
			}
			acceptedMerge = cloneMergeOutput(mergeOutput)
			mergeWarnings = append(cloneWarnings(mergeResult.Warnings), validationWarnings...)
			if err := writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
				Stage:     string(StageMerge),
				LaneID:    lane.ID,
				ModuleKey: merger.Key(),
				Attempt:   attempt,
				StartedAt: attemptStarted,
				Payload: map[string]any{
					"output":   debugMergeOutputEnvelope(mergeOutput),
					"warnings": mergeWarnings,
				},
			}, llmScope)); err != nil {
				return false, nil, err
			}
			return true, nil, nil
		})
		if err != nil {
			_ = checkpoints.MergeFailed(lane.ID, merger.Key(), mergeDependencies, err)
			return err
		}
		if !mergeAccepted {
			output.Rejected = append(output.Rejected, *mergeRejection)
			if err := checkpoints.MergeRejected(lane.ID, merger.Key(), mergeDependencies, *mergeRejection); err != nil {
				return fmt.Errorf("write merge checkpoint for lane %q: %w", lane.ID, err)
			}
			if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{
				Stage:     string(StageMerge),
				LaneID:    lane.ID,
				ModuleKey: merger.Key(),
				StartedAt: mergeStarted,
				Payload: map[string]any{
					"accepted":  false,
					"rejection": debugRejectedOutputEnvelope(*mergeRejection),
					"warnings":  mergeWarnings,
				},
			}); err != nil {
				return fmt.Errorf("write merge debug artifact for lane %q: %w", lane.ID, err)
			}
			return nil
		}
		output.Warnings = append(output.Warnings, mergeWarnings...)
		if err := checkpoints.MergeSucceeded(lane.ID, merger.Key(), mergeDependencies, acceptedMerge, mergeWarnings); err != nil {
			return fmt.Errorf("write merge checkpoint for lane %q: %w", lane.ID, err)
		}
	}
	if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{
		Stage:     string(StageMerge),
		LaneID:    lane.ID,
		ModuleKey: merger.Key(),
		StartedAt: mergeStarted,
		Payload: map[string]any{
			"reused":   mergeDecision.Reused,
			"accepted": true,
			"output":   debugMergeOutputEnvelope(acceptedMerge),
			"warnings": mergeWarnings,
		},
	}); err != nil {
		return fmt.Errorf("write merge debug artifact for lane %q: %w", lane.ID, err)
	}

	var acceptedNormalize contracts.NormalizeOutput
	var normalizeWarnings []contracts.Warning
	normalizeDependencies := rawOutputDigests([]contracts.RawPayload{acceptedMerge.Payload})
	normalizeCheckpoint, normalizeDecision := checkpointLoader.Normalize(lane.ID, normalizer.Key(), normalizeDependencies)
	recordCheckpointEvent(output, checkpointLoader, string(StageNormalize), lane.ID, normalizer.Key(), normalizeDecision)
	normalizeStarted := time.Now().UTC()
	if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{
		Stage:     string(StageNormalize),
		LaneID:    lane.ID,
		ModuleKey: normalizer.Key(),
		StartedAt: normalizeStarted,
		Payload: map[string]any{
			"reused":       normalizeDecision.Reused,
			"decision":     normalizeDecision,
			"source":       debugSourceDocumentEnvelope(doc),
			"merge_output": debugMergeOutputEnvelope(acceptedMerge),
			"options":      redactSensitiveMap(lane.Normalize.Options),
			"metadata":     redactSensitiveMap(input.Metadata),
		},
	}); err != nil {
		return fmt.Errorf("write normalize debug artifact for lane %q: %w", lane.ID, err)
	}
	if normalizeDecision.Reused {
		acceptedNormalize = cloneNormalizeOutput(normalizeCheckpoint.Output)
		normalizeWarnings = cloneWarnings(normalizeCheckpoint.Warnings)
		output.Warnings = append(output.Warnings, normalizeWarnings...)
	} else {
		if err := checkpoints.NormalizeRunning(lane.ID, normalizer.Key(), normalizeDependencies); err != nil {
			return fmt.Errorf("write normalize checkpoint for lane %q: %w", lane.ID, err)
		}
		normalizeAccepted, normalizeRejection, err := runWithRetry(ctx, lane.Normalize.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
			attemptStarted := time.Now().UTC()
			attemptPath := path.Join("normalize", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
			attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
			normalizeResult, err := normalizer.Normalize(attemptCtx, contracts.NormalizeRequest{
				Source:      doc,
				LaneID:      lane.ID,
				MergeOutput: cloneMergeOutput(acceptedMerge),
				SourceInput: sourceInput.Clone(),
				SessionID:   sessionID,
				References:  CloneReferenceSet(lane.NormalizeReferences.ReferenceSet),
				LLMClient:   input.LLMClient,
				LLMProfile:  lane.Normalize.LLMProfile,
				Options:     cloneOptions(lane.Normalize.Options),
				Metadata:    input.Metadata,
			})
			if err != nil {
				_ = writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
					Stage:     string(StageNormalize),
					LaneID:    lane.ID,
					ModuleKey: normalizer.Key(),
					Attempt:   attempt,
					StartedAt: attemptStarted,
					Error:     err.Error(),
				}, llmScope))
				return false, nil, fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, normalizer.Key(), err)
			}
			normalizeOutput := normalizeResult.Output
			normalizeOutput.LaneID = lane.ID
			normalizeOutput.NormalizerKey = normalizer.Key()
			normalizeOutput.SourceID = doc.ID
			normalizeOutput.Payload.Warnings = append(normalizeOutput.Payload.Warnings, normalizeResult.Warnings...)
			validationWarnings, rejection, err := r.validateRaw(attemptCtx, rawValidationTarget{
				stage:       StageNormalize,
				laneID:      lane.ID,
				moduleKey:   normalizer.Key(),
				source:      doc,
				sourceID:    doc.ID,
				sourceInput: sourceInput.Clone(),
				sessionID:   sessionID,
				references:  lane.NormalizeReferences.ReferenceSet,
				llmClient:   input.LLMClient,
				schema:      normalizeOutput.Schema,
				payload:     normalizeOutput.Payload,
				mergeOutput: acceptedMerge,
				metadata:    input.Metadata,
				chains:      input.Pipeline.ValidatorChains,
				attempt:     attempt,
				debug:       input.Debug,
			})
			if err != nil || rejection != nil {
				_ = writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
					Stage:     string(StageNormalize),
					LaneID:    lane.ID,
					ModuleKey: normalizer.Key(),
					Attempt:   attempt,
					StartedAt: attemptStarted,
					Payload: map[string]any{
						"output":    debugNormalizeOutputEnvelope(normalizeOutput),
						"warnings":  append(cloneWarnings(normalizeResult.Warnings), validationWarnings...),
						"rejection": debugRejectedOutputPtr(rejection),
					},
				}, llmScope))
				return false, rejection, err
			}
			acceptedNormalize = cloneNormalizeOutput(normalizeOutput)
			normalizeWarnings = append(cloneWarnings(normalizeResult.Warnings), validationWarnings...)
			if err := writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
				Stage:     string(StageNormalize),
				LaneID:    lane.ID,
				ModuleKey: normalizer.Key(),
				Attempt:   attempt,
				StartedAt: attemptStarted,
				Payload: map[string]any{
					"output":   debugNormalizeOutputEnvelope(normalizeOutput),
					"warnings": normalizeWarnings,
				},
			}, llmScope)); err != nil {
				return false, nil, err
			}
			return true, nil, nil
		})
		if err != nil {
			_ = checkpoints.NormalizeFailed(lane.ID, normalizer.Key(), normalizeDependencies, err)
			return err
		}
		if !normalizeAccepted {
			output.Rejected = append(output.Rejected, *normalizeRejection)
			if err := checkpoints.NormalizeRejected(lane.ID, normalizer.Key(), normalizeDependencies, *normalizeRejection); err != nil {
				return fmt.Errorf("write normalize checkpoint for lane %q: %w", lane.ID, err)
			}
			if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{
				Stage:     string(StageNormalize),
				LaneID:    lane.ID,
				ModuleKey: normalizer.Key(),
				StartedAt: normalizeStarted,
				Payload: map[string]any{
					"accepted":  false,
					"rejection": debugRejectedOutputEnvelope(*normalizeRejection),
					"warnings":  normalizeWarnings,
				},
			}); err != nil {
				return fmt.Errorf("write normalize debug artifact for lane %q: %w", lane.ID, err)
			}
			return nil
		}
		output.Warnings = append(output.Warnings, normalizeWarnings...)
		if err := checkpoints.NormalizeSucceeded(lane.ID, normalizer.Key(), normalizeDependencies, acceptedNormalize, normalizeWarnings); err != nil {
			return fmt.Errorf("write normalize checkpoint for lane %q: %w", lane.ID, err)
		}
	}
	if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{
		Stage:     string(StageNormalize),
		LaneID:    lane.ID,
		ModuleKey: normalizer.Key(),
		StartedAt: normalizeStarted,
		Payload: map[string]any{
			"reused":   normalizeDecision.Reused,
			"accepted": true,
			"output":   debugNormalizeOutputEnvelope(acceptedNormalize),
			"warnings": normalizeWarnings,
		},
	}); err != nil {
		return fmt.Errorf("write normalize debug artifact for lane %q: %w", lane.ID, err)
	}
	output.NormalizeOutputs = append(output.NormalizeOutputs, acceptedNormalize)
	return nil
}

type rawValidationTarget struct {
	stage          ModuleStage
	laneID         string
	moduleKey      string
	source         *source.SourceDocument
	sourceID       string
	sourceInput    contracts.LLMInputMaterial
	sessionID      string
	references     contracts.ReferenceSet
	llmClient      contracts.StructuredLLMClient
	chunkID        string
	chunkIndex     int
	chunk          *contracts.SourceChunk
	chunks         []contracts.SourceChunk
	schema         contracts.ResponseSchema
	payload        contracts.RawPayload
	extractOutputs []contracts.ExtractOutput
	mergeOutput    contracts.MergeOutput
	metadata       map[string]any
	chains         []ResolvedValidatorChain
	attempt        int
	debug          DebugRecorder
}

func runWithRetry(ctx context.Context, retries int, run func(attempt int) (bool, *contracts.RejectedOutput, error)) (bool, *contracts.RejectedOutput, error) {
	attempts := 1
	if retries > 0 {
		attempts += retries
	}

	var lastRejection *contracts.RejectedOutput
	for attempt := 1; attempt <= attempts; attempt++ {
		if err := ctx.Err(); err != nil {
			return false, nil, err
		}

		accepted, rejection, err := run(attempt)
		if err != nil {
			if ctxErr := ctx.Err(); ctxErr != nil {
				return false, nil, ctxErr
			}
			if attempt == attempts {
				return false, nil, fmt.Errorf("failed after %d attempt(s): %w", attempt, err)
			}
			continue
		}
		if accepted {
			return true, nil, nil
		}
		if rejection != nil {
			rejection.AttemptCount = attempt
			lastRejection = rejection
		}
		if ctxErr := ctx.Err(); ctxErr != nil {
			return false, nil, ctxErr
		}
		if attempt == attempts {
			if lastRejection == nil {
				lastRejection = &contracts.RejectedOutput{
					ReasonCode:   "raw_output_rejected",
					Message:      "raw output rejected",
					AttemptCount: attempt,
				}
			}
			return false, lastRejection, nil
		}
	}

	return false, lastRejection, nil
}

func (r *Runner) validateChunksRaw(ctx context.Context, doc *source.SourceDocument, moduleKey string, chunks []contracts.SourceChunk, sourceInput contracts.LLMInputMaterial, sessionID string, references contracts.ReferenceSet, llmClient contracts.StructuredLLMClient, metadata map[string]any, chains []ResolvedValidatorChain, attempt int, debug DebugRecorder) ([]contracts.Warning, *contracts.RejectedOutput, error) {
	return r.validateRaw(ctx, rawValidationTarget{
		stage:       StageChunk,
		moduleKey:   moduleKey,
		source:      doc,
		sourceID:    doc.ID,
		sourceInput: sourceInput.Clone(),
		sessionID:   sessionID,
		references:  references,
		llmClient:   llmClient,
		chunks:      chunks,
		metadata:    metadata,
		chains:      chains,
		attempt:     attempt,
		debug:       debug,
	})
}

func (r *Runner) validateRaw(ctx context.Context, target rawValidationTarget) ([]contracts.Warning, *contracts.RejectedOutput, error) {
	chain := resolvedValidatorChain(target.stage, target.laneID, target.moduleKey, target.chains)
	if len(chain.Validators) == 0 {
		return nil, nil, nil
	}
	if r.registries.Validators == nil {
		return nil, nil, fmt.Errorf("validator registry must not be nil")
	}

	var warnings []contracts.Warning
	for index, validatorBinding := range chain.Validators {
		validator, err := r.registries.Validators.Build(validatorBinding.Binding.Module)
		if err != nil {
			return nil, nil, fmt.Errorf("build validator %q: %w", validatorBinding.Binding.Module, err)
		}
		request := target.validationRequest(validatorBinding.Binding)
		started := time.Now().UTC()
		attemptPath := path.Join("validate", debugPathComponent(string(target.stage)), debugPathComponent(target.laneID), debugPathComponent(target.moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d", index+1, debugPathComponent(validator.Name()), target.attempt))
		validatorCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
		result, err := validator.Validate(validatorCtx, request)
		debugPayload := debugValidationCall{
			ValidatorName: validator.Name(),
			Request:       debugValidationRequestEnvelope(request),
			Result:        debugValidationResultEnvelope(result),
		}
		if err != nil {
			debugPayload.Error = err.Error()
		}
		if debugErr := writeDebugTimed(target.debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
			Stage:     string(target.stage),
			LaneID:    target.laneID,
			ModuleKey: target.moduleKey,
			Attempt:   target.attempt,
			StartedAt: started,
			Payload:   debugPayload,
			Error:     debugPayload.Error,
		}, llmScope)); debugErr != nil {
			return nil, nil, fmt.Errorf("write validation debug artifact: %w", debugErr)
		}
		if err != nil {
			return nil, nil, fmt.Errorf("validate raw %s output with validator %q: %w", target.stage, validator.Name(), err)
		}
		if !result.Approved {
			reasonCode := strings.TrimSpace(result.ReasonCode)
			if reasonCode == "" {
				reasonCode = "raw_output_rejected"
			}
			message := strings.TrimSpace(result.Message)
			if message == "" {
				message = "raw output rejected"
			}
			return nil, &contracts.RejectedOutput{
				Stage:                  string(target.stage),
				LaneID:                 target.laneID,
				ModuleKey:              target.moduleKey,
				ChunkID:                target.chunkID,
				ChunkIndex:             target.chunkIndex,
				ValidatorName:          validator.Name(),
				ReasonCode:             reasonCode,
				Message:                message,
				AttemptCount:           target.attempt,
				DiagnosticArtifactPath: result.DiagnosticArtifactPath,
			}, nil
		}
		warnings = append(warnings, result.Warnings...)
	}
	return warnings, nil, nil
}

func (target rawValidationTarget) validationRequest(binding ModuleBinding) contracts.ValidationRequest {
	return contracts.ValidationRequest{
		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),
		LLMClient:      target.llmClient,
		LLMProfile:     binding.LLMProfile,
		Options:        cloneOptions(binding.Options),
		Metadata:       cloneMetadata(target.metadata),
		Schema:         cloneResponseSchema(target.schema),
		Payload:        cloneRawPayload(target.payload),
		ChunkID:        target.chunkID,
		ChunkIndex:     target.chunkIndex,
		Chunk:          cloneSourceChunkPtr(target.chunk),
		Chunks:         cloneSourceChunks(target.chunks),
		ExtractOutputs: cloneExtractOutputs(target.extractOutputs),
		MergeOutput:    cloneMergeOutput(target.mergeOutput),
	}
}

func resolvedValidatorChain(stage ModuleStage, laneID string, moduleKey string, chains []ResolvedValidatorChain) ResolvedValidatorChain {
	for _, chain := range chains {
		if chain.Stage != stage {
			continue
		}
		if chain.ModuleKey != moduleKey {
			continue
		}
		if strings.TrimSpace(chain.LaneID) != strings.TrimSpace(laneID) {
			continue
		}
		return ResolvedValidatorChain{
			Stage:      chain.Stage,
			LaneID:     chain.LaneID,
			ModuleKey:  chain.ModuleKey,
			Validators: cloneResolvedValidators(chain.Validators),
		}
	}
	return ResolvedValidatorChain{
		Stage:     stage,
		LaneID:    strings.TrimSpace(laneID),
		ModuleKey: strings.TrimSpace(moduleKey),
	}
}

func (r *Runner) validateRegistries(pipeline ResolvedPipeline) error {
	if r.registries.Inputs == nil {
		return fmt.Errorf("input registry must not be nil")
	}
	if r.registries.Chunkers == nil {
		return fmt.Errorf("chunker registry must not be nil")
	}
	if r.registries.Extractors == nil {
		return fmt.Errorf("extractor registry must not be nil")
	}
	if r.registries.Mergers == nil {
		return fmt.Errorf("merger registry must not be nil")
	}
	if r.registries.Normalizers == nil {
		return fmt.Errorf("normalizer registry must not be nil")
	}
	if r.registries.Outputs == nil {
		return fmt.Errorf("output encoder registry must not be nil")
	}
	return nil
}

func validateRunInput(input RunInput) error {
	if input.Pipeline.ID == "" {
		return fmt.Errorf("resolved pipeline id must not be empty")
	}
	if input.Pipeline.Digest == "" {
		return fmt.Errorf("resolved pipeline digest must not be empty")
	}
	if input.Pipeline.Input.Module == "" {
		return fmt.Errorf("resolved pipeline input module must not be empty")
	}
	if input.Pipeline.Chunk.Module == "" {
		return fmt.Errorf("resolved pipeline chunk module must not be empty")
	}
	if input.Pipeline.Output.Module == "" {
		return fmt.Errorf("resolved pipeline output module must not be empty")
	}
	if len(input.Pipeline.ArtifactLanes) == 0 {
		return fmt.Errorf("resolved pipeline artifact lanes must not be empty")
	}
	for _, lane := range input.Pipeline.ArtifactLanes {
		if lane.ID == "" {
			return fmt.Errorf("resolved pipeline artifact lane id must not be empty")
		}
		if lane.Extract.Module == "" {
			return fmt.Errorf("resolved pipeline lane %q extract module must not be empty", lane.ID)
		}
		if lane.Merge.Module == "" {
			return fmt.Errorf("resolved pipeline lane %q merge module must not be empty", lane.ID)
		}
		if lane.Normalize.Module == "" {
			return fmt.Errorf("resolved pipeline lane %q normalize module must not be empty", lane.ID)
		}
		if len(lane.Validators) > 0 {
			return fmt.Errorf("resolved pipeline lane %q validators are not supported at artifact lane level; use extract.validators, merge.validators, or normalize.validators", lane.ID)
		}
	}
	return nil
}

func startedTime(t time.Time) time.Time {
	if t.IsZero() {
		return time.Now().UTC()
	}
	return t.UTC()
}

func manifestFromPipeline(input RunInput) artifacts.RunManifest {
	startedAt := input.StartedAt
	if startedAt.IsZero() {
		startedAt = time.Now().UTC()
	}
	runID := strings.TrimSpace(input.RunID)
	if runID == "" {
		runID = fmt.Sprintf("run-%d", startedAt.UnixNano())
	}

	pipeline := input.Pipeline
	manifest := artifacts.RunManifest{
		PipelineID:      pipeline.ID,
		PipelineDigest:  pipeline.Digest,
		InputModule:     pipeline.Input.Module,
		Chunker:         pipeline.Chunk.Module,
		OutputEncoder:   pipeline.Output.Module,
		ArtifactLanes:   make([]artifacts.ArtifactLaneManifest, 0, len(pipeline.ArtifactLanes)),
		ValidatorChains: validatorChainManifests(pipeline.ValidatorChains),
		RunID:           runID,
		StartedAt:       timePtr(startedAt),
		References:      ReferenceProvenance(pipeline),
		LLMProfiles:     cloneLLMProfiles(input.LLMProfiles),
	}
	// The runner does not currently maintain a cache or idempotency key. Reference
	// digests are recorded in manifest provenance and intentionally kept separate
	// from source_digests.

	for _, lane := range pipeline.ArtifactLanes {
		laneManifest := artifacts.ArtifactLaneManifest{
			ID:         lane.ID,
			Extractor:  lane.Extract.Module,
			Merger:     lane.Merge.Module,
			Normalizer: lane.Normalize.Module,
		}
		manifest.ArtifactLanes = append(manifest.ArtifactLanes, laneManifest)
	}
	return manifest
}

func validatorChainManifests(chains []ResolvedValidatorChain) []artifacts.ValidatorChainManifest {
	if len(chains) == 0 {
		return nil
	}
	manifests := make([]artifacts.ValidatorChainManifest, 0, len(chains))
	for _, chain := range chains {
		manifest := artifacts.ValidatorChainManifest{
			Stage:      string(chain.Stage),
			LaneID:     chain.LaneID,
			ModuleKey:  chain.ModuleKey,
			Validators: make([]artifacts.ValidatorManifest, 0, len(chain.Validators)),
		}
		for _, validator := range chain.Validators {
			manifest.Validators = append(manifest.Validators, artifacts.ValidatorManifest{
				Key:            validator.Binding.Module,
				ExecutionClass: string(validator.ExecutionClass),
			})
		}
		manifests = append(manifests, manifest)
	}
	return manifests
}

func failOutput(output RunOutput) RunOutput {
	if output.Manifest.PipelineID != "" {
		populateRawOutputManifest(&output)
		output.Manifest.ValidationStatus = "failed"
		output.Manifest.CompletedAt = timePtr(time.Now().UTC())
	}
	return output
}

func recordCheckpointEvent(output *RunOutput, loader CheckpointLoader, stage string, laneID string, moduleKey string, decision CheckpointDecision) {
	if output == nil || loader == nil || !loader.Enabled() {
		return
	}
	action := "executed"
	if decision.Reused {
		action = "reused"
	}
	output.CheckpointEvents = append(output.CheckpointEvents, CheckpointEvent{
		Stage:     stage,
		LaneID:    laneID,
		ModuleKey: moduleKey,
		Action:    action,
		Reason:    decision.Reason,
	})
}

func populateRawOutputManifest(output *RunOutput) {
	if output == nil {
		return
	}
	output.Manifest.NormalizedOutputs = normalizedOutputManifests(output.NormalizeOutputs)
	output.Manifest.RejectedOutputs = rejectedOutputManifests(output.Rejected)
}

func normalizedOutputManifests(outputs []contracts.NormalizeOutput) []artifacts.NormalizedOutputManifest {
	if len(outputs) == 0 {
		return nil
	}
	manifests := make([]artifacts.NormalizedOutputManifest, 0, len(outputs))
	for _, output := range outputs {
		manifests = append(manifests, artifacts.NormalizedOutputManifest{
			LaneID:    output.LaneID,
			ModuleKey: output.NormalizerKey,
			SourceID:  output.SourceID,
			MediaType: output.Payload.MediaType,
			Schema: artifacts.OutputSchemaProvenance{
				ID:      output.Schema.ID,
				Name:    output.Schema.Name,
				Version: output.Schema.Version,
			},
		})
	}
	return manifests
}

func rejectedOutputManifests(rejected []contracts.RejectedOutput) []artifacts.RejectedOutputManifest {
	if len(rejected) == 0 {
		return nil
	}
	manifests := make([]artifacts.RejectedOutputManifest, 0, len(rejected))
	for _, output := range rejected {
		manifests = append(manifests, artifacts.RejectedOutputManifest{
			Stage:                  output.Stage,
			LaneID:                 output.LaneID,
			ModuleKey:              output.ModuleKey,
			ChunkID:                output.ChunkID,
			ChunkIndex:             output.ChunkIndex,
			ValidatorName:          output.ValidatorName,
			ReasonCode:             output.ReasonCode,
			Message:                output.Message,
			AttemptCount:           output.AttemptCount,
			DiagnosticArtifactPath: output.DiagnosticArtifactPath,
		})
	}
	return manifests
}

func setLaneManifestMetadata(output *RunOutput, laneID string, modules ...any) {
	if output == nil {
		return
	}
	for i := range output.Manifest.ArtifactLanes {
		if output.Manifest.ArtifactLanes[i].ID != laneID {
			continue
		}

		metadata := make(map[string]any)
		for _, module := range modules {
			moduleMetadata, ok := moduleManifestMetadata(module)
			if !ok {
				continue
			}
			key := manifestMetadataKey(module)
			if key == "" {
				continue
			}
			metadata[key] = moduleMetadata
		}
		if len(metadata) > 0 {
			output.Manifest.ArtifactLanes[i].Metadata = metadata
		}
		return
	}
}

func attachModuleManifestMetadata(output *RunOutput, moduleKey string, module any) {
	if output == nil {
		return
	}
	moduleMetadata, ok := moduleManifestMetadata(module)
	if !ok {
		return
	}
	if output.Manifest.ModuleMetadata == nil {
		output.Manifest.ModuleMetadata = make(map[string]map[string]any)
	}
	output.Manifest.ModuleMetadata[moduleKey] = moduleMetadata
}

func manifestMetadataKey(module any) string {
	switch module.(type) {
	case contracts.Extractor:
		return "extractor"
	case contracts.Merger:
		return "merger"
	case contracts.Normalizer:
		return "normalizer"
	default:
		return ""
	}
}

func moduleManifestMetadata(module any) (map[string]any, bool) {
	provider, ok := module.(contracts.ManifestMetadataProvider)
	if !ok {
		return nil, false
	}

	moduleMetadata := cloneMetadata(provider.ManifestMetadata())
	if len(moduleMetadata) == 0 {
		return nil, false
	}
	return moduleMetadata, true
}

func outputFilesFromResult(result contracts.OutputResult) ([]contracts.OutputFile, error) {
	out := make([]contracts.OutputFile, 0, len(result.Files))
	for _, file := range result.Files {
		if err := validateOutputFileName(file.Name); err != nil {
			return nil, err
		}
		out = append(out, contracts.OutputFile{
			Name:        file.Name,
			ContentType: file.ContentType,
			Bytes:       append([]byte(nil), file.Bytes...),
		})
	}
	return out, nil
}

func validateOutputFileName(name string) error {
	if strings.TrimSpace(name) == "" {
		return fmt.Errorf("output file name must not be empty")
	}
	if strings.Contains(name, "\\") {
		return fmt.Errorf("output file name %q must use slash-separated relative paths", name)
	}
	if path.IsAbs(name) {
		return fmt.Errorf("output file name %q must be relative", name)
	}
	if strings.Contains(name, "..") {
		return fmt.Errorf("output file name %q must not contain ..", name)
	}
	cleaned := path.Clean(name)
	if cleaned == "." || cleaned != name {
		return fmt.Errorf("output file name %q must be clean", name)
	}
	return nil
}

func cloneMetadata(metadata map[string]any) map[string]any {
	if len(metadata) == 0 {
		return nil
	}
	out := make(map[string]any, len(metadata))
	for key, value := range metadata {
		out[key] = value
	}
	return out
}

func cloneLLMProfiles(profiles []artifacts.LLMProfileManifest) []artifacts.LLMProfileManifest {
	if len(profiles) == 0 {
		return nil
	}
	return append([]artifacts.LLMProfileManifest(nil), profiles...)
}

func llmProfileManifests(client contracts.StructuredLLMClient) []artifacts.LLMProfileManifest {
	provider, ok := client.(contracts.LLMProfileManifestProvider)
	if !ok {
		return nil
	}
	return provider.LLMProfileManifests()
}

func mergeLLMProfileManifests(sources ...[]artifacts.LLMProfileManifest) []artifacts.LLMProfileManifest {
	merged := make(map[string]artifacts.LLMProfileManifest)
	for _, source := range sources {
		for _, profile := range source {
			id := strings.TrimSpace(profile.ID)
			provider := strings.TrimSpace(profile.Provider)
			model := strings.TrimSpace(profile.Model)
			key := id + "\x00" + provider + "\x00" + model
			if _, exists := merged[key]; exists {
				continue
			}
			merged[key] = artifacts.LLMProfileManifest{
				ID:       id,
				Provider: provider,
				Model:    model,
			}
		}
	}
	if len(merged) == 0 {
		return nil
	}
	keys := make([]string, 0, len(merged))
	for key := range merged {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	out := make([]artifacts.LLMProfileManifest, 0, len(keys))
	for _, key := range keys {
		out = append(out, merged[key])
	}
	return out
}

func sourceInputMaterial(inputPath string, content []byte) contracts.LLMInputMaterial {
	return contracts.NewLLMInputMaterial(
		"source",
		sourceInputMediaType(inputPath),
		content,
		sourceInputDigest(content),
		sourceInputOriginURI(inputPath),
	)
}

func chunkInputMaterial(sourceInput contracts.LLMInputMaterial, chunk contracts.SourceChunk) contracts.LLMInputMaterial {
	return contracts.NewLLMInputMaterial(
		"source",
		chunk.MediaType,
		chunk.Content,
		sourceInputDigest(chunk.Content),
		sourceInput.OriginURI,
	)
}

func sourceInputMediaType(inputPath string) string {
	extension := strings.ToLower(filepath.Ext(strings.TrimSpace(inputPath)))
	if extension == ".json" {
		return "application/json"
	}
	mediaType := mime.TypeByExtension(extension)
	if strings.TrimSpace(mediaType) == "" {
		return unknownMediaType
	}
	return canonicalMediaType(mediaType)
}

func sourceInputDigest(content []byte) string {
	sum := sha256.Sum256(content)
	return "sha256:" + hex.EncodeToString(sum[:])
}

func sourceInputOriginURI(inputPath string) string {
	if strings.TrimSpace(inputPath) == "" {
		return ""
	}
	return fileURI(inputPath)
}

func resolvedSessionID(explicit string, sourceDocumentID string) string {
	if trimmed := strings.TrimSpace(explicit); trimmed != "" {
		return trimmed
	}
	return strings.TrimSpace(sourceDocumentID)
}

func manifestMetadataWithSessionID(metadata map[string]any, sessionID string) map[string]any {
	out := cloneMetadata(metadata)
	if strings.TrimSpace(sessionID) == "" {
		return out
	}
	if out == nil {
		out = make(map[string]any)
	}
	out["session_id"] = sessionID
	return out
}

func cloneWarnings(warnings []contracts.Warning) []contracts.Warning {
	if len(warnings) == 0 {
		return nil
	}
	return append([]contracts.Warning(nil), warnings...)
}

func cloneRawPayload(payload contracts.RawPayload) contracts.RawPayload {
	return contracts.RawPayload{
		Content:   append([]byte(nil), payload.Content...),
		MediaType: payload.MediaType,
		Metadata:  cloneMetadata(payload.Metadata),
		Warnings:  cloneWarnings(payload.Warnings),
	}
}

func cloneResponseSchema(schema contracts.ResponseSchema) contracts.ResponseSchema {
	schema.JSONSchema = append([]byte(nil), schema.JSONSchema...)
	return schema
}

func cloneSourceChunkPtr(chunk *contracts.SourceChunk) *contracts.SourceChunk {
	if chunk == nil {
		return nil
	}
	cloned := cloneSourceChunk(*chunk)
	return &cloned
}

func cloneSourceChunk(chunk contracts.SourceChunk) contracts.SourceChunk {
	chunk.Content = append([]byte(nil), chunk.Content...)
	chunk.Units = cloneSourceUnits(chunk.Units)
	chunk.Metadata = cloneMetadata(chunk.Metadata)
	return chunk
}

func cloneSourceChunks(chunks []contracts.SourceChunk) []contracts.SourceChunk {
	if len(chunks) == 0 {
		return nil
	}
	out := make([]contracts.SourceChunk, 0, len(chunks))
	for _, chunk := range chunks {
		out = append(out, cloneSourceChunk(chunk))
	}
	return out
}

func cloneSourceUnits(units []source.SourceUnit) []source.SourceUnit {
	if len(units) == 0 {
		return nil
	}
	out := make([]source.SourceUnit, 0, len(units))
	for _, unit := range units {
		out = append(out, cloneSourceUnit(unit))
	}
	return out
}

func cloneExtractOutput(output contracts.ExtractOutput) contracts.ExtractOutput {
	output.Schema = cloneResponseSchema(output.Schema)
	output.Payload = cloneRawPayload(output.Payload)
	return output
}

func cloneExtractOutputs(outputs []contracts.ExtractOutput) []contracts.ExtractOutput {
	if len(outputs) == 0 {
		return nil
	}
	out := make([]contracts.ExtractOutput, 0, len(outputs))
	for _, output := range outputs {
		out = append(out, cloneExtractOutput(output))
	}
	return out
}

func cloneMergeOutput(output contracts.MergeOutput) contracts.MergeOutput {
	output.Schema = cloneResponseSchema(output.Schema)
	output.Payload = cloneRawPayload(output.Payload)
	return output
}

func cloneNormalizeOutput(output contracts.NormalizeOutput) contracts.NormalizeOutput {
	output.Schema = cloneResponseSchema(output.Schema)
	output.Payload = cloneRawPayload(output.Payload)
	return output
}

func cloneNormalizeOutputs(outputs []contracts.NormalizeOutput) []contracts.NormalizeOutput {
	if len(outputs) == 0 {
		return nil
	}
	out := make([]contracts.NormalizeOutput, 0, len(outputs))
	for _, output := range outputs {
		out = append(out, cloneNormalizeOutput(output))
	}
	return out
}

func cloneRejectedOutputs(rejected []contracts.RejectedOutput) []contracts.RejectedOutput {
	if len(rejected) == 0 {
		return nil
	}
	return append([]contracts.RejectedOutput(nil), rejected...)
}

func timePtr(t time.Time) *time.Time {
	return &t
}