From 4023c665080991f40034a2ab4e13888628ad4a6a Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Fri, 17 Jul 2026 08:21:24 +0000 Subject: [PATCH] Add extract worker configuration controls --- docs/config.md | 30 ++++++- docs/internal/modules.md | 4 +- docs/internal/pipeline.md | 9 ++- docs/policy/architecture.md | 4 +- examples/dnd-spells-production.config.yml | 2 + internal/cli/compatibility_test.go | 4 +- internal/core/config/config.go | 39 ++++++++- internal/core/config/config_test.go | 6 ++ internal/core/config/effective_config.go | 1 + internal/core/config/effective_config_test.go | 13 +++ internal/core/config/env.go | 12 +++ internal/core/config/env_test.go | 79 +++++++++++++++++-- internal/core/config/file_config.go | 32 +++++++- internal/core/config/file_config_test.go | 55 +++++++++++++ internal/core/config/redaction_test.go | 11 +++ internal/core/config/validation.go | 29 +++++++ internal/core/config/validation_test.go | 35 ++++++++ 17 files changed, 343 insertions(+), 22 deletions(-) diff --git a/docs/config.md b/docs/config.md index 5decacf..3413c3c 100644 --- a/docs/config.md +++ b/docs/config.md @@ -41,6 +41,7 @@ rejected; execution profiles now come from Scriptorium. Built-in defaults: - `concurrency.total_llm`: `1` +- `concurrency.stage_workers.extract`: effective `concurrency.total_llm` - `diagnostics.work_dir`: `/tmp/notarius` - `diagnostics.retention`: `auto` - `workspace.directory`: unset @@ -88,6 +89,7 @@ These environment variables are applied after the config file: - `NOTARIUS_CONFIG`: config discovery path. - `NOTARIUS_TOTAL_LLM_CONCURRENCY`: integer global LLM concurrency. +- `NOTARIUS_STAGE_WORKERS_EXTRACT`: integer extract worker limit. - `NOTARIUS_WORKSPACE_DIR`: workspace directory. - `NOTARIUS_WORKSPACE_DIAGNOSTICS_ENABLED`: boolean diagnostics enablement. - `NOTARIUS_WORKSPACE_DIAGNOSTICS_RETENTION`: workspace diagnostics retention @@ -107,6 +109,24 @@ The removed `NOTARIUS_LLM_DEFAULT_*` variables are not read. Configure provider endpoint, model, and credential environment variable names through Scriptorium profiles. +## Concurrency + +`concurrency` fields: + +- `total_llm`: positive integer ceiling on concurrent provider calls. +- `stage_workers`: optional map of framework worker limits. The only supported + key is `extract`. + +`stage_workers.extract` defaults to the effective `total_llm` value after file +and environment precedence. It must be between `1` and `total_llm`, inclusive. +Unknown or empty stage-worker keys are rejected. The environment override +`NOTARIUS_STAGE_WORKERS_EXTRACT` takes precedence over the file value, as does +`NOTARIUS_TOTAL_LLM_CONCURRENCY` for the global ceiling. + +The worker value is present in effective and redacted configuration. The +current runner remains serial; this setting does not yet change execution +scheduling. + ## Pipelines A pipeline selects implementations for the fixed workflow defined by @@ -262,9 +282,9 @@ production validators do not call the LLM and must not set `llm_profile`. | input | `seriatim` | Reads Seriatim transcript JSON. | | chunk | `generic` | Splits source units into ordered chunks. | | chunk | `dnd/scenes` | Uses an LLM to split transcript source units into D&D scenes. | -| extract | `dnd/spells` | Extracts D&D spell raw outputs. | -| merge | `appendorder` | Merges JSON raw extract outputs in chunk order. | -| normalize | `noop` | Passes merged raw outputs through unchanged. | +| extract | `dnd/spells` | Extracts typed D&D spell-list artifacts. | +| merge | `appendorder` | Combines typed artifacts in chunk order. | +| normalize | `noop` | Passes merged typed artifacts through unchanged. | | output | `json` | Produces JSON output files for normalized `application/json` lanes. | ## Implemented Production Validators @@ -275,7 +295,7 @@ production validators do not call the LLM and must not set `llm_profile`. | `generic/always_reject` | deterministic | Rejects returned module output with reason `always_reject`. | | `generic/valid_json` | deterministic | Rejects payloads that are not syntactically valid JSON. | | `generic/valid_json_schema` | deterministic | Rejects invalid JSON or JSON that does not conform to the module response schema. | -| `extract/dnd/spells/shape` | deterministic | Rejects malformed D&D spell-cast JSON payloads. | +| `extract/dnd/spells/shape` | deterministic | Rejects malformed D&D spell-list artifacts. | | `extract/dnd/spells/source_refs` | deterministic | Rejects missing or invalid D&D spell source references. | | `extract/dnd/spells/source_relatedness` | deterministic | Emits warnings when a spell name is not found near its cited source text. | @@ -366,6 +386,8 @@ Configuration validation checks: - mutually exclusive `scriptorium.profile_dir` and `scriptorium.profile_file`; - non-empty, non-duplicated IDs after trimming; - positive global LLM concurrency; +- supported stage-worker keys and an effective extract worker count in the + inclusive range `1..concurrency.total_llm`; - supported diagnostics retention and non-empty work directory; - stale removed fields such as `llm_profiles`. diff --git a/docs/internal/modules.md b/docs/internal/modules.md index 48f3df6..5dfdfa5 100644 --- a/docs/internal/modules.md +++ b/docs/internal/modules.md @@ -209,8 +209,8 @@ does not inventory implementations. ## Tests To Inspect - Package-local `*_test.go` files under the module or validator being changed. -- `internal/framework/pipeline/registry_integration_test.go`: registry and spec - composition. +- `internal/framework/pipeline/typed_resolution_test.go`: typed registry, spec, + and heterogeneous artifact composition. - `internal/framework/pipeline/default_modules_test.go`: framework binding defaults. - `internal/cli/run_test.go`: production catalog, config resolution, and diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 801b77d..a03e1dc 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -245,9 +245,10 @@ durable manifest and logical file schemas are defined in the construction order, dependency failures, and the before-source-work boundary. - `internal/framework/pipeline/references_test.go`: target resolution and materialization. -- `internal/framework/pipeline/runner_test.go`: stage transitions, retries, - rejections, warnings, checkpoints, debug hooks, and manifests. -- `internal/framework/pipeline/walking_skeleton_test.go`: fake-backed complete - workflow composition. +- `internal/cli/run_test.go`: production stage transitions, retries, rejections, + warnings, debug hooks, manifests, and end-to-end composition. +- `internal/modules/integration/*_test.go` and + `internal/modules/seriatim/input/transcript/runner_test.go`: typed runner + composition across concrete module families. - `internal/framework/checkpoint/*_test.go`: checkpoint serialization and reuse collaborators. diff --git a/docs/policy/architecture.md b/docs/policy/architecture.md index 25d413f..dd7976e 100644 --- a/docs/policy/architecture.md +++ b/docs/policy/architecture.md @@ -111,8 +111,8 @@ file placement, checkpoints, or diagnostics. ## Validation -Validation is a framework-managed boundary around raw outputs from chunk, -extract, merge, and normalize stages. Validators receive immutable stage output +Validation is a framework-managed boundary around outputs from chunk, extract, +merge, and normalize stages. Validators receive immutable stage output and make an explicit whole-output decision: approve, approve with warnings, or reject. diff --git a/examples/dnd-spells-production.config.yml b/examples/dnd-spells-production.config.yml index 2ccfc9e..e3cb08f 100644 --- a/examples/dnd-spells-production.config.yml +++ b/examples/dnd-spells-production.config.yml @@ -1,6 +1,8 @@ version: 2 concurrency: total_llm: 1 + stage_workers: + extract: 1 workspace: directory: /var/lib/notarius diagnostics: diff --git a/internal/cli/compatibility_test.go b/internal/cli/compatibility_test.go index d581f6a..dd30de0 100644 --- a/internal/cli/compatibility_test.go +++ b/internal/cli/compatibility_test.go @@ -172,8 +172,8 @@ func TestProductionCompatibilitySnapshot(t *testing.T) { t.Fatalf("Resolve(production example) error = %v, want nil", err) } productionResolved := productionEffective.ResolvedPipeline - if productionConfig.Concurrency.TotalLLM != 1 || !reflect.DeepEqual(productionResolved.Chunk.Options, map[string]any{"max_units": 50}) { - t.Fatalf("production example concurrency/options = %d/%#v, want compatibility snapshot", productionConfig.Concurrency.TotalLLM, productionResolved.Chunk.Options) + if productionConfig.Concurrency.TotalLLM != 1 || productionConfig.Concurrency.StageWorkers["extract"] != 1 || !reflect.DeepEqual(productionResolved.Chunk.Options, map[string]any{"max_units": 50}) { + t.Fatalf("production example concurrency/options = %#v/%#v, want compatibility snapshot", productionConfig.Concurrency, productionResolved.Chunk.Options) } bindings := productionResolved.ArtifactLanes[0].ExtractReferences.Bindings if len(bindings) != 2 || bindings[0].SlotName != "glossary" || bindings[0].Source != "./dnd-spells-glossary.txt" || bindings[1].SlotName != "party" || bindings[1].Source != "./dnd-spells-roster.txt" { diff --git a/internal/core/config/config.go b/internal/core/config/config.go index 9dd99ac..e091f85 100644 --- a/internal/core/config/config.go +++ b/internal/core/config/config.go @@ -24,7 +24,11 @@ type ScriptoriumConfig struct { } type ConcurrencyConfig struct { - TotalLLM int `json:"total_llm"` + TotalLLM int `json:"total_llm"` + StageWorkers map[string]int `json:"stage_workers"` + + extractWorkersConfigured bool + defaultedExtractWorkers int } type DiagnosticsConfig struct { @@ -58,7 +62,9 @@ func Default() Config { return Config{ Pipelines: map[string]pipeline.PipelineProfile{}, Concurrency: ConcurrencyConfig{ - TotalLLM: 1, + TotalLLM: 1, + StageWorkers: map[string]int{"extract": 1}, + defaultedExtractWorkers: 1, }, Diagnostics: DiagnosticsConfig{ WorkDir: "/tmp/notarius", @@ -101,6 +107,7 @@ func (c Config) workspaceDirectory() string { func cloneConfig(in Config) Config { out := in + out.Concurrency.StageWorkers = cloneIntMap(in.Concurrency.StageWorkers) out.Pipelines = make(map[string]pipeline.PipelineProfile, len(in.Pipelines)) for key, profile := range in.Pipelines { out.Pipelines[key] = clonePipelineProfile(profile) @@ -108,6 +115,34 @@ func cloneConfig(in Config) Config { return out } +func cloneIntMap(in map[string]int) map[string]int { + if len(in) == 0 { + return nil + } + out := make(map[string]int, len(in)) + for key, value := range in { + out[key] = value + } + 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 + } + c.StageWorkers["extract"] = c.TotalLLM + c.defaultedExtractWorkers = c.TotalLLM + } +} + func clonePipelineProfile(in pipeline.PipelineProfile) pipeline.PipelineProfile { out := in out.Input = cloneModuleBinding(in.Input) diff --git a/internal/core/config/config_test.go b/internal/core/config/config_test.go index f97968a..3ae9b03 100644 --- a/internal/core/config/config_test.go +++ b/internal/core/config/config_test.go @@ -18,6 +18,9 @@ func TestDefaultValues(t *testing.T) { if cfg.Concurrency.TotalLLM != 1 { t.Fatalf("unexpected total LLM concurrency: %d", cfg.Concurrency.TotalLLM) } + if got := cfg.Concurrency.StageWorkers["extract"]; got != 1 { + t.Fatalf("unexpected extract workers: %d", got) + } if cfg.Diagnostics.WorkDir != "/tmp/notarius" { t.Fatalf("unexpected diagnostics work dir: %q", cfg.Diagnostics.WorkDir) } @@ -68,6 +71,9 @@ pipelines: if cfg.Concurrency.TotalLLM != 1 { t.Fatalf("expected default concurrency preserved, got %d", cfg.Concurrency.TotalLLM) } + if got := cfg.Concurrency.StageWorkers["extract"]; got != 1 { + t.Fatalf("expected default extract workers preserved, got %d", got) + } if cfg.Diagnostics.Retention != diagnostics.RetentionAuto { t.Fatalf("expected default diagnostics retention preserved, got %q", cfg.Diagnostics.Retention) } diff --git a/internal/core/config/effective_config.go b/internal/core/config/effective_config.go index 68c22cb..e9c4615 100644 --- a/internal/core/config/effective_config.go +++ b/internal/core/config/effective_config.go @@ -26,6 +26,7 @@ type EffectiveConfig struct { } func (c Config) Resolve(input ResolveInput) (EffectiveConfig, error) { + c.Concurrency.recomputeStageWorkerDefaults() if err := c.Validate(); err != nil { return EffectiveConfig{}, err } diff --git a/internal/core/config/effective_config_test.go b/internal/core/config/effective_config_test.go index ea39f21..352e624 100644 --- a/internal/core/config/effective_config_test.go +++ b/internal/core/config/effective_config_test.go @@ -27,6 +27,19 @@ func TestResolveRejectsEmptyAndUnknownPipelineID(t *testing.T) { } } +func TestResolveMaterializesDefaultExtractWorkersFromEffectiveTotal(t *testing.T) { + cfg := validConfig() + cfg.Concurrency.TotalLLM = 4 + + effective, err := cfg.Resolve(ResolveInput{PipelineID: "example", Catalog: fakeCatalog(t)}) + if err != nil { + t.Fatalf("Resolve() error = %v, want nil", err) + } + if got := effective.Config.Concurrency.StageWorkers["extract"]; got != 4 { + t.Fatalf("effective extract workers = %d, want total concurrency 4", got) + } +} + func TestResolveLaneFilteringSuccessAndFailure(t *testing.T) { effective, err := validConfig().Resolve(ResolveInput{ PipelineID: " example ", diff --git a/internal/core/config/env.go b/internal/core/config/env.go index 89caa58..fb33a28 100644 --- a/internal/core/config/env.go +++ b/internal/core/config/env.go @@ -36,6 +36,18 @@ func (c *Config) applyEnvOverridesWithLookup(lookup func(string) (string, bool)) } 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_WORK_DIR"); ok { c.Diagnostics.WorkDir = strings.TrimSpace(raw) } diff --git a/internal/core/config/env_test.go b/internal/core/config/env_test.go index 3618676..77cd5eb 100644 --- a/internal/core/config/env_test.go +++ b/internal/core/config/env_test.go @@ -14,6 +14,7 @@ func TestApplyEnvOverridesOperationalValues(t *testing.T) { err := cfg.applyEnvOverridesWithLookup(mapLookup(map[string]string{ "NOTARIUS_TOTAL_LLM_CONCURRENCY": "3", + "NOTARIUS_STAGE_WORKERS_EXTRACT": "2", "NOTARIUS_WORK_DIR": "/tmp/notarius-env", "NOTARIUS_DIAGNOSTICS_RETENTION": "never", "NOTARIUS_WORKSPACE_DIR": "/var/lib/notarius-env", @@ -33,6 +34,9 @@ func TestApplyEnvOverridesOperationalValues(t *testing.T) { if cfg.Concurrency.TotalLLM != 3 { t.Fatalf("unexpected total concurrency: %d", cfg.Concurrency.TotalLLM) } + if got := cfg.Concurrency.StageWorkers["extract"]; got != 2 { + t.Fatalf("extract workers = %d, want 2", got) + } if cfg.Workspace.Directory != "/var/lib/notarius-env" { t.Fatalf("unexpected workspace directory: %q", cfg.Workspace.Directory) } @@ -54,12 +58,74 @@ func TestApplyEnvOverridesOperationalValues(t *testing.T) { } func TestApplyEnvOverridesRejectsInvalidIntegers(t *testing.T) { + for _, name := range []string{"NOTARIUS_TOTAL_LLM_CONCURRENCY", "NOTARIUS_STAGE_WORKERS_EXTRACT"} { + t.Run(name, func(t *testing.T) { + cfg := Default() + err := cfg.applyEnvOverridesWithLookup(mapLookup(map[string]string{name: "many"})) + if err == nil || !strings.Contains(err.Error(), name) { + t.Fatalf("expected named integer error, got %v", err) + } + }) + } +} + +func TestStageWorkerEnvironmentPrecedenceAndDefaulting(t *testing.T) { + fileCfg, err := ParseFileConfigYAML([]byte(` +version: 2 +concurrency: + total_llm: 4 + stage_workers: + extract: 2 +`)) + if err != nil { + t.Fatalf("ParseFileConfigYAML() error = %v", err) + } cfg := Default() - err := cfg.applyEnvOverridesWithLookup(mapLookup(map[string]string{ - "NOTARIUS_TOTAL_LLM_CONCURRENCY": "many", - })) - if err == nil || !strings.Contains(err.Error(), "NOTARIUS_TOTAL_LLM_CONCURRENCY") { - t.Fatalf("expected named integer error, got %v", err) + if err := cfg.applyFileConfigWithLookup(fileCfg, emptyLookup); err != nil { + t.Fatalf("ApplyFileConfig() error = %v", err) + } + if err := cfg.applyEnvOverridesWithLookup(mapLookup(map[string]string{ + "NOTARIUS_TOTAL_LLM_CONCURRENCY": "5", + "NOTARIUS_STAGE_WORKERS_EXTRACT": "3", + })); err != nil { + t.Fatalf("ApplyEnvOverrides() error = %v", err) + } + if cfg.Concurrency.TotalLLM != 5 || cfg.Concurrency.StageWorkers["extract"] != 3 { + t.Fatalf("effective concurrency = %#v, want total 5 and extract 3", cfg.Concurrency) + } + if err := cfg.Validate(); err != nil { + t.Fatalf("Validate(overridden) error = %v, want nil", err) + } + + defaulted := Default() + if err := defaulted.applyEnvOverridesWithLookup(mapLookup(map[string]string{"NOTARIUS_TOTAL_LLM_CONCURRENCY": "6"})); err != nil { + t.Fatalf("ApplyEnvOverrides(defaulted) error = %v", err) + } + if got := defaulted.Concurrency.StageWorkers["extract"]; got != 6 { + t.Fatalf("defaulted extract workers = %d, want effective total 6", got) + } +} + +func TestStageWorkerRangeValidationUsesFinalEnvironmentTotal(t *testing.T) { + fileCfg, err := ParseFileConfigYAML([]byte(` +version: 2 +concurrency: + total_llm: 4 + stage_workers: + extract: 5 +`)) + if err != nil { + t.Fatalf("ParseFileConfigYAML() error = %v", err) + } + cfg := Default() + if err := cfg.applyFileConfigWithLookup(fileCfg, emptyLookup); err != nil { + t.Fatalf("ApplyFileConfig() error = %v", err) + } + if err := cfg.applyEnvOverridesWithLookup(mapLookup(map[string]string{"NOTARIUS_TOTAL_LLM_CONCURRENCY": "6"})); err != nil { + t.Fatalf("ApplyEnvOverrides() error = %v", err) + } + if err := cfg.Validate(); err != nil { + t.Fatalf("Validate() error = %v, want final total to make extract workers valid", err) } } @@ -111,4 +177,7 @@ func TestLoadFromEnvUsesDefaultConfig(t *testing.T) { if cfg.Concurrency.TotalLLM != 2 { t.Fatalf("expected env concurrency override, got %+v", cfg.Concurrency) } + if got := cfg.Concurrency.StageWorkers["extract"]; got != 2 { + t.Fatalf("expected extract workers to default to total, got %d", got) + } } diff --git a/internal/core/config/file_config.go b/internal/core/config/file_config.go index 398940b..dcfa571 100644 --- a/internal/core/config/file_config.go +++ b/internal/core/config/file_config.go @@ -43,7 +43,8 @@ type FileArtifactLaneProfile struct { } type FileConcurrencyConfig struct { - TotalLLM *int `yaml:"total_llm,omitempty"` + TotalLLM *int `yaml:"total_llm,omitempty"` + StageWorkers map[string]int `yaml:"stage_workers,omitempty"` } type FileDiagnosticsConfig struct { @@ -316,6 +317,15 @@ func (c *Config) applyFileConfigWithLookup(fileCfg FileConfig, lookup func(strin 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.Diagnostics != nil { if fileCfg.Diagnostics.WorkDir != nil { c.Diagnostics.WorkDir = strings.TrimSpace(*fileCfg.Diagnostics.WorkDir) @@ -350,6 +360,26 @@ func (c *Config) applyFileConfigWithLookup(fileCfg FileConfig, lookup func(strin return nil } +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)) diff --git a/internal/core/config/file_config_test.go b/internal/core/config/file_config_test.go index fda4777..b99d3ec 100644 --- a/internal/core/config/file_config_test.go +++ b/internal/core/config/file_config_test.go @@ -546,6 +546,9 @@ diagnostics: if cfg.Concurrency.TotalLLM != 4 { t.Fatalf("unexpected total concurrency: %d", cfg.Concurrency.TotalLLM) } + if got := cfg.Concurrency.StageWorkers["extract"]; got != 4 { + t.Fatalf("default extract workers = %d, want total concurrency", got) + } if cfg.Diagnostics.WorkDir != "/tmp/notarius-test" { t.Fatalf("unexpected work dir: %q", cfg.Diagnostics.WorkDir) } @@ -554,6 +557,58 @@ diagnostics: } } +func TestApplyFileConfigStageWorkers(t *testing.T) { + cfg := parseAndApplyConfig(t, ` +version: 2 +concurrency: + total_llm: 4 + stage_workers: + extract: 3 +`) + + if got := cfg.Concurrency.StageWorkers["extract"]; got != 3 { + t.Fatalf("extract workers = %d, want 3", got) + } + if err := cfg.Validate(); err != nil { + t.Fatalf("Validate() error = %v, want nil", err) + } +} + +func TestApplyFileConfigEmptyStageWorkersDefaultsExtractToTotal(t *testing.T) { + cfg := parseAndApplyConfig(t, ` +version: 2 +concurrency: + total_llm: 4 + stage_workers: {} +`) + if got := cfg.Concurrency.StageWorkers["extract"]; got != 4 { + t.Fatalf("extract workers = %d, want total concurrency 4", got) + } +} + +func TestApplyFileConfigRejectsUnsupportedStageWorkerKeys(t *testing.T) { + for _, test := range []struct { + name string + key string + want string + }{ + {name: "empty", key: "' '", want: "must not be empty"}, + {name: "unknown", key: "merge", want: "not supported"}, + } { + t.Run(test.name, func(t *testing.T) { + fileCfg, err := ParseFileConfigYAML([]byte("version: 2\nconcurrency:\n stage_workers:\n " + test.key + ": 1\n")) + if err != nil { + t.Fatalf("ParseFileConfigYAML() error = %v", err) + } + cfg := Default() + err = cfg.applyFileConfigWithLookup(fileCfg, emptyLookup) + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("ApplyFileConfig() error = %v, want %q", err, test.want) + } + }) + } +} + func TestApplyFileConfigWorkspaceSection(t *testing.T) { cfg := parseAndApplyConfig(t, ` version: 2 diff --git a/internal/core/config/redaction_test.go b/internal/core/config/redaction_test.go index 70ee21b..409c2f5 100644 --- a/internal/core/config/redaction_test.go +++ b/internal/core/config/redaction_test.go @@ -12,6 +12,7 @@ func TestRedactedConfigCopiesScriptoriumConfig(t *testing.T) { cfg.Scriptorium.ProfileDir = "./profiles" cfg.Workspace.Directory = "/var/lib/notarius" cfg.Workspace.Resume.Enabled = true + cfg.Concurrency.StageWorkers["extract"] = 1 redacted := cfg.Redacted() @@ -29,6 +30,10 @@ func TestRedactedConfigCopiesScriptoriumConfig(t *testing.T) { if cfg.Workspace.Directory != "/var/lib/notarius" { t.Fatalf("redaction mutated original workspace config") } + redacted.Concurrency.StageWorkers["extract"] = 9 + if cfg.Concurrency.StageWorkers["extract"] != 1 { + t.Fatalf("redaction aliased stage worker map") + } } func TestConfigRedactedDiagnosticsPayloadCopiesConfig(t *testing.T) { @@ -46,6 +51,8 @@ func TestConfigRedactedDiagnosticsPayloadCopiesConfig(t *testing.T) { func TestEffectiveConfigRedactedDiagnosticsPayloadCopies(t *testing.T) { cfg := validConfig() + cfg.Concurrency.TotalLLM = 4 + cfg.Concurrency.StageWorkers["extract"] = 2 lane := cfg.Pipelines["example"].Artifacts["events"] lane.Extract.Options = map[string]any{"temperature": 0.2} lane.References = map[string]string{"roster": "./roster.yml"} @@ -101,6 +108,10 @@ func TestEffectiveConfigRedactedDiagnosticsPayloadCopies(t *testing.T) { if payload.PipelineID != effective.PipelineID || payload.ResolvedPipeline.Digest != effective.ResolvedPipeline.Digest { t.Fatalf("expected pipeline metadata preserved, got %+v", payload) } + payload.Config.Concurrency.StageWorkers["extract"] = 4 + if effective.Config.Concurrency.StageWorkers["extract"] != 2 { + t.Fatalf("expected effective stage worker map to be copied") + } payload.Only[0] = "changed" if effective.Only[0] != "events" { diff --git a/internal/core/config/validation.go b/internal/core/config/validation.go index 59b7a88..7742b1e 100644 --- a/internal/core/config/validation.go +++ b/internal/core/config/validation.go @@ -2,6 +2,7 @@ package config import ( "fmt" + "sort" "strings" "gitea.maximumdirect.net/eric/notarius/internal/core/diagnostics" @@ -9,6 +10,7 @@ import ( ) func (c Config) Validate() error { + c.Concurrency.recomputeStageWorkerDefaults() if err := validateScriptorium(c.Scriptorium); err != nil { return err } @@ -21,9 +23,36 @@ func (c Config) Validate() error { if c.Concurrency.TotalLLM <= 0 { return fmt.Errorf("total LLM concurrency must be greater than zero") } + if err := validateStageWorkers(c.Concurrency); err != nil { + return err + } return validatePipelineProfiles(c.Pipelines) } +func validateStageWorkers(cfg ConcurrencyConfig) error { + keys := make([]string, 0, len(cfg.StageWorkers)) + for key := range cfg.StageWorkers { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + if strings.TrimSpace(key) == "" { + return fmt.Errorf("concurrency.stage_workers key must not be empty") + } + if key != "extract" { + return fmt.Errorf("concurrency.stage_workers key %q is not supported", key) + } + } + extractWorkers, ok := cfg.StageWorkers["extract"] + if !ok { + extractWorkers = cfg.TotalLLM + } + if extractWorkers < 1 || extractWorkers > cfg.TotalLLM { + return fmt.Errorf("concurrency.stage_workers.extract must be between 1 and concurrency.total_llm (%d)", cfg.TotalLLM) + } + return nil +} + func validateScriptorium(cfg ScriptoriumConfig) error { if strings.TrimSpace(cfg.ProfileDir) != "" && strings.TrimSpace(cfg.ProfileFile) != "" { return fmt.Errorf("scriptorium profile_dir and profile_file are mutually exclusive") diff --git a/internal/core/config/validation_test.go b/internal/core/config/validation_test.go index a600d77..1f0a00b 100644 --- a/internal/core/config/validation_test.go +++ b/internal/core/config/validation_test.go @@ -79,6 +79,41 @@ func TestValidateRejectsInvalidNumericFields(t *testing.T) { } } +func TestValidateStageWorkerBoundaries(t *testing.T) { + for _, test := range []struct { + name string + workers int + wantErr bool + }{ + {name: "below minimum", workers: 0, wantErr: true}, + {name: "minimum", workers: 1}, + {name: "maximum", workers: 4}, + {name: "above maximum", workers: 5, wantErr: true}, + } { + t.Run(test.name, func(t *testing.T) { + cfg := validConfig() + cfg.Concurrency.TotalLLM = 4 + cfg.Concurrency.StageWorkers["extract"] = test.workers + err := cfg.Validate() + if test.wantErr && (err == nil || !strings.Contains(err.Error(), "stage_workers.extract")) { + t.Fatalf("Validate() error = %v, want extract worker range error", err) + } + if !test.wantErr && err != nil { + t.Fatalf("Validate() error = %v, want nil", err) + } + }) + } +} + +func TestValidateRejectsUnknownEffectiveStageWorkerKey(t *testing.T) { + cfg := validConfig() + cfg.Concurrency.StageWorkers["merge"] = 1 + err := cfg.Validate() + if err == nil || !strings.Contains(err.Error(), "stage_workers key") || !strings.Contains(err.Error(), "merge") { + t.Fatalf("Validate() error = %v, want unknown stage worker key", err) + } +} + func TestValidateRejectsMutuallyExclusiveScriptoriumProfileSources(t *testing.T) { cfg := validConfig() cfg.Scriptorium.ProfileDir = "./profiles"