notarius/internal/core/config/validation.go

package config

import (
	"fmt"
	"sort"
	"strings"

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

func (c Config) Validate() error {
	c.Concurrency.recomputeStageWorkerDefaults()
	if err := validateScriptorium(c.Scriptorium); err != nil {
		return err
	}
	if err := validateStateSurfaces(c); err != nil {
		return err
	}
	if c.Concurrency.TotalLLM <= 0 {
		return fmt.Errorf("total LLM concurrency must be greater than zero")
	}
	if err := validateStageWorkers(c.Concurrency); err != nil {
		return err
	}
	return validatePipelineProfiles(c.Pipelines)
}

func validateStageWorkers(cfg ConcurrencyConfig) error {
	keys := make([]string, 0, len(cfg.StageWorkers))
	for key := range cfg.StageWorkers {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	for _, key := range keys {
		if strings.TrimSpace(key) == "" {
			return fmt.Errorf("concurrency.stage_workers key must not be empty")
		}
		if key != "extract" {
			return fmt.Errorf("concurrency.stage_workers key %q is not supported", key)
		}
	}
	extractWorkers, ok := cfg.StageWorkers["extract"]
	if !ok {
		extractWorkers = cfg.TotalLLM
	}
	if extractWorkers < 1 || extractWorkers > cfg.TotalLLM {
		return fmt.Errorf("concurrency.stage_workers.extract must be between 1 and concurrency.total_llm (%d)", cfg.TotalLLM)
	}
	return nil
}

func validateScriptorium(cfg ScriptoriumConfig) error {
	if strings.TrimSpace(cfg.ProfileDir) != "" && strings.TrimSpace(cfg.ProfileFile) != "" {
		return fmt.Errorf("scriptorium profile_dir and profile_file are mutually exclusive")
	}
	return nil
}

func validateStateSurfaces(cfg Config) error {
	if strings.TrimSpace(cfg.Output.Directory) == "" {
		return fmt.Errorf("output.directory must not be empty")
	}
	if strings.TrimSpace(cfg.Debug.Directory) == "" {
		return fmt.Errorf("debug.directory must not be empty")
	}
	if err := cfg.Cache.ChunkPlans.Mode.Validate(); err != nil {
		return fmt.Errorf("cache.chunk_plans.mode: %w", err)
	}
	for name, value := range map[string]string{
		"output.directory":            cfg.Output.Directory,
		"cache.chunk_plans.directory": cfg.Cache.ChunkPlans.Directory,
		"cache.checkpoints.directory": cfg.Cache.Checkpoints.Directory,
		"debug.directory":             cfg.Debug.Directory,
	} {
		if strings.ContainsRune(value, '\x00') {
			return fmt.Errorf("%s must not contain NUL", name)
		}
	}
	return nil
}

func validatePipelineProfiles(profiles map[string]pipeline.PipelineProfile) error {
	seen := make(map[string]struct{}, len(profiles))
	for rawID, profile := range profiles {
		id := strings.TrimSpace(rawID)
		if id == "" {
			return fmt.Errorf("pipeline id must not be empty")
		}
		if _, ok := seen[id]; ok {
			return fmt.Errorf("pipeline id %q is duplicated after trimming", id)
		}
		seen[id] = struct{}{}

		if profile.ID != "" && strings.TrimSpace(profile.ID) != id {
			return fmt.Errorf("pipeline %q profile id %q does not match map key", id, profile.ID)
		}
		if err := validateBinding(id, "", "input", profile.Input, false); err != nil {
			return err
		}
		if err := validateBinding(id, "", "chunk", profile.Chunk, true); err != nil {
			return err
		}
		if err := validateBinding(id, "", "output", profile.Output, false); err != nil {
			return err
		}
		if err := validateReferenceMap(id, "", profile.References); err != nil {
			return err
		}
		for rawLaneID, lane := range profile.Artifacts {
			laneID := strings.TrimSpace(rawLaneID)
			if laneID == "" {
				return fmt.Errorf("pipeline %q artifact lane id must not be empty", id)
			}
			if err := validateReferenceMap(id, laneID, lane.References); err != nil {
				return err
			}
			if err := validateBinding(id, laneID, "extract", lane.Extract, true); err != nil {
				return err
			}
			if err := validateBinding(id, laneID, "merge", lane.Merge, true); err != nil {
				return err
			}
			if err := validateBinding(id, laneID, "normalize", lane.Normalize, true); err != nil {
				return err
			}
			if len(lane.Validators) > 0 {
				return fmt.Errorf("pipeline %q lane %q validators are not supported at artifact lane level; use extract.validators, merge.validators, or normalize.validators", id, laneID)
			}
		}
	}
	return nil
}

func validateBinding(
	pipelineID string,
	laneID string,
	slot string,
	binding pipeline.ModuleBinding,
	referencesAllowed bool,
) error {
	if err := validateBindingLLMProfile(pipelineID, laneID, slot, binding); err != nil {
		return err
	}
	if binding.Retries < 0 {
		if laneID != "" {
			return fmt.Errorf("pipeline %q lane %q %s retries must be greater than or equal to zero", pipelineID, laneID, slot)
		}
		return fmt.Errorf("pipeline %q %s retries must be greater than or equal to zero", pipelineID, slot)
	}
	if err := validateValidatorOverride(pipelineID, laneID, slot, binding.Validators); err != nil {
		return err
	}
	if len(binding.References) == 0 {
		return nil
	}
	if !referencesAllowed {
		if laneID != "" {
			return fmt.Errorf("pipeline %q lane %q %s references are not supported", pipelineID, laneID, slot)
		}
		return fmt.Errorf("pipeline %q %s references are not supported", pipelineID, slot)
	}
	return validateReferenceMapForContext(pipelineID, laneID, slot, binding.References)
}

func validateValidatorOverride(pipelineID string, laneID string, slot string, override pipeline.ValidatorOverride) error {
	if !override.Set {
		return nil
	}
	switch slot {
	case "chunk", "extract", "merge", "normalize":
	default:
		return fmt.Errorf("%s validators are not supported", referenceContext(pipelineID, laneID, slot))
	}
	for i, validator := range override.Validators {
		context := fmt.Sprintf("%s validators[%d]", referenceContext(pipelineID, laneID, slot), i)
		if strings.TrimSpace(validator.Module) == "" {
			return fmt.Errorf("%s module must not be empty", context)
		}
		if len(validator.References) > 0 {
			return fmt.Errorf("%s references are not supported", context)
		}
		if validator.Validators.Set {
			return fmt.Errorf("%s nested validators are not supported", context)
		}
		if validator.Retries != 0 {
			return fmt.Errorf("%s retries are not supported", context)
		}
		if validator.LLMProfile != "" && strings.TrimSpace(validator.LLMProfile) == "" {
			return fmt.Errorf("%s llm_profile must not be empty when set", context)
		}
	}
	return nil
}

func validateReferenceMap(pipelineID string, laneID string, references map[string]string) error {
	return validateReferenceMapForContext(pipelineID, laneID, "", references)
}

func validateReferenceMapForContext(pipelineID string, laneID string, slot string, references map[string]string) error {
	seen := make(map[string]struct{}, len(references))
	for rawSlotName, rawSource := range references {
		slotName := strings.TrimSpace(rawSlotName)
		if slotName == "" {
			return fmt.Errorf("%s reference slot name must not be empty", referenceContext(pipelineID, laneID, slot))
		}
		if _, ok := seen[slotName]; ok {
			return fmt.Errorf("%s reference slot %q is duplicated after trimming", referenceContext(pipelineID, laneID, slot), slotName)
		}
		seen[slotName] = struct{}{}
		if strings.TrimSpace(rawSource) == "" {
			return fmt.Errorf("%s reference slot %q source must not be empty", referenceContext(pipelineID, laneID, slot), slotName)
		}
	}
	return nil
}

func referenceContext(pipelineID string, laneID string, slot string) string {
	if laneID != "" && slot != "" {
		return fmt.Sprintf("pipeline %q lane %q %s", pipelineID, laneID, slot)
	}
	if laneID != "" {
		return fmt.Sprintf("pipeline %q lane %q", pipelineID, laneID)
	}
	if slot != "" {
		return fmt.Sprintf("pipeline %q %s", pipelineID, slot)
	}
	return fmt.Sprintf("pipeline %q", pipelineID)
}

func validateBindingLLMProfile(
	pipelineID string,
	laneID string,
	slot string,
	binding pipeline.ModuleBinding,
) error {
	if binding.LLMProfile != "" && strings.TrimSpace(binding.LLMProfile) == "" {
		if laneID != "" {
			return fmt.Errorf("pipeline %q lane %q %s llm_profile must not be empty when set", pipelineID, laneID, slot)
		}
		return fmt.Errorf("pipeline %q %s llm_profile must not be empty when set", pipelineID, slot)
	}
	return nil
}