notarius/internal/framework/pipeline/runner.go

package pipeline

import (
	"context"
	"fmt"
	"path"
	"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"
	"gitea.maximumdirect.net/eric/notarius/internal/framework/validate"
)

type Registries struct {
	Inputs      *InputAdapterRegistry
	Chunkers    *ChunkerRegistry
	Extractors  *ExtractorRegistry
	Mergers     *MergerRegistry
	Normalizers *NormalizerRegistry
	Validators  *ValidatorRegistry
	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
	RunID       string
	StartedAt   time.Time
	LLMProfiles []artifacts.LLMProfileManifest
	Metadata    map[string]any
	Warnings    []contracts.Warning
}

type RunOutput struct {
	Manifest    artifacts.RunManifest        `json:"manifest"`
	Approved    []artifacts.Artifact         `json:"approved,omitempty"`
	Rejected    []artifacts.RejectedArtifact `json:"rejected,omitempty"`
	Warnings    []contracts.Warning          `json:"warnings,omitempty"`
	OutputFiles []contracts.OutputFile       `json:"-"`
}

func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
	var output RunOutput
	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.Warnings = append(output.Warnings, cloneWarnings(input.Warnings)...)
	output.Manifest = manifestFromPipeline(input)

	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)
	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 {
		return failOutput(output), fmt.Errorf("parse input with adapter %q: %w", adapter.Key(), err)
	}
	if err := source.ValidateDocument(doc); err != nil {
		return failOutput(output), fmt.Errorf("validate source document: %w", err)
	}
	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)
	chunkResult, err := chunker.Chunk(ctx, contracts.ChunkRequest{
		Source:     doc,
		References: CloneReferenceSet(input.Pipeline.ChunkReferences.ReferenceSet),
		LLMClient:  input.LLMClient,
		LLMProfile: input.Pipeline.Chunk.LLMProfile,
		Options:    cloneOptions(input.Pipeline.Chunk.Options),
		Metadata:   input.Metadata,
	})
	output.Warnings = append(output.Warnings, chunkResult.Warnings...)
	if err != nil {
		return failOutput(output), fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), err)
	}
	if len(chunkResult.Chunks) == 0 {
		return failOutput(output), fmt.Errorf("chunker %q returned no chunks", chunker.Key())
	}
	canonicalChunks, err := validateAndCanonicalizeChunkResult(doc, chunkResult.Chunks)
	if err != nil {
		return failOutput(output), fmt.Errorf("validate chunks from chunker %q: %w", chunker.Key(), err)
	}

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

	if len(output.Rejected) > 0 {
		output.Manifest.ValidationStatus = "rejected"
	} else {
		output.Manifest.ValidationStatus = "approved"
	}
	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)
	encoded, err := encoder.Encode(ctx, contracts.OutputRequest{
		Manifest:   output.Manifest,
		Approved:   output.Approved,
		Rejected:   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

	return output, nil
}

func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.SourceDocument, chunks []contracts.SourceChunk, lane ResolvedArtifactLane, output *RunOutput, nextCandidateIndex *int) 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)

	var validators []validatorExecution
	if len(lane.Validators) > 0 {
		validators, err = r.buildConfiguredValidators(lane)
		if err != nil {
			return err
		}
	} else {
		for _, validator := range extractor.Validators() {
			validators = append(validators, validatorExecution{validator: validator})
		}
	}

	chunkArtifacts := make([]contracts.ChunkArtifacts, 0, len(chunks))
	for index := range chunks {
		chunk := chunks[index]
		result, err := extractor.Extract(ctx, contracts.ExtractionRequest{
			Source:     doc,
			Chunk:      &chunk,
			References: CloneReferenceSet(lane.ExtractReferences.ReferenceSet),
			LLMClient:  input.LLMClient,
			LLMProfile: lane.Extract.LLMProfile,
			Options:    cloneOptions(lane.Extract.Options),
			Metadata:   input.Metadata,
		})
		output.Warnings = append(output.Warnings, result.Warnings...)
		if err != nil {
			return fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, extractor.Key(), err)
		}

		candidates, err := normalizeCandidates(extractor, result.Candidates, nextCandidateIndex)
		if err != nil {
			return err
		}
		chunkArtifacts = append(chunkArtifacts, contracts.ChunkArtifacts{
			Chunk:      chunk,
			Candidates: candidates,
		})
	}

	mergeResult, err := merger.Merge(ctx, contracts.MergeRequest{
		Source:         doc,
		LaneID:         lane.ID,
		ChunkArtifacts: chunkArtifacts,
		LLMProfile:     lane.Merge.LLMProfile,
		Options:        cloneOptions(lane.Merge.Options),
		Metadata:       input.Metadata,
	})
	output.Warnings = append(output.Warnings, mergeResult.Warnings...)
	if err != nil {
		return fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, merger.Key(), err)
	}

	normalizeResult, err := normalizer.Normalize(ctx, contracts.NormalizeRequest{
		Source:     doc,
		LaneID:     lane.ID,
		Candidates: mergeResult.Candidates,
		References: CloneReferenceSet(lane.NormalizeReferences.ReferenceSet),
		LLMClient:  input.LLMClient,
		LLMProfile: lane.Normalize.LLMProfile,
		Options:    cloneOptions(lane.Normalize.Options),
		Metadata:   input.Metadata,
	})
	output.Warnings = append(output.Warnings, normalizeResult.Warnings...)
	if err != nil {
		return fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, normalizer.Key(), err)
	}

	if err := validateCandidateEnvelope(extractor, normalizeResult.Candidates); err != nil {
		return fmt.Errorf("validate normalized candidates for lane %q: %w", lane.ID, err)
	}

	approved, rejected, warnings, err := runValidators(ctx, extractor.Key(), validators, doc, normalizeResult.Candidates, input.Metadata)
	output.Warnings = append(output.Warnings, warnings...)
	output.Rejected = append(output.Rejected, rejected...)
	if err != nil {
		return err
	}

	for _, candidate := range approved {
		output.Approved = append(output.Approved, artifacts.ArtifactFromCandidate(candidate))
	}
	return nil
}

type validatorExecution struct {
	validator contracts.Validator
	binding   ModuleBinding
}

func (r *Runner) buildConfiguredValidators(lane ResolvedArtifactLane) ([]validatorExecution, error) {
	validators := make([]validatorExecution, 0, len(lane.Validators))
	for _, binding := range lane.Validators {
		validator, err := r.registries.Validators.Build(binding.Module)
		if err != nil {
			return nil, fmt.Errorf("build validator %q for lane %q: %w", binding.Module, lane.ID, err)
		}
		validators = append(validators, validatorExecution{
			validator: validator,
			binding:   binding,
		})
	}
	return validators, nil
}

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")
	}
	if pipelineUsesConfiguredValidators(pipeline) && r.registries.Validators == nil {
		return fmt.Errorf("validator 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)
		}
		for _, validator := range lane.Validators {
			if validator.Module == "" {
				return fmt.Errorf("resolved pipeline lane %q validator module must not be empty", lane.ID)
			}
		}
	}
	return nil
}

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)),
		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,
		}
		for _, validator := range lane.Validators {
			laneManifest.Validators = append(laneManifest.Validators, validator.Module)
		}
		manifest.ArtifactLanes = append(manifest.ArtifactLanes, laneManifest)
	}
	return manifest
}

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

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 cloneWarnings(warnings []contracts.Warning) []contracts.Warning {
	if len(warnings) == 0 {
		return nil
	}
	return append([]contracts.Warning(nil), warnings...)
}

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

func pipelineUsesConfiguredValidators(pipeline ResolvedPipeline) bool {
	for _, lane := range pipeline.ArtifactLanes {
		if len(lane.Validators) > 0 {
			return true
		}
	}
	return false
}

func normalizeCandidates(extractor contracts.Extractor, candidates []artifacts.ArtifactCandidate, nextIndex *int) ([]artifacts.ArtifactCandidate, error) {
	normalized := make([]artifacts.ArtifactCandidate, 0, len(candidates))
	for _, candidate := range candidates {
		candidate.Index = *nextIndex
		*nextIndex = *nextIndex + 1

		if candidate.ExtractorKey == "" {
			candidate.ExtractorKey = extractor.Key()
		} else if candidate.ExtractorKey != extractor.Key() {
			return nil, fmt.Errorf("candidate extractor_key %q does not match extractor %q", candidate.ExtractorKey, extractor.Key())
		}

		if candidate.ArtifactType == "" {
			candidate.ArtifactType = extractor.ArtifactType()
		} else if candidate.ArtifactType != extractor.ArtifactType() {
			return nil, fmt.Errorf("candidate artifact_type %q does not match extractor %q artifact type %q", candidate.ArtifactType, extractor.Key(), extractor.ArtifactType())
		}

		if candidate.SchemaVersion == "" {
			candidate.SchemaVersion = extractor.SchemaVersion()
		} else if candidate.SchemaVersion != extractor.SchemaVersion() {
			return nil, fmt.Errorf("candidate schema_version %q does not match extractor %q schema version %q", candidate.SchemaVersion, extractor.Key(), extractor.SchemaVersion())
		}

		normalized = append(normalized, candidate)
	}
	return normalized, nil
}

func validateCandidateEnvelope(extractor contracts.Extractor, candidates []artifacts.ArtifactCandidate) error {
	seen := make(map[int]struct{}, len(candidates))
	for _, candidate := range candidates {
		if _, ok := seen[candidate.Index]; ok {
			return fmt.Errorf("candidate index %d is duplicated", candidate.Index)
		}
		seen[candidate.Index] = struct{}{}

		if candidate.ExtractorKey == "" {
			return fmt.Errorf("candidate index %d extractor_key must not be empty", candidate.Index)
		}
		if candidate.ExtractorKey != extractor.Key() {
			return fmt.Errorf("candidate index %d extractor_key %q does not match extractor %q", candidate.Index, candidate.ExtractorKey, extractor.Key())
		}
		if candidate.ArtifactType == "" {
			return fmt.Errorf("candidate index %d artifact_type must not be empty", candidate.Index)
		}
		if candidate.ArtifactType != extractor.ArtifactType() {
			return fmt.Errorf("candidate index %d artifact_type %q does not match extractor %q artifact type %q", candidate.Index, candidate.ArtifactType, extractor.Key(), extractor.ArtifactType())
		}
		if candidate.SchemaVersion == "" {
			return fmt.Errorf("candidate index %d schema_version must not be empty", candidate.Index)
		}
		if candidate.SchemaVersion != extractor.SchemaVersion() {
			return fmt.Errorf("candidate index %d schema_version %q does not match extractor %q schema version %q", candidate.Index, candidate.SchemaVersion, extractor.Key(), extractor.SchemaVersion())
		}
	}
	return nil
}

func runValidators(ctx context.Context, extractorKey string, validators []validatorExecution, doc *source.SourceDocument, candidates []artifacts.ArtifactCandidate, metadata map[string]any) ([]artifacts.ArtifactCandidate, []artifacts.RejectedArtifact, []contracts.Warning, error) {
	eligible := candidates
	var rejected []artifacts.RejectedArtifact
	var warnings []contracts.Warning

	for validatorIndex, execution := range validators {
		validator := execution.validator
		if validator == nil {
			return nil, rejected, warnings, fmt.Errorf("extractor %q validator[%d] must not be nil", extractorKey, validatorIndex)
		}
		result, err := validator.Validate(ctx, contracts.ValidationRequest{
			Source:     doc,
			Candidates: eligible,
			LLMProfile: execution.binding.LLMProfile,
			Options:    cloneOptions(execution.binding.Options),
			Metadata:   metadata,
		})
		warnings = append(warnings, result.Warnings...)
		if err != nil {
			return nil, rejected, warnings, fmt.Errorf("validate extractor %q with validator %q: %w", extractorKey, validator.Name(), err)
		}
		if result.ValidatorName != validator.Name() {
			return nil, rejected, warnings, fmt.Errorf("validator %q returned result for %q", validator.Name(), result.ValidatorName)
		}
		if err := validate.EnforceDecisionCardinality(eligible, result.Decisions); err != nil {
			return nil, rejected, warnings, fmt.Errorf("validate extractor %q with validator %q: %w", extractorKey, validator.Name(), err)
		}

		decisions := make(map[int]contracts.ValidationDecision, len(result.Decisions))
		for _, decision := range result.Decisions {
			decisions[decision.CandidateIndex] = decision
		}

		nextEligible := make([]artifacts.ArtifactCandidate, 0, len(eligible))
		for _, candidate := range eligible {
			decision := decisions[candidate.Index]
			if decision.Approved {
				nextEligible = append(nextEligible, candidate)
				continue
			}
			rejected = append(rejected, artifacts.RejectedArtifact{
				Candidate:     candidate,
				ValidatorName: result.ValidatorName,
				ReasonCode:    decision.ReasonCode,
				Message:       decision.Message,
			})
		}
		eligible = nextEligible
	}

	return eligible, rejected, warnings, nil
}