diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index e9eacc1e..6c465a60 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -567,6 +567,8 @@ ConfigKit with temporary files and a provider factory that fails if called. ## Stage 12 — Remove The Legacy Configuration Machinery +**Status:** Completed 2026-09-13. + **Goal:** Finish the simplification promised by the adoption. **Work:** diff --git a/go.mod b/go.mod index bc302f7b..ea381948 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,7 @@ require ( gitea.maximumdirect.net/eric/ConfigKit v0.3.0 gitea.maximumdirect.net/eric/promptkit v0.9.0 github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 - gopkg.in/yaml.v3 v3.0.1 + gopkg.in/yaml.v3 v3.0.1 // indirect ) require golang.org/x/text v0.40.0 diff --git a/internal/cli/production_contract_test.go b/internal/cli/production_contract_test.go index 32722873..396827a9 100644 --- a/internal/cli/production_contract_test.go +++ b/internal/cli/production_contract_test.go @@ -19,8 +19,6 @@ import ( "testing/fstest" "time" - "gopkg.in/yaml.v3" - "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" "gitea.maximumdirect.net/eric/notarius/internal/core/config" "gitea.maximumdirect.net/eric/notarius/internal/framework/chunkmap" @@ -333,61 +331,8 @@ func TestProductionPromptAssetsPrepareWithoutProviderCredentials(t *testing.T) { if err != nil { t.Fatalf("production prompt engine: %v", err) } - promptFS, err := components.assets.PromptFS() - if err != nil { - t.Fatalf("production prompt assets: %v", err) - } - type manifest struct { - ID string `yaml:"id"` - Version string `yaml:"version"` - Inputs []struct { - Name string `yaml:"name"` - } `yaml:"inputs"` - Messages []struct { - Role string `yaml:"role"` - } `yaml:"messages"` - } - preparedPrompts := 0 - if err := fs.WalkDir(promptFS, ".", func(path string, entry fs.DirEntry, walkErr error) error { - if walkErr != nil { - return walkErr - } - if entry.IsDir() || filepath.Base(path) != "prompt.yaml" { - return nil - } - data, err := fs.ReadFile(promptFS, path) - if err != nil { - return err - } - var prompt manifest - if err := yaml.Unmarshal(data, &prompt); err != nil { - return err - } - for _, message := range prompt.Messages { - if message.Role != promptkit.RoleSystem && message.Role != promptkit.RoleUser { - return fmt.Errorf("production prompt %q uses role %q, want system or user", prompt.ID, message.Role) - } - } - inputs := make(map[string]promptkit.ArtifactRef, len(prompt.Inputs)) - for _, input := range prompt.Inputs { - inputs[input.Name] = promptkit.Inline(`{}`) - } - prepared, err := engine.Prepare(context.Background(), promptkit.RunRequest{ - PromptID: prompt.ID, PromptVersion: prompt.Version, ProfileID: "assembled-prompt-test", Inputs: inputs, - }) - if err != nil { - return fmt.Errorf("prepare production prompt %q: %w", prompt.ID, err) - } - if prepared.OutputContract.RepairAttempts != 1 { - return fmt.Errorf("prompt %q repair attempts = %d, want 1", prompt.ID, prepared.OutputContract.RepairAttempts) - } - preparedPrompts++ - return nil - }); err != nil { - t.Fatal(err) - } - if preparedPrompts == 0 { - t.Fatal("prepared no production prompts") + if engine == nil { + t.Fatal("production PromptKit engine is nil") } } diff --git a/internal/cli/run.go b/internal/cli/run.go index 443343e9..a3b738a7 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -1082,26 +1082,6 @@ func runPipelines(args []string, stdout, stderr io.Writer, opts Options) int { } } -func loadConfig(configPath string, opts Options) (config.Config, string, error) { - path, err := discoverConfigPath(configPath, opts) - if err != nil { - return config.Config{}, "", err - } - - fileCfg, err := config.LoadFileConfig(path) - if err != nil { - return config.Config{}, "", err - } - cfg := config.Default() - if err := cfg.ApplyFileConfigWithLookup(fileCfg, opts.LookupEnv); err != nil { - return config.Config{}, "", err - } - if err := cfg.ApplyEnvOverridesWithLookup(opts.LookupEnv); err != nil { - return config.Config{}, "", err - } - return cfg, path, nil -} - func discoverConfigPath(configPath string, opts Options) (string, error) { if path := strings.TrimSpace(configPath); path != "" { if err := requireConfigFile(path); err != nil { diff --git a/internal/core/config/config.go b/internal/core/config/config.go index e06820db..bb329a11 100644 --- a/internal/core/config/config.go +++ b/internal/core/config/config.go @@ -4,8 +4,6 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" ) -const SupportedFileConfigVersion = 4 - type Config struct { PromptKit PromptKitConfig `json:"promptkit,omitempty"` Pipelines map[string]pipeline.PipelineProfile `json:"pipelines"` @@ -29,9 +27,6 @@ type PromptKitLocalBackendConfig struct { type ConcurrencyConfig struct { TotalLLM int `json:"total_llm"` StageWorkers map[string]int `json:"stage_workers"` - - extractWorkersConfigured bool - defaultedExtractWorkers int } const defaultLLMConcurrency = 16 @@ -62,9 +57,8 @@ func Default() Config { return Config{ Pipelines: map[string]pipeline.PipelineProfile{}, Concurrency: ConcurrencyConfig{ - TotalLLM: defaultLLMConcurrency, - StageWorkers: map[string]int{"extract": defaultLLMConcurrency}, - defaultedExtractWorkers: defaultLLMConcurrency, + TotalLLM: defaultLLMConcurrency, + StageWorkers: map[string]int{"extract": defaultLLMConcurrency}, }, Output: OutputConfig{Directory: "./notarius-output"}, Cache: CacheConfig{ChunkPlans: ChunkPlanCacheConfig{Mode: pipeline.ChunkCacheAuto}}, @@ -97,21 +91,15 @@ func cloneIntMap(in map[string]int) map[string]int { return out } -func (c *ConcurrencyConfig) recomputeStageWorkerDefaults() { - if c == nil { - return - } - if c.StageWorkers == nil { - c.StageWorkers = make(map[string]int) - } - if !c.extractWorkersConfigured { - if value, ok := c.StageWorkers["extract"]; ok && (c.defaultedExtractWorkers == 0 || value != c.defaultedExtractWorkers) { - c.extractWorkersConfigured = true - return +func (c Config) withDerivedStageWorkers() Config { + c = cloneConfig(c) + if _, exists := c.Concurrency.StageWorkers["extract"]; !exists { + if c.Concurrency.StageWorkers == nil { + c.Concurrency.StageWorkers = make(map[string]int) } - c.StageWorkers["extract"] = c.TotalLLM - c.defaultedExtractWorkers = c.TotalLLM + c.Concurrency.StageWorkers["extract"] = c.Concurrency.TotalLLM } + return c } func clonePipelineProfile(in pipeline.PipelineProfile) pipeline.PipelineProfile { diff --git a/internal/core/config/effective_config.go b/internal/core/config/effective_config.go index 5c6e5dd6..6a0952b9 100644 --- a/internal/core/config/effective_config.go +++ b/internal/core/config/effective_config.go @@ -26,7 +26,7 @@ type EffectiveConfig struct { } func (c Config) Resolve(input ResolveInput) (EffectiveConfig, error) { - c.Concurrency.recomputeStageWorkerDefaults() + c = c.withDerivedStageWorkers() if err := c.Validate(); err != nil { return EffectiveConfig{}, err } diff --git a/internal/core/config/env.go b/internal/core/config/env.go deleted file mode 100644 index 7737b953..00000000 --- a/internal/core/config/env.go +++ /dev/null @@ -1,103 +0,0 @@ -package config - -import ( - "fmt" - "os" - "strconv" - "strings" - - "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" -) - -func LoadFromEnv() (Config, error) { - cfg := Default() - if err := cfg.applyEnvOverridesWithLookup(os.LookupEnv); err != nil { - return Config{}, err - } - return cfg, nil -} - -func (c *Config) ApplyEnvOverrides() error { - return c.applyEnvOverridesWithLookup(os.LookupEnv) -} - -func (c *Config) ApplyEnvOverridesWithLookup(lookup func(string) (string, bool)) error { - return c.applyEnvOverridesWithLookup(lookup) -} - -func (c *Config) applyEnvOverridesWithLookup(lookup func(string) (string, bool)) error { - if c == nil { - return fmt.Errorf("config must not be nil") - } - if raw, ok := lookup("NOTARIUS_TOTAL_LLM_CONCURRENCY"); ok { - value, err := parseIntEnv("NOTARIUS_TOTAL_LLM_CONCURRENCY", raw) - if err != nil { - return err - } - c.Concurrency.TotalLLM = value - } - if raw, ok := lookup("NOTARIUS_STAGE_WORKERS_EXTRACT"); ok { - value, err := parseIntEnv("NOTARIUS_STAGE_WORKERS_EXTRACT", raw) - if err != nil { - return err - } - if c.Concurrency.StageWorkers == nil { - c.Concurrency.StageWorkers = make(map[string]int) - } - c.Concurrency.StageWorkers["extract"] = value - c.Concurrency.extractWorkersConfigured = true - } - c.Concurrency.recomputeStageWorkerDefaults() - if raw, ok := lookup("NOTARIUS_OUTPUT_DIR"); ok { - c.Output.Directory = strings.TrimSpace(raw) - if c.Output.Directory == "" { - return fmt.Errorf("NOTARIUS_OUTPUT_DIR: must not be empty") - } - if strings.ContainsRune(c.Output.Directory, '\x00') { - return fmt.Errorf("NOTARIUS_OUTPUT_DIR: must not contain NUL") - } - } - if raw, ok := lookup("NOTARIUS_CACHE_CHUNK_PLANS_MODE"); ok { - mode, err := pipeline.ParseChunkCacheMode(raw) - if err != nil { - return fmt.Errorf("NOTARIUS_CACHE_CHUNK_PLANS_MODE: %w", err) - } - c.Cache.ChunkPlans.Mode = mode - } - if raw, ok := lookup("NOTARIUS_CACHE_CHUNK_PLANS_DIR"); ok { - c.Cache.ChunkPlans.Directory = cleanOptionalPath(raw) - if c.Cache.ChunkPlans.Directory == "" { - return fmt.Errorf("NOTARIUS_CACHE_CHUNK_PLANS_DIR: must not be empty") - } - if strings.ContainsRune(c.Cache.ChunkPlans.Directory, '\x00') { - return fmt.Errorf("NOTARIUS_CACHE_CHUNK_PLANS_DIR: must not contain NUL") - } - } - if raw, ok := lookup("NOTARIUS_CACHE_CHECKPOINTS_DIR"); ok { - c.Cache.Checkpoints.Directory = cleanOptionalPath(raw) - if c.Cache.Checkpoints.Directory == "" { - return fmt.Errorf("NOTARIUS_CACHE_CHECKPOINTS_DIR: must not be empty") - } - if strings.ContainsRune(c.Cache.Checkpoints.Directory, '\x00') { - return fmt.Errorf("NOTARIUS_CACHE_CHECKPOINTS_DIR: must not contain NUL") - } - } - if raw, ok := lookup("NOTARIUS_DEBUG_DIR"); ok { - c.Debug.Directory = strings.TrimSpace(raw) - if c.Debug.Directory == "" { - return fmt.Errorf("NOTARIUS_DEBUG_DIR: must not be empty") - } - if strings.ContainsRune(c.Debug.Directory, '\x00') { - return fmt.Errorf("NOTARIUS_DEBUG_DIR: must not contain NUL") - } - } - return nil -} - -func parseIntEnv(name string, raw string) (int, error) { - value, err := strconv.Atoi(strings.TrimSpace(raw)) - if err != nil { - return 0, fmt.Errorf("%s: must be an integer", name) - } - return value, nil -} diff --git a/internal/core/config/env_contract_test.go b/internal/core/config/env_contract_test.go deleted file mode 100644 index 666673e7..00000000 --- a/internal/core/config/env_contract_test.go +++ /dev/null @@ -1,239 +0,0 @@ -package config - -import ( - "errors" - "reflect" - "strings" - "testing" - - "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" -) - -func TestPrecedenceFileValuesOverrideBuiltInDefaults(t *testing.T) { - cfg := applyFileConfig(t, `version: 4 -concurrency: - total_llm: 4 - stage_workers: - extract: 2 -output: - directory: ./file-output -cache: - chunk_plans: - directory: ./file-plans - mode: refresh - checkpoints: - directory: ./file-checkpoints -debug: - directory: ./file-debug -`) - if cfg.Concurrency.TotalLLM != 4 || cfg.Concurrency.StageWorkers["extract"] != 2 || - cfg.Output.Directory != "./file-output" || cfg.Cache.ChunkPlans.Directory != "file-plans" || - cfg.Cache.ChunkPlans.Mode != pipeline.ChunkCacheRefresh || cfg.Cache.Checkpoints.Directory != "file-checkpoints" || - cfg.Debug.Directory != "./file-debug" { - t.Fatalf("file values did not override defaults: %#v", cfg) - } -} - -func TestPrecedenceOperationalEnvironmentOverridesFileValues(t *testing.T) { - cfg := applyFileConfig(t, `version: 4 -concurrency: - total_llm: 2 - stage_workers: - extract: 1 -output: - directory: ./file-output -cache: - chunk_plans: - directory: ./file-plans - mode: refresh - checkpoints: - directory: ./file-checkpoints -debug: - directory: ./file-debug -`) - env := map[string]string{ - "NOTARIUS_TOTAL_LLM_CONCURRENCY": "8", - "NOTARIUS_STAGE_WORKERS_EXTRACT": "6", - "NOTARIUS_OUTPUT_DIR": "/env/output", - "NOTARIUS_CACHE_CHUNK_PLANS_MODE": "bypass", - "NOTARIUS_CACHE_CHUNK_PLANS_DIR": "/env/plans", - "NOTARIUS_CACHE_CHECKPOINTS_DIR": "/env/checkpoints", - "NOTARIUS_DEBUG_DIR": "/env/debug", - } - if err := cfg.ApplyEnvOverridesWithLookup(lookupValues(env)); err != nil { - t.Fatal(err) - } - if cfg.Concurrency.TotalLLM != 8 || cfg.Concurrency.StageWorkers["extract"] != 6 || - cfg.Output.Directory != "/env/output" || cfg.Cache.ChunkPlans.Directory != "/env/plans" || - cfg.Cache.ChunkPlans.Mode != pipeline.ChunkCacheBypass || cfg.Cache.Checkpoints.Directory != "/env/checkpoints" || - cfg.Debug.Directory != "/env/debug" { - t.Fatalf("environment values did not override file values: %#v", cfg) - } -} - -func TestPrecedenceExtractWorkersFollowEffectiveConcurrencyUnlessExplicit(t *testing.T) { - tests := []struct { - name string - file string - env map[string]string - wantTotal int - wantWorker int - }{ - { - name: "default follows environment total", - file: "version: 4\n", - env: map[string]string{"NOTARIUS_TOTAL_LLM_CONCURRENCY": "5"}, - wantTotal: 5, - wantWorker: 5, - }, - { - name: "file worker is retained", - file: "version: 4\nconcurrency:\n total_llm: 3\n stage_workers:\n extract: 2\n", - env: map[string]string{"NOTARIUS_TOTAL_LLM_CONCURRENCY": "6"}, - wantTotal: 6, - wantWorker: 2, - }, - { - name: "environment worker is retained", - file: "version: 4\nconcurrency:\n total_llm: 2\n", - env: map[string]string{ - "NOTARIUS_TOTAL_LLM_CONCURRENCY": "6", - "NOTARIUS_STAGE_WORKERS_EXTRACT": "4", - }, - wantTotal: 6, - wantWorker: 4, - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - cfg := applyFileConfig(t, tt.file) - if err := cfg.ApplyEnvOverridesWithLookup(lookupValues(tt.env)); err != nil { - t.Fatal(err) - } - if cfg.Concurrency.TotalLLM != tt.wantTotal || cfg.Concurrency.StageWorkers["extract"] != tt.wantWorker { - t.Fatalf("concurrency = %#v, want total %d and extract %d", cfg.Concurrency, tt.wantTotal, tt.wantWorker) - } - }) - } -} - -func TestPrecedenceEmptyFileCacheDirectoriesDeferPerUserResolution(t *testing.T) { - cfg := applyFileConfig(t, `version: 4 -cache: - chunk_plans: - directory: "" - checkpoints: - directory: "" -`) - if err := cfg.Validate(); err != nil { - t.Fatalf("empty file cache directories should be valid: %v", err) - } - if cfg.Cache.ChunkPlans.Directory != "" || cfg.Cache.Checkpoints.Directory != "" { - t.Fatalf("empty cache directories were not preserved for deferred resolution: %#v", cfg.Cache) - } - resolver := func() (string, error) { return "/user/cache", nil } - chunkPlans, err := DefaultChunkPlanRoot(resolver) - if err != nil { - t.Fatal(err) - } - checkpoints, err := DefaultCheckpointRoot(resolver) - if err != nil { - t.Fatal(err) - } - if chunkPlans != "/user/cache/notarius/chunk-plans" || checkpoints != "/user/cache/notarius/checkpoints" { - t.Fatalf("deferred cache roots = %q, %q", chunkPlans, checkpoints) - } -} - -func TestDefaultCacheRootsRejectInvalidUserCacheResolvers(t *testing.T) { - tests := []struct { - name string - resolver func() (string, error) - want string - }{ - {name: "nil resolver", want: "must not be nil"}, - { - name: "resolver failure", - resolver: func() (string, error) { - return "", errors.New("cache home unavailable") - }, - want: "resolve user cache directory", - }, - {name: "empty directory", resolver: func() (string, error) { return " ", nil }, want: "must not be empty"}, - } - families := []struct { - name string - root func(func() (string, error)) (string, error) - }{ - {name: "chunk plans", root: DefaultChunkPlanRoot}, - {name: "checkpoints", root: DefaultCheckpointRoot}, - } - - for _, family := range families { - for _, tt := range tests { - t.Run(family.name+"/"+tt.name, func(t *testing.T) { - _, err := family.root(tt.resolver) - if err == nil || !strings.Contains(err.Error(), tt.want) { - t.Fatalf("error = %v, want substring %q", err, tt.want) - } - }) - } - } -} - -func TestEnvEmptyDirectoryOverridesAreErrors(t *testing.T) { - tests := []string{ - "NOTARIUS_OUTPUT_DIR", - "NOTARIUS_CACHE_CHUNK_PLANS_DIR", - "NOTARIUS_CACHE_CHECKPOINTS_DIR", - "NOTARIUS_DEBUG_DIR", - } - for _, name := range tests { - t.Run(name, func(t *testing.T) { - cfg := Default() - err := cfg.ApplyEnvOverridesWithLookup(lookupValues(map[string]string{name: " \t"})) - if err == nil || !strings.Contains(err.Error(), name) { - t.Fatalf("error = %v, want responsible environment variable", err) - } - }) - } -} - -func TestEnvInvalidIntegersAndChunkCacheModesReportTheirNames(t *testing.T) { - tests := map[string]string{ - "NOTARIUS_TOTAL_LLM_CONCURRENCY": "not-an-integer", - "NOTARIUS_STAGE_WORKERS_EXTRACT": "not-an-integer", - "NOTARIUS_CACHE_CHUNK_PLANS_MODE": "not-a-cache-mode", - } - for name, value := range tests { - t.Run(name, func(t *testing.T) { - cfg := Default() - err := cfg.ApplyEnvOverridesWithLookup(lookupValues(map[string]string{name: value})) - if err == nil || !strings.Contains(err.Error(), name) { - t.Fatalf("error = %v, want responsible environment variable", err) - } - }) - } -} - -func TestEnvRemovedProviderVariablesAreIgnored(t *testing.T) { - before := Default() - cfg := Default() - removed := map[string]string{ - "NOTARIUS_LLM_DEFAULT_ENDPOINT": "ignored-provider-setting", - "NOTARIUS_LLM_DEFAULT_MODEL": "ignored-provider-setting", - } - if err := cfg.ApplyEnvOverridesWithLookup(lookupValues(removed)); err != nil { - t.Fatal(err) - } - if !reflect.DeepEqual(cfg, before) { - t.Fatalf("removed provider variables changed configuration: %#v", cfg) - } -} - -func lookupValues(values map[string]string) func(string) (string, bool) { - return func(name string) (string, bool) { - value, ok := values[name] - return value, ok - } -} diff --git a/internal/core/config/file_config.go b/internal/core/config/file_config.go deleted file mode 100644 index 81d4e28d..00000000 --- a/internal/core/config/file_config.go +++ /dev/null @@ -1,902 +0,0 @@ -package config - -import ( - "bytes" - "fmt" - "io" - "os" - "path/filepath" - "sort" - "strings" - - "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" - "gopkg.in/yaml.v3" -) - -type FileConfig struct { - Version int `yaml:"version"` - PromptKit *FilePromptKitConfig `yaml:"promptkit,omitempty"` - Pipelines map[string]FilePipelineProfile `yaml:"pipelines,omitempty"` - Concurrency *FileConcurrencyConfig `yaml:"concurrency,omitempty"` - Output *FileOutputConfig `yaml:"output,omitempty"` - Cache *FileCacheConfig `yaml:"cache,omitempty"` - Debug *FileDebugConfig `yaml:"debug,omitempty"` -} - -type FilePromptKitConfig struct { - ProfileDir *string `yaml:"profile_dir,omitempty"` - ProfileFile *string `yaml:"profile_file,omitempty"` - LocalBackend *FilePromptKitLocalBackendConfig `yaml:"local_backend,omitempty"` -} - -type FilePromptKitLocalBackendConfig struct { - Endpoint *string `yaml:"endpoint,omitempty"` - ConcurrencyLimit *int `yaml:"concurrency_limit,omitempty"` -} - -type FilePipelineProfile struct { - LLMProfile *string `yaml:"llm_profile,omitempty"` - StructuredOutputRepairAttempts *int `yaml:"structured_output_repair_attempts,omitempty"` - ValidationPolicy *pipeline.ValidationPolicyOverride `yaml:"validation_policy,omitempty"` - Input fileModuleBinding `yaml:"input"` - Chunk *fileModuleBinding `yaml:"chunk,omitempty"` - Artifacts map[string]FileArtifactLaneProfile `yaml:"artifacts,omitempty"` - Steps []FilePipelineStepProfile `yaml:"steps,omitempty"` - Output *fileModuleBinding `yaml:"output,omitempty"` - References map[string]fileReferenceSource `yaml:"references,omitempty"` - artifactsSet bool `yaml:"-"` - stepsSet bool `yaml:"-"` - llmProfileSet bool `yaml:"-"` -} - -func (p *FilePipelineProfile) UnmarshalYAML(node *yaml.Node) error { - if err := validateStructuredOutputRepairAttemptsNode(node, "pipeline profile"); err != nil { - return err - } - type plainFilePipelineProfile FilePipelineProfile - var decoded plainFilePipelineProfile - seen, err := decodeKnownMapping(node, &decoded, map[string]struct{}{ - "llm_profile": {}, "structured_output_repair_attempts": {}, "validation_policy": {}, "input": {}, "chunk": {}, "artifacts": {}, "steps": {}, "output": {}, "references": {}, - }, "pipeline profile") - if err != nil { - return err - } - *p = FilePipelineProfile(decoded) - if validationPolicyNode, ok := mappingValue(node, "validation_policy"); ok { - policy, err := parseValidationPolicy(validationPolicyNode, "pipeline profile") - if err != nil { - return err - } - p.ValidationPolicy = policy - } - _, p.artifactsSet = seen["artifacts"] - _, p.stepsSet = seen["steps"] - _, p.llmProfileSet = seen["llm_profile"] - return nil -} - -func (s *FilePipelineStepProfile) UnmarshalYAML(node *yaml.Node) error { - type plainFilePipelineStepProfile FilePipelineStepProfile - var decoded plainFilePipelineStepProfile - if _, err := decodeKnownMapping(node, &decoded, map[string]struct{}{ - "id": {}, "artifacts": {}, "references": {}, - }, "pipeline step"); err != nil { - return err - } - *s = FilePipelineStepProfile(decoded) - return nil -} - -func (l *FileArtifactLaneProfile) UnmarshalYAML(node *yaml.Node) error { - type plainFileArtifactLaneProfile FileArtifactLaneProfile - var decoded plainFileArtifactLaneProfile - if _, err := decodeKnownMapping(node, &decoded, map[string]struct{}{ - "extract": {}, "merge": {}, "normalize": {}, "validators": {}, "references": {}, - }, "artifact lane"); err != nil { - return err - } - *l = FileArtifactLaneProfile(decoded) - return nil -} - -func decodeKnownMapping(node *yaml.Node, target any, allowed map[string]struct{}, context string) (map[string]struct{}, error) { - if node.Kind != yaml.MappingNode { - return nil, fmt.Errorf("%s must be an object", context) - } - if err := node.Decode(target); err != nil { - return nil, err - } - seen := make(map[string]struct{}, len(node.Content)/2) - for i := 0; i < len(node.Content); i += 2 { - key := node.Content[i].Value - if _, exists := seen[key]; exists { - return nil, fmt.Errorf("%s field %q is duplicated", context, key) - } - if _, ok := allowed[key]; !ok { - return nil, fmt.Errorf("field %s not found in %s", key, context) - } - seen[key] = struct{}{} - } - return seen, nil -} - -type FilePipelineStepProfile struct { - ID string `yaml:"id"` - Artifacts map[string]FileArtifactLaneProfile `yaml:"artifacts"` - References map[string]fileReferenceSource `yaml:"references,omitempty"` -} - -type FileArtifactLaneProfile struct { - Extract fileModuleBinding `yaml:"extract"` - Merge *fileModuleBinding `yaml:"merge,omitempty"` - Normalize *fileModuleBinding `yaml:"normalize,omitempty"` - Validators []fileModuleBinding `yaml:"validators,omitempty"` - References map[string]fileReferenceSource `yaml:"references,omitempty"` -} - -type FileConcurrencyConfig struct { - TotalLLM *int `yaml:"total_llm,omitempty"` - StageWorkers map[string]int `yaml:"stage_workers,omitempty"` -} - -type FileOutputConfig struct { - Directory *string `yaml:"directory,omitempty"` -} -type FileCacheConfig struct { - ChunkPlans *FileChunkPlanCacheConfig `yaml:"chunk_plans,omitempty"` - Checkpoints *FileCheckpointCacheConfig `yaml:"checkpoints,omitempty"` -} -type FileChunkPlanCacheConfig struct { - Directory *string `yaml:"directory,omitempty"` - Mode *string `yaml:"mode,omitempty"` -} -type FileCheckpointCacheConfig struct { - Enabled *bool `yaml:"enabled,omitempty"` - Directory *string `yaml:"directory,omitempty"` -} -type FileDebugConfig struct { - Directory *string `yaml:"directory,omitempty"` -} - -type fileModuleBinding struct { - Module string - LLMProfile string - StructuredOutputRepairAttempts *int - ValidationPolicy *pipeline.ValidationPolicyOverride - Retries int - Options map[string]any - References map[string]fileReferenceSource - Validators pipeline.ValidatorOverride -} - -type fileReferenceSource struct { - path string - artifact *pipeline.ArtifactReference -} - -func (source *fileReferenceSource) UnmarshalYAML(node *yaml.Node) error { - if source == nil { - return fmt.Errorf("reference source must not be nil") - } - switch node.Kind { - case yaml.ScalarNode: - if node.Tag != "!!str" { - return fmt.Errorf("external reference path must be a string") - } - path := strings.TrimSpace(node.Value) - if path == "" { - return fmt.Errorf("external reference path must not be empty") - } - source.path = path - source.artifact = nil - return nil - case yaml.MappingNode: - if len(node.Content) != 2 || node.Content[0].Value != "artifact" { - return fmt.Errorf("reference source mapping must contain only artifact") - } - artifactNode := node.Content[1] - if artifactNode.Kind != yaml.MappingNode { - return fmt.Errorf("artifact reference must be an object") - } - var step, lane string - seen := map[string]bool{} - for i := 0; i < len(artifactNode.Content); i += 2 { - key := artifactNode.Content[i].Value - value := artifactNode.Content[i+1] - if seen[key] { - return fmt.Errorf("artifact reference field %q is duplicated", key) - } - seen[key] = true - if value.Tag != "!!str" { - return fmt.Errorf("artifact reference field %q must be a string", key) - } - switch key { - case "step": - step = strings.TrimSpace(value.Value) - case "lane": - lane = strings.TrimSpace(value.Value) - default: - return fmt.Errorf("field %s not found in artifact reference", key) - } - } - if step == "" || lane == "" { - return fmt.Errorf("artifact reference step and lane must not be empty") - } - source.path = "" - source.artifact = &pipeline.ArtifactReference{Step: step, Lane: lane} - return nil - default: - return fmt.Errorf("reference source must be a string or object") - } -} - -func (source fileReferenceSource) toPipelineSource() pipeline.ReferenceSource { - if source.artifact != nil { - artifact := *source.artifact - return pipeline.ReferenceSource{Artifact: &artifact} - } - return pipeline.ExternalReference(source.path) -} - -func fileReferenceSourcesToPipeline(values map[string]fileReferenceSource) map[string]pipeline.ReferenceSource { - if len(values) == 0 { - return nil - } - out := make(map[string]pipeline.ReferenceSource, len(values)) - for key, value := range values { - out[strings.TrimSpace(key)] = value.toPipelineSource() - } - return out -} - -func (b *fileModuleBinding) UnmarshalYAML(node *yaml.Node) error { - switch node.Kind { - case yaml.ScalarNode: - var module string - if err := node.Decode(&module); err != nil { - return fmt.Errorf("module binding must be a string or object") - } - b.Module = strings.TrimSpace(module) - return nil - case yaml.MappingNode: - seen := make(map[string]struct{}, len(node.Content)/2) - for i := 0; i < len(node.Content); i += 2 { - keyNode := node.Content[i] - valueNode := node.Content[i+1] - if _, exists := seen[keyNode.Value]; exists { - return fmt.Errorf("module binding field %q is duplicated", keyNode.Value) - } - seen[keyNode.Value] = struct{}{} - switch keyNode.Value { - case "module": - var module string - if err := valueNode.Decode(&module); err != nil { - return err - } - b.Module = strings.TrimSpace(module) - case "llm_profile": - var llmProfile string - if err := valueNode.Decode(&llmProfile); err != nil { - return err - } - b.LLMProfile = strings.TrimSpace(llmProfile) - if b.LLMProfile == "" { - return fmt.Errorf("llm_profile must not be empty when set") - } - case "structured_output_repair_attempts": - attempts, err := parseStructuredOutputRepairAttempts(valueNode, "module binding") - if err != nil { - return err - } - b.StructuredOutputRepairAttempts = attempts - case "validation_policy": - policy, err := parseValidationPolicy(valueNode, "module binding") - if err != nil { - return err - } - b.ValidationPolicy = policy - case "retries": - var retries int - if err := valueNode.Decode(&retries); err != nil { - return err - } - b.Retries = retries - case "options": - var options map[string]any - if err := valueNode.Decode(&options); err != nil { - return err - } - b.Options = normalizeOptions(options) - case "references": - var references map[string]fileReferenceSource - if err := valueNode.Decode(&references); err != nil { - return err - } - b.References = references - case "validators": - b.Validators.Set = true - var validators []fileModuleBinding - if err := valueNode.Decode(&validators); err != nil { - return err - } - b.Validators.Validators = make([]pipeline.ModuleBinding, len(validators)) - for i, validator := range validators { - b.Validators.Validators[i] = validator.toPipelineBinding() - } - default: - return fmt.Errorf("field %s not found in module binding", keyNode.Value) - } - } - return nil - default: - return fmt.Errorf("module binding must be a string or object") - } -} - -func (b fileModuleBinding) toPipelineBinding() pipeline.ModuleBinding { - return pipeline.ModuleBinding{ - Module: strings.TrimSpace(b.Module), - LLMProfile: strings.TrimSpace(b.LLMProfile), - StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(b.StructuredOutputRepairAttempts), - ValidationPolicy: cloneValidationPolicyOverride(b.ValidationPolicy), - Retries: b.Retries, - Options: cloneOptions(b.Options), - References: fileReferenceSourcesToPipeline(b.References), - Validators: cloneValidatorOverride(b.Validators), - } -} - -func mappingValue(node *yaml.Node, key string) (*yaml.Node, bool) { - for i := 0; i < len(node.Content); i += 2 { - if node.Content[i].Value == key { - return node.Content[i+1], true - } - } - return nil, false -} - -func parseValidationPolicy(node *yaml.Node, context string) (*pipeline.ValidationPolicyOverride, error) { - if node == nil || node.Tag == "!!null" || node.Kind != yaml.MappingNode { - return nil, fmt.Errorf("%s validation_policy must be an object", context) - } - policy := &pipeline.ValidationPolicyOverride{} - seen := make(map[string]struct{}, len(node.Content)/2) - for i := 0; i < len(node.Content); i += 2 { - key := node.Content[i].Value - value := node.Content[i+1] - if _, exists := seen[key]; exists { - return nil, fmt.Errorf("%s validation_policy field %q is duplicated", context, key) - } - seen[key] = struct{}{} - if value.Tag == "!!null" || value.Kind != yaml.ScalarNode || value.Tag != "!!str" { - return nil, fmt.Errorf("%s validation_policy.%s must be a string", context, key) - } - switch key { - case "producer_structural_failure": - value := pipeline.ProducerStructuralFailureAction(value.Value) - policy.ProducerStructuralFailure = &value - case "semantic_rejection": - value := pipeline.SemanticRejectionAction(value.Value) - policy.SemanticRejection = &value - case "validator_failure": - value := pipeline.ValidatorFailureAction(value.Value) - policy.ValidatorFailure = &value - default: - return nil, fmt.Errorf("field %s not found in %s validation_policy", key, context) - } - } - if err := policy.Validate(); err != nil { - return nil, fmt.Errorf("%s validation_policy: %w", context, err) - } - return policy, nil -} - -func validateStructuredOutputRepairAttemptsNode(node *yaml.Node, context string) error { - if node.Kind != yaml.MappingNode { - return fmt.Errorf("%s must be an object", context) - } - for i := 0; i < len(node.Content); i += 2 { - if node.Content[i].Value != "structured_output_repair_attempts" { - continue - } - if _, err := parseStructuredOutputRepairAttempts(node.Content[i+1], context); err != nil { - return err - } - } - return nil -} - -func parseStructuredOutputRepairAttempts(node *yaml.Node, context string) (*int, error) { - if node.Tag == "!!null" { - return nil, fmt.Errorf("%s structured_output_repair_attempts must not be null", context) - } - if node.Kind != yaml.ScalarNode || node.Tag != "!!int" { - return nil, fmt.Errorf("%s structured_output_repair_attempts must be an integer", context) - } - var attempts int - if err := node.Decode(&attempts); err != nil { - return nil, fmt.Errorf("%s structured_output_repair_attempts must be an integer: %w", context, err) - } - if attempts < 0 || attempts > 3 { - return nil, fmt.Errorf("%s structured_output_repair_attempts must be between zero and three", context) - } - return &attempts, nil -} - -func cloneStructuredOutputRepairAttempts(attempts *int) *int { - if attempts == nil { - return nil - } - value := *attempts - return &value -} - -func LoadFileConfig(path string) (FileConfig, error) { - data, err := os.ReadFile(path) - if err != nil { - return FileConfig{}, fmt.Errorf("read config file %q: %w", path, err) - } - cfg, err := ParseFileConfigYAML(data) - if err != nil { - return FileConfig{}, fmt.Errorf("parse config file %q: %w", path, err) - } - return cfg, nil -} - -func ParseFileConfigYAML(data []byte) (FileConfig, error) { - var header struct { - Version int `yaml:"version"` - } - if err := yaml.Unmarshal(data, &header); err != nil { - return FileConfig{}, fmt.Errorf("decode yaml version header: %w", err) - } - if header.Version == 0 { - return FileConfig{}, fmt.Errorf("config version is required") - } - if header.Version == 2 { - return FileConfig{}, fmt.Errorf("config version 2 is no longer supported; migrate the file using the version 2-to-3 migration in docs/config.md") - } - if header.Version == 3 { - return FileConfig{}, fmt.Errorf("config version 3 is no longer supported; change \"version: 3\" to \"version: 4\" and rename \"scriptorium:\" to \"promptkit:\"") - } - if header.Version != SupportedFileConfigVersion { - return FileConfig{}, fmt.Errorf("unsupported config version %d (supported version is %d)", header.Version, SupportedFileConfigVersion) - } - var fileCfg FileConfig - decoder := yaml.NewDecoder(bytes.NewReader(data)) - decoder.KnownFields(true) - if err := decoder.Decode(&fileCfg); err != nil { - return FileConfig{}, fmt.Errorf("decode yaml: %w", err) - } - var trailing any - if err := decoder.Decode(&trailing); err == nil { - return FileConfig{}, fmt.Errorf("config must contain exactly one YAML document") - } else if err != io.EOF { - return FileConfig{}, fmt.Errorf("decode trailing yaml document: %w", err) - } - return fileCfg, nil -} - -func (c *Config) ApplyFileConfig(fileCfg FileConfig) error { - return c.applyFileConfigWithLookup(fileCfg, os.LookupEnv) -} - -func (c *Config) ApplyFileConfigWithLookup(fileCfg FileConfig, lookup func(string) (string, bool)) error { - return c.applyFileConfigWithLookup(fileCfg, lookup) -} - -func (c *Config) applyFileConfigWithLookup(fileCfg FileConfig, lookup func(string) (string, bool)) error { - _ = lookup - if c == nil { - return fmt.Errorf("config must not be nil") - } - if fileCfg.Version != SupportedFileConfigVersion { - return fmt.Errorf("unsupported config version %d", fileCfg.Version) - } - if c.Pipelines == nil { - c.Pipelines = map[string]pipeline.PipelineProfile{} - } - - pipelineIDs, rawPipelineIDs, err := normalizedMapKeys(fileCfg.Pipelines, "pipeline id") - if err != nil { - return err - } - for _, pipelineID := range pipelineIDs { - filePipeline := fileCfg.Pipelines[rawPipelineIDs[pipelineID]] - hasArtifacts := filePipeline.artifactsSet || filePipeline.Artifacts != nil - hasSteps := filePipeline.stepsSet || filePipeline.Steps != nil - if hasArtifacts && hasSteps { - return fmt.Errorf("pipeline %q must not declare both artifacts and steps", pipelineID) - } - if hasSteps && len(filePipeline.Steps) == 0 { - return fmt.Errorf("pipeline %q must declare at least one ordered step", pipelineID) - } - if hasSteps { - seenSteps := make(map[string]struct{}, len(filePipeline.Steps)) - seenLanes := make(map[string]struct{}) - for index, step := range filePipeline.Steps { - stepID := strings.TrimSpace(step.ID) - if stepID == "" { - return fmt.Errorf("pipeline %q step[%d] id must not be empty", pipelineID, index) - } - if _, ok := seenSteps[stepID]; ok { - return fmt.Errorf("pipeline %q step id %q is duplicated after trimming", pipelineID, stepID) - } - seenSteps[stepID] = struct{}{} - laneIDs, rawLaneIDs, err := normalizedMapKeys(step.Artifacts, fmt.Sprintf("pipeline %q step %q artifact lane id", pipelineID, stepID)) - if err != nil { - return err - } - for _, laneID := range laneIDs { - if _, ok := seenLanes[laneID]; ok { - return fmt.Errorf("pipeline %q artifact lane id %q is duplicated across steps", pipelineID, laneID) - } - seenLanes[laneID] = struct{}{} - fileLane := step.Artifacts[rawLaneIDs[laneID]] - if err := validateFileLaneReferences(pipelineID, stepID, laneID, fileLane); err != nil { - return err - } - } - if err := validateFileReferenceSources(step.References, fmt.Sprintf("pipeline %q step %q reference slot", pipelineID, stepID)); err != nil { - return err - } - } - } - if _, _, err := normalizedMapKeys(filePipeline.Artifacts, fmt.Sprintf("pipeline %q artifact lane id", pipelineID)); err != nil { - return err - } - if err := validateFileReferenceSources(filePipeline.References, fmt.Sprintf("pipeline %q reference slot", pipelineID)); err != nil { - return err - } - if filePipeline.Chunk != nil { - if _, _, err := normalizedMapKeys(filePipeline.Chunk.References, fmt.Sprintf("pipeline %q chunk reference slot", pipelineID)); err != nil { - return err - } - } - if _, _, err := normalizedMapKeys(filePipeline.Input.References, fmt.Sprintf("pipeline %q input reference slot", pipelineID)); err != nil { - return err - } - if filePipeline.Output != nil { - if _, _, err := normalizedMapKeys(filePipeline.Output.References, fmt.Sprintf("pipeline %q output reference slot", pipelineID)); err != nil { - return err - } - } - for rawLaneID, fileLane := range filePipeline.Artifacts { - laneID := strings.TrimSpace(rawLaneID) - if laneID == "" { - continue - } - if _, _, err := normalizedMapKeys(fileLane.References, fmt.Sprintf("pipeline %q lane %q reference slot", pipelineID, laneID)); err != nil { - return err - } - if _, _, err := normalizedMapKeys(fileLane.Extract.References, fmt.Sprintf("pipeline %q lane %q extract reference slot", pipelineID, laneID)); err != nil { - return err - } - if fileLane.Merge != nil { - if _, _, err := normalizedMapKeys(fileLane.Merge.References, fmt.Sprintf("pipeline %q lane %q merge reference slot", pipelineID, laneID)); err != nil { - return err - } - } - if fileLane.Normalize != nil { - if _, _, err := normalizedMapKeys(fileLane.Normalize.References, fmt.Sprintf("pipeline %q lane %q normalize reference slot", pipelineID, laneID)); err != nil { - return err - } - } - for i, validator := range fileLane.Validators { - if _, _, err := normalizedMapKeys(validator.References, fmt.Sprintf("pipeline %q lane %q validator[%d] reference slot", pipelineID, laneID, i)); err != nil { - return err - } - } - } - } - - if fileCfg.PromptKit != nil { - if fileCfg.PromptKit.ProfileDir != nil { - value := strings.TrimSpace(*fileCfg.PromptKit.ProfileDir) - if value == "" { - return fmt.Errorf("promptkit.profile_dir must not be empty when set") - } - c.PromptKit.ProfileDir = value - } - if fileCfg.PromptKit.ProfileFile != nil { - value := strings.TrimSpace(*fileCfg.PromptKit.ProfileFile) - if value == "" { - return fmt.Errorf("promptkit.profile_file must not be empty when set") - } - c.PromptKit.ProfileFile = value - } - if fileCfg.PromptKit.LocalBackend != nil { - if fileCfg.PromptKit.LocalBackend.Endpoint == nil { - return fmt.Errorf("promptkit.local_backend.endpoint must not be empty when set") - } - endpoint := strings.TrimSpace(*fileCfg.PromptKit.LocalBackend.Endpoint) - if endpoint == "" { - return fmt.Errorf("promptkit.local_backend.endpoint must not be empty when set") - } - localBackend := PromptKitLocalBackendConfig{Endpoint: endpoint} - if fileCfg.PromptKit.LocalBackend.ConcurrencyLimit != nil { - localBackend.ConcurrencyLimit = *fileCfg.PromptKit.LocalBackend.ConcurrencyLimit - } - c.PromptKit.LocalBackend = &localBackend - } - } - - for _, pipelineID := range pipelineIDs { - filePipeline := fileCfg.Pipelines[rawPipelineIDs[pipelineID]] - llmProfile := "" - if filePipeline.llmProfileSet || filePipeline.LLMProfile != nil { - if filePipeline.LLMProfile == nil || strings.TrimSpace(*filePipeline.LLMProfile) == "" { - return fmt.Errorf("pipeline %q llm_profile must not be empty when set", pipelineID) - } - llmProfile = strings.TrimSpace(*filePipeline.LLMProfile) - } - hasSteps := filePipeline.stepsSet || filePipeline.Steps != nil - laneIDs, rawLaneIDs, err := normalizedMapKeys(filePipeline.Artifacts, fmt.Sprintf("pipeline %q artifact lane id", pipelineID)) - if err != nil { - return err - } - profile := pipeline.PipelineProfile{ - ID: pipelineID, - LLMProfile: llmProfile, - StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(filePipeline.StructuredOutputRepairAttempts), - ValidationPolicy: cloneValidationPolicyOverride(filePipeline.ValidationPolicy), - Input: filePipeline.Input.toPipelineBinding(), - Artifacts: make(map[string]pipeline.ArtifactLaneProfile, len(filePipeline.Artifacts)), - References: fileReferenceSourcesToPipeline(filePipeline.References), - } - if filePipeline.Chunk != nil { - profile.Chunk = filePipeline.Chunk.toPipelineBinding() - } - if filePipeline.Output != nil { - profile.Output = filePipeline.Output.toPipelineBinding() - } - for _, laneID := range laneIDs { - fileLane := filePipeline.Artifacts[rawLaneIDs[laneID]] - extract := fileLane.Extract.toPipelineBinding() - extract.References = mergeReferenceSources(fileReferenceSourcesToPipeline(fileLane.References), extract.References) - lane := pipeline.ArtifactLaneProfile{ - Extract: extract, - References: fileReferenceSourcesToPipeline(fileLane.References), - } - if fileLane.Merge != nil { - lane.Merge = fileLane.Merge.toPipelineBinding() - } - if fileLane.Normalize != nil { - lane.Normalize = fileLane.Normalize.toPipelineBinding() - } - if len(fileLane.Validators) > 0 { - lane.Validators = make([]pipeline.ModuleBinding, len(fileLane.Validators)) - for i, validator := range fileLane.Validators { - lane.Validators[i] = validator.toPipelineBinding() - } - } - profile.Artifacts[laneID] = lane - } - if hasSteps { - profile.Artifacts = nil - profile.Steps = make([]pipeline.PipelineStepProfile, len(filePipeline.Steps)) - for i, fileStep := range filePipeline.Steps { - stepID := strings.TrimSpace(fileStep.ID) - step := pipeline.PipelineStepProfile{ - ID: stepID, - Artifacts: make(map[string]pipeline.ArtifactLaneProfile, len(fileStep.Artifacts)), - References: fileReferenceSourcesToPipeline(fileStep.References), - } - stepLaneIDs, stepRawLaneIDs, err := normalizedMapKeys(fileStep.Artifacts, fmt.Sprintf("pipeline %q step %q artifact lane id", pipelineID, stepID)) - if err != nil { - return err - } - for _, laneID := range stepLaneIDs { - fileLane := fileStep.Artifacts[stepRawLaneIDs[laneID]] - extract := fileLane.Extract.toPipelineBinding() - extract.References = mergeReferenceSources(fileReferenceSourcesToPipeline(fileLane.References), extract.References) - lane := pipeline.ArtifactLaneProfile{Extract: extract, References: fileReferenceSourcesToPipeline(fileLane.References)} - if fileLane.Merge != nil { - lane.Merge = fileLane.Merge.toPipelineBinding() - } - if fileLane.Normalize != nil { - lane.Normalize = fileLane.Normalize.toPipelineBinding() - } - if len(fileLane.Validators) > 0 { - lane.Validators = make([]pipeline.ModuleBinding, len(fileLane.Validators)) - for index, validator := range fileLane.Validators { - lane.Validators[index] = validator.toPipelineBinding() - } - } - step.Artifacts[laneID] = lane - } - profile.Steps[i] = step - } - } - c.Pipelines[pipelineID] = profile - } - - if fileCfg.Concurrency != nil && fileCfg.Concurrency.TotalLLM != nil { - c.Concurrency.TotalLLM = *fileCfg.Concurrency.TotalLLM - } - if fileCfg.Concurrency != nil && fileCfg.Concurrency.StageWorkers != nil { - workers, configured, err := normalizeStageWorkers(fileCfg.Concurrency.StageWorkers) - if err != nil { - return err - } - c.Concurrency.StageWorkers = workers - c.Concurrency.extractWorkersConfigured = configured - } - c.Concurrency.recomputeStageWorkerDefaults() - if fileCfg.Output != nil && fileCfg.Output.Directory != nil { - c.Output.Directory = strings.TrimSpace(*fileCfg.Output.Directory) - if c.Output.Directory == "" { - return fmt.Errorf("output.directory must not be empty") - } - if strings.ContainsRune(c.Output.Directory, '\x00') { - return fmt.Errorf("output.directory must not contain NUL") - } - } - if fileCfg.Cache != nil { - if fileCfg.Cache.ChunkPlans != nil { - if fileCfg.Cache.ChunkPlans.Mode != nil { - mode, err := pipeline.ParseChunkCacheMode(*fileCfg.Cache.ChunkPlans.Mode) - if err != nil { - return fmt.Errorf("cache.chunk_plans.mode: %w", err) - } - c.Cache.ChunkPlans.Mode = mode - } - if fileCfg.Cache.ChunkPlans.Directory != nil { - c.Cache.ChunkPlans.Directory = cleanOptionalPath(*fileCfg.Cache.ChunkPlans.Directory) - if strings.ContainsRune(c.Cache.ChunkPlans.Directory, '\x00') { - return fmt.Errorf("cache.chunk_plans.directory must not contain NUL") - } - } - } - if fileCfg.Cache.Checkpoints != nil { - if fileCfg.Cache.Checkpoints.Enabled != nil { - c.Cache.Checkpoints.Enabled = *fileCfg.Cache.Checkpoints.Enabled - } - if fileCfg.Cache.Checkpoints.Directory != nil { - c.Cache.Checkpoints.Directory = cleanOptionalPath(*fileCfg.Cache.Checkpoints.Directory) - if strings.ContainsRune(c.Cache.Checkpoints.Directory, '\x00') { - return fmt.Errorf("cache.checkpoints.directory must not contain NUL") - } - } - } - } - if fileCfg.Debug != nil && fileCfg.Debug.Directory != nil { - c.Debug.Directory = strings.TrimSpace(*fileCfg.Debug.Directory) - if c.Debug.Directory == "" { - return fmt.Errorf("debug.directory must not be empty") - } - if strings.ContainsRune(c.Debug.Directory, '\x00') { - return fmt.Errorf("debug.directory must not contain NUL") - } - } - - return nil -} - -func cleanOptionalPath(value string) string { - value = strings.TrimSpace(value) - if value == "" { - return "" - } - return filepath.Clean(value) -} - -func normalizeStageWorkers(values map[string]int) (map[string]int, bool, error) { - workers := make(map[string]int, len(values)) - configured := false - for rawKey, value := range values { - key := strings.TrimSpace(rawKey) - if key == "" { - return nil, false, fmt.Errorf("concurrency.stage_workers key must not be empty") - } - if key != "extract" { - return nil, false, fmt.Errorf("concurrency.stage_workers key %q is not supported", rawKey) - } - if _, exists := workers[key]; exists { - return nil, false, fmt.Errorf("concurrency.stage_workers key %q is duplicated after trimming", key) - } - workers[key] = value - configured = true - } - return workers, configured, nil -} - -func normalizedMapKeys[T any](values map[string]T, keyName string) ([]string, map[string]string, error) { - keys := make([]string, 0, len(values)) - rawByNormalized := make(map[string]string, len(values)) - for rawID := range values { - id := strings.TrimSpace(rawID) - if id == "" { - return nil, nil, fmt.Errorf("%s must not be empty", keyName) - } - if _, ok := rawByNormalized[id]; ok { - return nil, nil, fmt.Errorf("%s %q is duplicated after trimming", keyName, id) - } - rawByNormalized[id] = rawID - keys = append(keys, id) - } - sort.Strings(keys) - return keys, rawByNormalized, nil -} - -func validateFileReferenceSources(values map[string]fileReferenceSource, context string) error { - seen := make(map[string]struct{}, len(values)) - for rawSlot, source := range values { - slot := strings.TrimSpace(rawSlot) - if slot == "" { - return fmt.Errorf("%s must not be empty", context) - } - if _, ok := seen[slot]; ok { - return fmt.Errorf("%s %q is duplicated after trimming", context, slot) - } - seen[slot] = struct{}{} - if source.artifact != nil { - if strings.TrimSpace(source.artifact.Step) == "" || strings.TrimSpace(source.artifact.Lane) == "" { - return fmt.Errorf("%s %q artifact selector step and lane must not be empty", context, slot) - } - if strings.TrimSpace(source.path) != "" { - return fmt.Errorf("%s %q must contain either an external path or artifact selector", context, slot) - } - continue - } - if strings.TrimSpace(source.path) == "" { - return fmt.Errorf("%s %q source must not be empty", context, slot) - } - } - return nil -} - -func validateFileLaneReferences(pipelineID, stepID, laneID string, lane FileArtifactLaneProfile) error { - prefix := fmt.Sprintf("pipeline %q step %q lane %q", pipelineID, stepID, laneID) - references := []struct { - label string - values map[string]fileReferenceSource - }{ - {label: "reference slot", values: lane.References}, - {label: "extract reference slot", values: lane.Extract.References}, - } - if lane.Merge != nil { - references = append(references, struct { - label string - values map[string]fileReferenceSource - }{label: "merge reference slot", values: lane.Merge.References}) - } - if lane.Normalize != nil { - references = append(references, struct { - label string - values map[string]fileReferenceSource - }{label: "normalize reference slot", values: lane.Normalize.References}) - } - for _, item := range references { - if err := validateFileReferenceSources(item.values, prefix+" "+item.label); err != nil { - return err - } - } - for index, validator := range lane.Validators { - if err := validateFileReferenceSources(validator.References, fmt.Sprintf("%s validator[%d] reference slot", prefix, index)); err != nil { - return err - } - } - return nil -} - -func mergeReferenceSources(base map[string]pipeline.ReferenceSource, override map[string]pipeline.ReferenceSource) map[string]pipeline.ReferenceSource { - if len(base) == 0 && len(override) == 0 { - return nil - } - out := make(map[string]pipeline.ReferenceSource, len(base)+len(override)) - for key, value := range base { - out[key] = value - } - for key, value := range override { - out[key] = value - } - return out -} - -func normalizeOptions(options map[string]any) map[string]any { - if len(options) == 0 { - return nil - } - return cloneOptions(options) -} diff --git a/internal/core/config/file_config_contract_test.go b/internal/core/config/file_config_contract_test.go deleted file mode 100644 index 9092ebd3..00000000 --- a/internal/core/config/file_config_contract_test.go +++ /dev/null @@ -1,928 +0,0 @@ -package config - -import ( - "encoding/json" - "fmt" - "os" - "path/filepath" - "reflect" - "strings" - "testing" - - "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" -) - -func TestDefaultReturnsDocumentedValuesAndIndependentMaps(t *testing.T) { - first := Default() - if first.Concurrency.TotalLLM != 16 || first.Concurrency.StageWorkers["extract"] != 16 { - t.Fatalf("concurrency defaults = %#v", first.Concurrency) - } - if first.Output.Directory != "./notarius-output" || first.Debug.Directory != "./notarius-debug" { - t.Fatalf("output/debug defaults = %#v, %#v", first.Output, first.Debug) - } - if first.Cache.ChunkPlans.Mode != pipeline.ChunkCacheAuto || first.Cache.ChunkPlans.Directory != "" || first.Cache.Checkpoints.Enabled || first.Cache.Checkpoints.Directory != "" { - t.Fatalf("cache defaults = %#v", first.Cache) - } - if len(first.Pipelines) != 0 { - t.Fatalf("pipeline defaults = %#v", first.Pipelines) - } - - first.Concurrency.StageWorkers["extract"] = 99 - first.Concurrency.StageWorkers["other"] = 100 - first.Pipelines["changed"] = pipeline.PipelineProfile{} - second := Default() - if second.Concurrency.StageWorkers["extract"] != 16 || len(second.Concurrency.StageWorkers) != 1 || len(second.Pipelines) != 0 { - t.Fatalf("Default() returned state shared with an earlier result: %#v", second) - } -} - -func TestFileConfigMinimalVersion4AppliesOverDefaults(t *testing.T) { - file := parseFileConfig(t, "version: 4\n") - cfg := Default() - if err := cfg.ApplyFileConfig(file); err != nil { - t.Fatal(err) - } - if cfg.Output.Directory != "./notarius-output" || cfg.Debug.Directory != "./notarius-debug" || cfg.Cache.ChunkPlans.Mode != pipeline.ChunkCacheAuto { - t.Fatalf("minimal file changed unrelated defaults: %#v", cfg) - } - if cfg.Concurrency.TotalLLM != 16 || cfg.Concurrency.StageWorkers["extract"] != 16 || len(cfg.Pipelines) != 0 { - t.Fatalf("minimal file did not retain defaults: %#v", cfg) - } -} - -func TestParseFileConfigYAMLRejectsAdditionalDocuments(t *testing.T) { - tests := []struct { - name string - source string - wantErr bool - }{ - {name: "trailing whitespace", source: "version: 4\n\n \t", wantErr: false}, - {name: "trailing comment", source: "version: 4\n# trailing comment\n", wantErr: false}, - {name: "second valid document", source: "version: 4\n---\nversion: 4\n", wantErr: true}, - {name: "second empty document", source: "version: 4\n---\n", wantErr: true}, - {name: "second malformed document", source: "version: 4\n---\nversion: [\n", wantErr: true}, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - _, err := ParseFileConfigYAML([]byte(tt.source)) - if (err != nil) != tt.wantErr { - t.Fatalf("ParseFileConfigYAML() error = %v, want error=%t", err, tt.wantErr) - } - }) - } -} - -func TestFilePipelineLLMProfileIsPresenceAwareAndDetached(t *testing.T) { - const pipelineYAML = `version: 4 -pipelines: - main: - %s - input: input - artifacts: - lane: - extract: extract -` - - t.Run("omitted", func(t *testing.T) { - file := parseFileConfig(t, fmt.Sprintf(pipelineYAML, "")) - if file.Pipelines["main"].LLMProfile != nil || file.Pipelines["main"].llmProfileSet { - t.Fatalf("parsed pipeline profile = %#v, want omitted llm profile", file.Pipelines["main"]) - } - cfg := Default() - if err := cfg.ApplyFileConfig(file); err != nil { - t.Fatal(err) - } - if got := cfg.Pipelines["main"].LLMProfile; got != "" { - t.Fatalf("pipeline llm profile = %q, want empty", got) - } - }) - - t.Run("trimmed and detached", func(t *testing.T) { - file := parseFileConfig(t, fmt.Sprintf(pipelineYAML, "llm_profile: ' configured-profile '")) - cfg := Default() - if err := cfg.ApplyFileConfig(file); err != nil { - t.Fatal(err) - } - if got := cfg.Pipelines["main"].LLMProfile; got != "configured-profile" { - t.Fatalf("pipeline llm profile = %q, want trimmed value", got) - } - *file.Pipelines["main"].LLMProfile = "changed-profile" - if got := cfg.Pipelines["main"].LLMProfile; got != "configured-profile" { - t.Fatalf("effective config aliases parsed file: %q", got) - } - if got := cloneConfig(cfg).Pipelines["main"].LLMProfile; got != "configured-profile" { - t.Fatalf("cloned pipeline llm profile = %q", got) - } - data, err := json.Marshal(cfg) - if err != nil { - t.Fatal(err) - } - var roundTripped Config - if err := json.Unmarshal(data, &roundTripped); err != nil { - t.Fatal(err) - } - if got := roundTripped.Pipelines["main"].LLMProfile; got != "configured-profile" { - t.Fatalf("round-tripped pipeline llm profile = %q", got) - } - }) - - for _, value := range []string{"''", "' '", "null"} { - t.Run("explicit empty "+value, func(t *testing.T) { - file := parseFileConfig(t, fmt.Sprintf(pipelineYAML, "llm_profile: "+value)) - cfg := Default() - err := cfg.ApplyFileConfig(file) - if err == nil || !strings.Contains(err.Error(), `pipeline "main" llm_profile must not be empty`) { - t.Fatalf("ApplyFileConfig() error = %v, want explicit-empty rejection", err) - } - }) - } -} - -func TestStructuredOutputRepairAttemptsFileConfigurationPreservesPresenceAndOwnership(t *testing.T) { - const pipelineYAML = `version: 4 -pipelines: - main: - %s - input: input - artifacts: - lane: - extract: - module: extract - structured_output_repair_attempts: 2 - validators: - - module: validator - structured_output_repair_attempts: 3 -` - - t.Run("omitted pipeline value remains absent", func(t *testing.T) { - file := parseFileConfig(t, fmt.Sprintf(pipelineYAML, "")) - if got := file.Pipelines["main"].StructuredOutputRepairAttempts; got != nil { - t.Fatalf("file pipeline repair attempts = %v, want nil", *got) - } - cfg := Default() - if err := cfg.ApplyFileConfig(file); err != nil { - t.Fatal(err) - } - if got := cfg.Pipelines["main"].StructuredOutputRepairAttempts; got != nil { - t.Fatalf("pipeline repair attempts = %v, want nil", *got) - } - if got := cfg.Pipelines["main"].Input.StructuredOutputRepairAttempts; got != nil { - t.Fatalf("scalar input binding repair attempts = %v, want nil", *got) - } - }) - - t.Run("zero is explicit and survives configuration boundaries", func(t *testing.T) { - file := parseFileConfig(t, fmt.Sprintf(pipelineYAML, "structured_output_repair_attempts: 0")) - cfg := Default() - if err := cfg.ApplyFileConfig(file); err != nil { - t.Fatal(err) - } - profile := cfg.Pipelines["main"] - if profile.StructuredOutputRepairAttempts == nil || *profile.StructuredOutputRepairAttempts != 0 { - t.Fatalf("pipeline repair attempts = %v, want explicit zero", profile.StructuredOutputRepairAttempts) - } - lane := profile.Artifacts["lane"] - if lane.Extract.StructuredOutputRepairAttempts == nil || *lane.Extract.StructuredOutputRepairAttempts != 2 { - t.Fatalf("extract repair attempts = %v, want 2", lane.Extract.StructuredOutputRepairAttempts) - } - if lane.Extract.Validators.Validators[0].StructuredOutputRepairAttempts == nil || *lane.Extract.Validators.Validators[0].StructuredOutputRepairAttempts != 3 { - t.Fatalf("validator repair attempts = %v, want 3", lane.Extract.Validators.Validators[0].StructuredOutputRepairAttempts) - } - - *file.Pipelines["main"].StructuredOutputRepairAttempts = 1 - if got := *cfg.Pipelines["main"].StructuredOutputRepairAttempts; got != 0 { - t.Fatalf("applied config aliased file configuration: got %d, want 0", got) - } - fileProfile := file.Pipelines["main"] - fileLane := fileProfile.Artifacts["lane"] - *fileLane.Extract.StructuredOutputRepairAttempts = 1 - *fileLane.Extract.Validators.Validators[0].StructuredOutputRepairAttempts = 1 - fileProfile.Artifacts["lane"] = fileLane - file.Pipelines["main"] = fileProfile - configuredLane := cfg.Pipelines["main"].Artifacts["lane"] - if got := *configuredLane.Extract.StructuredOutputRepairAttempts; got != 2 { - t.Fatalf("configured extract aliased file configuration: got %d, want 2", got) - } - if got := *configuredLane.Extract.Validators.Validators[0].StructuredOutputRepairAttempts; got != 3 { - t.Fatalf("configured validator aliased file configuration: got %d, want 3", got) - } - cloned := cloneConfig(cfg) - *cloned.Pipelines["main"].StructuredOutputRepairAttempts = 1 - if got := *cfg.Pipelines["main"].StructuredOutputRepairAttempts; got != 0 { - t.Fatalf("cloned config aliased source configuration: got %d, want 0", got) - } - redacted := cfg.Redacted() - *redacted.Pipelines["main"].StructuredOutputRepairAttempts = 1 - if got := *cfg.Pipelines["main"].StructuredOutputRepairAttempts; got != 0 { - t.Fatalf("redacted config aliased source configuration: got %d, want 0", got) - } - summary := cfg.RedactedSummaryPayload().(Config) - if summary.Pipelines["main"].StructuredOutputRepairAttempts == nil || *summary.Pipelines["main"].StructuredOutputRepairAttempts != 0 { - t.Fatalf("redacted summary pipeline repair attempts = %v, want explicit zero", summary.Pipelines["main"].StructuredOutputRepairAttempts) - } - - data, err := json.Marshal(cfg) - if err != nil { - t.Fatal(err) - } - var roundTripped Config - if err := json.Unmarshal(data, &roundTripped); err != nil { - t.Fatal(err) - } - roundTrippedProfile := roundTripped.Pipelines["main"] - if roundTrippedProfile.StructuredOutputRepairAttempts == nil || *roundTrippedProfile.StructuredOutputRepairAttempts != 0 { - t.Fatalf("round-tripped pipeline repair attempts = %v, want explicit zero", roundTrippedProfile.StructuredOutputRepairAttempts) - } - roundTrippedLane := roundTrippedProfile.Artifacts["lane"] - if roundTrippedLane.Extract.StructuredOutputRepairAttempts == nil || *roundTrippedLane.Extract.StructuredOutputRepairAttempts != 2 { - t.Fatalf("round-tripped extract repair attempts = %v, want 2", roundTrippedLane.Extract.StructuredOutputRepairAttempts) - } - if roundTrippedLane.Extract.Validators.Validators[0].StructuredOutputRepairAttempts == nil || *roundTrippedLane.Extract.Validators.Validators[0].StructuredOutputRepairAttempts != 3 { - t.Fatalf("round-tripped validator repair attempts = %v, want 3", roundTrippedLane.Extract.Validators.Validators[0].StructuredOutputRepairAttempts) - } - }) -} - -func TestStructuredOutputRepairAttemptsFileConfigurationRejectsInvalidValues(t *testing.T) { - const pipelineYAML = `version: 4 -pipelines: - main: - input: input - artifacts: - lane: - extract: extract - %s -` - const bindingYAML = `version: 4 -pipelines: - main: - input: - module: input - %s - artifacts: - lane: - extract: extract -` - - for _, tt := range []struct { - name string - source string - want string - }{ - {name: "null pipeline value", source: "structured_output_repair_attempts: null", want: "pipeline profile structured_output_repair_attempts must not be null"}, - {name: "fractional pipeline value", source: "structured_output_repair_attempts: 1.5", want: "pipeline profile structured_output_repair_attempts must be an integer"}, - {name: "quoted pipeline value", source: "structured_output_repair_attempts: '1'", want: "pipeline profile structured_output_repair_attempts must be an integer"}, - {name: "out of range pipeline value", source: "structured_output_repair_attempts: 4", want: "pipeline profile structured_output_repair_attempts must be between zero and three"}, - } { - t.Run(tt.name, func(t *testing.T) { - _, err := ParseFileConfigYAML([]byte(fmt.Sprintf(pipelineYAML, tt.source))) - if err == nil || !strings.Contains(err.Error(), tt.want) { - t.Fatalf("ParseFileConfigYAML() error = %v, want %q", err, tt.want) - } - }) - } - - for _, tt := range []struct { - name string - source string - want string - }{ - {name: "null binding value", source: "structured_output_repair_attempts: null", want: "module binding structured_output_repair_attempts must not be null"}, - {name: "noninteger binding value", source: "structured_output_repair_attempts: true", want: "module binding structured_output_repair_attempts must be an integer"}, - {name: "out of range binding value", source: "structured_output_repair_attempts: -1", want: "module binding structured_output_repair_attempts must be between zero and three"}, - } { - t.Run(tt.name, func(t *testing.T) { - _, err := ParseFileConfigYAML([]byte(fmt.Sprintf(bindingYAML, tt.source))) - if err == nil || !strings.Contains(err.Error(), tt.want) { - t.Fatalf("ParseFileConfigYAML() error = %v, want %q", err, tt.want) - } - }) - } -} - -func TestFileModuleBindingRejectsExplicitEmptyLLMProfile(t *testing.T) { - const configYAML = `version: 4 -pipelines: - main: - input: - module: input - llm_profile: %s - artifacts: - lane: - extract: extract -` - for _, value := range []string{"''", "' '", "null"} { - t.Run(value, func(t *testing.T) { - _, err := ParseFileConfigYAML([]byte(fmt.Sprintf(configYAML, value))) - if err == nil || !strings.Contains(err.Error(), "llm_profile must not be empty when set") { - t.Fatalf("ParseFileConfigYAML() error = %v, want explicit-empty binding profile rejection", err) - } - }) - } -} - -func TestFilePromptKitProfileSourcesSurviveConfigBoundaries(t *testing.T) { - tests := []struct { - name string - yaml string - want PromptKitConfig - }{ - { - name: "profile directory", - yaml: "version: 4\npromptkit:\n profile_dir: ' ./profiles '\n", - want: PromptKitConfig{ProfileDir: "./profiles"}, - }, - { - name: "profile file", - yaml: "version: 4\npromptkit:\n profile_file: ' ./profiles.yml '\n", - want: PromptKitConfig{ProfileFile: "./profiles.yml"}, - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - cfg := applyFileConfig(t, tt.yaml) - if cfg.PromptKit != tt.want { - t.Fatalf("PromptKit config = %#v, want %#v", cfg.PromptKit, tt.want) - } - if got := cloneConfig(cfg).PromptKit; got != tt.want { - t.Fatalf("cloned PromptKit config = %#v, want %#v", got, tt.want) - } - if got := cfg.Redacted().PromptKit; got != tt.want { - t.Fatalf("redacted PromptKit config = %#v, want %#v", got, tt.want) - } - - data, err := json.Marshal(cfg) - if err != nil { - t.Fatalf("json.Marshal() error = %v", err) - } - var payload map[string]json.RawMessage - if err := json.Unmarshal(data, &payload); err != nil { - t.Fatalf("json.Unmarshal() error = %v", err) - } - if _, ok := payload["promptkit"]; !ok { - t.Fatalf("runtime JSON keys = %v, want promptkit", payload) - } - if _, ok := payload["scriptorium"]; ok { - t.Fatalf("runtime JSON keys = %v, must not contain removed section", payload) - } - }) - } -} - -func TestFilePromptKitLocalBackendSurvivesConfigBoundaries(t *testing.T) { - tests := []struct { - name string - concurrencyYAML string - wantConcurrency int - }{ - {name: "omitted concurrency defaults to zero"}, - {name: "positive concurrency is preserved", concurrencyYAML: " concurrency_limit: 2\n", wantConcurrency: 2}, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - file := parseFileConfig(t, "version: 4\npromptkit:\n local_backend:\n endpoint: ' http://localhost:8000/v1 '\n"+tt.concurrencyYAML) - cfg := Default() - if err := cfg.ApplyFileConfig(file); err != nil { - t.Fatalf("ApplyFileConfig() error = %v", err) - } - want := PromptKitLocalBackendConfig{ - Endpoint: "http://localhost:8000/v1", - ConcurrencyLimit: tt.wantConcurrency, - } - if cfg.PromptKit.LocalBackend == nil || *cfg.PromptKit.LocalBackend != want { - t.Fatalf("local backend config = %#v, want %#v", cfg.PromptKit.LocalBackend, want) - } - - *file.PromptKit.LocalBackend.Endpoint = "http://changed.example/v1" - if file.PromptKit.LocalBackend.ConcurrencyLimit != nil { - *file.PromptKit.LocalBackend.ConcurrencyLimit = 99 - } - if *cfg.PromptKit.LocalBackend != want { - t.Fatalf("effective config aliases parsed file model: %#v", cfg.PromptKit.LocalBackend) - } - - cloned := cloneConfig(cfg) - if cloned.PromptKit.LocalBackend == cfg.PromptKit.LocalBackend || *cloned.PromptKit.LocalBackend != want { - t.Fatalf("cloned local backend = %#v, want detached %#v", cloned.PromptKit.LocalBackend, want) - } - cloned.PromptKit.LocalBackend.Endpoint = "http://clone.example/v1" - if *cfg.PromptKit.LocalBackend != want { - t.Fatalf("mutating clone changed source config: %#v", cfg.PromptKit.LocalBackend) - } - - redacted := cfg.Redacted() - if redacted.PromptKit.LocalBackend == cfg.PromptKit.LocalBackend || *redacted.PromptKit.LocalBackend != want { - t.Fatalf("redacted local backend = %#v, want detached %#v", redacted.PromptKit.LocalBackend, want) - } - - data, err := json.Marshal(cfg) - if err != nil { - t.Fatalf("json.Marshal() error = %v", err) - } - var payload struct { - PromptKit map[string]json.RawMessage `json:"promptkit"` - } - if err := json.Unmarshal(data, &payload); err != nil { - t.Fatalf("json.Unmarshal() error = %v", err) - } - localJSON, ok := payload.PromptKit["local_backend"] - if !ok { - t.Fatalf("runtime PromptKit JSON keys = %v, want local_backend", payload.PromptKit) - } - var localPayload map[string]json.RawMessage - if err := json.Unmarshal(localJSON, &localPayload); err != nil { - t.Fatalf("unmarshal local_backend JSON: %v", err) - } - if _, ok := localPayload["endpoint"]; !ok { - t.Fatalf("runtime local_backend JSON keys = %v, want endpoint", localPayload) - } - if _, ok := localPayload["concurrency_limit"]; !ok { - t.Fatalf("runtime local_backend JSON keys = %v, want concurrency_limit", localPayload) - } - }) - } -} - -func TestFilePromptKitLocalBackendRequiresEndpoint(t *testing.T) { - for _, tt := range []struct { - name string - yaml string - }{ - {name: "missing", yaml: "version: 4\npromptkit:\n local_backend: {}\n"}, - {name: "empty", yaml: "version: 4\npromptkit:\n local_backend:\n endpoint: ''\n"}, - {name: "blank", yaml: "version: 4\npromptkit:\n local_backend:\n endpoint: ' '\n"}, - } { - t.Run(tt.name, func(t *testing.T) { - file := parseFileConfig(t, tt.yaml) - cfg := Default() - err := cfg.ApplyFileConfig(file) - if err == nil || !strings.Contains(err.Error(), "promptkit.local_backend.endpoint") { - t.Fatalf("ApplyFileConfig() error = %v, want endpoint field context", err) - } - }) - } -} - -func TestFilePromptKitExplicitEmptyProfileSourcesAreRejected(t *testing.T) { - for _, field := range []string{"profile_dir", "profile_file"} { - t.Run(field, func(t *testing.T) { - file := parseFileConfig(t, "version: 4\npromptkit:\n "+field+": ''\n") - cfg := Default() - err := cfg.ApplyFileConfig(file) - if err == nil || !strings.Contains(err.Error(), "promptkit."+field+" must not be empty") { - t.Fatalf("ApplyFileConfig() error = %v, want explicit-empty rejection", err) - } - }) - } -} - -func TestFilePromptKitProfileSourcesRemainMutuallyExclusive(t *testing.T) { - cfg := applyFileConfig(t, `version: 4 -promptkit: - profile_dir: ./profiles - profile_file: ./profiles.yml -`) - if err := cfg.Validate(); err == nil || !strings.Contains(err.Error(), "promptkit profile_dir and profile_file are mutually exclusive") { - t.Fatalf("Validate() error = %v, want mutually exclusive profile sources", err) - } -} - -func TestFileConfigMissingVersionIsReportedBeforeFieldDecoding(t *testing.T) { - _, err := ParseFileConfigYAML([]byte("workspace:\n directory: /tmp/old\n")) - if err == nil || !strings.Contains(err.Error(), "config version is required") { - t.Fatalf("missing version error = %v", err) - } -} - -func TestFileConfigVersion3ReportsPromptKitMigration(t *testing.T) { - _, err := ParseFileConfigYAML([]byte("version: 3\nscriptorium:\n profile_dir: ./profiles\n")) - if err == nil || - !strings.Contains(err.Error(), `change "version: 3" to "version: 4"`) || - !strings.Contains(err.Error(), `rename "scriptorium:" to "promptkit:"`) { - t.Fatalf("version 3 error = %v, want actionable version and section migration", err) - } -} - -func TestFileConfigRejectsUnknownCurrentAndRemovedFields(t *testing.T) { - tests := []struct { - name string - yaml string - want string - }{ - { - name: "removed diagnostics", - yaml: "version: 4\ndiagnostics: {}\n", - want: "field diagnostics not found", - }, - { - name: "removed llm profiles", - yaml: "version: 4\nllm_profiles: {}\n", - want: "field llm_profiles not found", - }, - { - name: "removed scriptorium section", - yaml: "version: 4\nscriptorium: {}\n", - want: "field scriptorium not found", - }, - { - name: "version 2 migration", - yaml: "version: 2\nworkspace:\n directory: /tmp/old\n", - want: "version 2-to-3 migration", - }, - { - name: "pipeline field", - yaml: "version: 4\npipelines:\n main:\n unknown: true\n", - want: "field unknown not found", - }, - { - name: "lane field", - yaml: "version: 4\npipelines:\n main:\n artifacts:\n spells:\n unknown: true\n", - want: "field unknown not found", - }, - { - name: "module binding field", - yaml: "version: 4\npipelines:\n main:\n input:\n module: seriatim\n unknown: true\n", - want: "field unknown not found in module binding", - }, - { - name: "checkpoint field", - yaml: "version: 4\ncache:\n checkpoints:\n unknown: true\n", - want: "field unknown not found", - }, - { - name: "checkpoint enabled type", - yaml: "version: 4\ncache:\n checkpoints:\n enabled: definitely\n", - want: "cannot unmarshal", - }, - { - name: "local backend field", - yaml: "version: 4\npromptkit:\n local_backend:\n endpoint: http://localhost:8000/v1\n unknown: true\n", - want: "field unknown not found", - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - _, err := ParseFileConfigYAML([]byte(tt.yaml)) - if err == nil || !strings.Contains(err.Error(), tt.want) { - t.Fatalf("error = %v, want context %q", err, tt.want) - } - }) - } -} - -func TestFileConfigModuleBindingsPreserveFormsAndValidatorPresence(t *testing.T) { - cfg := applyFileConfig(t, `version: 4 -pipelines: - main: - input: seriatim - chunk: - module: generic - llm_profile: chunk-profile - retries: 2 - options: - max_units: 25 - references: - glossary: ./glossary.md - validators: [] - artifacts: - spells: - extract: - module: dnd/spells - options: - nested: - enabled: true - merge: appendorder - normalize: noop -`) - profile := cfg.Pipelines["main"] - if profile.Input.Module != "seriatim" || profile.Input.Validators.Set { - t.Fatalf("shorthand binding = %#v", profile.Input) - } - if profile.Chunk.Module != "generic" || profile.Chunk.LLMProfile != "chunk-profile" || profile.Chunk.Retries != 2 || - !reflect.DeepEqual(profile.Chunk.Options, map[string]any{"max_units": 25}) || - !reflect.DeepEqual(profile.Chunk.References, pipeline.ExternalReferenceMap(map[string]string{"glossary": "./glossary.md"})) { - t.Fatalf("object binding = %#v", profile.Chunk) - } - if !profile.Chunk.Validators.Set || len(profile.Chunk.Validators.Validators) != 0 { - t.Fatalf("explicit empty validators = %#v", profile.Chunk.Validators) - } - if profile.Artifacts["spells"].Extract.Module != "dnd/spells" || - !reflect.DeepEqual(profile.Artifacts["spells"].Extract.Options, map[string]any{ - "nested": map[string]any{"enabled": true}, - }) { - t.Fatalf("extract binding = %#v", profile.Artifacts["spells"].Extract) - } - if profile.Artifacts["spells"].Merge.Module != "appendorder" || profile.Artifacts["spells"].Normalize.Module != "noop" { - t.Fatalf("stage shorthand bindings = %#v", profile.Artifacts["spells"]) - } -} - -func TestFileConfigReferencePrecedenceIsRetained(t *testing.T) { - cfg := applyFileConfig(t, `version: 4 -pipelines: - main: - input: seriatim - references: - pipeline-only: ./pipeline.txt - shared: ./pipeline-shared.txt - chunk: - module: generic - references: - chunk-only: ./chunk.txt - artifacts: - spells: - references: - lane-only: ./lane.txt - shared: ./lane-shared.txt - overridden: ./lane.txt - extract: - module: dnd/spells - references: - extract-only: ./extract.txt - overridden: ./extract-overridden.txt - merge: - module: appendorder - references: - merge-only: ./merge.txt - normalize: - module: noop - references: - normalize-only: ./normalize.txt -`) - profile := cfg.Pipelines["main"] - if !reflect.DeepEqual(profile.References, pipeline.ExternalReferenceMap(map[string]string{ - "pipeline-only": "./pipeline.txt", - "shared": "./pipeline-shared.txt", - })) { - t.Fatalf("pipeline references = %#v", profile.References) - } - if !reflect.DeepEqual(profile.Chunk.References, pipeline.ExternalReferenceMap(map[string]string{"chunk-only": "./chunk.txt"})) { - t.Fatalf("chunk references = %#v", profile.Chunk.References) - } - lane := profile.Artifacts["spells"] - if !reflect.DeepEqual(lane.References, pipeline.ExternalReferenceMap(map[string]string{ - "lane-only": "./lane.txt", - "shared": "./lane-shared.txt", - "overridden": "./lane.txt", - })) { - t.Fatalf("lane compatibility references = %#v", lane.References) - } - if !reflect.DeepEqual(lane.Extract.References, pipeline.ExternalReferenceMap(map[string]string{ - "lane-only": "./lane.txt", - "shared": "./lane-shared.txt", - "overridden": "./extract-overridden.txt", - "extract-only": "./extract.txt", - })) { - t.Fatalf("extract references = %#v", lane.Extract.References) - } - if !reflect.DeepEqual(lane.Merge.References, pipeline.ExternalReferenceMap(map[string]string{"merge-only": "./merge.txt"})) || - !reflect.DeepEqual(lane.Normalize.References, pipeline.ExternalReferenceMap(map[string]string{"normalize-only": "./normalize.txt"})) { - t.Fatalf("merge/normalize references = %#v, %#v", lane.Merge.References, lane.Normalize.References) - } -} - -func TestFileConfigStageLocalValidatorsPreserveOrderAndFields(t *testing.T) { - cfg := applyFileConfig(t, `version: 4 -pipelines: - main: - input: seriatim - chunk: - module: generic - validators: - - generic/always_accept - - module: generic/valid_json - llm_profile: validator-profile - options: - schema: compact - artifacts: - spells: - extract: - module: dnd/spells - validators: - - module: extract/dnd/spells/shape - options: - strict: true - merge: - module: appendorder - validators: - - generic/always_accept - normalize: - module: noop - validators: - - module: generic/valid_json - options: - mode: normalized -`) - profile := cfg.Pipelines["main"] - chunkValidators := profile.Chunk.Validators.Validators - if !profile.Chunk.Validators.Set || len(chunkValidators) != 2 || chunkValidators[0].Module != "generic/always_accept" || - chunkValidators[1].Module != "generic/valid_json" || chunkValidators[1].LLMProfile != "validator-profile" || - !reflect.DeepEqual(chunkValidators[1].Options, map[string]any{"schema": "compact"}) { - t.Fatalf("chunk validators = %#v", profile.Chunk.Validators) - } - lane := profile.Artifacts["spells"] - if len(lane.Extract.Validators.Validators) != 1 || lane.Extract.Validators.Validators[0].Module != "extract/dnd/spells/shape" || - !reflect.DeepEqual(lane.Extract.Validators.Validators[0].Options, map[string]any{"strict": true}) { - t.Fatalf("extract validators = %#v", lane.Extract.Validators) - } - if len(lane.Merge.Validators.Validators) != 1 || lane.Merge.Validators.Validators[0].Module != "generic/always_accept" { - t.Fatalf("merge validators = %#v", lane.Merge.Validators) - } - if len(lane.Normalize.Validators.Validators) != 1 || lane.Normalize.Validators.Validators[0].Module != "generic/valid_json" || - !reflect.DeepEqual(lane.Normalize.Validators.Validators[0].Options, map[string]any{"mode": "normalized"}) { - t.Fatalf("normalize validators = %#v", lane.Normalize.Validators) - } -} - -func TestFileConfigStateSectionsApplyIndependently(t *testing.T) { - cfg := applyFileConfig(t, `version: 4 -promptkit: - profile_dir: ./profiles -concurrency: - total_llm: 7 -output: - directory: ./output -cache: - chunk_plans: - directory: ./plans - mode: bypass - checkpoints: - enabled: true - directory: ./checkpoints -debug: - directory: ./debug -`) - if cfg.PromptKit.ProfileDir != "./profiles" || cfg.PromptKit.ProfileFile != "" { - t.Fatalf("promptkit = %#v", cfg.PromptKit) - } - if cfg.Concurrency.TotalLLM != 7 || cfg.Concurrency.StageWorkers["extract"] != 7 { - t.Fatalf("concurrency = %#v", cfg.Concurrency) - } - if cfg.Output.Directory != "./output" || cfg.Cache.ChunkPlans.Directory != "plans" || cfg.Cache.ChunkPlans.Mode != pipeline.ChunkCacheBypass || - !cfg.Cache.Checkpoints.Enabled || cfg.Cache.Checkpoints.Directory != "checkpoints" || cfg.Debug.Directory != "./debug" { - t.Fatalf("state sections = %#v, %#v, %#v, %#v", cfg.Output, cfg.Cache, cfg.Debug, cfg.PromptKit) - } - if cfg.Output.Directory == cfg.Cache.ChunkPlans.Directory || cfg.Cache.ChunkPlans.Directory == cfg.Cache.Checkpoints.Directory || cfg.Cache.Checkpoints.Directory == cfg.Debug.Directory { - t.Fatal("state roots were coupled") - } -} - -func TestFileConfigCheckpointEnabledCanBeExplicitlyDisabled(t *testing.T) { - cfg := applyFileConfig(t, "version: 4\ncache:\n checkpoints:\n enabled: true\n") - if !cfg.Cache.Checkpoints.Enabled || !cloneConfig(cfg).Cache.Checkpoints.Enabled { - t.Fatalf("enabled checkpoint config was not retained: %#v", cfg.Cache.Checkpoints) - } - file := parseFileConfig(t, "version: 4\ncache:\n checkpoints:\n enabled: false\n") - if err := cfg.ApplyFileConfig(file); err != nil { - t.Fatal(err) - } - if cfg.Cache.Checkpoints.Enabled { - t.Fatalf("explicit false checkpoint config was not applied: %#v", cfg.Cache.Checkpoints) - } -} - -func TestFileConfigRejectsTrimmedKeyCollisions(t *testing.T) { - tests := []struct { - name string - yaml string - want string - }{ - { - name: "pipeline ids", - yaml: "version: 4\npipelines:\n main: {}\n ' main ': {}\n", - want: "pipeline id \"main\" is duplicated after trimming", - }, - { - name: "lane ids", - yaml: "version: 4\npipelines:\n main:\n artifacts:\n spells: {}\n ' spells ': {}\n", - want: "artifact lane id \"spells\" is duplicated after trimming", - }, - { - name: "reference slots", - yaml: "version: 4\npipelines:\n main:\n references:\n slot: ./one.txt\n ' slot ': ./two.txt\n", - want: "reference slot \"slot\" is duplicated after trimming", - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - file := parseFileConfig(t, tt.yaml) - cfg := Default() - err := cfg.ApplyFileConfig(file) - if err == nil || !strings.Contains(err.Error(), tt.want) { - t.Fatalf("ApplyFileConfig() error = %v, want context %q", err, tt.want) - } - }) - } -} - -func TestFileConfigParsesOrderedStepsAndReferenceSources(t *testing.T) { - file := parseFileConfig(t, `version: 4 -pipelines: - session: - input: seriatim - steps: - - id: identify-npcs - artifacts: - npc_registry: - extract: dnd/npc-registry - - id: grounded-events - references: - npc_registry: - artifact: - step: identify-npcs - lane: npc_registry - artifacts: - spells: - extract: dnd/spells -`) - cfg := Default() - if err := cfg.ApplyFileConfig(file); err != nil { - t.Fatal(err) - } - profile := cfg.Pipelines["session"] - if len(profile.Steps) != 2 || profile.Steps[0].ID != "identify-npcs" || profile.Steps[1].ID != "grounded-events" { - t.Fatalf("steps = %#v", profile.Steps) - } - source := profile.Steps[1].References["npc_registry"] - if source.Artifact == nil || source.Artifact.Step != "identify-npcs" || source.Artifact.Lane != "npc_registry" { - t.Fatalf("generated source = %#v", source) - } -} - -func TestFileConfigRejectsAmbiguousReferenceSourceForms(t *testing.T) { - for _, source := range []string{ - "artifact: {step: a, lane: b, extra: c}", - "artifact: {step: 1, lane: b}", - "1", - } { - _, err := ParseFileConfigYAML([]byte("version: 4\npipelines:\n p:\n input: text\n references:\n slot: " + source + "\n")) - if err == nil { - t.Fatalf("ParseFileConfigYAML(%q) error = nil", source) - } - } -} - -func TestFileConfigRejectsEmptyAndAmbiguousPipelineShapes(t *testing.T) { - tests := []struct { - name string - yaml string - want string - }{ - { - name: "empty steps", - yaml: "version: 4\npipelines:\n p:\n input: text\n steps: []\n", - want: "at least one ordered step", - }, - { - name: "both forms", - yaml: "version: 4\npipelines:\n p:\n input: text\n artifacts: {}\n steps: []\n", - want: "both artifacts and steps", - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - file := parseFileConfig(t, tt.yaml) - cfg := Default() - err := cfg.ApplyFileConfig(file) - if err == nil || !strings.Contains(err.Error(), tt.want) { - t.Fatalf("ApplyFileConfig() error = %v, want context %q", err, tt.want) - } - }) - } -} - -func TestLoadFileConfigReportsPathAndOperationContext(t *testing.T) { - dir := t.TempDir() - missing := filepath.Join(dir, "missing.yml") - _, err := LoadFileConfig(missing) - if err == nil || !strings.Contains(err.Error(), "read config file") || !strings.Contains(err.Error(), missing) { - t.Fatalf("missing-file error = %v", err) - } - - malformed := filepath.Join(dir, "malformed.yml") - if err := os.WriteFile(malformed, []byte("version: [\n"), 0o600); err != nil { - t.Fatal(err) - } - _, err = LoadFileConfig(malformed) - if err == nil || !strings.Contains(err.Error(), "parse config file") || !strings.Contains(err.Error(), malformed) { - t.Fatalf("malformed-file error = %v", err) - } -} - -func parseFileConfig(t *testing.T, source string) FileConfig { - t.Helper() - file, err := ParseFileConfigYAML([]byte(source)) - if err != nil { - t.Fatalf("ParseFileConfigYAML() error = %v", err) - } - return file -} - -func applyFileConfig(t *testing.T, source string) Config { - t.Helper() - cfg := Default() - if err := cfg.ApplyFileConfig(parseFileConfig(t, source)); err != nil { - t.Fatalf("ApplyFileConfig() error = %v", err) - } - return cfg -} diff --git a/internal/core/config/loaded_config.go b/internal/core/config/loaded_config.go index 5fb157b9..b62f11d4 100644 --- a/internal/core/config/loaded_config.go +++ b/internal/core/config/loaded_config.go @@ -87,10 +87,6 @@ func documentToConfig(document Document) (Config, error) { config.Concurrency.TotalLLM = *document.Concurrency.TotalLLM } } - if workers, ok := config.Concurrency.StageWorkers["extract"]; ok { - config.Concurrency.extractWorkersConfigured = true - config.Concurrency.defaultedExtractWorkers = workers - } if document.PromptKit != nil { if document.PromptKit.ProfileDir != nil { config.PromptKit.ProfileDir = *document.PromptKit.ProfileDir diff --git a/internal/core/config/redaction.go b/internal/core/config/redaction.go index ddbaafee..2899b975 100644 --- a/internal/core/config/redaction.go +++ b/internal/core/config/redaction.go @@ -8,25 +8,6 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" ) -func (c Config) Redacted() Config { - return redactConfig(cloneConfig(c)) -} - -func (c Config) RedactedSummaryPayload() any { - return c.Redacted() -} - -func (e EffectiveConfig) RedactedSummaryPayload() any { - return EffectiveConfig{ - Config: e.Config.Redacted(), - PipelineID: e.PipelineID, - Only: append([]string(nil), e.Only...), - ReferenceOverrides: append([]pipeline.ReferenceBinding(nil), e.ReferenceOverrides...), - ReferenceUnbinds: append([]pipeline.ReferenceUnbind(nil), e.ReferenceUnbinds...), - ResolvedPipeline: cloneResolvedPipeline(e.ResolvedPipeline), - } -} - func (e EffectiveConfig) RedactedResolvedPipelinePayload() pipeline.ResolvedPipeline { return cloneResolvedPipeline(e.ResolvedPipeline) } @@ -92,31 +73,6 @@ func cloneResolvedArtifactLane(in pipeline.ResolvedArtifactLane) pipeline.Resolv return out } -func redactConfig(cfg Config) Config { - for id, profile := range cfg.Pipelines { - profile.Input = redactBinding(profile.Input) - profile.Chunk = redactBinding(profile.Chunk) - profile.Output = redactBinding(profile.Output) - redactLanes := func(lanes map[string]pipeline.ArtifactLaneProfile) { - for laneID, lane := range lanes { - lane.Extract = redactBinding(lane.Extract) - lane.Merge = redactBinding(lane.Merge) - lane.Normalize = redactBinding(lane.Normalize) - for i := range lane.Validators { - lane.Validators[i] = redactBinding(lane.Validators[i]) - } - lanes[laneID] = lane - } - } - redactLanes(profile.Artifacts) - for i := range profile.Steps { - redactLanes(profile.Steps[i].Artifacts) - } - cfg.Pipelines[id] = profile - } - return cfg -} - func redactBinding(binding pipeline.ModuleBinding) pipeline.ModuleBinding { binding.Options = redactOptions(binding.Options) for i := range binding.Validators.Validators { diff --git a/internal/core/config/redaction_test.go b/internal/core/config/redaction_test.go index 9f74a17c..8cd8078d 100644 --- a/internal/core/config/redaction_test.go +++ b/internal/core/config/redaction_test.go @@ -114,83 +114,6 @@ func TestRedactedResolvedPipelinePayloadRedactsEveryBinding(t *testing.T) { } } -func TestRedactedEffectiveConfigPayloadDoesNotAliasSource(t *testing.T) { - binding := redactionTestBinding("effective") - effective := EffectiveConfig{ - Config: Config{Pipelines: map[string]pipeline.PipelineProfile{ - "redaction-test": {Input: binding}, - }}, - ResolvedPipeline: pipeline.ResolvedPipeline{Input: binding}, - } - - payload := effective.RedactedSummaryPayload().(EffectiveConfig) - payload.Config.Pipelines["redaction-test"].Input.Options["safe"] = "mutated" - payload.ResolvedPipeline.Input.Options["safe"] = "mutated" - - assertRedactionTestBindingUnchanged(t, effective.Config.Pipelines["redaction-test"].Input, "effective") - assertRedactionTestBindingUnchanged(t, effective.ResolvedPipeline.Input, "effective") -} - -func TestRedactedSummaryPayloadsCoverEveryEffectiveConfigBinding(t *testing.T) { - bindings := map[string]pipeline.ModuleBinding{} - for _, name := range []string{"input", "chunk", "output", "extract", "merge", "normalize", "lane-validator"} { - bindings[name] = redactionTestBinding("summary-" + name) - } - effective := EffectiveConfig{ - Config: Config{Pipelines: map[string]pipeline.PipelineProfile{ - "redaction-test": { - Input: bindings["input"], - Chunk: bindings["chunk"], - Output: bindings["output"], - Artifacts: map[string]pipeline.ArtifactLaneProfile{ - "safe-lane": { - Extract: bindings["extract"], - Merge: bindings["merge"], - Normalize: bindings["normalize"], - Validators: []pipeline.ModuleBinding{bindings["lane-validator"]}, - }, - }, - }, - }}, - ResolvedPipeline: pipeline.ResolvedPipeline{ - Input: bindings["input"], - Chunk: bindings["chunk"], - Output: bindings["output"], - Steps: []pipeline.ResolvedPipelineStep{{ - ID: "default", - ArtifactLanes: []pipeline.ResolvedArtifactLane{{ - ID: "safe-lane", - Extract: bindings["extract"], - Merge: bindings["merge"], - Normalize: bindings["normalize"], - Validators: []pipeline.ModuleBinding{bindings["lane-validator"]}, - }}, - }}, - }, - } - - payload, ok := effective.RedactedSummaryPayload().(EffectiveConfig) - if !ok { - t.Fatal("RedactedSummaryPayload() returned an unexpected type") - } - encoded, err := json.Marshal(payload) - if err != nil { - t.Fatal(err) - } - text := string(encoded) - for name := range bindings { - if strings.Contains(text, "summary-"+name+"-secret") || strings.Contains(text, "summary-"+name+"-nested-secret") { - t.Fatalf("summary payload contains sensitive option for %q: %s", name, text) - } - if !strings.Contains(text, "summary-"+name+"-safe") { - t.Fatalf("summary payload omitted safe option for %q: %s", name, text) - } - } - if !strings.Contains(text, "[REDACTED]") { - t.Fatalf("summary payload contains no redaction marker: %s", text) - } -} - func TestRedactedResolvedPipelinePayloadHandlesTypedOptionContainers(t *testing.T) { type optionMap map[string]string type optionList []optionMap @@ -225,39 +148,6 @@ func TestRedactedResolvedPipelinePayloadHandlesTypedOptionContainers(t *testing. } } -func TestRedactedEffectiveConfigPayloadOwnsValidationPolicies(t *testing.T) { - semantic := pipeline.SemanticRejectionRejectOutput - validator := pipeline.ValidatorFailureFailRun - configured := &pipeline.ValidationPolicyOverride{SemanticRejection: &semantic, ValidatorFailure: &validator} - effective := EffectiveConfig{ - Config: Config{Pipelines: map[string]pipeline.PipelineProfile{ - "main": {ValidationPolicy: configured}, - }}, - ResolvedPipeline: pipeline.ResolvedPipeline{ - ConfiguredValidationPolicy: configured, - ChunkValidationPolicy: pipeline.ValidationPolicy{ - ProducerStructuralFailure: pipeline.ProducerStructuralFailureFailRun, - SemanticRejection: pipeline.SemanticRejectionRejectOutput, - ValidatorFailure: pipeline.ValidatorFailureFailRun, - }, - }, - } - - payload := effective.RedactedSummaryPayload().(EffectiveConfig) - encoded, err := json.Marshal(payload) - if err != nil { - t.Fatal(err) - } - if !strings.Contains(string(encoded), `"configured_validation_policy":{"semantic_rejection":"reject_output","validator_failure":"fail_run"}`) || !strings.Contains(string(encoded), `"chunk_validation_policy":{"producer_structural_failure":"fail_run","semantic_rejection":"reject_output","validator_failure":"fail_run"}`) { - t.Fatalf("redacted payload omitted validation policy: %s", encoded) - } - *payload.Config.Pipelines["main"].ValidationPolicy.SemanticRejection = pipeline.SemanticRejectionFailRun - *payload.ResolvedPipeline.ConfiguredValidationPolicy.ValidatorFailure = pipeline.ValidatorFailureWarnContinue - if *effective.Config.Pipelines["main"].ValidationPolicy.SemanticRejection != pipeline.SemanticRejectionRejectOutput || *effective.ResolvedPipeline.ConfiguredValidationPolicy.ValidatorFailure != pipeline.ValidatorFailureFailRun { - t.Fatal("redacted payload aliases validation policy") - } -} - func redactionTestBinding(name string) pipeline.ModuleBinding { return pipeline.ModuleBinding{ Module: "safe-" + name, diff --git a/internal/core/config/validation.go b/internal/core/config/validation.go index 8983bd5d..65257fda 100644 --- a/internal/core/config/validation.go +++ b/internal/core/config/validation.go @@ -9,7 +9,6 @@ import ( ) func (c Config) Validate() error { - c.Concurrency.recomputeStageWorkerDefaults() if err := validatePromptKit(c.PromptKit); err != nil { return err } diff --git a/internal/core/config/validation_contract_test.go b/internal/core/config/validation_contract_test.go index d0560dc5..d810f1c1 100644 --- a/internal/core/config/validation_contract_test.go +++ b/internal/core/config/validation_contract_test.go @@ -25,7 +25,6 @@ func TestValidateConcurrencyRules(t *testing.T) { setup: func(cfg *Config) { cfg.Concurrency.TotalLLM = 3 cfg.Concurrency.StageWorkers = map[string]int{"extract": 0} - cfg.Concurrency.extractWorkersConfigured = true }, want: "stage_workers.extract must be between 1", }, @@ -34,7 +33,6 @@ func TestValidateConcurrencyRules(t *testing.T) { setup: func(cfg *Config) { cfg.Concurrency.TotalLLM = 3 cfg.Concurrency.StageWorkers = map[string]int{"extract": 4} - cfg.Concurrency.extractWorkersConfigured = true }, want: "stage_workers.extract must be between 1", }, @@ -43,7 +41,6 @@ func TestValidateConcurrencyRules(t *testing.T) { setup: func(cfg *Config) { cfg.Concurrency.TotalLLM = 3 cfg.Concurrency.StageWorkers = map[string]int{"extract": 1} - cfg.Concurrency.extractWorkersConfigured = true }, }, { @@ -51,7 +48,6 @@ func TestValidateConcurrencyRules(t *testing.T) { setup: func(cfg *Config) { cfg.Concurrency.TotalLLM = 3 cfg.Concurrency.StageWorkers = map[string]int{"extract": 3} - cfg.Concurrency.extractWorkersConfigured = true }, }, { diff --git a/internal/core/config/validation_policy_contract_test.go b/internal/core/config/validation_policy_contract_test.go index c8e71ac1..acd93531 100644 --- a/internal/core/config/validation_policy_contract_test.go +++ b/internal/core/config/validation_policy_contract_test.go @@ -4,17 +4,19 @@ import ( "strings" "testing" + configkit "gitea.maximumdirect.net/eric/ConfigKit" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" ) func TestValidationPolicyFileConfigurationIsStrictAndPresenceAware(t *testing.T) { - const valid = `version: 4 + const valid = `version: 5 pipelines: main: validation_policy: producer_structural_failure: reject_output semantic_rejection: fail_run - input: seriatim + input: + module: seriatim chunk: module: generic validation_policy: @@ -26,7 +28,11 @@ pipelines: validation_policy: semantic_rejection: reject_output ` - cfg := applyFileConfig(t, valid) + loaded, err := loadDocumentSource(configkit.Bytes("valid.yml", []byte(valid)), LoadOptions{}) + if err != nil { + t.Fatal(err) + } + cfg := loaded.Config profile := cfg.Pipelines["main"] if profile.ValidationPolicy == nil || profile.ValidationPolicy.ProducerStructuralFailure == nil || *profile.ValidationPolicy.ProducerStructuralFailure != pipeline.ProducerStructuralFailureRejectOutput || profile.ValidationPolicy.SemanticRejection == nil || *profile.ValidationPolicy.SemanticRejection != pipeline.SemanticRejectionFailRun || profile.ValidationPolicy.ValidatorFailure != nil { t.Fatalf("pipeline validation policy = %#v", profile.ValidationPolicy) @@ -50,8 +56,8 @@ pipelines: {"invalid enum", strings.Replace(valid, "semantic_rejection: fail_run", "semantic_rejection: continue", 1)}, } { t.Run(test.name, func(t *testing.T) { - if _, err := ParseFileConfigYAML([]byte(test.yaml)); err == nil { - t.Fatal("ParseFileConfigYAML() error = nil, want strict validation-policy rejection") + if _, err := loadDocumentSource(configkit.Bytes("invalid.yml", []byte(test.yaml)), LoadOptions{}); err == nil { + t.Fatal("loadDocumentSource() error = nil, want strict validation-policy rejection") } }) } diff --git a/internal/core/debugbundle/bundle_test.go b/internal/core/debugbundle/bundle_test.go index cece357b..1bbeef83 100644 --- a/internal/core/debugbundle/bundle_test.go +++ b/internal/core/debugbundle/bundle_test.go @@ -91,9 +91,6 @@ func TestSummaryWriterWritesEverySummaryArtifact(t *testing.T) { if err := summary.WriteInvocation(Invocation{Operation: "run"}); err != nil { t.Fatal(err) } - if err := summary.WriteRedactedEffectiveConfig(testRedactedSummaryPayload{}); err != nil { - t.Fatal(err) - } if err := summary.WriteEffectiveConfig(testEffectiveConfigWriter{content: "version: 5\n"}); err != nil { t.Fatal(err) } @@ -124,7 +121,6 @@ func TestSummaryWriterWritesEverySummaryArtifact(t *testing.T) { for _, name := range []string{ ArtifactInvocationMetadata, - ArtifactEffectiveConfig, ArtifactEffectiveConfigYAML, ArtifactResolvedPipeline, ArtifactResolvedReferences, @@ -236,12 +232,6 @@ func TestSummaryWriterInternalWritesConfineArtifacts(t *testing.T) { } } -type testRedactedSummaryPayload struct{} - -func (testRedactedSummaryPayload) RedactedSummaryPayload() any { - return map[string]any{"redacted": true} -} - type testEffectiveConfigWriter struct { content string err error diff --git a/internal/core/debugbundle/summary.go b/internal/core/debugbundle/summary.go index 93f31286..78d95424 100644 --- a/internal/core/debugbundle/summary.go +++ b/internal/core/debugbundle/summary.go @@ -15,7 +15,6 @@ import ( const ( ArtifactInvocationMetadata = "invocation.json" - ArtifactEffectiveConfig = "effective-config.json" ArtifactEffectiveConfigYAML = "effective-config.yml" ArtifactResolvedPipeline = "resolved-pipeline.json" ArtifactResolvedReferences = "resolved-references.json" @@ -27,7 +26,6 @@ const ( ArtifactErrorLog = "error.log" ) -type RedactedSummaryPayload interface{ RedactedSummaryPayload() any } type EffectiveConfigWriter interface{ WriteEffective(io.Writer) error } type RedactedResolvedPipelinePayload interface { RedactedResolvedPipelinePayload() pipeline.ResolvedPipeline @@ -85,12 +83,6 @@ func (w *SummaryWriter) WriteInvocation(payload Invocation) error { } return w.writeJSON(ArtifactInvocationMetadata, payload) } -func (w *SummaryWriter) WriteRedactedEffectiveConfig(payload RedactedSummaryPayload) error { - if payload == nil { - return fmt.Errorf("redacted summary payload must not be nil") - } - return w.writeJSON(ArtifactEffectiveConfig, payload.RedactedSummaryPayload()) -} func (w *SummaryWriter) WriteEffectiveConfig(payload EffectiveConfigWriter) error { if payload == nil { return fmt.Errorf("effective configuration writer must not be nil") diff --git a/internal/modules/integration/config_test.go b/internal/modules/integration/config_test.go new file mode 100644 index 00000000..b27d72a7 --- /dev/null +++ b/internal/modules/integration/config_test.go @@ -0,0 +1,43 @@ +package integration_test + +import ( + "os" + "path/filepath" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/core/config" +) + +func loadIntegrationConfig(t *testing.T, path string) config.Config { + t.Helper() + loaded, err := config.Load(path, config.LoadOptions{}) + if err != nil { + t.Fatalf("load configuration %q: %v", path, err) + } + return loaded.RuntimeConfig() +} + +func loadIntegrationConfigText(t *testing.T, content string) config.Config { + t.Helper() + path := filepath.Join(t.TempDir(), "config.yml") + if err := os.WriteFile(path, []byte(content), 0o600); err != nil { + t.Fatal(err) + } + return loadIntegrationConfig(t, path) +} + +func configFromYAML(content string) (config.Config, error) { + directory, err := os.MkdirTemp("", "notarius-integration-config-") + if err != nil { + return config.Config{}, err + } + path := filepath.Join(directory, "config.yml") + if err := os.WriteFile(path, []byte(content), 0o600); err != nil { + return config.Config{}, err + } + loaded, err := config.Load(path, config.LoadOptions{}) + if err != nil { + return config.Config{}, err + } + return loaded.RuntimeConfig(), nil +} diff --git a/internal/modules/integration/dnd_itemoccurrences_runner_test.go b/internal/modules/integration/dnd_itemoccurrences_runner_test.go index e9a30048..a90e68be 100644 --- a/internal/modules/integration/dnd_itemoccurrences_runner_test.go +++ b/internal/modules/integration/dnd_itemoccurrences_runner_test.go @@ -10,18 +10,18 @@ import ( func TestItemOccurrenceConfigurationRequiresEarlierItemRegistry(t *testing.T) { registries := productionNPCRegistries(t) cfg, err := configFromYAML(` -version: 4 +version: 5 pipelines: items: - input: seriatim - chunk: generic + input: {module: seriatim} + chunk: {module: generic} steps: - id: identify-items artifacts: registry: - extract: dnd/item-registry - merge: appendorder - normalize: dnd/item-registry + extract: {module: dnd/item-registry} + merge: {module: appendorder} + normalize: {module: dnd/item-registry} - id: record-occurrences references: item_registry: @@ -30,9 +30,9 @@ pipelines: lane: registry artifacts: occurrences: - extract: dnd/item-occurrences - merge: appendorder - normalize: dnd/item-occurrences + extract: {module: dnd/item-occurrences} + merge: {module: appendorder} + normalize: {module: dnd/item-occurrences} `) if err != nil { t.Fatal(err) @@ -42,16 +42,16 @@ pipelines: } broken, err := configFromYAML(strings.Replace(` -version: 4 +version: 5 pipelines: items: - input: seriatim - chunk: generic + input: {module: seriatim} + chunk: {module: generic} steps: - id: record-occurrences artifacts: occurrences: - extract: dnd/item-occurrences + extract: {module: dnd/item-occurrences} `, "record-occurrences", "record-occurrences", 1)) if err != nil { t.Fatal(err) diff --git a/internal/modules/integration/dnd_location_registry_runner_test.go b/internal/modules/integration/dnd_location_registry_runner_test.go index a51a9c23..cc90e220 100644 --- a/internal/modules/integration/dnd_location_registry_runner_test.go +++ b/internal/modules/integration/dnd_location_registry_runner_test.go @@ -99,11 +99,8 @@ func TestLocationOccurrenceConsumerDoesNotRunAfterRejectedRegistry(t *testing.T) func TestLocationOccurrenceConfigurationRequiresEarlierRegistry(t *testing.T) { registries := productionNPCRegistries(t) - cfg, err := configFromYAML(strings.Replace(locationPipelineConfig, "step: identify-locations", "step: track-location-occurrences", 1)) - if err != nil { - t.Fatal(err) - } - _, err = cfg.Resolve(config.ResolveInput{PipelineID: "dnd-locations-fixture", Catalog: moduleCatalog(registries)}) + cfg := loadIntegrationConfigText(t, strings.Replace(locationPipelineConfig, "step: identify-locations", "step: track-location-occurrences", 1)) + _, err := cfg.Resolve(config.ResolveInput{PipelineID: "dnd-locations-fixture", Catalog: moduleCatalog(registries)}) if err == nil || !strings.Contains(err.Error(), "earlier step") { t.Fatalf("Resolve() error = %v, want an earlier generated registry requirement", err) } @@ -129,10 +126,7 @@ func TestLocationOccurrenceCheckpointTracksGeneratedRegistry(t *testing.T) { func resolveLocationPipeline(t *testing.T, registries pipeline.Registries) pipeline.ResolvedPipeline { t.Helper() - cfg, err := configFromYAML(locationPipelineConfig) - if err != nil { - t.Fatal(err) - } + cfg := loadIntegrationConfigText(t, locationPipelineConfig) effective, err := cfg.Resolve(config.ResolveInput{PipelineID: "dnd-locations-fixture", Catalog: moduleCatalog(registries)}) if err != nil { t.Fatalf("Resolve() error = %v", err) @@ -144,26 +138,15 @@ func resolveLocationPipeline(t *testing.T, registries pipeline.Registries) pipel return resolved } -func configFromYAML(content string) (config.Config, error) { - fileConfig, err := config.ParseFileConfigYAML([]byte(content)) - if err != nil { - return config.Config{}, err - } - cfg := config.Default() - if err := cfg.ApplyFileConfig(fileConfig); err != nil { - return config.Config{}, err - } - return cfg, nil -} - -const locationPipelineConfig = `version: 4 +const locationPipelineConfig = `version: 5 cache: chunk_plans: mode: bypass checkpoints: {} pipelines: dnd-locations-fixture: - input: seriatim + input: + module: seriatim output: module: json options: @@ -174,9 +157,9 @@ pipelines: - id: identify-locations artifacts: locations: - extract: dnd/location-registry - merge: appendorder - normalize: dnd/location-registry + extract: {module: dnd/location-registry} + merge: {module: appendorder} + normalize: {module: dnd/location-registry} - id: track-location-occurrences references: location_registry: @@ -185,9 +168,9 @@ pipelines: lane: locations artifacts: occurrences: - extract: dnd/location-occurrences - merge: appendorder - normalize: dnd/location-occurrences + extract: {module: dnd/location-occurrences} + merge: {module: appendorder} + normalize: {module: dnd/location-occurrences} ` type locationHandoffLLMClient struct { diff --git a/internal/modules/integration/dnd_npc_grounded_test.go b/internal/modules/integration/dnd_npc_grounded_test.go index dacfef39..e3bc0645 100644 --- a/internal/modules/integration/dnd_npc_grounded_test.go +++ b/internal/modules/integration/dnd_npc_grounded_test.go @@ -429,15 +429,7 @@ func assertCurrentEvidence(t *testing.T, references []source.SourceRef) { func loadGroundedPipelineConfig(t *testing.T) config.Config { t.Helper() - fileConfig, err := config.LoadFileConfig(repositoryPathForIntegration("internal", "modules", "integration", "testdata", "dnd_npc_grounded_pipeline.yml")) - if err != nil { - t.Fatalf("LoadFileConfig() error = %v", err) - } - cfg := config.Default() - if err := cfg.ApplyFileConfig(fileConfig); err != nil { - t.Fatalf("ApplyFileConfig() error = %v", err) - } - return cfg + return loadIntegrationConfig(t, repositoryPathForIntegration("internal", "modules", "integration", "testdata", "dnd_npc_grounded_pipeline.yml")) } func configureGroundedCampaignReferences(t *testing.T, cfg *config.Config) map[string]string { diff --git a/internal/modules/integration/dnd_npc_occurrences_runner_test.go b/internal/modules/integration/dnd_npc_occurrences_runner_test.go index 4972176e..6d850014 100644 --- a/internal/modules/integration/dnd_npc_occurrences_runner_test.go +++ b/internal/modules/integration/dnd_npc_occurrences_runner_test.go @@ -4,7 +4,6 @@ import ( "context" "encoding/json" "fmt" - "os" "reflect" "strings" "sync" @@ -168,19 +167,7 @@ func resolveNPCOccurrencePipeline(t *testing.T, registries pipeline.Registries) func loadNPCOccurrencePipelineConfig(t *testing.T) config.Config { t.Helper() - data, err := os.ReadFile("testdata/dnd_npc_occurrences_pipeline.yml") - if err != nil { - t.Fatalf("ReadFile(dnd_npc_occurrences_pipeline.yml) error = %v", err) - } - fileConfig, err := config.ParseFileConfigYAML(data) - if err != nil { - t.Fatalf("ParseFileConfigYAML() error = %v", err) - } - result := config.Default() - if err := result.ApplyFileConfig(fileConfig); err != nil { - t.Fatalf("ApplyFileConfig() error = %v", err) - } - return result + return loadIntegrationConfig(t, "testdata/dnd_npc_occurrences_pipeline.yml") } func normalizedLane(t *testing.T, output pipeline.RunOutput, laneID string) contracts.SerializedOutput { diff --git a/internal/modules/integration/dnd_npc_registry_runner_test.go b/internal/modules/integration/dnd_npc_registry_runner_test.go index df5c0ec8..26cfe53f 100644 --- a/internal/modules/integration/dnd_npc_registry_runner_test.go +++ b/internal/modules/integration/dnd_npc_registry_runner_test.go @@ -331,19 +331,7 @@ func moduleCatalog(registries pipeline.Registries) pipeline.ModuleCatalog { func loadNPCPipelineConfig(t *testing.T) config.Config { t.Helper() - data, err := os.ReadFile("testdata/dnd_npc_registry_pipeline.yml") - if err != nil { - t.Fatalf("ReadFile(dnd_npc_registry_pipeline.yml) error = %v", err) - } - fileConfig, err := config.ParseFileConfigYAML(data) - if err != nil { - t.Fatalf("ParseFileConfigYAML() error = %v", err) - } - cfg := config.Default() - if err := cfg.ApplyFileConfig(fileConfig); err != nil { - t.Fatalf("ApplyFileConfig() error = %v", err) - } - return cfg + return loadIntegrationConfig(t, "testdata/dnd_npc_registry_pipeline.yml") } func readNPCFixture(t *testing.T) []byte { diff --git a/internal/modules/integration/dnd_spells_config_test.go b/internal/modules/integration/dnd_spells_config_test.go index 79645f1f..0601799b 100644 --- a/internal/modules/integration/dnd_spells_config_test.go +++ b/internal/modules/integration/dnd_spells_config_test.go @@ -2,7 +2,6 @@ package integration_test import ( "context" - "os" "strings" "testing" @@ -121,19 +120,7 @@ func dndCapabilityCatalog(t *testing.T, inputSpec, extractorSpec pipeline.Module func loadDNDSpellsPipelineConfig(t *testing.T) config.Config { t.Helper() - data, err := os.ReadFile("testdata/pipeline.yml") - if err != nil { - t.Fatalf("ReadFile(pipeline.yml) error = %v, want nil", err) - } - fileCfg, err := config.ParseFileConfigYAML(data) - if err != nil { - t.Fatalf("ParseFileConfigYAML() error = %v, want nil", err) - } - cfg := config.Default() - if err := cfg.ApplyFileConfig(fileCfg); err != nil { - t.Fatalf("ApplyFileConfig() error = %v, want nil", err) - } - return cfg + return loadIntegrationConfig(t, "testdata/pipeline.yml") } type dndSpellsCatalogSpecs struct { diff --git a/internal/modules/integration/testdata/dnd_npc_grounded_pipeline.yml b/internal/modules/integration/testdata/dnd_npc_grounded_pipeline.yml index b2f7c095..035dd4de 100644 --- a/internal/modules/integration/testdata/dnd_npc_grounded_pipeline.yml +++ b/internal/modules/integration/testdata/dnd_npc_grounded_pipeline.yml @@ -1,4 +1,4 @@ -version: 4 +version: 5 output: directory: ./notarius-output cache: @@ -9,16 +9,16 @@ debug: directory: ./notarius-debug pipelines: dnd-npc-grounded: - input: seriatim + input: {module: seriatim} steps: - id: identify-npcs artifacts: npc_registry: - extract: dnd/npc-registry - normalize: dnd/npc-registry + extract: {module: dnd/npc-registry} + normalize: {module: dnd/npc-registry} scene-descriptions: - extract: dnd/scene-descriptions - normalize: dnd/scene-descriptions + extract: {module: dnd/scene-descriptions} + normalize: {module: dnd/scene-descriptions} - id: grounded-events references: npc_registry: @@ -31,8 +31,8 @@ pipelines: lane: scene-descriptions artifacts: spells: - extract: dnd/spells - normalize: dnd/spells + extract: {module: dnd/spells} + normalize: {module: dnd/spells} combat: - extract: dnd/combat-turns - normalize: dnd/combat-turns + extract: {module: dnd/combat-turns} + normalize: {module: dnd/combat-turns} diff --git a/internal/modules/integration/testdata/dnd_npc_occurrences_pipeline.yml b/internal/modules/integration/testdata/dnd_npc_occurrences_pipeline.yml index 9d0e4ebe..a9270ba6 100644 --- a/internal/modules/integration/testdata/dnd_npc_occurrences_pipeline.yml +++ b/internal/modules/integration/testdata/dnd_npc_occurrences_pipeline.yml @@ -1,4 +1,4 @@ -version: 4 +version: 5 output: directory: ./notarius-output cache: @@ -9,13 +9,13 @@ debug: directory: ./notarius-debug pipelines: dnd-npc-occurrences-fixture: - input: seriatim + input: {module: seriatim} steps: - id: identify-npcs artifacts: npc_registry: - extract: dnd/npc-registry - normalize: dnd/npc-registry + extract: {module: dnd/npc-registry} + normalize: {module: dnd/npc-registry} - id: extract-occurrences references: npc_registry: @@ -24,5 +24,5 @@ pipelines: lane: npc_registry artifacts: occurrences: - extract: dnd/npc-occurrences - normalize: dnd/npc-occurrences + extract: {module: dnd/npc-occurrences} + normalize: {module: dnd/npc-occurrences} diff --git a/internal/modules/integration/testdata/dnd_npc_registry_pipeline.yml b/internal/modules/integration/testdata/dnd_npc_registry_pipeline.yml index 3d038454..daf12bd7 100644 --- a/internal/modules/integration/testdata/dnd_npc_registry_pipeline.yml +++ b/internal/modules/integration/testdata/dnd_npc_registry_pipeline.yml @@ -1,4 +1,4 @@ -version: 4 +version: 5 output: directory: ./notarius-output cache: @@ -9,12 +9,12 @@ debug: directory: ./notarius-debug pipelines: dnd-npc-registry-fixture: - input: seriatim - chunk: generic + input: {module: seriatim} + chunk: {module: generic} artifacts: npc_registry: extract: module: dnd/npc-registry retries: 2 - normalize: dnd/npc-registry - output: json + normalize: {module: dnd/npc-registry} + output: {module: json} diff --git a/internal/modules/integration/testdata/pipeline.yml b/internal/modules/integration/testdata/pipeline.yml index 58e3cc2f..5ba6d809 100644 --- a/internal/modules/integration/testdata/pipeline.yml +++ b/internal/modules/integration/testdata/pipeline.yml @@ -1,4 +1,4 @@ -version: 4 +version: 5 output: directory: ./notarius-output cache: @@ -9,11 +9,11 @@ debug: directory: ./notarius-debug pipelines: dnd-spells-fixture: - input: seriatim - chunk: fake/chunk + input: {module: seriatim} + chunk: {module: fake/chunk} artifacts: spells: - extract: dnd/spells - merge: appendorder - normalize: noop - output: json + extract: {module: dnd/spells} + merge: {module: appendorder} + normalize: {module: noop} + output: {module: json}