notarius/internal/framework/pipeline/runner.go

package pipeline

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"mime"
	"path"
	"path/filepath"
	"sort"
	"strings"
	"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/chunkmap"
	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)

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

type Runner struct{}

func New() *Runner {
	return &Runner{}
}

type RunInput struct {
	Prepared         *PreparedPipeline
	SourceID         string
	Path             string
	RawInput         []byte
	SessionID        string
	RunID            string
	StartedAt        time.Time
	LLMProfiles      []artifacts.LLMProfileManifest
	Metadata         map[string]any
	Diagnostics      []contracts.ProducerDiagnostic
	ChunkCacheMode   ChunkCacheMode
	ChunkPlans       ChunkPlanStore
	Checkpoints      CheckpointRecorder
	Checkpoint       CheckpointLoader
	CheckpointPolicy CheckpointExecutionPolicy
	Debug            DebugRecorder
	// ExtractWorkers bounds run-wide extract jobs. Values less than one use a
	// single worker so direct framework callers retain deterministic behavior.
	ExtractWorkers int

	pipeline                  ResolvedPipeline
	llmClient                 contracts.StructuredLLMClient
	stepID                    string
	references                map[referenceTargetKey]contracts.ReferenceSet
	referenceReuseEligibility map[referenceTargetKey]bool
}

type RunOutput struct {
	Manifest            artifacts.RunManifest          `json:"manifest"`
	ChunkPlan           *artifacts.ChunkPlanSummary    `json:"chunk_plan,omitempty"`
	NormalizeOutputs    []contracts.SerializedOutput   `json:"normalize_outputs,omitempty"`
	Rejected            []contracts.RejectedOutput     `json:"rejected,omitempty"`
	Diagnostics         contracts.DiagnosticCollection `json:"diagnostics,omitempty"`
	OutputFiles         []contracts.OutputFile         `json:"-"`
	CheckpointEvents    []CheckpointEvent              `json:"checkpoint_events,omitempty"`
	ValidationSummaries []artifacts.ValidationSummary  `json:"validation_summaries,omitempty"`

	normalizeReuseEligibility map[generatedOutputKey]bool
	diagnosticGroups          []contracts.DiagnosticGroup
}

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 := ctx.Err(); err != nil {
		return output, err
	}
	if err := validateRunInput(input); err != nil {
		return output, err
	}
	input.Metadata, err = cloneMetadata(input.Metadata)
	if err != nil {
		return output, fmt.Errorf("clone run metadata: %w", err)
	}
	input.pipeline = input.Prepared.resolved
	input.llmClient = input.Prepared.dependencies.LLM

	output.Manifest = manifestFromPipeline(input)
	sessionID := strings.TrimSpace(input.SessionID)
	output.Manifest.Metadata, err = manifestMetadataWithSessionID(output.Manifest.Metadata, sessionID)
	if err != nil {
		return failOutput(output), err
	}
	output.ChunkPlan = &artifacts.ChunkPlanSummary{
		Mode: string(effectiveChunkCacheMode(input.ChunkCacheMode)), RequestedModule: input.pipeline.Chunk.Module,
		LookupStatus: "skipped", LookupReason: "chunk plan lookup skipped",
		ValidationStatus: "not_run", PublicationStatus: "not_requested",
	}
	checkpoints := input.Checkpoints
	if checkpoints == nil {
		checkpoints = NoopCheckpointRecorder()
	}
	checkpointLoader := input.Checkpoint
	if checkpointLoader == nil {
		checkpointLoader = NoopCheckpointLoader()
	}
	debugRecorder := input.Debug
	if debugRecorder == nil {
		debugRecorder = NoopDebugRecorder()
	}
	checkpoints = synchronizedCheckpointRecorder(checkpoints)
	checkpointLoader = synchronizedCheckpointLoader(checkpointLoader)
	debugRecorder = synchronizedDebugRecorder(debugRecorder)
	input.Debug = debugRecorder
	input.llmClient = wrapDebugLLMClient(input.llmClient, debugRecorder)
	defer func() {
		output.Manifest.LLMProfiles = mergeLLMProfileManifests(input.LLMProfiles, llmProfileManifests(input.llmClient))
	}()
	inputDiagnostics, diagnosticErr := promoteProducerDiagnostics(input.Diagnostics, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageReferences}, nil)
	if diagnosticErr != nil {
		return failOutput(output), fmt.Errorf("promote input diagnostics: %w", diagnosticErr)
	}
	appendDiagnosticGroups(&output, inputDiagnostics)
	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 := input.Prepared.input
	if err := attachModuleManifestMetadata(&output, "input", adapter); err != nil {
		return failOutput(output), err
	}
	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)
		}
		requestMetadata, metadataErr := cloneMetadata(input.Metadata)
		if metadataErr != nil {
			return failOutput(output), fmt.Errorf("clone input adapter metadata: %w", metadataErr)
		}
		if err := ctx.Err(); err != nil {
			_ = checkpoints.SourceFailed(adapter.Key(), err)
			return failOutput(output), err
		}
		doc, err = adapter.Parse(ctx, contracts.ParseRequest{
			SourceID:                       input.SourceID,
			Path:                           input.Path,
			Raw:                            input.RawInput,
			LLMProfile:                     input.pipeline.Input.LLMProfile,
			StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(input.pipeline.Input.StructuredOutputRepairAttempts),
			Metadata:                       requestMetadata,
		})
		if ctxErr := ctx.Err(); ctxErr != nil {
			_ = checkpoints.SourceFailed(adapter.Key(), ctxErr)
			return failOutput(output), ctxErr
		}
		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)
	output.Manifest.SourceDigests = []string{doc.Digest}

	chunker := input.Prepared.chunker
	if err := attachModuleManifestMetadata(&output, "chunker", chunker); err != nil {
		return failOutput(output), err
	}
	chunkStarted := time.Now().UTC()
	chunkMode := effectiveChunkCacheMode(input.ChunkCacheMode)
	if err := writeDebugTimed(debugRecorder, "chunk/input.json", debugTimedEnvelope{
		Stage:     string(StageChunk),
		ModuleKey: chunker.Key(),
		StartedAt: chunkStarted,
		Payload: map[string]any{
			"cache_mode":   chunkMode,
			"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)
	}
	chunkResult, err := r.runChunkPlan(ctx, input, doc, sourceInput, sessionID)
	if applyErr := applyChunkPlanExecution(&output, chunkResult); applyErr != nil {
		return failOutput(output), applyErr
	}
	if err != nil {
		return failOutput(output), err
	}
	if chunkResult.rejection != nil {
		output.Rejected = append(output.Rejected, *chunkResult.rejection)
	}
	if chunkResult.validation != nil {
		output.ValidationSummaries = append(output.ValidationSummaries, artifacts.CloneValidationSummary(*chunkResult.validation))
	}
	appendDiagnosticGroups(&output, chunkResult.diagnostics)
	chunkDebugPayload := map[string]any{
		"cache_mode":          chunkMode,
		"lookup":              chunkResult.lookup,
		"accepted":            chunkResult.accepted,
		"materialized_chunks": debugSourceChunkEnvelopes(chunkResult.chunks),
	}
	if chunkResult.plan != nil {
		chunkDebugPayload["plan"] = debugChunkPlanEnvelope(*chunkResult.plan)
	}
	if chunkResult.rejection != nil {
		chunkDebugPayload["rejection"] = debugRejectedOutputEnvelope(*chunkResult.rejection)
	}
	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)
	}

	var acceptedChunkMap *contracts.SerializedArtifact
	if chunkResult.accepted {
		if chunkResult.plan == nil || chunkResult.record == nil {
			return failOutput(output), fmt.Errorf("accepted chunk plan is missing plan or producer record")
		}
		artifact, buildErr := chunkmap.Serialize(chunkmap.BuildRequest{
			Source:           doc,
			Plan:             *chunkResult.plan,
			Chunks:           chunkResult.chunks,
			RequestedChunker: input.pipeline.Chunk.Module,
			Producer: chunkmap.Producer{
				InputModule: chunkResult.record.Producer.InputModule,
				ChunkModule: chunkResult.record.Producer.ChunkModule,
				LLMProfile:  chunkResult.record.Producer.LLMProfile,
			},
		})
		if buildErr != nil {
			return failOutput(output), fmt.Errorf("serialize accepted chunk map: %w", buildErr)
		}
		acceptedChunkMap = &artifact
	}

	if chunkResult.accepted {
		if err := r.runPreparedSteps(ctx, input, checkpoints, checkpointLoader, doc, sourceInput, sessionID, chunkResult.chunks, &output); err != nil {
			return failOutput(output), err
		}
	}
	if err := ctx.Err(); err != nil {
		return failOutput(output), err
	}

	if len(output.Rejected) > 0 {
		output.Manifest.ValidationStatus = "rejected"
	} else if hasIncompleteValidation(output.ValidationSummaries) {
		output.Manifest.ValidationStatus = "incomplete"
	} else {
		output.Manifest.ValidationStatus = "approved"
	}
	populateOutputManifest(&output)
	output.Manifest.CompletedAt = timePtr(time.Now().UTC())
	if err := finalizeDiagnostics(&output); err != nil {
		return failOutput(output), fmt.Errorf("aggregate diagnostics: %w", err)
	}

	encoder := input.Prepared.output
	if err := attachModuleManifestMetadata(&output, "output", encoder); err != nil {
		return failOutput(output), err
	}
	if err := ctx.Err(); err != nil {
		return failOutput(output), err
	}
	outputStarted := time.Now().UTC()
	evidenceArtifact, evidenceSummary, err := buildOutputEvidenceContext(input.Prepared, doc, output.NormalizeOutputs)
	if err != nil {
		return failOutput(output), err
	}
	if err := ctx.Err(); err != nil {
		return failOutput(output), err
	}
	if evidenceSummary != nil {
		if err := writeDebugTimed(debugRecorder, "output/evidence-context.json", debugTimedEnvelope{
			Stage:     string(StageOutput),
			ModuleKey: encoder.Key(),
			StartedAt: outputStarted,
			Payload:   *evidenceSummary,
		}); err != nil {
			return failOutput(output), fmt.Errorf("write evidence context debug artifact: %w", err)
		}
		if err := ctx.Err(); err != nil {
			return failOutput(output), err
		}
	}
	outputDebugPayload := map[string]any{
		"manifest":          output.Manifest,
		"normalize_outputs": debugSerializedOutputEnvelopes(output.NormalizeOutputs),
		"rejected":          debugRejectedOutputEnvelopes(output.Rejected),
		"options":           redactSensitiveMap(input.pipeline.Output.Options),
		"metadata":          redactSensitiveMap(input.Metadata),
	}
	if acceptedChunkMap != nil {
		outputDebugPayload["chunk_map"] = debugSerializedOutputEnvelope(contracts.SerializedOutput{Artifact: *acceptedChunkMap})
	}
	if err := writeDebugTimed(debugRecorder, "output/input.json", debugTimedEnvelope{
		Stage:     string(StageOutput),
		ModuleKey: encoder.Key(),
		StartedAt: outputStarted,
		Payload:   outputDebugPayload,
	}); err != nil {
		return failOutput(output), fmt.Errorf("write output debug artifact: %w", err)
	}
	if err := ctx.Err(); err != nil {
		return failOutput(output), err
	}
	outputMetadata, err := cloneMetadata(input.Metadata)
	if err != nil {
		return failOutput(output), fmt.Errorf("clone output encoder metadata: %w", err)
	}
	if err := ctx.Err(); err != nil {
		return failOutput(output), err
	}
	encoded, err := encoder.Encode(ctx, contracts.OutputRequest{
		Manifest:                       output.Manifest,
		NormalizeOutputs:               cloneSerializedOutputs(output.NormalizeOutputs),
		Rejected:                       cloneRejectedOutputs(output.Rejected),
		Diagnostics:                    contracts.CloneDiagnosticCollection(output.Diagnostics),
		LLMProfile:                     input.pipeline.Output.LLMProfile,
		StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(input.pipeline.Output.StructuredOutputRepairAttempts),
		Metadata:                       outputMetadata,
		ChunkMap:                       contracts.CloneSerializedArtifactPointer(acceptedChunkMap),
		EvidenceContext:                contracts.CloneSerializedArtifactPointer(evidenceArtifact),
	})
	if ctxErr := ctx.Err(); ctxErr != nil {
		return failOutput(output), ctxErr
	}
	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)
	}
	if err := writeDebugTimed(debugRecorder, "output/output.json", debugTimedEnvelope{
		Stage:     string(StageOutput),
		ModuleKey: encoder.Key(),
		StartedAt: outputStarted,
		Payload: map[string]any{
			"files": debugOutputFiles(files),
		},
	}); err != nil {
		return failOutput(output), fmt.Errorf("write output debug artifact: %w", err)
	}
	if err := ctx.Err(); err != nil {
		return failOutput(output), err
	}
	output.OutputFiles = files

	return output, nil
}

func hasIncompleteValidation(summaries []artifacts.ValidationSummary) bool {
	for _, summary := range summaries {
		if summary.Status == "incomplete" {
			return true
		}
	}
	return false
}

func (r *Runner) runPreparedSteps(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, output *RunOutput) error {
	for _, step := range input.Prepared.Steps {
		stepInput := input
		stepInput.stepID = step.ID
		stepReferences, referenceProvenance, err := buildStepReferenceSets(input, step, output.NormalizeOutputs)
		if err != nil {
			return fmt.Errorf("prepare generated references for pipeline step %q: %w", step.ID, err)
		}
		stepInput.references = stepReferences
		stepInput.referenceReuseEligibility = buildStepReferenceReuseEligibility(step, output.normalizeReuseEligibility)
		output.Manifest.References = append(output.Manifest.References, referenceProvenance...)
		laneOutput, laneErr := r.runLanes(ctx, stepInput, step, checkpoints, loader, doc, sourceInput, sessionID, chunks)
		if err := mergeLaneOutput(output, laneOutput); err != nil {
			return err
		}
		if laneErr != nil {
			return fmt.Errorf("execute pipeline step %q: %w", step.ID, laneErr)
		}
	}
	return nil
}

type retryAttemptResult struct {
	accepted  bool
	rejection *contracts.RejectedOutput
}

func runSimpleRetry(ctx context.Context, retries int, run func(attempt int) (retryAttemptResult, error)) (retryAttemptResult, error) {
	attempts := retries + 1
	var last retryAttemptResult
	for attempt := 1; attempt <= attempts; attempt++ {
		if err := ctx.Err(); err != nil {
			return retryAttemptResult{}, err
		}
		result, err := run(attempt)
		if err != nil {
			var debugErr *attemptDebugPersistenceError
			if errors.As(err, &debugErr) {
				return retryAttemptResult{}, fmt.Errorf("failed after %d attempt(s): %w", attempt, err)
			}
			if attempt == attempts {
				return retryAttemptResult{}, fmt.Errorf("failed after %d attempt(s): %w", attempt, err)
			}
			if ctxErr := ctx.Err(); ctxErr != nil {
				return retryAttemptResult{}, ctxErr
			}
			continue
		}
		if result.accepted {
			return result, nil
		}
		if result.rejection != nil {
			rejection := *result.rejection
			rejection.AttemptCount = attempt
			last = retryAttemptResult{rejection: &rejection}
		}
		if ctxErr := ctx.Err(); ctxErr != nil {
			return retryAttemptResult{}, ctxErr
		}
		if attempt == attempts {
			if last.rejection == nil {
				last.rejection = &contracts.RejectedOutput{ReasonCode: "output_rejected", Message: "output rejected", AttemptCount: attempt}
			}
			return last, nil
		}
	}
	return last, nil
}

func (r *Runner) validateChunkReport(ctx context.Context, doc *source.SourceDocument, moduleKey string, chunks []source.Chunk, sourceInput contracts.LLMInputMaterial, sessionID string, references contracts.ReferenceSet, metadata map[string]any, prepared preparedValidatorChain, attempt int, debug DebugRecorder) (validationReport, error) {
	content, err := json.Marshal(chunks)
	if err != nil {
		return validationReport{}, fmt.Errorf("encode canonical chunks for validation: %w", err)
	}
	schema := contracts.ArtifactSchema{ID: "notarius.source.chunks", Name: "notarius_source_chunks", Version: "v1", JSONSchema: []byte(`{"$schema":"https://json-schema.org/draft/2020-12/schema","type":"array"}`)}
	report, err := executeValidationChain(ctx, prepared, func(validatorCtx context.Context, item preparedValidator, validatorAttempt int) (validationInvocation, error) {
		binding := item.resolved.Binding
		started := time.Now().UTC()
		attemptPath := validatorAttemptPath(path.Join("validate", fileio.EncodePathComponent(string(StageChunk)), "", fileio.EncodePathComponent(moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d", item.position, fileio.EncodePathComponent(binding.Module), attempt)), validatorAttempt)
		validatorCtx, llmScope := withIsolatedDebugLLMScope(validatorCtx, attemptPath)
		var result contracts.ValidationResult
		requestMetadata, cloneErr := cloneMetadata(metadata)
		if cloneErr != nil {
			return validationInvocation{}, fatalValidationError(fmt.Errorf("clone chunk validation metadata: %w", cloneErr))
		}
		requestChunks, cloneErr := cloneSourceChunks(chunks)
		if cloneErr != nil {
			return validationInvocation{}, fatalValidationError(fmt.Errorf("clone chunks for validation: %w", cloneErr))
		}
		switch item.resolved.Target {
		case ValidatorTargetChunk:
			result, err = item.chunk.Validate(validatorCtx, contracts.ChunkValidationRequest{ModuleKey: moduleKey, Source: doc, SourceID: doc.ID, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(references), LLMProfile: binding.LLMProfile, StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(binding.StructuredOutputRepairAttempts), Metadata: requestMetadata, Chunks: requestChunks})
		case ValidatorTargetSerialized:
			result, err = item.serialized.Validate(validatorCtx, contracts.SerializedValidationRequest{Stage: string(StageChunk), ModuleKey: moduleKey, Source: doc, SourceID: doc.ID, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(references), LLMProfile: binding.LLMProfile, StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(binding.StructuredOutputRepairAttempts), Metadata: requestMetadata, Chunks: requestChunks, Schema: contracts.CloneArtifactSchema(schema), MediaType: "application/json", Content: append([]byte(nil), content...)})
		default:
			return validationInvocation{}, fatalValidationError(fmt.Errorf("validator %q is incompatible with chunk validation", binding.Module))
		}
		if err == nil {
			err = contracts.ValidateValidationResult(result)
		}
		debugContent := debugContentEnvelope(content, "application/json", nil, nil)
		debugContent.ContentDigest = debugContentDigest(content)
		debugCall := debugValidationCall{ValidatorName: binding.Module, Request: map[string]any{"stage": string(StageChunk), "module_key": moduleKey, "source_id": doc.ID, "schema": schema, "schema_digest": contracts.DigestArtifactSchema(schema), "content": debugContent, "metadata": redactSensitiveMap(metadata)}, Result: debugValidationResultEnvelope(result)}
		if err != nil {
			debugCall.Error = err.Error()
		}
		if err != nil {
			validationErr := fmt.Errorf("validate chunks with validator %q: %w", binding.Module, err)
			if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope); debugErr != nil {
				return validationInvocation{}, fatalValidationError(errors.Join(validationErr, fmt.Errorf("write chunk validator attempt debug artifact: %w", debugErr)))
			}
			return validationInvocation{}, validationErr
		}
		if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall}, llmScope); debugErr != nil {
			return validationInvocation{}, fatalValidationError(fmt.Errorf("write chunk validator attempt debug artifact: %w", debugErr))
		}
		return validationInvocation{result: result}, nil
	})
	if err != nil {
		return report, err
	}
	return report, nil
}

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 validateRunInput(input RunInput) error {
	if input.Prepared == nil {
		return fmt.Errorf("prepared pipeline must not be nil")
	}
	mode := effectiveChunkCacheMode(input.ChunkCacheMode)
	if err := mode.Validate(); err != nil {
		return err
	}
	if mode != ChunkCacheBypass && input.ChunkPlans == nil {
		return fmt.Errorf("chunk plan store is required for %q mode", mode)
	}
	if err := validateResolvedPipeline(input.Prepared.resolved); err != nil {
		return err
	}
	return nil
}

func validateResolvedPipeline(pipeline ResolvedPipeline) error {
	if pipeline.ID == "" {
		return fmt.Errorf("resolved pipeline id must not be empty")
	}
	if pipeline.Digest == "" {
		return fmt.Errorf("resolved pipeline digest must not be empty")
	}
	if pipeline.Input.Module == "" {
		return fmt.Errorf("resolved pipeline input module must not be empty")
	}
	if pipeline.Chunk.Module == "" {
		return fmt.Errorf("resolved pipeline chunk module must not be empty")
	}
	if pipeline.Output.Module == "" {
		return fmt.Errorf("resolved pipeline output module must not be empty")
	}
	if len(pipeline.Steps) == 0 {
		return fmt.Errorf("resolved pipeline artifact lanes must not be empty")
	}
	for _, lane := range pipeline.AllArtifactLanes() {
		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,
		ChunkPlan: &artifacts.ChunkPlanManifest{
			Mode:            string(effectiveChunkCacheMode(input.ChunkCacheMode)),
			RequestedModule: pipeline.Chunk.Module,
		},
		OutputEncoder:   pipeline.Output.Module,
		ArtifactLanes:   make([]artifacts.ArtifactLaneManifest, 0, len(pipeline.AllArtifactLanes())),
		ValidatorChains: validatorChainManifests(pipeline.ValidatorChains),
		RunID:           runID,
		StartedAt:       timePtr(startedAt),
		References:      ReferenceProvenance(pipeline),
		LLMProfiles:     cloneLLMProfiles(input.LLMProfiles),
	}
	for _, lane := range pipeline.AllArtifactLanes() {
		laneManifest := artifacts.ArtifactLaneManifest{
			StepID:     lane.StepID,
			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 {
	// Preserve any diagnostics already accepted by the pipeline when a later
	// operation fails. Producers and validators validate each diagnostic before
	// it is appended, so finalization here cannot introduce a new failure path.
	_ = finalizeDiagnostics(&output)
	if output.Manifest.PipelineID != "" {
		populateOutputManifest(&output)
		output.Manifest.ValidationStatus = "failed"
		output.Manifest.CompletedAt = timePtr(time.Now().UTC())
	}
	return output
}

func recordCheckpointEvent(output *RunOutput, loader CheckpointLoader, stage string, stepID string, laneID string, moduleKey string, decision CheckpointDecision) {
	if output == nil || loader == nil || !loader.Enabled() {
		return
	}
	action := checkpointDecisionCategory(decision)
	output.CheckpointEvents = append(output.CheckpointEvents, CheckpointEvent{
		Stage:      stage,
		StepID:     stepID,
		LaneID:     laneID,
		ModuleKey:  moduleKey,
		Action:     action,
		Category:   checkpointDecisionCategory(decision),
		ReasonCode: decision.ReasonCode,
		Detail:     decision.Detail,
		Reason:     decision.Reason,
	})
}

func checkpointDecisionCategory(decision CheckpointDecision) CheckpointDecisionCategory {
	if decision.Category != "" {
		return decision.Category
	}
	if decision.Reused {
		return CheckpointDecisionReused
	}
	return CheckpointDecisionExecuted
}

func populateOutputManifest(output *RunOutput) {
	if output == nil {
		return
	}
	output.Manifest.NormalizedOutputs = normalizedOutputManifests(output.NormalizeOutputs)
	output.Manifest.RejectedOutputs = rejectedOutputManifests(output.Rejected)
	if len(output.ValidationSummaries) > 0 {
		output.Manifest.ValidationSummaries = cloneValidationSummaries(output.ValidationSummaries)
	}
	if len(output.CheckpointEvents) > 0 {
		decisions := make([]artifacts.CheckpointDecisionManifest, 0, len(output.CheckpointEvents))
		for _, event := range output.CheckpointEvents {
			decisions = append(decisions, artifacts.CheckpointDecisionManifest{Stage: event.Stage, StepID: event.StepID, LaneID: event.LaneID, ModuleKey: event.ModuleKey, Category: string(event.Category), ReasonCode: string(event.ReasonCode), Detail: event.Detail})
		}
		output.Manifest.CheckpointDecisions = decisions
	}
}

func normalizedOutputManifests(outputs []contracts.SerializedOutput) []artifacts.NormalizedOutputManifest {
	if len(outputs) == 0 {
		return nil
	}
	manifests := make([]artifacts.NormalizedOutputManifest, 0, len(outputs))
	for _, output := range outputs {
		manifests = append(manifests, artifacts.NormalizedOutputManifest{
			StepID:    output.StepID,
			LaneID:    output.LaneID,
			ModuleKey: output.NormalizerKey,
			SourceID:  output.SourceID,
			MediaType: output.Artifact.MediaType,
			Schema: artifacts.OutputSchemaProvenance{
				ID:      output.Artifact.Schema.ID,
				Name:    output.Artifact.Schema.Name,
				Version: output.Artifact.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,
			StepID:                 output.StepID,
			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,
			Validation:             cloneValidationSummaryPtr(output.Validation),
		})
	}
	return manifests
}

func attachModuleManifestMetadata(output *RunOutput, moduleKey string, module any) error {
	if output == nil {
		return nil
	}
	metadata, ok, err := moduleManifestMetadata(module)
	if err != nil {
		return fmt.Errorf("clone manifest metadata for module %q: %w", moduleKey, err)
	}
	if !ok {
		return nil
	}
	if output.Manifest.ModuleMetadata == nil {
		output.Manifest.ModuleMetadata = make(map[string]map[string]any)
	}
	output.Manifest.ModuleMetadata[moduleKey] = metadata
	return nil
}

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

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

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, error) {
	return source.CloneMetadata(metadata)
}

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 {
			profile = profile.Normalized()
			key := profile.IdentityKey()
			if _, exists := merged[key]; exists {
				continue
			}
			merged[key] = profile
		}
	}
	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 source.Chunk) 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 manifestMetadataWithSessionID(metadata map[string]any, sessionID string) (map[string]any, error) {
	out, err := cloneMetadata(metadata)
	if err != nil {
		return nil, fmt.Errorf("clone run manifest metadata: %w", err)
	}
	if strings.TrimSpace(sessionID) == "" {
		return out, nil
	}
	if out == nil {
		out = make(map[string]any)
	}
	out["session_id"] = sessionID
	return out, nil
}

func cloneSourceChunkPtr(chunk *source.Chunk) (*source.Chunk, error) {
	if chunk == nil {
		return nil, nil
	}
	cloned, err := cloneSourceChunk(*chunk)
	if err != nil {
		return nil, err
	}
	return &cloned, nil
}

func cloneSourceChunk(chunk source.Chunk) (source.Chunk, error) {
	chunk.Content = append([]byte(nil), chunk.Content...)
	var err error
	chunk.Units, err = cloneSourceUnits(chunk.Units)
	if err != nil {
		return source.Chunk{}, err
	}
	chunk.Metadata, err = cloneMetadata(chunk.Metadata)
	if err != nil {
		return source.Chunk{}, fmt.Errorf("clone chunk metadata: %w", err)
	}
	chunk.Annotations = source.CloneChunkAnnotations(chunk.Annotations)
	chunk.PlanAnnotations = source.CloneChunkAnnotations(chunk.PlanAnnotations)
	return chunk, nil
}

func cloneSourceChunks(chunks []source.Chunk) ([]source.Chunk, error) {
	if len(chunks) == 0 {
		return nil, nil
	}
	out := make([]source.Chunk, 0, len(chunks))
	for _, chunk := range chunks {
		cloned, err := cloneSourceChunk(chunk)
		if err != nil {
			return nil, err
		}
		out = append(out, cloned)
	}
	return out, nil
}

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

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

func cloneRejectedOutputs(rejected []contracts.RejectedOutput) []contracts.RejectedOutput {
	if len(rejected) == 0 {
		return nil
	}
	cloned := make([]contracts.RejectedOutput, len(rejected))
	for index, item := range rejected {
		cloned[index] = item
		cloned[index].Validation = cloneValidationSummaryPtr(item.Validation)
	}
	return cloned
}

func cloneValidationSummaryPtr(summary *artifacts.ValidationSummary) *artifacts.ValidationSummary {
	if summary == nil {
		return nil
	}
	cloned := artifacts.CloneValidationSummary(*summary)
	return &cloned
}

func cloneValidationSummaries(summaries []artifacts.ValidationSummary) []artifacts.ValidationSummary {
	if len(summaries) == 0 {
		return nil
	}
	cloned := make([]artifacts.ValidationSummary, len(summaries))
	for index, summary := range summaries {
		cloned[index] = artifacts.CloneValidationSummary(summary)
	}
	return cloned
}

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