Apply ConfigKit operational configuration defaults
This commit is contained in:
@@ -148,7 +148,7 @@ func TestDocumentDefinitionReturnsIndependentDefaults(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(second.Config.Pipelines) != 0 || len(second.Config.Concurrency.StageWorkers) != 0 {
|
||||
if len(second.Config.Pipelines) != 0 || second.Config.Concurrency.StageWorkers["extract"] != defaultLLMConcurrency || len(second.Config.Concurrency.StageWorkers) != 1 {
|
||||
t.Fatalf("defaults aliased across loads: %#v", second.Config)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -29,14 +29,9 @@ func loadDocumentSource(source configkit.Source, options LoadOptions) (*configki
|
||||
|
||||
func documentDefinition() configkit.Definition[Document] {
|
||||
return configkit.Definition[Document]{
|
||||
Defaults: func() Document {
|
||||
return Document{
|
||||
Pipelines: map[string]PipelineDocument{},
|
||||
Concurrency: &ConcurrencyDocument{
|
||||
StageWorkers: map[string]int{},
|
||||
},
|
||||
}
|
||||
},
|
||||
Defaults: documentDefaults,
|
||||
Resolve: resolveDocumentOperations,
|
||||
Validate: validateDocumentOperations,
|
||||
Version: &configkit.VersionRule{
|
||||
Path: configkit.Keys("version"),
|
||||
Check: validateDocumentVersion,
|
||||
@@ -44,6 +39,30 @@ func documentDefinition() configkit.Definition[Document] {
|
||||
}
|
||||
}
|
||||
|
||||
func documentDefaults() Document {
|
||||
return Document{
|
||||
Pipelines: map[string]PipelineDocument{},
|
||||
Concurrency: &ConcurrencyDocument{
|
||||
TotalLLM: documentValue(defaultLLMConcurrency),
|
||||
StageWorkers: map[string]int{},
|
||||
},
|
||||
Output: &OutputDocument{Directory: documentValue("./notarius-output")},
|
||||
Cache: &CacheDocument{
|
||||
ChunkPlans: &ChunkPlanCacheDocument{
|
||||
Directory: documentValue(""),
|
||||
Mode: documentValue("auto"),
|
||||
},
|
||||
Checkpoints: &CheckpointCacheDocument{
|
||||
Enabled: documentValue(false),
|
||||
Directory: documentValue(""),
|
||||
},
|
||||
},
|
||||
Debug: &DebugDocument{Directory: documentValue("./notarius-debug")},
|
||||
}
|
||||
}
|
||||
|
||||
func documentValue[T any](value T) *T { return &value }
|
||||
|
||||
func validateDocumentVersion(version configkit.Version, validation *configkit.Validation) error {
|
||||
if !version.Present || version.Null {
|
||||
validation.Error(configkit.Keys("version"), "configuration version is required")
|
||||
|
||||
292
internal/core/config/operational_document.go
Normal file
292
internal/core/config/operational_document.go
Normal file
@@ -0,0 +1,292 @@
|
||||
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) error {
|
||||
if document == nil {
|
||||
return nil
|
||||
}
|
||||
normalizeDocumentPromptKit(document.PromptKit, resolution)
|
||||
normalizeDocumentState(document, resolution)
|
||||
deriveDocumentExtractWorkers(document, resolution)
|
||||
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) {
|
||||
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")
|
||||
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)
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
184
internal/core/config/operational_document_test.go
Normal file
184
internal/core/config/operational_document_test.go
Normal file
@@ -0,0 +1,184 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
configkit "gitea.maximumdirect.net/eric/ConfigKit"
|
||||
)
|
||||
|
||||
func TestDocumentLoadAppliesOperationalDefaultsAndNormalization(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
loaded, err := loadDocumentSource(configkit.Bytes("document.yml", []byte(`version: 5
|
||||
concurrency:
|
||||
total_llm: 4
|
||||
promptkit:
|
||||
profile_dir: " ./profiles "
|
||||
local_backend:
|
||||
endpoint: " https://localhost:8080/v1 "
|
||||
output:
|
||||
directory: " ./output "
|
||||
cache:
|
||||
chunk_plans:
|
||||
mode: " refresh "
|
||||
directory: " "
|
||||
checkpoints:
|
||||
directory: " "
|
||||
debug:
|
||||
directory: " ./debug "
|
||||
`)), LoadOptions{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if loaded.Config.Concurrency.TotalLLM == nil || *loaded.Config.Concurrency.TotalLLM != 4 || loaded.Config.Concurrency.StageWorkers["extract"] != 4 {
|
||||
t.Fatalf("concurrency = %#v", loaded.Config.Concurrency)
|
||||
}
|
||||
if loaded.Config.PromptKit.ProfileDir == nil || *loaded.Config.PromptKit.ProfileDir != "./profiles" ||
|
||||
loaded.Config.PromptKit.LocalBackend == nil || loaded.Config.PromptKit.LocalBackend.Endpoint == nil || *loaded.Config.PromptKit.LocalBackend.Endpoint != "https://localhost:8080/v1" ||
|
||||
loaded.Config.PromptKit.LocalBackend.ConcurrencyLimit == nil || *loaded.Config.PromptKit.LocalBackend.ConcurrencyLimit != 0 {
|
||||
t.Fatalf("promptkit = %#v", loaded.Config.PromptKit)
|
||||
}
|
||||
if loaded.Config.Output.Directory == nil || *loaded.Config.Output.Directory != "./output" ||
|
||||
loaded.Config.Cache.ChunkPlans.Mode == nil || *loaded.Config.Cache.ChunkPlans.Mode != "refresh" ||
|
||||
loaded.Config.Cache.ChunkPlans.Directory == nil || *loaded.Config.Cache.ChunkPlans.Directory != "" ||
|
||||
loaded.Config.Cache.Checkpoints.Directory == nil || *loaded.Config.Cache.Checkpoints.Directory != "" ||
|
||||
loaded.Config.Debug.Directory == nil || *loaded.Config.Debug.Directory != "./debug" {
|
||||
t.Fatalf("state = %#v, %#v, %#v", loaded.Config.Output, loaded.Config.Cache, loaded.Config.Debug)
|
||||
}
|
||||
if diagnostics := loaded.Diagnostics(); len(diagnostics) != 0 {
|
||||
t.Fatalf("diagnostics = %#v", diagnostics)
|
||||
}
|
||||
extractHistory := loaded.Metadata().History(configkit.Keys("concurrency", "stage_workers", "extract"))
|
||||
if len(extractHistory) == 0 || extractHistory[len(extractHistory)-1].Origin.Operation != configkit.ChangeDerivedDefault {
|
||||
t.Fatalf("extract history = %#v", extractHistory)
|
||||
}
|
||||
outputHistory := loaded.Metadata().History(configkit.Keys("output", "directory"))
|
||||
if len(outputHistory) == 0 || outputHistory[len(outputHistory)-1].Origin.Operation != configkit.ChangeNormalization {
|
||||
t.Fatalf("output history = %#v", outputHistory)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDocumentLoadDerivesWorkersAfterProfileAndCLIOverrides(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
directory := t.TempDir()
|
||||
main := filepath.Join(directory, "config.yml")
|
||||
profiles := filepath.Join(directory, "profiles")
|
||||
if err := os.Mkdir(profiles, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(main, []byte("version: 5\nconcurrency:\n total_llm: 4\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(profiles, "development.yml"), []byte("concurrency:\n total_llm: 3\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
profiled, err := loadDocument(main, LoadOptions{Profile: "development"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if profiled.Config.Concurrency.TotalLLM == nil || *profiled.Config.Concurrency.TotalLLM != 3 || profiled.Config.Concurrency.StageWorkers["extract"] != 3 {
|
||||
t.Fatalf("profiled concurrency = %#v", profiled.Config.Concurrency)
|
||||
}
|
||||
|
||||
overridden, err := loadDocument(main, LoadOptions{Overrides: []configkit.Override{{
|
||||
Path: configkit.Keys("concurrency", "total_llm"), Value: 2, Flag: "--total-llm",
|
||||
}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if overridden.Config.Concurrency.TotalLLM == nil || *overridden.Config.Concurrency.TotalLLM != 2 || overridden.Config.Concurrency.StageWorkers["extract"] != 2 {
|
||||
t.Fatalf("overridden concurrency = %#v", overridden.Config.Concurrency)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDocumentLoadRetainsExplicitWorkerAndRejectsInvalidOperationalValues(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
loaded, err := loadDocumentSource(configkit.Bytes("explicit.yml", []byte(`version: 5
|
||||
concurrency:
|
||||
total_llm: 4
|
||||
stage_workers:
|
||||
extract: 0
|
||||
`)), LoadOptions{})
|
||||
if loaded != nil {
|
||||
t.Fatalf("loaded = %#v, want validation failure", loaded.Config)
|
||||
}
|
||||
assertDocumentDiagnostic(t, err, configkit.CodeValidateError, "/concurrency/stage_workers/extract")
|
||||
|
||||
_, err = loadDocumentSource(configkit.Bytes("invalid.yml", []byte(`version: 5
|
||||
concurrency:
|
||||
total_llm: 0
|
||||
output:
|
||||
directory: ""
|
||||
promptkit:
|
||||
local_backend:
|
||||
endpoint: "ftp://localhost"
|
||||
concurrency_limit: -1
|
||||
`)), LoadOptions{})
|
||||
var loadError *configkit.LoadError
|
||||
if !errors.As(err, &loadError) {
|
||||
t.Fatalf("error = %v, want ConfigKit load error", err)
|
||||
}
|
||||
seen := make(map[string]bool)
|
||||
for _, diagnostic := range loadError.Diagnostics() {
|
||||
if diagnostic.Code != configkit.CodeValidateError || diagnostic.Primary == nil {
|
||||
t.Fatalf("diagnostic = %#v", diagnostic)
|
||||
}
|
||||
seen[diagnostic.Path.String()] = true
|
||||
}
|
||||
for _, path := range []string{"/concurrency/total_llm", "/output/directory", "/promptkit/local_backend/endpoint", "/promptkit/local_backend/concurrency_limit"} {
|
||||
if !seen[path] {
|
||||
t.Fatalf("diagnostics missing %s: %#v", path, loadError.Diagnostics())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestDocumentLoadDistinguishesRequiredAndOptionalPathValues(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
loaded, err := loadDocumentSource(configkit.Bytes("cache.yml", []byte(`version: 5
|
||||
cache:
|
||||
chunk_plans:
|
||||
directory: " "
|
||||
checkpoints:
|
||||
directory: " "
|
||||
`)), LoadOptions{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if loaded.Config.Cache.ChunkPlans.Directory == nil || *loaded.Config.Cache.ChunkPlans.Directory != "" ||
|
||||
loaded.Config.Cache.Checkpoints.Directory == nil || *loaded.Config.Cache.Checkpoints.Directory != "" {
|
||||
t.Fatalf("cache directories = %#v", loaded.Config.Cache)
|
||||
}
|
||||
|
||||
_, err = loadDocumentSource(configkit.Bytes("output.yml", []byte("version: 5\noutput:\n directory: ' '\n")), LoadOptions{})
|
||||
assertDocumentDiagnostic(t, err, configkit.CodeValidateError, "/output/directory")
|
||||
|
||||
_, err = loadDocumentSource(configkit.Bytes("promptkit.yml", []byte("version: 5\npromptkit:\n profile_file: null\n local_backend: {}\n")), LoadOptions{})
|
||||
assertDocumentDiagnostic(t, err, configkit.CodeResolveError, "/promptkit/profile_file")
|
||||
|
||||
_, err = loadDocumentSource(configkit.Bytes("empty-source.yml", []byte("version: 5\npromptkit:\n profile_dir: ' '\n")), LoadOptions{})
|
||||
assertDocumentDiagnostic(t, err, configkit.CodeValidateError, "/promptkit/profile_dir")
|
||||
|
||||
_, err = loadDocumentSource(configkit.Bytes("missing-endpoint.yml", []byte("version: 5\npromptkit:\n local_backend: {}\n")), LoadOptions{})
|
||||
assertDocumentDiagnostic(t, err, configkit.CodeValidateError, "/promptkit/local_backend/endpoint")
|
||||
}
|
||||
|
||||
func assertDocumentDiagnostic(t *testing.T, err error, code configkit.Code, path string) {
|
||||
t.Helper()
|
||||
var loadError *configkit.LoadError
|
||||
if !errors.As(err, &loadError) {
|
||||
t.Fatalf("error = %v, want ConfigKit load error", err)
|
||||
}
|
||||
for _, diagnostic := range loadError.Diagnostics() {
|
||||
if diagnostic.Code == code && diagnostic.Path.String() == path {
|
||||
return
|
||||
}
|
||||
}
|
||||
t.Fatalf("diagnostics = %#v, want %s at %s", loadError.Diagnostics(), code, path)
|
||||
}
|
||||
@@ -2,7 +2,6 @@ package config
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/url"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
@@ -17,8 +16,8 @@ func (c Config) Validate() error {
|
||||
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 := validateTotalLLMConcurrency(c.Concurrency.TotalLLM); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := validateStageWorkers(c.Concurrency); err != nil {
|
||||
return err
|
||||
@@ -33,70 +32,42 @@ func validateStageWorkers(cfg ConcurrencyConfig) error {
|
||||
}
|
||||
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)
|
||||
if err := validateStageWorkerKey(key); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
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
|
||||
return validateExtractWorkerCount(extractWorkers, cfg.TotalLLM)
|
||||
}
|
||||
|
||||
func validatePromptKit(cfg PromptKitConfig) error {
|
||||
if strings.TrimSpace(cfg.ProfileDir) != "" && strings.TrimSpace(cfg.ProfileFile) != "" {
|
||||
return fmt.Errorf("promptkit profile_dir and profile_file are mutually exclusive")
|
||||
if err := validatePromptKitProfileSources(cfg.ProfileDir, cfg.ProfileFile); err != nil {
|
||||
return err
|
||||
}
|
||||
if cfg.LocalBackend == nil {
|
||||
return nil
|
||||
}
|
||||
endpoint := strings.TrimSpace(cfg.LocalBackend.Endpoint)
|
||||
if endpoint == "" {
|
||||
return fmt.Errorf("promptkit.local_backend.endpoint must not be empty when set")
|
||||
}
|
||||
parsed, err := url.Parse(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")
|
||||
}
|
||||
if cfg.LocalBackend.ConcurrencyLimit < 0 {
|
||||
return fmt.Errorf("promptkit.local_backend.concurrency_limit must not be negative")
|
||||
}
|
||||
return nil
|
||||
return validatePromptKitLocalBackend(cfg.LocalBackend.Endpoint, cfg.LocalBackend.ConcurrencyLimit)
|
||||
}
|
||||
|
||||
func validateStateSurfaces(cfg Config) error {
|
||||
if strings.TrimSpace(cfg.Output.Directory) == "" {
|
||||
return fmt.Errorf("output.directory must not be empty")
|
||||
if err := validateRequiredPath("output.directory", cfg.Output.Directory); err != nil {
|
||||
return err
|
||||
}
|
||||
if strings.TrimSpace(cfg.Debug.Directory) == "" {
|
||||
return fmt.Errorf("debug.directory must not be empty")
|
||||
if err := validateRequiredPath("debug.directory", cfg.Debug.Directory); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := cfg.Cache.ChunkPlans.Mode.Validate(); err != nil {
|
||||
return fmt.Errorf("cache.chunk_plans.mode: %w", err)
|
||||
if err := validateChunkPlanCacheMode(string(cfg.Cache.ChunkPlans.Mode)); err != nil {
|
||||
return 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)
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user