318 lines
11 KiB
Go
318 lines
11 KiB
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 := validatePromptKit(c.PromptKit); err != nil {
|
|
return err
|
|
}
|
|
if err := validateStateSurfaces(c); err != nil {
|
|
return err
|
|
}
|
|
if err := validateTotalLLMConcurrency(c.Concurrency.TotalLLM); err != nil {
|
|
return err
|
|
}
|
|
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 err := validateStageWorkerKey(key); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
extractWorkers, ok := cfg.StageWorkers["extract"]
|
|
if !ok {
|
|
extractWorkers = cfg.TotalLLM
|
|
}
|
|
return validateExtractWorkerCount(extractWorkers, cfg.TotalLLM)
|
|
}
|
|
|
|
func validatePromptKit(cfg PromptKitConfig) error {
|
|
if err := validatePromptKitProfileSources(cfg.ProfileDir, cfg.ProfileFile); err != nil {
|
|
return err
|
|
}
|
|
if cfg.LocalBackend == nil {
|
|
return nil
|
|
}
|
|
return validatePromptKitLocalBackend(cfg.LocalBackend.Endpoint, cfg.LocalBackend.ConcurrencyLimit)
|
|
}
|
|
|
|
func validateStateSurfaces(cfg Config) error {
|
|
if err := validateRequiredPath("output.directory", cfg.Output.Directory); err != nil {
|
|
return err
|
|
}
|
|
if err := validateRequiredPath("debug.directory", cfg.Debug.Directory); err != nil {
|
|
return err
|
|
}
|
|
if err := validateChunkPlanCacheMode(string(cfg.Cache.ChunkPlans.Mode)); err != nil {
|
|
return err
|
|
}
|
|
if err := validateOptionalPath("cache.chunk_plans.directory", cfg.Cache.ChunkPlans.Directory); err != nil {
|
|
return err
|
|
}
|
|
if err := validateOptionalPath("cache.checkpoints.directory", cfg.Cache.Checkpoints.Directory); err != nil {
|
|
return err
|
|
}
|
|
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 := validateStructuredOutputRepairAttempts(fmt.Sprintf("pipeline %q", id), profile.StructuredOutputRepairAttempts); err != nil {
|
|
return err
|
|
}
|
|
if profile.ValidationPolicy != nil {
|
|
if err := profile.ValidationPolicy.Validate(); err != nil {
|
|
return fmt.Errorf("pipeline %q validation_policy: %w", id, err)
|
|
}
|
|
}
|
|
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
|
|
}
|
|
explicitSteps := profile.Steps != nil
|
|
if len(profile.Artifacts) > 0 && explicitSteps {
|
|
return fmt.Errorf("pipeline %q must not declare both artifacts and steps", id)
|
|
}
|
|
steps := profile.Steps
|
|
if !explicitSteps {
|
|
steps = []pipeline.PipelineStepProfile{{ID: "default", Artifacts: profile.Artifacts}}
|
|
}
|
|
if explicitSteps && len(steps) == 0 {
|
|
return fmt.Errorf("pipeline %q must declare at least one ordered step", id)
|
|
}
|
|
seenSteps := make(map[string]struct{}, len(steps))
|
|
seenLanes := make(map[string]struct{})
|
|
for index, step := range steps {
|
|
stepID := strings.TrimSpace(step.ID)
|
|
if stepID == "" {
|
|
return fmt.Errorf("pipeline %q step[%d] id must not be empty", id, index)
|
|
}
|
|
if _, ok := seenSteps[stepID]; ok {
|
|
return fmt.Errorf("pipeline %q step id %q is duplicated after trimming", id, stepID)
|
|
}
|
|
seenSteps[stepID] = struct{}{}
|
|
if explicitSteps {
|
|
if err := validateReferenceMapForContext(id, "", "step "+stepID, step.References, true); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
for rawLaneID, lane := range step.Artifacts {
|
|
laneID := strings.TrimSpace(rawLaneID)
|
|
if laneID == "" {
|
|
return fmt.Errorf("pipeline %q artifact lane id must not be empty", id)
|
|
}
|
|
if _, ok := seenLanes[laneID]; ok {
|
|
if !explicitSteps {
|
|
return fmt.Errorf("pipeline %q artifact lane id %q is duplicated after trimming", id, laneID)
|
|
}
|
|
return fmt.Errorf("pipeline %q artifact lane id %q is duplicated across steps", id, laneID)
|
|
}
|
|
seenLanes[laneID] = struct{}{}
|
|
if err := validateReferenceMapForContext(id, laneID, "", lane.References, true); 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 binding.ValidationPolicy != nil {
|
|
switch slot {
|
|
case "chunk", "extract", "merge", "normalize":
|
|
default:
|
|
return fmt.Errorf("%s validation_policy is not supported", referenceContext(pipelineID, laneID, slot))
|
|
}
|
|
if err := binding.ValidationPolicy.Validate(); err != nil {
|
|
return fmt.Errorf("%s validation_policy: %w", referenceContext(pipelineID, laneID, slot), err)
|
|
}
|
|
}
|
|
if err := validateStructuredOutputRepairAttempts(referenceContext(pipelineID, laneID, slot), binding.StructuredOutputRepairAttempts); err != nil {
|
|
return err
|
|
}
|
|
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, true)
|
|
}
|
|
|
|
func validateStructuredOutputRepairAttempts(context string, attempts *int) error {
|
|
if attempts != nil && (*attempts < 0 || *attempts > 3) {
|
|
return fmt.Errorf("%s structured_output_repair_attempts must be between zero and three", context)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
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 err := validateStructuredOutputRepairAttempts(context, validator.StructuredOutputRepairAttempts); err != nil {
|
|
return err
|
|
}
|
|
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.ValidationPolicy != nil {
|
|
return fmt.Errorf("%s validation_policy is not supported", context)
|
|
}
|
|
if validator.Retries < 0 {
|
|
return fmt.Errorf("%s retries must be greater than or equal to zero", 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]pipeline.ReferenceSource) error {
|
|
return validateReferenceMapForContext(pipelineID, laneID, "", references, false)
|
|
}
|
|
|
|
func validateReferenceMapForContext(pipelineID string, laneID string, slot string, references map[string]pipeline.ReferenceSource, generatedAllowed bool) error {
|
|
seen := make(map[string]struct{}, len(references))
|
|
for rawSlotName, source := 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 source.Artifact != nil {
|
|
if !generatedAllowed {
|
|
return fmt.Errorf("%s reference slot %q must use an external path", referenceContext(pipelineID, laneID, slot), slotName)
|
|
}
|
|
if strings.TrimSpace(source.Artifact.Step) == "" || strings.TrimSpace(source.Artifact.Lane) == "" {
|
|
return fmt.Errorf("%s reference slot %q artifact selector step and lane must not be empty", referenceContext(pipelineID, laneID, slot), slotName)
|
|
}
|
|
if strings.TrimSpace(source.Path) != "" {
|
|
return fmt.Errorf("%s reference slot %q must contain exactly one source form", referenceContext(pipelineID, laneID, slot), slotName)
|
|
}
|
|
continue
|
|
}
|
|
if strings.TrimSpace(source.Path) == "" {
|
|
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
|
|
}
|