Add extract worker configuration controls

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. |
+| extract | `dnd/spells` | Extracts typed D&D spell-list artifacts. |
-| merge | `appendorder` | Merges JSON raw extract outputs in chunk order. |
+| merge | `appendorder` | Combines typed artifacts in chunk order. |
-| normalize | `noop` | Passes merged raw outputs through unchanged. |
+| 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`.
 

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
+- `internal/framework/pipeline/typed_resolution_test.go`: typed registry, spec,
-  composition.
+  and heterogeneous artifact composition.
 - `internal/framework/pipeline/default_modules_test.go`: framework binding
   defaults.
 - `internal/cli/run_test.go`: production catalog, config resolution, and

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,
+- `internal/cli/run_test.go`: production stage transitions, retries, rejections,
-  rejections, warnings, checkpoints, debug hooks, and manifests.
+  warnings, debug hooks, manifests, and end-to-end composition.
-- `internal/framework/pipeline/walking_skeleton_test.go`: fake-backed complete
+- `internal/modules/integration/*_test.go` and
-  workflow composition.
+  `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.

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,
+Validation is a framework-managed boundary around outputs from chunk, extract,
-extract, merge, and normalize stages. Validators receive immutable stage output
+merge, and normalize stages. Validators receive immutable stage output
 and make an explicit whole-output decision: approve, approve with warnings, or
 reject.
 

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:

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}) {
+	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 = %d/%#v, want compatibility snapshot", productionConfig.Concurrency.TotalLLM, productionResolved.Chunk.Options)
+		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" {

internal/core/config/config.go

@@ -25,6 +25,10 @@ type ScriptoriumConfig struct {
 
 type ConcurrencyConfig struct {
 	TotalLLM     int            `json:"total_llm"`
+	StageWorkers map[string]int `json:"stage_workers"`
+
+	extractWorkersConfigured bool
+	defaultedExtractWorkers  int
 }
 
 type DiagnosticsConfig struct {
@@ -59,6 +63,8 @@ func Default() Config {
 		Pipelines: map[string]pipeline.PipelineProfile{},
 		Concurrency: ConcurrencyConfig{
 			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)

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)
 	}

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
 	}

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 ",

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)
 	}

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,13 +58,75 @@ 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{
+			err := cfg.applyEnvOverridesWithLookup(mapLookup(map[string]string{name: "many"}))
-		"NOTARIUS_TOTAL_LLM_CONCURRENCY": "many",
+			if err == nil || !strings.Contains(err.Error(), name) {
-	}))
-	if err == nil || !strings.Contains(err.Error(), "NOTARIUS_TOTAL_LLM_CONCURRENCY") {
 				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()
+	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)
+	}
 }
 
 func TestApplyEnvOverridesRejectsInvalidBooleans(t *testing.T) {
@@ -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)
+	}
 }

internal/core/config/file_config.go

@@ -44,6 +44,7 @@ type FileArtifactLaneProfile struct {
 
 type FileConcurrencyConfig struct {
 	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))

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

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" {

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")

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"