306 lines
12 KiB
Go
306 lines
12 KiB
Go
package config
|
|
|
|
import (
|
|
"fmt"
|
|
"net/url"
|
|
"sort"
|
|
"strings"
|
|
|
|
configkit "gitea.maximumdirect.net/eric/ConfigKit"
|
|
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
|
|
)
|
|
|
|
func resolveDocumentOperations(document *Document, resolution *configkit.Resolution, state *documentLoadState) error {
|
|
if document == nil {
|
|
return nil
|
|
}
|
|
normalizeDocumentPromptKit(document.PromptKit, resolution)
|
|
normalizeDocumentState(document, resolution)
|
|
normalizeDocumentPipelines(document, resolution, state)
|
|
normalizeDocumentPaths(document, resolution, state)
|
|
if document.Concurrency != nil {
|
|
document.Concurrency.StageWorkers = normalizeDocumentMap(document.Concurrency.StageWorkers,
|
|
configkit.Keys("concurrency", "stage_workers"), "stage worker identifier",
|
|
documentNormalizer{Resolution: resolution}, state)
|
|
}
|
|
deriveDocumentExtractWorkers(document, resolution, state)
|
|
return nil
|
|
}
|
|
|
|
func normalizeDocumentPromptKit(promptKit *PromptKitDocument, resolution *configkit.Resolution) {
|
|
if promptKit == nil {
|
|
return
|
|
}
|
|
profileDirPath := configkit.Keys("promptkit", "profile_dir")
|
|
profileFilePath := configkit.Keys("promptkit", "profile_file")
|
|
if promptKit.ProfileDir == nil && resolution.IsExplicit(profileDirPath) {
|
|
resolution.Error(profileDirPath, "must not be null")
|
|
}
|
|
if promptKit.ProfileFile == nil && resolution.IsExplicit(profileFilePath) {
|
|
resolution.Error(profileFilePath, "must not be null")
|
|
}
|
|
normalizeDocumentString(resolution, profileDirPath, &promptKit.ProfileDir, "trimmed PromptKit profile directory")
|
|
normalizeDocumentString(resolution, profileFilePath, &promptKit.ProfileFile, "trimmed PromptKit profile file")
|
|
|
|
if promptKit.LocalBackend == nil {
|
|
return
|
|
}
|
|
endpointPath := configkit.Keys("promptkit", "local_backend", "endpoint")
|
|
limitPath := configkit.Keys("promptkit", "local_backend", "concurrency_limit")
|
|
normalizeDocumentString(resolution, endpointPath, &promptKit.LocalBackend.Endpoint, "trimmed PromptKit local backend endpoint")
|
|
if promptKit.LocalBackend.ConcurrencyLimit == nil && !resolution.IsExplicit(limitPath) {
|
|
configkit.Default(resolution, limitPath, &promptKit.LocalBackend.ConcurrencyLimit,
|
|
documentValue(0), "defaulted PromptKit local backend concurrency limit")
|
|
}
|
|
}
|
|
|
|
func normalizeDocumentState(document *Document, resolution *configkit.Resolution) {
|
|
if document.Output != nil {
|
|
normalizeDocumentString(resolution, configkit.Keys("output", "directory"), &document.Output.Directory, "trimmed output directory")
|
|
}
|
|
if document.Cache != nil {
|
|
if document.Cache.ChunkPlans != nil {
|
|
normalizeDocumentString(resolution, configkit.Keys("cache", "chunk_plans", "directory"), &document.Cache.ChunkPlans.Directory, "trimmed chunk-plan cache directory")
|
|
normalizeDocumentString(resolution, configkit.Keys("cache", "chunk_plans", "mode"), &document.Cache.ChunkPlans.Mode, "trimmed chunk-plan cache mode")
|
|
}
|
|
if document.Cache.Checkpoints != nil {
|
|
normalizeDocumentString(resolution, configkit.Keys("cache", "checkpoints", "directory"), &document.Cache.Checkpoints.Directory, "trimmed checkpoint cache directory")
|
|
}
|
|
}
|
|
if document.Debug != nil {
|
|
normalizeDocumentString(resolution, configkit.Keys("debug", "directory"), &document.Debug.Directory, "trimmed debug directory")
|
|
}
|
|
}
|
|
|
|
func normalizeDocumentString(resolution *configkit.Resolution, path configkit.Path, target **string, reason string) {
|
|
if target == nil || *target == nil || !resolution.IsExplicit(path) {
|
|
return
|
|
}
|
|
configkit.Normalize(resolution, path, target, documentValue(strings.TrimSpace(**target)), reason)
|
|
}
|
|
|
|
func deriveDocumentExtractWorkers(document *Document, resolution *configkit.Resolution, state *documentLoadState) {
|
|
if document.Concurrency == nil || document.Concurrency.TotalLLM == nil {
|
|
return
|
|
}
|
|
if document.Concurrency.StageWorkers == nil {
|
|
document.Concurrency.StageWorkers = make(map[string]int)
|
|
}
|
|
path := configkit.Keys("concurrency", "stage_workers", "extract")
|
|
// Presence belongs to the input spelling, not the key produced by trimming.
|
|
// A moved explicit count must never be replaced by a derived default.
|
|
if resolution.IsExplicit(state.originalPath(path)) {
|
|
return
|
|
}
|
|
workers := document.Concurrency.StageWorkers
|
|
value := workers["extract"]
|
|
configkit.Default(resolution, path, &value, *document.Concurrency.TotalLLM,
|
|
"derived extraction workers from total LLM concurrency", configkit.Keys("concurrency", "total_llm"))
|
|
workers["extract"] = value
|
|
}
|
|
|
|
func validateDocumentOperations(document *Document, validation *configkit.Validation) {
|
|
if document == nil {
|
|
return
|
|
}
|
|
validateDocumentConcurrency(document.Concurrency, validation)
|
|
validateDocumentPromptKit(document.PromptKit, validation)
|
|
validateDocumentState(document, validation)
|
|
validateDocumentPipelines(document, validation)
|
|
}
|
|
|
|
func validateDocumentConcurrency(concurrency *ConcurrencyDocument, validation *configkit.Validation) {
|
|
if concurrency == nil {
|
|
validation.Error(configkit.Keys("concurrency"), "must not be null")
|
|
return
|
|
}
|
|
totalPath := configkit.Keys("concurrency", "total_llm")
|
|
if concurrency.TotalLLM == nil {
|
|
validation.Error(totalPath, "must not be null")
|
|
return
|
|
}
|
|
totalValid := true
|
|
if err := validateTotalLLMConcurrency(*concurrency.TotalLLM); err != nil {
|
|
validation.Error(totalPath, err.Error())
|
|
totalValid = false
|
|
}
|
|
keys := make([]string, 0, len(concurrency.StageWorkers))
|
|
for key := range concurrency.StageWorkers {
|
|
keys = append(keys, key)
|
|
}
|
|
sort.Strings(keys)
|
|
for _, key := range keys {
|
|
if err := validateStageWorkerKey(key); err != nil {
|
|
validation.Error(configkit.Keys("concurrency", "stage_workers").Key(key), err.Error())
|
|
}
|
|
}
|
|
if !totalValid {
|
|
return
|
|
}
|
|
if workers, ok := concurrency.StageWorkers["extract"]; ok {
|
|
if err := validateExtractWorkerCount(workers, *concurrency.TotalLLM); err != nil {
|
|
validation.Error(configkit.Keys("concurrency", "stage_workers", "extract"), err.Error())
|
|
}
|
|
}
|
|
}
|
|
|
|
func validateDocumentPromptKit(promptKit *PromptKitDocument, validation *configkit.Validation) {
|
|
if promptKit == nil {
|
|
return
|
|
}
|
|
profileDirPath := configkit.Keys("promptkit", "profile_dir")
|
|
profileFilePath := configkit.Keys("promptkit", "profile_file")
|
|
profileDir := ""
|
|
if promptKit.ProfileDir != nil {
|
|
profileDir = *promptKit.ProfileDir
|
|
if strings.TrimSpace(profileDir) == "" {
|
|
validation.Error(profileDirPath, "must not be empty when set")
|
|
}
|
|
}
|
|
profileFile := ""
|
|
if promptKit.ProfileFile != nil {
|
|
profileFile = *promptKit.ProfileFile
|
|
if strings.TrimSpace(profileFile) == "" {
|
|
validation.Error(profileFilePath, "must not be empty when set")
|
|
}
|
|
}
|
|
if err := validatePromptKitProfileSources(profileDir, profileFile); err != nil {
|
|
validation.Error(configkit.Keys("promptkit"), err.Error())
|
|
}
|
|
if promptKit.LocalBackend == nil {
|
|
return
|
|
}
|
|
endpointPath := configkit.Keys("promptkit", "local_backend", "endpoint")
|
|
limitPath := configkit.Keys("promptkit", "local_backend", "concurrency_limit")
|
|
if promptKit.LocalBackend.Endpoint == nil {
|
|
validation.Error(endpointPath, "must be nonempty when local backend is configured")
|
|
}
|
|
if promptKit.LocalBackend.ConcurrencyLimit == nil {
|
|
validation.Error(limitPath, "must be an integer when local backend is configured")
|
|
}
|
|
if promptKit.LocalBackend.Endpoint != nil {
|
|
if err := validatePromptKitLocalBackendEndpoint(*promptKit.LocalBackend.Endpoint); err != nil {
|
|
validation.Error(endpointPath, err.Error())
|
|
}
|
|
}
|
|
if promptKit.LocalBackend.ConcurrencyLimit != nil {
|
|
if err := validatePromptKitLocalBackendConcurrencyLimit(*promptKit.LocalBackend.ConcurrencyLimit); err != nil {
|
|
validation.Error(limitPath, err.Error())
|
|
}
|
|
}
|
|
}
|
|
|
|
func validateDocumentState(document *Document, validation *configkit.Validation) {
|
|
if document.Output == nil || document.Output.Directory == nil {
|
|
validation.Error(configkit.Keys("output", "directory"), "must not be empty")
|
|
} else if err := validateRequiredPath("output.directory", *document.Output.Directory); err != nil {
|
|
validation.Error(configkit.Keys("output", "directory"), err.Error())
|
|
}
|
|
|
|
if document.Cache == nil || document.Cache.ChunkPlans == nil || document.Cache.ChunkPlans.Mode == nil {
|
|
validation.Error(configkit.Keys("cache", "chunk_plans", "mode"), "must be a supported cache mode")
|
|
} else if err := validateChunkPlanCacheMode(*document.Cache.ChunkPlans.Mode); err != nil {
|
|
validation.Error(configkit.Keys("cache", "chunk_plans", "mode"), err.Error())
|
|
}
|
|
if document.Cache != nil && document.Cache.ChunkPlans != nil && document.Cache.ChunkPlans.Directory != nil {
|
|
if err := validateOptionalPath("cache.chunk_plans.directory", *document.Cache.ChunkPlans.Directory); err != nil {
|
|
validation.Error(configkit.Keys("cache", "chunk_plans", "directory"), err.Error())
|
|
}
|
|
}
|
|
if document.Cache != nil && document.Cache.Checkpoints != nil && document.Cache.Checkpoints.Directory != nil {
|
|
if err := validateOptionalPath("cache.checkpoints.directory", *document.Cache.Checkpoints.Directory); err != nil {
|
|
validation.Error(configkit.Keys("cache", "checkpoints", "directory"), err.Error())
|
|
}
|
|
}
|
|
|
|
if document.Debug == nil || document.Debug.Directory == nil {
|
|
validation.Error(configkit.Keys("debug", "directory"), "must not be empty")
|
|
} else if err := validateRequiredPath("debug.directory", *document.Debug.Directory); err != nil {
|
|
validation.Error(configkit.Keys("debug", "directory"), err.Error())
|
|
}
|
|
}
|
|
|
|
func validateTotalLLMConcurrency(total int) error {
|
|
if total <= 0 {
|
|
return fmt.Errorf("total LLM concurrency must be greater than zero")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validateStageWorkerKey(key string) error {
|
|
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)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validateExtractWorkerCount(workers, total int) error {
|
|
if workers < 1 || workers > total {
|
|
return fmt.Errorf("concurrency.stage_workers.extract must be between 1 and concurrency.total_llm (%d)", total)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validatePromptKitProfileSources(profileDir, profileFile string) error {
|
|
if strings.TrimSpace(profileDir) != "" && strings.TrimSpace(profileFile) != "" {
|
|
return fmt.Errorf("promptkit profile_dir and profile_file are mutually exclusive")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validatePromptKitLocalBackend(endpoint string, concurrencyLimit int) error {
|
|
if err := validatePromptKitLocalBackendEndpoint(endpoint); err != nil {
|
|
return err
|
|
}
|
|
return validatePromptKitLocalBackendConcurrencyLimit(concurrencyLimit)
|
|
}
|
|
|
|
func validatePromptKitLocalBackendEndpoint(endpoint string) error {
|
|
if strings.TrimSpace(endpoint) == "" {
|
|
return fmt.Errorf("promptkit.local_backend.endpoint must not be empty when set")
|
|
}
|
|
parsed, err := url.Parse(strings.TrimSpace(endpoint))
|
|
if err != nil ||
|
|
(!strings.EqualFold(parsed.Scheme, "http") && !strings.EqualFold(parsed.Scheme, "https")) ||
|
|
!parsed.IsAbs() ||
|
|
parsed.Hostname() == "" ||
|
|
parsed.User != nil ||
|
|
parsed.RawQuery != "" ||
|
|
parsed.ForceQuery ||
|
|
strings.Contains(endpoint, "#") {
|
|
return fmt.Errorf("promptkit.local_backend.endpoint must be an absolute HTTP or HTTPS URL with a host and no user information, query, or fragment")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validatePromptKitLocalBackendConcurrencyLimit(concurrencyLimit int) error {
|
|
if concurrencyLimit < 0 {
|
|
return fmt.Errorf("promptkit.local_backend.concurrency_limit must not be negative")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validateRequiredPath(name, value string) error {
|
|
if strings.TrimSpace(value) == "" {
|
|
return fmt.Errorf("%s must not be empty", name)
|
|
}
|
|
return validateOptionalPath(name, value)
|
|
}
|
|
|
|
func validateOptionalPath(name, value string) error {
|
|
if strings.ContainsRune(value, '\x00') {
|
|
return fmt.Errorf("%s must not contain NUL", name)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validateChunkPlanCacheMode(value string) error {
|
|
if _, err := pipeline.ParseChunkCacheMode(value); err != nil {
|
|
return fmt.Errorf("cache.chunk_plans.mode: %w", err)
|
|
}
|
|
return nil
|
|
}
|