Remove obsolete sequential extraction path

docs/internal/modules.md

@@ -189,6 +189,11 @@ Seriatim registrars own their production leaf registrations. The D&D registrar
 owns D&D leaf registrations, the spell default-validator chain, and D&D
 prompt/schema asset collection.
 
+Concrete implementation packages do not import generic implementation
+packages directly. A concrete family's `register` package is its composition
+point for specializing reusable generic implementations, while the generic
+registrar composes only generic children.
+
 Framework packages must not import production extensions. Tests may compose
 registries and catalogs directly with fakes.
 

docs/internal/pipeline.md

@@ -31,13 +31,19 @@ calls `pipeline.ResolvePipeline`.
 7. validates each selected module and validator option set through its registry
    entry; and
 8. calculates a digest over the resolved structure, including typed artifact
-   kind and schema identity.
+   kind and schema identity and the effective validator policy in its resolved
+   execution order.
 
 Resolution returns a `ResolvedPipeline` containing ordered lanes, concrete
 bindings, validator chains, reference targets, and the digest. It does not read
 reference bytes or construct runtime modules. CLI lane and reference selector
 syntax is defined in the [CLI reference](../cli.md#run).
 
+The digest includes each resolved validator chain's stage, lane, owning module,
+ordered validator bindings, execution classes, targets, and artifact kinds.
+Changing a default chain or an explicit override therefore changes pipeline
+identity whenever it changes the effective validator policy.
+
 ## Reference Materialization
 
 The CLI calls `MaterializeReferences` after resolution and before constructing
@@ -76,6 +82,11 @@ separate target namespaces; serialized registrations declare whether they
 support chunks, artifacts, or both. Duplicate variants and exact Go-type
 mismatches are rejected deterministically.
 
+Lane-sensitive merger and normalizer spec discovery always supplies the
+extractor's artifact kind, so variants under one reusable key may declare
+different capabilities and reference slots. Kind-neutral registry inspection
+selects the first registered artifact kind in sorted order.
+
 Production composition registers the D&D spell-list codec and typed extractor,
 matching typed merge, normalize, and semantic-validator variants, and
 serialized JSON validators. Every artifact lane resolves through the typed
@@ -217,11 +228,13 @@ normally. Dependency fingerprints and debug content digests use the same stable
 codec bytes that cross those boundaries.
 
 Debug instrumentation wraps run, stage, attempt, validator, and structured LLM
-boundaries. Context scopes associate nested LLM calls with the module or
+boundaries. Every executed module retry has an attempt envelope containing its
-validator attempt that made them. Debug-write failures are framework errors;
+candidate, accepted-attempt warnings, rejection or error, and only the LLM
-debug data is never used as a checkpoint source. Typed artifact debug envelopes
+calls made by that module attempt. Validator attempts retain independent scopes
-are domain-neutral, redact sensitive metadata and bytes through the common
+under `validate/`. Debug-write failures are framework errors; debug data is
-debug policy, and record codec identity plus schema and content digests.
+never used as a checkpoint source. Typed artifact debug envelopes are
+domain-neutral, redact sensitive metadata and bytes through the common debug
+policy, and record codec identity plus schema and content digests.
 
 Checkpoint identity, physical layout, reuse behavior, and debug artifact
 handling are operator contracts in [Operations](../operations.md). Serialization

docs/operations.md

@@ -89,7 +89,10 @@ Treat checkpoint directories as sensitive local state.
 A checkpoint is reused only when its stored status, dependencies, payloads, and
 digests match the current invocation. Changes to input bytes, the resolved
 pipeline, selected lanes, the runtime LLM profile override, or bound reference
-content invalidate reuse.
+content invalidate reuse. The resolved pipeline identity includes effective
+default and explicitly overridden validator chains, so adding, removing,
+reordering, or reconfiguring a validator invalidates checkpoints even when the
+pipeline profile itself is unchanged.
 
 Typed artifact checkpoints additionally record codec-owned bytes, artifact
 kind, schema ID and version, exact schema digest, and media type. A missing or
@@ -130,11 +133,27 @@ write `prompt-000N.json`, `response-000N.json`, and
 `response-content-000N.*` files under that attempt directory and are linked from
 the attempt `llm_calls` array. Prompt content is written inline in the prompt
 artifact. The response metadata and body use the paired files described above;
-the body is pretty-printed JSON when possible and raw text otherwise. Debug
+the body is pretty-printed JSON when possible and raw text otherwise. Merge and
-artifacts may contain source material, reference material, prompt inputs, model
+normalize retries use these stable paths:
-outputs, and other sensitive data. Typed artifact envelopes include
+
-domain-neutral codec identity, redacted metadata and content, and digests of
+```text
-the stable codec bytes. API keys are not written, and obvious
+merge/<lane-id>/attempt-<NN>.json
+merge/<lane-id>/attempt-<NN>/prompt-<NNNN>.json
+merge/<lane-id>/attempt-<NN>/response-<NNNN>.json
+merge/<lane-id>/attempt-<NN>/response-content-<NNNN>.<ext>
+
+normalize/<lane-id>/attempt-<NN>.json
+normalize/<lane-id>/attempt-<NN>/prompt-<NNNN>.json
+normalize/<lane-id>/attempt-<NN>/response-<NNNN>.json
+normalize/<lane-id>/attempt-<NN>/response-content-<NNNN>.<ext>
+```
+
+Checkpoint-reused merge and normalize work retains the stage-level input and
+output artifacts but has no retry-attempt artifacts because no module attempt
+executed. Debug artifacts may contain source material, reference material,
+prompt inputs, model outputs, and other sensitive data. Typed artifact
+envelopes include domain-neutral codec identity, redacted metadata and content,
+and digests of the stable codec bytes. API keys are not written, and obvious
 credential-shaped values and sensitive map keys are redacted, but debug
 directories should still be protected as sensitive local state.
 

internal/cli/run.go

@@ -1117,6 +1117,10 @@ func selectedReferenceTargets(cfg config.Config, pipelineID string, only []strin
 		if err != nil {
 			return nil, fmt.Errorf("pipeline %q lane %q extract module %q: %w", strings.TrimSpace(pipelineID), laneID, extractModule, err)
 		}
+		artifactKind := extractSpec.ArtifactKind
+		if artifactKind == "" {
+			return nil, fmt.Errorf("pipeline %q lane %q extract module %q does not declare an artifact kind", strings.TrimSpace(pipelineID), laneID, extractModule)
+		}
 		targets = append(targets, selectedReferenceTarget{
 			laneID: laneID,
 			stage:  pipeline.StageExtract,
@@ -1128,7 +1132,7 @@ func selectedReferenceTargets(cfg config.Config, pipelineID string, only []strin
 		if mergeModule == "" {
 			mergeModule = pipeline.DefaultMergeModule
 		}
-		mergeSpec, err := cliReferenceMergerSpec(catalog, mergeModule)
+		mergeSpec, err := cliReferenceMergerSpec(catalog, mergeModule, artifactKind)
 		if err != nil {
 			return nil, fmt.Errorf("pipeline %q lane %q merge module %q: %w", strings.TrimSpace(pipelineID), laneID, mergeModule, err)
 		}
@@ -1143,7 +1147,7 @@ func selectedReferenceTargets(cfg config.Config, pipelineID string, only []strin
 		if normalizeModule == "" {
 			normalizeModule = pipeline.DefaultNormalizeModule
 		}
-		normalizeSpec, err := cliReferenceNormalizerSpec(catalog, normalizeModule)
+		normalizeSpec, err := cliReferenceNormalizerSpec(catalog, normalizeModule, artifactKind)
 		if err != nil {
 			return nil, fmt.Errorf("pipeline %q lane %q normalize module %q: %w", strings.TrimSpace(pipelineID), laneID, normalizeModule, err)
 		}
@@ -1189,28 +1193,47 @@ func cliReferenceExtractorSpec(catalog pipeline.ModuleCatalog, module string) (p
 	return spec, nil
 }
 
-func cliReferenceMergerSpec(catalog pipeline.ModuleCatalog, module string) (pipeline.ModuleSpec, error) {
+func cliReferenceMergerSpec(catalog pipeline.ModuleCatalog, module string, kind contracts.ArtifactKind) (pipeline.ModuleSpec, error) {
 	if catalog.Mergers == nil {
 		return pipeline.ModuleSpec{}, fmt.Errorf("module %q is not registered", module)
 	}
-	spec, ok := catalog.Mergers.Spec(module)
+	registered := catalog.Mergers.RegisteredArtifactKinds(module)
-	if !ok {
+	if len(registered) == 0 {
 		return pipeline.ModuleSpec{}, fmt.Errorf("module %q is not registered", module)
 	}
+	spec, ok := catalog.Mergers.SpecForArtifact(module, kind)
+	if !ok {
+		return pipeline.ModuleSpec{}, cliReferenceArtifactVariantError("merger", module, kind, registered)
+	}
 	return spec, nil
 }
 
-func cliReferenceNormalizerSpec(catalog pipeline.ModuleCatalog, module string) (pipeline.ModuleSpec, error) {
+func cliReferenceNormalizerSpec(catalog pipeline.ModuleCatalog, module string, kind contracts.ArtifactKind) (pipeline.ModuleSpec, error) {
 	if catalog.Normalizers == nil {
 		return pipeline.ModuleSpec{}, fmt.Errorf("module %q is not registered", module)
 	}
-	spec, ok := catalog.Normalizers.Spec(module)
+	registered := catalog.Normalizers.RegisteredArtifactKinds(module)
-	if !ok {
+	if len(registered) == 0 {
 		return pipeline.ModuleSpec{}, fmt.Errorf("module %q is not registered", module)
 	}
+	spec, ok := catalog.Normalizers.SpecForArtifact(module, kind)
+	if !ok {
+		return pipeline.ModuleSpec{}, cliReferenceArtifactVariantError("normalizer", module, kind, registered)
+	}
 	return spec, nil
 }
 
+func cliReferenceArtifactVariantError(moduleType string, module string, kind contracts.ArtifactKind, registered []contracts.ArtifactKind) error {
+	values := make([]string, len(registered))
+	for i, value := range registered {
+		values[i] = string(value)
+	}
+	if len(values) == 0 {
+		return fmt.Errorf("%s %q has no typed variant for artifact kind %q", moduleType, module, kind)
+	}
+	return fmt.Errorf("%s %q has no typed variant for artifact kind %q; registered kinds: %s", moduleType, module, kind, strings.Join(values, ", "))
+}
+
 func referenceSlotSet(slots []contracts.ReferenceSlot) map[string]struct{} {
 	slotSet := make(map[string]struct{}, len(slots))
 	for _, slot := range slots {

internal/cli/run_test.go

@@ -1268,6 +1268,111 @@ func TestRunPipelineReferenceFlagBindsUnambiguousSlot(t *testing.T) {
 	}
 }
 
+func TestCLIReferenceDiscoveryUsesLaneArtifactVariant(t *testing.T) {
+	catalog := referenceVariantCatalog(t)
+	cfg := config.Config{Pipelines: map[string]pipeline.PipelineProfile{
+		"variants": {
+			ID:    "variants",
+			Chunk: pipeline.Binding("generic"),
+			Artifacts: map[string]pipeline.ArtifactLaneProfile{
+				"alpha": {
+					Extract:   pipeline.Binding("extract/alpha"),
+					Merge:     pipeline.Binding("shared/merge"),
+					Normalize: pipeline.Binding("shared/normalize"),
+				},
+				"beta": {
+					Extract:   pipeline.Binding("extract/beta"),
+					Merge:     pipeline.Binding("shared/merge"),
+					Normalize: pipeline.Binding("shared/normalize"),
+				},
+			},
+		},
+	}}
+	targets, err := selectedReferenceTargets(cfg, "variants", nil, catalog)
+	if err != nil {
+		t.Fatalf("selectedReferenceTargets() error = %v, want nil", err)
+	}
+
+	tests := []struct {
+		name     string
+		selector cliReferenceSelector
+		laneID   string
+		stage    pipeline.ModuleStage
+	}{
+		{
+			name:     "qualified merger",
+			selector: cliReferenceSelector{LaneID: "alpha", Stage: pipeline.StageMerge, SlotName: "alpha_merge"},
+			laneID:   "alpha",
+			stage:    pipeline.StageMerge,
+		},
+		{
+			name:     "unqualified merger",
+			selector: cliReferenceSelector{Stage: pipeline.StageMerge, SlotName: "beta_merge"},
+			laneID:   "beta",
+			stage:    pipeline.StageMerge,
+		},
+		{
+			name:     "flat normalizer",
+			selector: cliReferenceSelector{SlotName: "alpha_normalize"},
+			laneID:   "alpha",
+			stage:    pipeline.StageNormalize,
+		},
+		{
+			name:     "lane normalizer",
+			selector: cliReferenceSelector{LaneID: "beta", SlotName: "beta_normalize"},
+			laneID:   "beta",
+			stage:    pipeline.StageNormalize,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			target, err := resolveCLIReferenceTarget(targets, tc.selector)
+			if err != nil {
+				t.Fatalf("resolveCLIReferenceTarget() error = %v, want nil", err)
+			}
+			if target.laneID != tc.laneID || target.stage != tc.stage {
+				t.Fatalf("target = %#v, want lane %q stage %q", target, tc.laneID, tc.stage)
+			}
+		})
+	}
+
+	_, err = resolveCLIReferenceTarget(targets, cliReferenceSelector{LaneID: "beta", Stage: pipeline.StageMerge, SlotName: "alpha_merge"})
+	if err == nil || !strings.Contains(err.Error(), "not declared") {
+		t.Fatalf("beta alpha-variant reference error = %v, want slot rejection", err)
+	}
+}
+
+func TestCLIReferenceDiscoveryReportsMissingArtifactVariant(t *testing.T) {
+	catalog := referenceVariantCatalog(t)
+	if err := pipeline.RegisterExtractor(catalog.Extractors, pipeline.ModuleSpec{Key: "extract/missing", Stage: pipeline.StageExtract, ArtifactKind: "test/missing"}, func() (contracts.Extractor[fakeRunArtifact], error) {
+		return referenceVariantExtractor{key: "extract/missing"}, nil
+	}); err != nil {
+		t.Fatalf("register missing-kind extractor: %v", err)
+	}
+	cfg := config.Config{Pipelines: map[string]pipeline.PipelineProfile{
+		"variants": {
+			ID: "variants",
+			Artifacts: map[string]pipeline.ArtifactLaneProfile{
+				"missing": {
+					Extract:   pipeline.Binding("extract/missing"),
+					Merge:     pipeline.Binding("shared/merge"),
+					Normalize: pipeline.Binding("shared/normalize"),
+				},
+			},
+		},
+	}}
+	_, err := selectedReferenceTargets(cfg, "variants", nil, catalog)
+	want := []string{"pipeline \"variants\"", "lane \"missing\"", "merge module \"shared/merge\"", "artifact kind \"test/missing\"", "registered kinds: test/alpha, test/beta"}
+	if err == nil {
+		t.Fatal("selectedReferenceTargets() error = nil, want missing variant error")
+	}
+	for _, value := range want {
+		if !strings.Contains(err.Error(), value) {
+			t.Fatalf("selectedReferenceTargets() error = %q, want %q", err, value)
+		}
+	}
+}
+
 func TestRunPipelineReferenceFlagBindsLaneQualifiedSlot(t *testing.T) {
 	configPath := writeTestConfig(t, testConfigYAML("example", "events", "notes"))
 	inputPath := filepath.Join(t.TempDir(), "missing.json")
@@ -3652,6 +3757,30 @@ const fakeRunArtifactKind contracts.ArtifactKind = "test/fake"
 type fakeRunArtifact struct {
 	Value bool `json:"value"`
 }
+
+type referenceVariantExtractor struct{ key string }
+
+func (e referenceVariantExtractor) Key() string                             { return e.key }
+func (referenceVariantExtractor) ReferenceSlots() []contracts.ReferenceSlot { return nil }
+func (referenceVariantExtractor) Extract(context.Context, contracts.TypedExtractionRequest) (contracts.TypedExtractionResult[fakeRunArtifact], error) {
+	return contracts.TypedExtractionResult[fakeRunArtifact]{}, nil
+}
+
+type referenceVariantMerger struct{ key string }
+
+func (m referenceVariantMerger) Key() string { return m.key }
+func (referenceVariantMerger) Merge(context.Context, contracts.TypedMergeRequest[fakeRunArtifact]) (contracts.TypedMergeResult[fakeRunArtifact], error) {
+	return contracts.TypedMergeResult[fakeRunArtifact]{}, nil
+}
+
+type referenceVariantNormalizer struct{ key string }
+
+func (n referenceVariantNormalizer) Key() string                             { return n.key }
+func (referenceVariantNormalizer) ReferenceSlots() []contracts.ReferenceSlot { return nil }
+func (referenceVariantNormalizer) Normalize(context.Context, contracts.TypedNormalizeRequest[fakeRunArtifact]) (contracts.TypedNormalizeResult[fakeRunArtifact], error) {
+	return contracts.TypedNormalizeResult[fakeRunArtifact]{}, nil
+}
+
 type fakeRunCodec struct{}
 
 func (fakeRunCodec) Kind() contracts.ArtifactKind { return fakeRunArtifactKind }
@@ -3994,6 +4123,52 @@ func fakeCatalog(t *testing.T, overrides ...pipeline.ModuleSpec) pipeline.Module
 	}
 }
 
+func referenceVariantCatalog(t *testing.T) pipeline.ModuleCatalog {
+	t.Helper()
+	chunkers := pipeline.NewChunkerRegistry()
+	extractors := pipeline.NewExtractorRegistry()
+	mergers := pipeline.NewMergerRegistry()
+	normalizers := pipeline.NewNormalizerRegistry()
+	mustRegisterChunker(t, chunkers, pipeline.ModuleSpec{Key: "generic", Stage: pipeline.StageChunk})
+
+	for _, item := range []struct {
+		key  string
+		kind contracts.ArtifactKind
+	}{
+		{key: "extract/alpha", kind: "test/alpha"},
+		{key: "extract/beta", kind: "test/beta"},
+	} {
+		item := item
+		if err := pipeline.RegisterExtractor(extractors, pipeline.ModuleSpec{Key: item.key, Stage: pipeline.StageExtract, ArtifactKind: item.kind}, func() (contracts.Extractor[fakeRunArtifact], error) {
+			return referenceVariantExtractor{key: item.key}, nil
+		}); err != nil {
+			t.Fatalf("register extractor %q: %v", item.key, err)
+		}
+	}
+	for _, item := range []struct {
+		kind          contracts.ArtifactKind
+		mergeSlot     string
+		normalizeSlot string
+	}{
+		{kind: "test/beta", mergeSlot: "beta_merge", normalizeSlot: "beta_normalize"},
+		{kind: "test/alpha", mergeSlot: "alpha_merge", normalizeSlot: "alpha_normalize"},
+	} {
+		mergeSpec := pipeline.ModuleSpec{Key: "shared/merge", Stage: pipeline.StageMerge, ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.mergeSlot}}}
+		if err := pipeline.RegisterMerger(mergers, mergeSpec, func() (contracts.Merger[fakeRunArtifact], error) {
+			return referenceVariantMerger{key: "shared/merge"}, nil
+		}); err != nil {
+			t.Fatalf("register merger %q: %v", item.kind, err)
+		}
+		normalizeSpec := pipeline.ModuleSpec{Key: "shared/normalize", Stage: pipeline.StageNormalize, ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.normalizeSlot}}}
+		if err := pipeline.RegisterNormalizer(normalizers, normalizeSpec, func() (contracts.Normalizer[fakeRunArtifact], error) {
+			return referenceVariantNormalizer{key: "shared/normalize"}, nil
+		}); err != nil {
+			t.Fatalf("register normalizer %q: %v", item.kind, err)
+		}
+	}
+	return pipeline.ModuleCatalog{Chunkers: chunkers, Extractors: extractors, Mergers: mergers, Normalizers: normalizers}
+}
+
 func mustRegisterInput(t *testing.T, registry *pipeline.InputAdapterRegistry, spec pipeline.ModuleSpec) {
 	t.Helper()
 	if err := registry.RegisterWithSpec(spec, func() (contracts.InputAdapter, error) { return fakeRunInputAdapter{}, nil }); err != nil {

internal/framework/pipeline/debug.go

@@ -302,6 +302,15 @@ func withDebugLLMScope(ctx context.Context, prefix string) (context.Context, *de
 	return context.WithValue(ctx, debugLLMScopeContextKey{}, scope), scope
 }
 
+func withIsolatedDebugLLMScope(ctx context.Context, prefix string) (context.Context, *debugLLMScope) {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	prefix = cleanDebugPath(prefix)
+	scope := &debugLLMScope{prefix: prefix}
+	return context.WithValue(ctx, debugLLMScopeContextKey{}, scope), scope
+}
+
 func debugLLMScopeFromContext(ctx context.Context) *debugLLMScope {
 	if ctx == nil {
 		return nil
@@ -388,6 +397,10 @@ func debugEnvelopeWithLLMCalls(envelope debugTimedEnvelope, scope *debugLLMScope
 	return envelope
 }
 
+func writeDebugAttempt(recorder DebugRecorder, attemptPath string, envelope debugTimedEnvelope, scope *debugLLMScope) error {
+	return writeDebugTimed(recorder, attemptPath+".json", debugEnvelopeWithLLMCalls(envelope, scope))
+}
+
 func debugContentEnvelope(content []byte, mediaType string, metadata map[string]any, warnings []contracts.Warning) debugBinaryEnvelope {
 	content = redactSecretBytes(content)
 	return debugBinaryEnvelope{

internal/framework/pipeline/merger_registry.go

@@ -108,16 +108,28 @@ func (r *MergerRegistry) validateOptions(key string, kind contracts.ArtifactKind
 }
 
 func (r *MergerRegistry) Spec(key string) (ModuleSpec, bool) {
-	if r == nil {
+	// Spec is for kind-neutral catalog inspection. Behavior-sensitive callers
+	// must use SpecForArtifact so they select the lane's exact typed variant.
+	kinds := r.registeredKinds(key)
+	if len(kinds) == 0 {
+		return ModuleSpec{}, false
+	}
+	return r.SpecForArtifact(key, kinds[0])
+}
+
+// SpecForArtifact returns the merger spec registered for an exact artifact kind.
+func (r *MergerRegistry) SpecForArtifact(key string, kind contracts.ArtifactKind) (ModuleSpec, bool) {
+	entry, ok := r.typedEntry(key, kind)
+	if !ok {
 		return ModuleSpec{}, false
 	}
-	module := strings.TrimSpace(key)
-	for variant, entry := range r.typedEntries {
-		if variant.module == module {
 	return cloneModuleSpec(entry.spec), true
 }
-	}
+
-	return ModuleSpec{}, false
+// RegisteredArtifactKinds returns the sorted artifact kinds registered for a
+// merger key.
+func (r *MergerRegistry) RegisteredArtifactKinds(key string) []contracts.ArtifactKind {
+	return r.registeredKinds(key)
 }
 
 func (r *MergerRegistry) typedEntry(key string, kind contracts.ArtifactKind) (typedMergerEntry, bool) {

internal/framework/pipeline/normalizer_registry.go

@@ -99,16 +99,28 @@ func (r *NormalizerRegistry) validateOptions(key string, kind contracts.Artifact
 }
 
 func (r *NormalizerRegistry) Spec(key string) (ModuleSpec, bool) {
-	if r == nil {
+	// Spec is for kind-neutral catalog inspection. Behavior-sensitive callers
+	// must use SpecForArtifact so they select the lane's exact typed variant.
+	kinds := r.registeredKinds(key)
+	if len(kinds) == 0 {
+		return ModuleSpec{}, false
+	}
+	return r.SpecForArtifact(key, kinds[0])
+}
+
+// SpecForArtifact returns the normalizer spec registered for an exact artifact kind.
+func (r *NormalizerRegistry) SpecForArtifact(key string, kind contracts.ArtifactKind) (ModuleSpec, bool) {
+	entry, ok := r.typedEntry(key, kind)
+	if !ok {
 		return ModuleSpec{}, false
 	}
-	module := strings.TrimSpace(key)
-	for variant, entry := range r.typedEntries {
-		if variant.module == module {
 	return cloneModuleSpec(entry.spec), true
 }
-	}
+
-	return ModuleSpec{}, false
+// RegisteredArtifactKinds returns the sorted artifact kinds registered for a
+// normalizer key.
+func (r *NormalizerRegistry) RegisteredArtifactKinds(key string) []contracts.ArtifactKind {
+	return r.registeredKinds(key)
 }
 
 func (r *NormalizerRegistry) typedEntry(key string, kind contracts.ArtifactKind) (typedNormalizerEntry, bool) {

internal/framework/pipeline/profile.go

@@ -417,7 +417,7 @@ func resolveArtifactIdentity(pipelineID, laneID string, lane *ResolvedArtifactLa
 
 func mergerSpecForArtifact(catalog ModuleCatalog, key string, kind contracts.ArtifactKind, expectedType reflect.Type) (ModuleSpec, error) {
 	if kind == "" {
-		return mergerSpec(catalog, key)
+		return ModuleSpec{}, fmt.Errorf("merger %q cannot be resolved without an artifact kind", key)
 	}
 	if catalog.Mergers == nil {
 		return ModuleSpec{}, fmt.Errorf("module %q is not registered", key)
@@ -434,7 +434,7 @@ func mergerSpecForArtifact(catalog ModuleCatalog, key string, kind contracts.Art
 
 func normalizerSpecForArtifact(catalog ModuleCatalog, key string, kind contracts.ArtifactKind, expectedType reflect.Type) (ModuleSpec, error) {
 	if kind == "" {
-		return normalizerSpec(catalog, key)
+		return ModuleSpec{}, fmt.Errorf("normalizer %q cannot be resolved without an artifact kind", key)
 	}
 	if catalog.Normalizers == nil {
 		return ModuleSpec{}, fmt.Errorf("module %q is not registered", key)
@@ -1010,6 +1010,7 @@ func resolvedPipelineDigest(resolved ResolvedPipeline) (string, error) {
 		Chunk           ModuleBinding
 		ChunkReferences ResolvedReferenceTarget
 		ArtifactLanes   []ResolvedArtifactLane
+		ValidatorChains []ResolvedValidatorChain
 		Output          ModuleBinding
 	}{
 		ID:              resolved.ID,
@@ -1017,6 +1018,7 @@ func resolvedPipelineDigest(resolved ResolvedPipeline) (string, error) {
 		Chunk:           resolved.Chunk,
 		ChunkReferences: resolved.ChunkReferences,
 		ArtifactLanes:   resolved.ArtifactLanes,
+		ValidatorChains: resolved.ValidatorChains,
 		Output:          resolved.Output,
 	}
 	encoded, err := json.Marshal(withoutDigest)
@@ -1040,14 +1042,6 @@ func extractorSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
 	return registrySpec(catalog.Extractors, key)
 }
 
-func mergerSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
-	return registrySpec(catalog.Mergers, key)
-}
-
-func normalizerSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
-	return registrySpec(catalog.Normalizers, key)
-}
-
 func outputSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
 	return registrySpec(catalog.Outputs, key)
 }

internal/framework/pipeline/profile_test.go

@@ -1062,6 +1062,186 @@ func TestResolvePipelineDigestChangesWhenBindingChanges(t *testing.T) {
 	}
 }
 
+func TestResolvePipelineDigestIncludesEffectiveValidatorChain(t *testing.T) {
+	resolve := func(t *testing.T, validators []ModuleBinding, explicitEmpty bool) ResolvedPipeline {
+		t.Helper()
+		catalog := newProfileCatalog(t)
+		registerProfileValidatorSpec(t, catalog, ValidatorSpec{Key: "second-validator", ExecutionClass: contracts.ExecutionClassDeterministic})
+		if len(validators) > 0 {
+			if err := catalog.ValidatorChains.Register(ValidatorChainMapping{
+				Stage:      StageExtract,
+				Module:     "event-extractor",
+				Validators: validators,
+			}); err != nil {
+				t.Fatalf("register validator chain: %v", err)
+			}
+		}
+
+		profile := baselineProfile()
+		if explicitEmpty {
+			lane := profile.Artifacts["events"]
+			lane.Extract.Validators = ValidatorOverride{Set: true}
+			profile.Artifacts["events"] = lane
+		}
+		resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog)
+		if err != nil {
+			t.Fatalf("ResolvePipeline() error = %v, want nil", err)
+		}
+		return resolved
+	}
+
+	inherited := resolve(t, []ModuleBinding{Binding("grounded"), Binding("second-validator")}, false)
+	repeated := resolve(t, []ModuleBinding{Binding("grounded"), Binding("second-validator")}, false)
+	if inherited.Digest != repeated.Digest {
+		t.Fatalf("same resolved validator policy produced digests %q and %q", inherited.Digest, repeated.Digest)
+	}
+
+	tests := []struct {
+		name  string
+		left  ResolvedPipeline
+		right ResolvedPipeline
+	}{
+		{
+			name:  "different default",
+			left:  resolve(t, []ModuleBinding{Binding("grounded")}, false),
+			right: resolve(t, []ModuleBinding{Binding("second-validator")}, false),
+		},
+		{
+			name:  "added validator",
+			left:  resolve(t, nil, false),
+			right: resolve(t, []ModuleBinding{Binding("grounded")}, false),
+		},
+		{
+			name:  "removed validator",
+			left:  resolve(t, []ModuleBinding{Binding("grounded")}, false),
+			right: resolve(t, nil, false),
+		},
+		{
+			name:  "reordered chain",
+			left:  inherited,
+			right: resolve(t, []ModuleBinding{Binding("second-validator"), Binding("grounded")}, false),
+		},
+		{
+			name:  "explicit empty",
+			left:  inherited,
+			right: resolve(t, []ModuleBinding{Binding("grounded"), Binding("second-validator")}, true),
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if tc.left.Digest == tc.right.Digest {
+				t.Fatalf("digest = %q for both validator policies, want changed digest", tc.left.Digest)
+			}
+		})
+	}
+}
+
+func TestResolvedPipelineDigestIncludesCompleteValidatorPolicy(t *testing.T) {
+	base := validatorDigestFixture()
+	baseDigest, err := resolvedPipelineDigest(base)
+	if err != nil {
+		t.Fatalf("resolvedPipelineDigest(base) error = %v, want nil", err)
+	}
+
+	tests := []struct {
+		name   string
+		mutate func(*ResolvedPipeline)
+	}{
+		{name: "stage", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Stage = StageMerge }},
+		{name: "lane", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].LaneID = "other" }},
+		{name: "owner module", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].ModuleKey = "other-extractor" }},
+		{name: "validator order", mutate: func(value *ResolvedPipeline) {
+			value.ValidatorChains[0].Validators[0], value.ValidatorChains[0].Validators[1] = value.ValidatorChains[0].Validators[1], value.ValidatorChains[0].Validators[0]
+		}},
+		{name: "validator module", mutate: func(value *ResolvedPipeline) {
+			value.ValidatorChains[0].Validators[0].Binding.Module = "other-validator"
+		}},
+		{name: "llm profile", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Validators[0].Binding.LLMProfile = "fast" }},
+		{name: "retries", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Validators[0].Binding.Retries++ }},
+		{name: "options", mutate: func(value *ResolvedPipeline) {
+			value.ValidatorChains[0].Validators[0].Binding.Options["threshold"] = 0.75
+		}},
+		{name: "references", mutate: func(value *ResolvedPipeline) {
+			value.ValidatorChains[0].Validators[0].Binding.References["rules"] = "other.md"
+		}},
+		{name: "execution class", mutate: func(value *ResolvedPipeline) {
+			value.ValidatorChains[0].Validators[0].ExecutionClass = contracts.ExecutionClassDeterministic
+		}},
+		{name: "target", mutate: func(value *ResolvedPipeline) {
+			value.ValidatorChains[0].Validators[0].Target = ValidatorTargetSerialized
+		}},
+		{name: "artifact kind", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Validators[0].ArtifactKind = "test/other" }},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			changed := validatorDigestFixture()
+			tc.mutate(&changed)
+			digest, err := resolvedPipelineDigest(changed)
+			if err != nil {
+				t.Fatalf("resolvedPipelineDigest(changed) error = %v, want nil", err)
+			}
+			if digest == baseDigest {
+				t.Fatalf("digest = %q after %s change, want different digest", digest, tc.name)
+			}
+		})
+	}
+}
+
+func TestResolvedPipelineDigestCanonicalizesValidatorBindingMaps(t *testing.T) {
+	left := validatorDigestFixture()
+	right := validatorDigestFixture()
+	right.ValidatorChains[0].Validators[0].Binding.Options = map[string]any{}
+	right.ValidatorChains[0].Validators[0].Binding.Options["threshold"] = 0.5
+	right.ValidatorChains[0].Validators[0].Binding.Options["mode"] = "strict"
+	right.ValidatorChains[0].Validators[0].Binding.References = map[string]string{}
+	right.ValidatorChains[0].Validators[0].Binding.References["examples"] = "examples.md"
+	right.ValidatorChains[0].Validators[0].Binding.References["rules"] = "rules.md"
+
+	leftDigest, err := resolvedPipelineDigest(left)
+	if err != nil {
+		t.Fatalf("resolvedPipelineDigest(left) error = %v, want nil", err)
+	}
+	rightDigest, err := resolvedPipelineDigest(right)
+	if err != nil {
+		t.Fatalf("resolvedPipelineDigest(right) error = %v, want nil", err)
+	}
+	if leftDigest != rightDigest {
+		t.Fatalf("equivalent validator maps produced digests %q and %q", leftDigest, rightDigest)
+	}
+}
+
+func validatorDigestFixture() ResolvedPipeline {
+	return ResolvedPipeline{
+		ID: "validator-policy",
+		ValidatorChains: []ResolvedValidatorChain{{
+			Stage:     StageExtract,
+			LaneID:    "events",
+			ModuleKey: "event-extractor",
+			Validators: []ResolvedValidator{
+				{
+					Binding: ModuleBinding{
+						Module:     "semantic-validator",
+						LLMProfile: "careful",
+						Retries:    2,
+						Options:    map[string]any{"mode": "strict", "threshold": 0.5},
+						References: map[string]string{"rules": "rules.md", "examples": "examples.md"},
+					},
+					ExecutionClass: contracts.ExecutionClassLLMBacked,
+					Target:         ValidatorTargetTyped,
+					ArtifactKind:   "test/notes",
+				},
+				{
+					Binding:        Binding("grounded"),
+					ExecutionClass: contracts.ExecutionClassDeterministic,
+					Target:         ValidatorTargetTyped,
+					ArtifactKind:   "test/notes",
+				},
+			},
+		}},
+	}
+}
+
 func TestBindingTrimsModuleAndLeavesResolutionFieldsEmpty(t *testing.T) {
 	binding := Binding(" module ")
 	if binding.Module != "module" {

internal/framework/pipeline/references.go

@@ -30,7 +30,7 @@ type ReferenceMaterializationOptions struct {
 func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, options ReferenceMaterializationOptions) (ResolvedPipeline, []contracts.Warning, error) {
 	out := resolved
 	out.ChunkReferences = CloneReferenceTarget(resolved.ChunkReferences)
-	chunkReferenceSet, chunkWarnings, err := materializeReferenceTarget(resolved.ID, resolved.ChunkReferences, catalog, options)
+	chunkReferenceSet, chunkWarnings, err := materializeReferenceTarget(resolved.ID, resolved.ChunkReferences, "", catalog, options)
 	if err != nil {
 		return ResolvedPipeline{}, nil, err
 	}
@@ -46,21 +46,21 @@ func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, opt
 		materializedLane.ExtractReferences = CloneReferenceTarget(lane.ExtractReferences)
 		materializedLane.MergeReferences = CloneReferenceTarget(lane.MergeReferences)
 		materializedLane.NormalizeReferences = CloneReferenceTarget(lane.NormalizeReferences)
-		extractReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.ExtractReferences, catalog, options)
+		extractReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.ExtractReferences, lane.ArtifactKind, catalog, options)
 		if err != nil {
 			return ResolvedPipeline{}, nil, err
 		}
 		materializedLane.ExtractReferences.ReferenceSet = extractReferenceSet
 		warnings = append(warnings, laneWarnings...)
 
-		mergeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.MergeReferences, catalog, options)
+		mergeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.MergeReferences, lane.ArtifactKind, catalog, options)
 		if err != nil {
 			return ResolvedPipeline{}, nil, err
 		}
 		materializedLane.MergeReferences.ReferenceSet = mergeReferenceSet
 		warnings = append(warnings, laneWarnings...)
 
-		normalizeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.NormalizeReferences, catalog, options)
+		normalizeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.NormalizeReferences, lane.ArtifactKind, catalog, options)
 		if err != nil {
 			return ResolvedPipeline{}, nil, err
 		}
@@ -74,13 +74,14 @@ func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, opt
 func materializeReferenceTarget(
 	pipelineID string,
 	target ResolvedReferenceTarget,
+	artifactKind contracts.ArtifactKind,
 	catalog ModuleCatalog,
 	options ReferenceMaterializationOptions,
 ) (contracts.ReferenceSet, []contracts.Warning, error) {
 	if len(target.Bindings) == 0 {
 		return contracts.ReferenceSet{}, nil, nil
 	}
-	spec, err := referenceTargetSpec(target, catalog)
+	spec, err := referenceTargetSpec(target, artifactKind, catalog)
 	if err != nil {
 		return contracts.ReferenceSet{}, nil, fmt.Errorf("%s: %w", referenceTargetContext(pipelineID, target), err)
 	}
@@ -142,16 +143,28 @@ func materializeReferenceTarget(
 	return set, warnings, nil
 }
 
-func referenceTargetSpec(target ResolvedReferenceTarget, catalog ModuleCatalog) (ModuleSpec, error) {
+func referenceTargetSpec(target ResolvedReferenceTarget, artifactKind contracts.ArtifactKind, catalog ModuleCatalog) (ModuleSpec, error) {
 	switch target.Stage {
 	case StageChunk:
 		return registrySpec(catalog.Chunkers, target.Module)
 	case StageExtract:
 		return registrySpec(catalog.Extractors, target.Module)
 	case StageMerge:
-		return registrySpec(catalog.Mergers, target.Module)
+		if catalog.Mergers == nil {
+			return ModuleSpec{}, fmt.Errorf("module %q is not registered", target.Module)
+		}
+		if spec, ok := catalog.Mergers.SpecForArtifact(target.Module, artifactKind); ok {
+			return spec, nil
+		}
+		return ModuleSpec{}, missingArtifactVariantError("merger", target.Module, artifactKind, catalog.Mergers.registeredKinds(target.Module))
 	case StageNormalize:
-		return registrySpec(catalog.Normalizers, target.Module)
+		if catalog.Normalizers == nil {
+			return ModuleSpec{}, fmt.Errorf("module %q is not registered", target.Module)
+		}
+		if spec, ok := catalog.Normalizers.SpecForArtifact(target.Module, artifactKind); ok {
+			return spec, nil
+		}
+		return ModuleSpec{}, missingArtifactVariantError("normalizer", target.Module, artifactKind, catalog.Normalizers.registeredKinds(target.Module))
 	default:
 		return ModuleSpec{}, fmt.Errorf("reference target stage %q is not supported", target.Stage)
 	}

internal/framework/pipeline/references_test.go

@@ -175,6 +175,57 @@ func TestMaterializeReferencesStoresSetsAndProvenanceForAllTargets(t *testing.T)
 	}
 }
 
+func TestMaterializeReferencesUsesLaneArtifactVariant(t *testing.T) {
+	configDir := t.TempDir()
+	writeReferenceFile(t, filepath.Join(configDir, "merge.txt"), []byte("merge alpha"))
+	writeReferenceFile(t, filepath.Join(configDir, "normalize.txt"), []byte("normalize alpha"))
+	mergers := NewMergerRegistry()
+	normalizers := NewNormalizerRegistry()
+	for _, item := range []struct {
+		kind          contracts.ArtifactKind
+		mergeSlot     string
+		normalizeSlot string
+	}{
+		{kind: "test/beta", mergeSlot: "beta_merge", normalizeSlot: "beta_normalize"},
+		{kind: "test/alpha", mergeSlot: "alpha_merge", normalizeSlot: "alpha_normalize"},
+	} {
+		if err := RegisterMerger(mergers, ModuleSpec{Key: "shared/merge", Stage: StageMerge, ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.mergeSlot}}}, func() (contracts.Merger[codecNotes], error) {
+			return typedTestMerger[codecNotes]{key: "shared/merge"}, nil
+		}); err != nil {
+			t.Fatalf("RegisterMerger(%s): %v", item.kind, err)
+		}
+		if err := RegisterNormalizer(normalizers, ModuleSpec{Key: "shared/normalize", Stage: StageNormalize, ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.normalizeSlot}}}, func() (contracts.Normalizer[codecNotes], error) {
+			return typedTestNormalizer[codecNotes]{key: "shared/normalize"}, nil
+		}); err != nil {
+			t.Fatalf("RegisterNormalizer(%s): %v", item.kind, err)
+		}
+	}
+	resolved := ResolvedPipeline{
+		ID: "variants",
+		ArtifactLanes: []ResolvedArtifactLane{{
+			ID:           "alpha",
+			ArtifactKind: "test/alpha",
+			MergeReferences: referenceTarget(StageMerge, "alpha", "shared/merge", []ReferenceBinding{{
+				Stage: StageMerge, LaneID: "alpha", SlotName: "alpha_merge", Source: "merge.txt", BindingSource: contracts.ReferenceBindingSourceConfig,
+			}}),
+			NormalizeReferences: referenceTarget(StageNormalize, "alpha", "shared/normalize", []ReferenceBinding{{
+				Stage: StageNormalize, LaneID: "alpha", SlotName: "alpha_normalize", Source: "normalize.txt", BindingSource: contracts.ReferenceBindingSourceConfig,
+			}}),
+		}},
+	}
+	materialized, _, err := MaterializeReferences(resolved, ModuleCatalog{Mergers: mergers, Normalizers: normalizers}, ReferenceMaterializationOptions{ConfigPath: filepath.Join(configDir, "notarius.yml")})
+	if err != nil {
+		t.Fatalf("MaterializeReferences() error = %v, want nil", err)
+	}
+	lane := materialized.ArtifactLanes[0]
+	if got := string(lane.MergeReferences.ReferenceSet.Slots["alpha_merge"].Items[0].Content); got != "merge alpha" {
+		t.Fatalf("merge reference = %q", got)
+	}
+	if got := string(lane.NormalizeReferences.ReferenceSet.Slots["alpha_normalize"].Items[0].Content); got != "normalize alpha" {
+		t.Fatalf("normalize reference = %q", got)
+	}
+}
+
 func TestMaterializeReferencesRejectsNonUTF8Content(t *testing.T) {
 	configDir := t.TempDir()
 	path := filepath.Join(configDir, "bad.txt")

internal/framework/pipeline/runner.go

@@ -56,7 +56,6 @@ type RunInput struct {
 
 	pipeline  ResolvedPipeline
 	llmClient contracts.StructuredLLMClient
-	extractDecision *CheckpointDecision
 }
 
 type RunOutput struct {
@@ -217,30 +216,30 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
 				Metadata:    input.Metadata,
 			})
 			if err != nil {
-				_ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
+				_ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{
 					Stage:     string(StageChunk),
 					ModuleKey: chunker.Key(),
 					Attempt:   attempt,
 					StartedAt: attemptStarted,
 					Error:     err.Error(),
-				}, llmScope))
+				}, llmScope)
 				return false, nil, fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), err)
 			}
 			if len(chunkResult.Chunks) == 0 {
 				err := fmt.Errorf("chunker %q returned no chunks", chunker.Key())
-				_ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
+				_ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{
 					Stage:     string(StageChunk),
 					ModuleKey: chunker.Key(),
 					Attempt:   attempt,
 					StartedAt: attemptStarted,
 					Error:     err.Error(),
-				}, llmScope))
+				}, llmScope)
 				return false, nil, err
 			}
 			chunks, err := validateAndCanonicalizeChunkResult(doc, chunkResult.Chunks)
 			if err != nil {
 				err := fmt.Errorf("validate chunks from chunker %q: %w", chunker.Key(), err)
-				_ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
+				_ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{
 					Stage:     string(StageChunk),
 					ModuleKey: chunker.Key(),
 					Attempt:   attempt,
@@ -249,12 +248,12 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
 						"warnings": cloneWarnings(chunkResult.Warnings),
 					},
 					Error: err.Error(),
-				}, llmScope))
+				}, llmScope)
 				return false, nil, err
 			}
 			validationWarnings, rejection, err := r.validateChunks(attemptCtx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.Metadata, input.Prepared.chunkValidators, attempt, input.Debug)
 			if err != nil || rejection != nil {
-				_ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
+				_ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{
 					Stage:     string(StageChunk),
 					ModuleKey: chunker.Key(),
 					Attempt:   attempt,
@@ -264,12 +263,12 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
 						"warnings":  append(cloneWarnings(chunkResult.Warnings), validationWarnings...),
 						"rejection": debugRejectedOutputPtr(rejection),
 					},
-				}, llmScope))
+				}, llmScope)
 				return false, rejection, err
 			}
 			canonicalChunks = chunks
 			chunkWarnings = append(cloneWarnings(chunkResult.Warnings), validationWarnings...)
-			if err := writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
+			if err := writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{
 				Stage:     string(StageChunk),
 				ModuleKey: chunker.Key(),
 				Attempt:   attempt,
@@ -278,7 +277,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
 					"chunks":   debugSourceChunkEnvelopes(chunks),
 					"warnings": chunkWarnings,
 				},
-			}, llmScope)); err != nil {
+			}, llmScope); err != nil {
 				return false, nil, err
 			}
 			return true, nil, nil
@@ -431,7 +430,7 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument,
 		binding := item.resolved.Binding
 		started := time.Now().UTC()
 		attemptPath := path.Join("validate", debugPathComponent(string(StageChunk)), "", debugPathComponent(moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d", index+1, debugPathComponent(binding.Module), attempt))
-		validatorCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
+		validatorCtx, llmScope := withIsolatedDebugLLMScope(ctx, attemptPath)
 		var result contracts.ValidationResult
 		switch item.resolved.Target {
 		case ValidatorTargetChunk:
@@ -447,7 +446,7 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument,
 		if err != nil {
 			debugCall.Error = err.Error()
 		}
-		if debugErr := writeDebugTimed(debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope)); debugErr != nil {
+		if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope); debugErr != nil {
 			return nil, nil, debugErr
 		}
 		if err != nil {

internal/framework/pipeline/runner_attempt_debug_test.go

@@ -0,0 +1,430 @@
+package pipeline
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+)
+
+type capturedDebugRecorder struct {
+	mu       sync.Mutex
+	json     map[string][]byte
+	bytes    map[string][]byte
+	failPath string
+}
+
+func newCapturedDebugRecorder() *capturedDebugRecorder {
+	return &capturedDebugRecorder{json: make(map[string][]byte), bytes: make(map[string][]byte)}
+}
+
+func (*capturedDebugRecorder) Enabled() bool { return true }
+
+func (r *capturedDebugRecorder) WriteJSON(name string, payload any) error {
+	if name == r.failPath {
+		return errors.New("debug recorder failure")
+	}
+	data, err := json.Marshal(payload)
+	if err != nil {
+		return err
+	}
+	r.mu.Lock()
+	r.json[name] = data
+	r.mu.Unlock()
+	return nil
+}
+
+func (r *capturedDebugRecorder) WriteBytes(name string, data []byte) error {
+	if name == r.failPath {
+		return errors.New("debug recorder failure")
+	}
+	r.mu.Lock()
+	r.bytes[name] = append([]byte(nil), data...)
+	r.mu.Unlock()
+	return nil
+}
+
+func (r *capturedDebugRecorder) has(name string) bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	_, jsonOK := r.json[name]
+	_, bytesOK := r.bytes[name]
+	return jsonOK || bytesOK
+}
+
+func (r *capturedDebugRecorder) names() []string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	names := make([]string, 0, len(r.json)+len(r.bytes))
+	for name := range r.json {
+		names = append(names, name)
+	}
+	for name := range r.bytes {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+	return names
+}
+
+func (r *capturedDebugRecorder) envelope(t *testing.T, name string) debugTimedEnvelope {
+	t.Helper()
+	r.mu.Lock()
+	data := append([]byte(nil), r.json[name]...)
+	r.mu.Unlock()
+	if len(data) == 0 {
+		t.Fatalf("debug envelope %q was not written; names = %#v", name, r.names())
+	}
+	var envelope debugTimedEnvelope
+	if err := json.Unmarshal(data, &envelope); err != nil {
+		t.Fatalf("decode debug envelope %q: %v", name, err)
+	}
+	return envelope
+}
+
+type attemptDebugLLM struct{}
+
+func (attemptDebugLLM) CompleteStructured(_ context.Context, request contracts.StructuredCompletionRequest, _ any) (contracts.StructuredCompletionResponse, error) {
+	return contracts.StructuredCompletionResponse{
+		Content:   json.RawMessage(`{"accepted":true}`),
+		Provider:  "test",
+		Model:     "test-model",
+		ProfileID: request.ProfileID,
+		Debug: &contracts.LLMDebugMaterial{
+			Prompt: &contracts.LLMDebugPrompt{PromptID: request.PromptID, Messages: []contracts.LLMDebugMessage{{Role: "user", Content: "test"}}},
+			Response: &contracts.LLMDebugResponse{
+				Content:   `{"accepted":true}`,
+				PromptID:  request.PromptID,
+				ModelName: "test-model",
+			},
+		},
+	}, nil
+}
+
+func preparedAttemptDebugPipeline(t *testing.T) *PreparedPipeline {
+	t.Helper()
+	prepared := preparedConcurrentPipeline(t, 1)
+	prepared.lanes = prepared.lanes[:1]
+	prepared.resolved.ArtifactLanes = prepared.resolved.ArtifactLanes[:1]
+	prepared.ArtifactLanes = prepared.ArtifactLanes[:1]
+	prepared.lanes[0].mergeValidators = preparedValidatorChain{}
+	prepared.lanes[0].normalizeValidators = preparedValidatorChain{}
+	return prepared
+}
+
+func callAttemptDebugLLM(ctx context.Context, client contracts.StructuredLLMClient, name string) error {
+	_, err := client.CompleteStructured(ctx, contracts.StructuredCompletionRequest{StageName: "unscoped-" + name, PromptID: name, ProfileID: "test"}, nil)
+	return err
+}
+
+func TestRunnerWritesAttemptScopedMergeAndNormalizeDebug(t *testing.T) {
+	prepared := preparedAttemptDebugPipeline(t)
+	debug := newCapturedDebugRecorder()
+	client := WithDebugLLMRecording(attemptDebugLLM{}, debug)
+	prepared.lanes[0].typed.merge = func(ctx context.Context, _ any, _ contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
+		if err := callAttemptDebugLLM(ctx, client, "merge"); err != nil {
+			return erasedTypedResult{}, err
+		}
+		return erasedTypedResult{Value: codecNotes{Items: []string{"merged"}}, Warnings: []contracts.Warning{{Scope: "merge", ReasonCode: "observed", Message: "merge warning"}}}, nil
+	}
+	prepared.lanes[0].typed.normalize = func(ctx context.Context, _ any, _ contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
+		if err := callAttemptDebugLLM(ctx, client, "normalize"); err != nil {
+			return erasedTypedResult{}, err
+		}
+		return erasedTypedResult{Value: codecNotes{Items: []string{"normalized"}}, Warnings: []contracts.Warning{{Scope: "normalize", ReasonCode: "observed", Message: "normalize warning"}}}, nil
+	}
+
+	output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug})
+	if err != nil {
+		t.Fatalf("Run() error = %v, want nil", err)
+	}
+	if len(output.NormalizeOutputs) != 1 {
+		t.Fatalf("normalize outputs = %d, want one", len(output.NormalizeOutputs))
+	}
+
+	wantPaths := []string{
+		"merge/notes/input.json",
+		"merge/notes/attempt-01.json",
+		"merge/notes/attempt-01/prompt-0001.json",
+		"merge/notes/attempt-01/response-0001.json",
+		"merge/notes/attempt-01/response-content-0001.json",
+		"merge/notes/output.json",
+		"normalize/notes/input.json",
+		"normalize/notes/attempt-01.json",
+		"normalize/notes/attempt-01/prompt-0002.json",
+		"normalize/notes/attempt-01/response-0002.json",
+		"normalize/notes/attempt-01/response-content-0002.json",
+		"normalize/notes/output.json",
+	}
+	for _, name := range wantPaths {
+		if !debug.has(name) {
+			t.Errorf("debug artifact %q missing; names = %#v", name, debug.names())
+		}
+	}
+	for _, name := range debug.names() {
+		if strings.HasPrefix(name, "unscoped-merge/") || strings.HasPrefix(name, "unscoped-normalize/") {
+			t.Errorf("LLM call used fallback debug path %q", name)
+		}
+	}
+	mergeEnvelope := debug.envelope(t, "merge/notes/attempt-01.json")
+	normalizeEnvelope := debug.envelope(t, "normalize/notes/attempt-01.json")
+	if len(mergeEnvelope.LLMCalls) != 1 || mergeEnvelope.LLMCalls[0].ResponsePath != "merge/notes/attempt-01/response-0001.json" {
+		t.Fatalf("merge LLM calls = %#v, want attempt-scoped call", mergeEnvelope.LLMCalls)
+	}
+	if len(normalizeEnvelope.LLMCalls) != 1 || normalizeEnvelope.LLMCalls[0].ResponsePath != "normalize/notes/attempt-01/response-0002.json" {
+		t.Fatalf("normalize LLM calls = %#v, want attempt-scoped call", normalizeEnvelope.LLMCalls)
+	}
+}
+
+func TestRunnerRecordsDistinctRetryAttemptsAndPromotesAcceptedWarningsOnly(t *testing.T) {
+	for _, stage := range []ModuleStage{StageMerge, StageNormalize} {
+		t.Run(string(stage), func(t *testing.T) {
+			prepared := preparedAttemptDebugPipeline(t)
+			debug := newCapturedDebugRecorder()
+			client := WithDebugLLMRecording(attemptDebugLLM{}, debug)
+			lane := &prepared.lanes[0]
+			attempts := 0
+			operation := func(ctx context.Context) (erasedTypedResult, error) {
+				attempts++
+				if err := callAttemptDebugLLM(ctx, client, string(stage)); err != nil {
+					return erasedTypedResult{}, err
+				}
+				scope := "accepted"
+				if attempts == 1 {
+					scope = "discarded"
+				}
+				return erasedTypedResult{Value: codecNotes{Items: []string{scope}}, Warnings: []contracts.Warning{{Scope: scope, ReasonCode: "observed", Message: scope}}}, nil
+			}
+			validatorCalls := 0
+			validator := preparedValidator{
+				resolved: ResolvedValidator{Binding: Binding("retry-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"},
+				typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {
+					validatorCalls++
+					return contracts.ValidationResult{Approved: validatorCalls > 1, ReasonCode: "retry", Message: "retry candidate"}, nil
+				},
+			}
+			switch stage {
+			case StageMerge:
+				lane.resolved.Merge.Retries = 1
+				lane.mergeValidators.validators = []preparedValidator{validator}
+				lane.typed.merge = func(ctx context.Context, _ any, _ contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
+					return operation(ctx)
+				}
+			case StageNormalize:
+				lane.resolved.Normalize.Retries = 1
+				lane.normalizeValidators.validators = []preparedValidator{validator}
+				lane.typed.normalize = func(ctx context.Context, _ any, _ contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
+					return operation(ctx)
+				}
+			}
+
+			output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug})
+			if err != nil {
+				t.Fatalf("Run() error = %v, want nil", err)
+			}
+			laneID := lane.resolved.ID
+			firstPath := fmt.Sprintf("%s/%s/attempt-01.json", stage, laneID)
+			secondPath := fmt.Sprintf("%s/%s/attempt-02.json", stage, laneID)
+			first := debug.envelope(t, firstPath)
+			second := debug.envelope(t, secondPath)
+			if len(first.LLMCalls) != 1 || len(second.LLMCalls) != 1 || first.LLMCalls[0].CallID == second.LLMCalls[0].CallID {
+				t.Fatalf("retry LLM calls = first %#v, second %#v; want distinct calls", first.LLMCalls, second.LLMCalls)
+			}
+			if !strings.Contains(string(debug.json[firstPath]), "discarded") || !strings.Contains(string(debug.json[firstPath]), "rejection") {
+				t.Fatalf("first attempt envelope = %s, want discarded warning and rejection", debug.json[firstPath])
+			}
+			if len(output.Warnings) != 1 || output.Warnings[0].Scope != "accepted" {
+				t.Fatalf("promoted warnings = %#v, want accepted attempt only", output.Warnings)
+			}
+		})
+	}
+}
+
+func TestRunnerAttemptDebugRepresentsFailuresAndFinalRejection(t *testing.T) {
+	tests := []struct {
+		name      string
+		configure func(*PreparedPipeline)
+		path      string
+		wantError string
+		wantBody  string
+	}{
+		{
+			name: "merge module error",
+			configure: func(prepared *PreparedPipeline) {
+				prepared.lanes[0].typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
+					return erasedTypedResult{}, errors.New("merge exploded")
+				}
+			},
+			path:      "merge/notes/attempt-01.json",
+			wantError: "merge exploded",
+		},
+		{
+			name: "normalize validator error",
+			configure: func(prepared *PreparedPipeline) {
+				prepared.lanes[0].normalizeValidators.validators = []preparedValidator{{
+					resolved: ResolvedValidator{Binding: Binding("error-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"},
+					typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {
+						return contracts.ValidationResult{}, errors.New("validator exploded")
+					},
+				}}
+			},
+			path:      "normalize/notes/attempt-01.json",
+			wantError: "validator exploded",
+			wantBody:  "output",
+		},
+		{
+			name: "merge final rejection",
+			configure: func(prepared *PreparedPipeline) {
+				prepared.lanes[0].mergeValidators.validators = []preparedValidator{{
+					resolved: ResolvedValidator{Binding: Binding("reject-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"},
+					typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {
+						return contracts.ValidationResult{Approved: false, ReasonCode: "rejected", Message: "not accepted"}, nil
+					},
+				}}
+			},
+			path:     "merge/notes/attempt-01.json",
+			wantBody: "rejection",
+		},
+		{
+			name: "normalize serialization error",
+			configure: func(prepared *PreparedPipeline) {
+				prepared.lanes[0].typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
+					return erasedTypedResult{Value: "wrong artifact type"}, nil
+				}
+			},
+			path:      "normalize/notes/attempt-01.json",
+			wantError: "serialize normalize candidate",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			prepared := preparedAttemptDebugPipeline(t)
+			debug := newCapturedDebugRecorder()
+			tc.configure(prepared)
+			output, runErr := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug})
+			envelope := debug.envelope(t, tc.path)
+			if tc.wantError != "" {
+				if runErr == nil || !strings.Contains(runErr.Error(), tc.wantError) || !strings.Contains(envelope.Error, tc.wantError) {
+					t.Fatalf("run error = %v, envelope error = %q; want %q", runErr, envelope.Error, tc.wantError)
+				}
+			} else if runErr != nil {
+				t.Fatalf("Run() error = %v, want nil rejection outcome", runErr)
+			}
+			if tc.wantBody != "" && !strings.Contains(string(debug.json[tc.path]), tc.wantBody) {
+				t.Fatalf("attempt envelope = %s, want %q", debug.json[tc.path], tc.wantBody)
+			}
+			if tc.name == "merge final rejection" && len(output.Rejected) != 1 {
+				t.Fatalf("rejected outputs = %#v, want one", output.Rejected)
+			}
+		})
+	}
+}
+
+func TestRunnerKeepsValidatorLLMCallsOutOfModuleAttempt(t *testing.T) {
+	prepared := preparedAttemptDebugPipeline(t)
+	debug := newCapturedDebugRecorder()
+	client := WithDebugLLMRecording(attemptDebugLLM{}, debug)
+	prepared.lanes[0].typed.merge = func(ctx context.Context, _ any, _ contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
+		if err := callAttemptDebugLLM(ctx, client, "merge-module"); err != nil {
+			return erasedTypedResult{}, err
+		}
+		return erasedTypedResult{Value: codecNotes{Items: []string{"merged"}}}, nil
+	}
+	prepared.lanes[0].mergeValidators.validators = []preparedValidator{{
+		resolved: ResolvedValidator{Binding: Binding("llm-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"},
+		typedValidate: func(ctx context.Context, _ any, _ typedValidationTarget) (contracts.ValidationResult, error) {
+			if err := callAttemptDebugLLM(ctx, client, "merge-validator"); err != nil {
+				return contracts.ValidationResult{}, err
+			}
+			return contracts.ValidationResult{Approved: true}, nil
+		},
+	}}
+
+	if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug}); err != nil {
+		t.Fatalf("Run() error = %v, want nil", err)
+	}
+	moduleEnvelope := debug.envelope(t, "merge/notes/attempt-01.json")
+	validatorPath := "validate/merge/notes/typed~2fmerge/01-llm-check-attempt-01.json"
+	validatorEnvelope := debug.envelope(t, validatorPath)
+	if len(moduleEnvelope.LLMCalls) != 1 || !strings.Contains(moduleEnvelope.LLMCalls[0].ResponsePath, "merge/notes/attempt-01/") {
+		t.Fatalf("module LLM calls = %#v, want module call only", moduleEnvelope.LLMCalls)
+	}
+	if len(validatorEnvelope.LLMCalls) != 1 || !strings.Contains(validatorEnvelope.LLMCalls[0].ResponsePath, "validate/merge/notes/") {
+		t.Fatalf("validator LLM calls = %#v, want validator call only", validatorEnvelope.LLMCalls)
+	}
+	if moduleEnvelope.LLMCalls[0].CallID == validatorEnvelope.LLMCalls[0].CallID {
+		t.Fatalf("module and validator envelopes reference the same call: %#v", moduleEnvelope.LLMCalls)
+	}
+}
+
+type attemptReuseLoader struct {
+	CheckpointLoader
+	laneID    string
+	merge     MergeCheckpoint
+	normalize NormalizeCheckpoint
+}
+
+func (l attemptReuseLoader) Enabled() bool { return true }
+
+func (l attemptReuseLoader) Merge(laneID, _ string, _ []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) {
+	if laneID == l.laneID {
+		return l.merge, CheckpointDecision{Reused: true, Reason: "test reuse"}
+	}
+	return MergeCheckpoint{}, CheckpointDecision{Reason: "not found"}
+}
+
+func (l attemptReuseLoader) Normalize(laneID, _ string, _ []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) {
+	if laneID == l.laneID {
+		return l.normalize, CheckpointDecision{Reused: true, Reason: "test reuse"}
+	}
+	return NormalizeCheckpoint{}, CheckpointDecision{Reason: "not found"}
+}
+
+func TestRunnerCheckpointReuseDoesNotSynthesizeModuleAttempts(t *testing.T) {
+	prepared := preparedAttemptDebugPipeline(t)
+	lane := prepared.lanes[0]
+	merge, err := checkpointArtifact(lane.typed.codec, lane.resolved.ID, lane.resolved.Merge.Module, "source", codecNotes{Items: []string{"merged"}})
+	if err != nil {
+		t.Fatalf("checkpointArtifact(merge): %v", err)
+	}
+	normalize, err := checkpointArtifact(lane.typed.codec, lane.resolved.ID, lane.resolved.Normalize.Module, "source", codecNotes{Items: []string{"normalized"}})
+	if err != nil {
+		t.Fatalf("checkpointArtifact(normalize): %v", err)
+	}
+	loader := attemptReuseLoader{
+		CheckpointLoader: NoopCheckpointLoader(),
+		laneID:           lane.resolved.ID,
+		merge:            MergeCheckpoint{Output: merge},
+		normalize:        NormalizeCheckpoint{Output: normalize},
+	}
+	debug := newCapturedDebugRecorder()
+	if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug, Checkpoint: loader}); err != nil {
+		t.Fatalf("Run() error = %v, want nil", err)
+	}
+	for _, name := range debug.names() {
+		if (strings.HasPrefix(name, "merge/notes/attempt-") || strings.HasPrefix(name, "normalize/notes/attempt-")) && name != "" {
+			t.Fatalf("checkpoint reuse synthesized module attempt artifact %q", name)
+		}
+	}
+	for _, name := range []string{"merge/notes/input.json", "merge/notes/output.json", "normalize/notes/input.json", "normalize/notes/output.json"} {
+		if !debug.has(name) {
+			t.Errorf("checkpoint reuse missing stage-level artifact %q", name)
+		}
+	}
+}
+
+func TestRunnerTreatsModuleAttemptDebugWriteFailureAsFrameworkError(t *testing.T) {
+	prepared := preparedAttemptDebugPipeline(t)
+	debug := newCapturedDebugRecorder()
+	debug.failPath = "merge/notes/attempt-01.json"
+	_, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug})
+	if err == nil || !strings.Contains(err.Error(), "write merge attempt debug artifact") || !strings.Contains(err.Error(), "debug recorder failure") {
+		t.Fatalf("Run() error = %v, want merge attempt debug failure", err)
+	}
+}

internal/framework/pipeline/runner_concurrent.go

@@ -28,6 +28,22 @@ type laneExtractState struct {
 	failed     bool
 }
 
+type finalizedExtractResults struct {
+	accepted   []erasedExtractArtifact
+	serialized []CheckpointArtifact
+	warnings   []contracts.Warning
+	rejected   []contracts.RejectedOutput
+	decision   CheckpointDecision
+}
+
+func loadExtract(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) {
+	return loader.Extract(laneID, moduleKey, deps)
+}
+
+func recordExtract(recorder CheckpointRecorder, laneID, moduleKey string, deps []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error {
+	return recorder.ExtractSucceeded(laneID, moduleKey, deps, outputs, rejected, warnings)
+}
+
 type extractJob struct {
 	lane  *laneExtractState
 	chunk source.Chunk
@@ -56,19 +72,6 @@ type orderedRunError struct {
 	err   error
 }
 
-type completedExtractLoader struct {
-	CheckpointLoader
-	laneID     string
-	checkpoint ExtractCheckpoint
-}
-
-func (l completedExtractLoader) Extract(laneID, _ string, _ []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) {
-	if laneID == l.laneID {
-		return l.checkpoint, CheckpointDecision{Reused: true, Reason: "coordinated extract result"}
-	}
-	return ExtractCheckpoint{}, CheckpointDecision{Reason: "extract result unavailable"}
-}
-
 func (r *Runner) runLanes(parent context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk) (RunOutput, error) {
 	output := RunOutput{Manifest: manifestFromPipeline(input)}
 	states := make([]*laneExtractState, len(input.Prepared.lanes))
@@ -85,6 +88,8 @@ func (r *Runner) runLanes(parent context.Context, input RunInput, checkpoints Ch
 			if err := checkpoints.ExtractRunning(prepared.resolved.ID, prepared.resolved.Extract.Module, state.deps); err != nil {
 				return output, fmt.Errorf("write extract checkpoint for lane %q: %w", prepared.resolved.ID, err)
 			}
+		} else if err := finalizeLaneExtract(checkpoints, state); err != nil {
+			return output, err
 		}
 		states[i] = state
 	}
@@ -265,7 +270,7 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source.
 		attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
 		extracted, callErr := typed.extract(attemptCtx, typed.extractor, contracts.TypedExtractionRequest{Source: doc, Chunk: &chunk, SourceInput: chunkInputMaterial(sourceInput, chunk), SessionID: sessionID, References: CloneReferenceSet(lane.ExtractReferences.ReferenceSet), LLMProfile: lane.Extract.LLMProfile, Metadata: cloneMetadata(input.Metadata)})
 		if callErr != nil {
-			_ = writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Error: callErr.Error()}, llmScope))
+			_ = writeDebugAttempt(input.Debug, attemptPath, debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Error: callErr.Error()}, llmScope)
 			return false, nil, fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, lane.Extract.Module, callErr)
 		}
 		artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: chunk.ID, ChunkIndex: chunk.Index, ChunkRef: chunk.Ref, Value: extracted.Value}
@@ -280,7 +285,7 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source.
 		stored.ChunkID, stored.ChunkIndex, stored.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef
 		accepted, serialized = artifact, stored
 		acceptedWarnings = append(cloneWarnings(extracted.Warnings), warnings...)
-		if debugErr := writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Payload: map[string]any{"output": debugCheckpointArtifact(stored), "warnings": debugWarningEnvelopes(acceptedWarnings)}}, llmScope)); debugErr != nil {
+		if debugErr := writeDebugAttempt(input.Debug, attemptPath, debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Payload: map[string]any{"output": debugCheckpointArtifact(stored), "warnings": debugWarningEnvelopes(acceptedWarnings)}}, llmScope); debugErr != nil {
 			return false, nil, debugErr
 		}
 		return true, nil, nil
@@ -325,10 +330,26 @@ func finalizeLaneExtract(checkpoints CheckpointRecorder, state *laneExtractState
 func (r *Runner) continueLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, state *laneExtractState) (RunOutput, error) {
 	lane := state.prepared.resolved
 	local := RunOutput{Manifest: manifestFromPipeline(input)}
-	checkpoint := ExtractCheckpoint{Outputs: state.serialized, Rejected: state.rejected, Warnings: state.warnings}
+	results := finalizedExtractResults{
-	coordinatedLoader := completedExtractLoader{CheckpointLoader: loader, laneID: lane.ID, checkpoint: checkpoint}
+		accepted:   state.values,
-	input.extractDecision = &state.decision
+		serialized: state.serialized,
-	err := r.runTypedLane(ctx, input, checkpoints, coordinatedLoader, doc, sourceInput, sessionID, chunks, state.prepared, &local)
+		warnings:   state.warnings,
+		rejected:   state.rejected,
+		decision:   state.decision,
+	}
+	local.Warnings = append(local.Warnings, cloneWarnings(results.warnings)...)
+	local.Rejected = append(local.Rejected, cloneRejectedOutputs(results.rejected)...)
+	recordCheckpointEvent(&local, loader, string(StageExtract), lane.ID, lane.Extract.Module, results.decision)
+	if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": results.decision.Reused, "decision": results.decision, "source": debugSourceDocumentEnvelope(doc), "chunks": debugSourceChunkEnvelopes(chunks), "options": redactSensitiveMap(lane.Extract.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
+		return local, &laneRunError{stage: StageExtract, err: err}
+	}
+	if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": results.decision.Reused, "outputs": debugCheckpointArtifacts(results.serialized), "rejected": debugRejectedOutputEnvelopes(results.rejected), "warnings": debugWarningEnvelopes(results.warnings)}}); err != nil {
+		return local, &laneRunError{stage: StageExtract, err: err}
+	}
+	if len(results.accepted) == 0 {
+		return local, nil
+	}
+	err := r.continueTypedLane(ctx, input, checkpoints, loader, doc, sourceInput, sessionID, state.prepared, results, &local)
 	return local, err
 }
 

internal/framework/pipeline/runner_extract_handoff_test.go

@@ -0,0 +1,180 @@
+package pipeline
+
+import (
+	"context"
+	"reflect"
+	"strings"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+)
+
+type extractCaptureRecorder struct {
+	CheckpointRecorder
+	checkpoint ExtractCheckpoint
+}
+
+func (r *extractCaptureRecorder) ExtractSucceeded(_ string, _ string, _ []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error {
+	r.checkpoint = ExtractCheckpoint{
+		Outputs:  cloneCheckpointArtifacts(outputs),
+		Rejected: cloneRejectedOutputs(rejected),
+		Warnings: cloneWarnings(warnings),
+	}
+	return nil
+}
+
+type extractResultLoader struct {
+	CheckpointLoader
+	checkpoint ExtractCheckpoint
+	decision   CheckpointDecision
+}
+
+func (l *extractResultLoader) Enabled() bool { return true }
+
+func (l *extractResultLoader) Extract(string, string, []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) {
+	return ExtractCheckpoint{
+		Outputs:  cloneCheckpointArtifacts(l.checkpoint.Outputs),
+		Rejected: cloneRejectedOutputs(l.checkpoint.Rejected),
+		Warnings: cloneWarnings(l.checkpoint.Warnings),
+	}, l.decision
+}
+
+func cloneCheckpointArtifacts(values []CheckpointArtifact) []CheckpointArtifact {
+	if len(values) == 0 {
+		return nil
+	}
+	cloned := make([]CheckpointArtifact, len(values))
+	for i := range values {
+		cloned[i] = cloneCheckpointArtifact(values[i])
+	}
+	return cloned
+}
+
+func TestRunnerContinuesFromFreshAndReusedExtractResults(t *testing.T) {
+	prepared := preparedAttemptDebugPipeline(t)
+	extractCalls := 0
+	installExtractOperation(prepared, 0, func(_ context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
+		extractCalls++
+		return erasedTypedResult{
+			Value:    typedValueForLane(0, request.Chunk.Index),
+			Warnings: []contracts.Warning{{Scope: "extract", ReasonCode: "observed", Message: "accepted extract"}},
+		}, nil
+	})
+
+	freshDebug := newCapturedDebugRecorder()
+	recorder := &extractCaptureRecorder{CheckpointRecorder: NoopCheckpointRecorder()}
+	freshLoader := &extractResultLoader{
+		CheckpointLoader: NoopCheckpointLoader(),
+		decision:         CheckpointDecision{Reason: "extract checkpoint not found"},
+	}
+	fresh, err := New().Run(context.Background(), RunInput{
+		Prepared:    prepared,
+		RawInput:    []byte("input"),
+		Checkpoints: recorder,
+		Checkpoint:  freshLoader,
+		Debug:       freshDebug,
+	})
+	if err != nil {
+		t.Fatalf("fresh Run() error = %v, want nil", err)
+	}
+	if extractCalls != 1 {
+		t.Fatalf("fresh extract calls = %d, want 1", extractCalls)
+	}
+	assertExtractDecision(t, fresh.CheckpointEvents, "executed", "extract checkpoint not found")
+	assertExtractDebugPaths(t, freshDebug, true)
+
+	reusedDebug := newCapturedDebugRecorder()
+	reusedLoader := &extractResultLoader{
+		CheckpointLoader: NoopCheckpointLoader(),
+		checkpoint:       recorder.checkpoint,
+		decision:         CheckpointDecision{Reused: true, Reason: "extract checkpoint matched"},
+	}
+	reused, err := New().Run(context.Background(), RunInput{
+		Prepared:   prepared,
+		RawInput:   []byte("input"),
+		Checkpoint: reusedLoader,
+		Debug:      reusedDebug,
+	})
+	if err != nil {
+		t.Fatalf("reused Run() error = %v, want nil", err)
+	}
+	if extractCalls != 1 {
+		t.Fatalf("extract calls after reuse = %d, want 1", extractCalls)
+	}
+	assertExtractDecision(t, reused.CheckpointEvents, "reused", "extract checkpoint matched")
+	assertExtractDebugPaths(t, reusedDebug, false)
+	if !reflect.DeepEqual(reused.NormalizeOutputs, fresh.NormalizeOutputs) {
+		t.Fatalf("reused normalize outputs = %#v, want fresh outputs %#v", reused.NormalizeOutputs, fresh.NormalizeOutputs)
+	}
+	if !reflect.DeepEqual(reused.Warnings, fresh.Warnings) {
+		t.Fatalf("reused warnings = %#v, want fresh warnings %#v", reused.Warnings, fresh.Warnings)
+	}
+}
+
+func TestRunnerPromotesOnlyAcceptedExtractRetryWarnings(t *testing.T) {
+	prepared := preparedAttemptDebugPipeline(t)
+	prepared.lanes[0].resolved.Extract.Retries = 1
+	attempts := 0
+	installExtractOperation(prepared, 0, func(_ context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
+		attempts++
+		scope := "discarded"
+		if attempts == 2 {
+			scope = "accepted"
+		}
+		return erasedTypedResult{
+			Value:    typedValueForLane(0, request.Chunk.Index),
+			Warnings: []contracts.Warning{{Scope: scope, ReasonCode: "observed", Message: scope}},
+		}, nil
+	})
+	validatorCalls := 0
+	prepared.lanes[0].extractValidators.validators[0].typedValidate = func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {
+		validatorCalls++
+		return contracts.ValidationResult{Approved: validatorCalls == 2, ReasonCode: "retry", Message: "retry extract"}, nil
+	}
+	debug := newCapturedDebugRecorder()
+	output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug})
+	if err != nil {
+		t.Fatalf("Run() error = %v, want nil", err)
+	}
+	if attempts != 2 {
+		t.Fatalf("extract attempts = %d, want 2", attempts)
+	}
+	if len(output.Warnings) != 1 || output.Warnings[0].Scope != "accepted" {
+		t.Fatalf("promoted warnings = %#v, want accepted attempt only", output.Warnings)
+	}
+	name := "extract/notes/chunk-000001/attempt-02.json"
+	if !debug.has(name) {
+		t.Fatalf("debug artifact %q is missing; names = %#v", name, debug.names())
+	}
+}
+
+func assertExtractDecision(t *testing.T, events []CheckpointEvent, action string, reason string) {
+	t.Helper()
+	for _, event := range events {
+		if event.Stage == string(StageExtract) {
+			if event.Action != action || event.Reason != reason {
+				t.Fatalf("extract checkpoint event = %#v, want action %q reason %q", event, action, reason)
+			}
+			return
+		}
+	}
+	t.Fatalf("extract checkpoint event missing from %#v", events)
+}
+
+func assertExtractDebugPaths(t *testing.T, debug *capturedDebugRecorder, wantAttempt bool) {
+	t.Helper()
+	for _, name := range []string{"extract/notes/input.json", "extract/notes/output.json"} {
+		if !debug.has(name) {
+			t.Fatalf("debug artifact %q is missing; names = %#v", name, debug.names())
+		}
+	}
+	attemptPath := "extract/notes/chunk-000001/attempt-01.json"
+	if debug.has(attemptPath) != wantAttempt {
+		t.Fatalf("attempt debug path present = %t, want %t; names = %#v", debug.has(attemptPath), wantAttempt, debug.names())
+	}
+	input := debug.envelope(t, "extract/notes/input.json")
+	wantReuse := !wantAttempt
+	if !strings.Contains(string(debug.json["extract/notes/input.json"]), `"reused":`+map[bool]string{true: "true", false: "false"}[wantReuse]) {
+		t.Fatalf("extract input payload = %#v, want reused %t", input.Payload, wantReuse)
+	}
+}

internal/framework/pipeline/runner_typed.go

@@ -6,25 +6,18 @@ import (
 	"encoding/hex"
 	"fmt"
 	"path"
-	"sort"
 	"time"
 
 	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
 )
 
-func loadExtract(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) {
-	return loader.Extract(laneID, moduleKey, deps)
-}
 func loadMerge(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) {
 	return loader.Merge(laneID, moduleKey, deps)
 }
 func loadNormalize(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) {
 	return loader.Normalize(laneID, moduleKey, deps)
 }
-func recordExtract(recorder CheckpointRecorder, laneID, moduleKey string, deps []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error {
-	return recorder.ExtractSucceeded(laneID, moduleKey, deps, outputs, rejected, warnings)
-}
 func recordMerge(recorder CheckpointRecorder, laneID, moduleKey string, deps []CheckpointFingerprint, output CheckpointArtifact, warnings []contracts.Warning) error {
 	return recorder.MergeSucceeded(laneID, moduleKey, deps, output, warnings)
 }
@@ -126,8 +119,8 @@ type laneRunError struct {
 func (e *laneRunError) Error() string { return e.err.Error() }
 func (e *laneRunError) Unwrap() error { return e.err }
 
-func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, prepared preparedLaneExecutor, output *RunOutput) (err error) {
+func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, prepared preparedLaneExecutor, extracts finalizedExtractResults, output *RunOutput) (err error) {
-	activeStage := StageExtract
+	activeStage := StageMerge
 	defer func() {
 		if err != nil {
 			err = &laneRunError{stage: activeStage, err: err}
@@ -139,116 +132,11 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C
 	}
 	setTypedLaneManifestMetadata(output, lane.ID, typed.extractor, typed.merger, typed.normalizer)
 
-	values := make([]erasedExtractArtifact, 0, len(chunks))
+	mergeInputs := make([]contracts.ExtractArtifact[any], len(extracts.accepted))
-	serializedExtracts := make([]CheckpointArtifact, 0, len(chunks))
+	for i, value := range extracts.accepted {
-	extractWarnings := []contracts.Warning{}
-	rejectedStart := len(output.Rejected)
-	chunksDigest, err := joinedChunkDigest(chunks)
-	if err != nil {
-		return fmt.Errorf("digest chunks for lane %q: %w", lane.ID, err)
-	}
-	extractDeps := digestFingerprints("chunks", chunksDigest)
-	cp, decision := loadExtract(loader, lane.ID, lane.Extract.Module, extractDeps)
-	if decision.Reused {
-		for _, stored := range cp.Outputs {
-			if _, decodeErr := decodeCheckpointArtifact(typed.codec, stored); decodeErr != nil {
-				decision = CheckpointDecision{Reason: "extract artifact checkpoint codec is incompatible: " + decodeErr.Error()}
-				break
-			}
-		}
-	}
-	reportedDecision := decision
-	if input.extractDecision != nil {
-		reportedDecision = *input.extractDecision
-	}
-	recordCheckpointEvent(output, loader, string(StageExtract), lane.ID, lane.Extract.Module, reportedDecision)
-	if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": reportedDecision.Reused, "decision": reportedDecision, "source": debugSourceDocumentEnvelope(doc), "chunks": debugSourceChunkEnvelopes(chunks), "options": redactSensitiveMap(lane.Extract.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
-		return err
-	}
-	if decision.Reused {
-		for _, stored := range cp.Outputs {
-			value, decodeErr := decodeCheckpointArtifact(typed.codec, stored)
-			if decodeErr != nil {
-				return fmt.Errorf("decode extract checkpoint for lane %q: %w", lane.ID, decodeErr)
-			}
-			stored = hydrateCheckpointArtifact(typed.codec, stored, value)
-			artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: stored.ChunkID, ChunkIndex: stored.ChunkIndex, ChunkRef: stored.ChunkRef, Value: value}
-			if stored.ChunkIndex >= 0 && stored.ChunkIndex < len(chunks) && artifact.ChunkRef == (source.SourceRef{}) {
-				artifact.ChunkRef = chunks[stored.ChunkIndex].Ref
-			}
-			values = append(values, artifact)
-			serializedExtracts = append(serializedExtracts, cloneCheckpointArtifact(stored))
-		}
-		extractWarnings = cloneWarnings(cp.Warnings)
-		output.Warnings = append(output.Warnings, extractWarnings...)
-		output.Rejected = append(output.Rejected, cloneRejectedOutputs(cp.Rejected)...)
-	} else {
-		if err := checkpoints.ExtractRunning(lane.ID, lane.Extract.Module, extractDeps); err != nil {
-			return fmt.Errorf("write extract checkpoint for lane %q: %w", lane.ID, err)
-		}
-		for i := range chunks {
-			chunk := chunks[i]
-			var accepted erasedExtractArtifact
-			var serializedAccepted CheckpointArtifact
-			var acceptedWarnings []contracts.Warning
-			ok, rejection, runErr := runWithRetry(ctx, lane.Extract.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
-				started := time.Now().UTC()
-				attemptPath := path.Join("extract", debugPathComponent(lane.ID), fmt.Sprintf("chunk-%06d", chunk.Index+1), fmt.Sprintf("attempt-%02d", attempt))
-				attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
-				result, callErr := typed.extract(attemptCtx, typed.extractor, contracts.TypedExtractionRequest{Source: doc, Chunk: &chunk, SourceInput: chunkInputMaterial(sourceInput, chunk), SessionID: sessionID, References: CloneReferenceSet(lane.ExtractReferences.ReferenceSet), LLMProfile: lane.Extract.LLMProfile, Metadata: cloneMetadata(input.Metadata)})
-				if callErr != nil {
-					_ = writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Error: callErr.Error()}, llmScope))
-					return false, nil, fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, lane.Extract.Module, callErr)
-				}
-				artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: chunk.ID, ChunkIndex: chunk.Index, ChunkRef: chunk.Ref, Value: result.Value}
-				warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageExtract, laneID: lane.ID, moduleKey: lane.Extract.Module, source: doc, sourceID: doc.ID, sourceInput: chunkInputMaterial(sourceInput, chunk), sessionID: sessionID, references: lane.ExtractReferences.ReferenceSet, metadata: input.Metadata, chunk: &chunk, ref: chunk.Ref, value: result.Value}, prepared.extractValidators, attempt, input.Debug)
-				if validateErr != nil || rejected != nil {
-					return false, rejected, validateErr
-				}
-				stored, encodeErr := checkpointArtifact(typed.codec, artifact.LaneID, artifact.ExtractorKey, artifact.SourceID, artifact.Value)
-				if encodeErr != nil {
-					return false, nil, encodeErr
-				}
-				stored.ChunkID, stored.ChunkIndex, stored.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef
-				accepted, serializedAccepted = artifact, stored
-				acceptedWarnings = append(cloneWarnings(result.Warnings), warnings...)
-				if debugErr := writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Payload: map[string]any{"output": debugCheckpointArtifact(stored), "warnings": debugWarningEnvelopes(acceptedWarnings)}}, llmScope)); debugErr != nil {
-					return false, nil, debugErr
-				}
-				return true, nil, nil
-			})
-			if runErr != nil {
-				_ = checkpoints.ExtractFailed(lane.ID, lane.Extract.Module, extractDeps, runErr)
-				return runErr
-			}
-			if !ok {
-				output.Rejected = append(output.Rejected, *rejection)
-				continue
-			}
-			values = append(values, accepted)
-			serializedExtracts = append(serializedExtracts, serializedAccepted)
-			extractWarnings = append(extractWarnings, acceptedWarnings...)
-			output.Warnings = append(output.Warnings, acceptedWarnings...)
-		}
-		if err := recordExtract(checkpoints, lane.ID, lane.Extract.Module, extractDeps, serializedExtracts, cloneRejectedOutputs(output.Rejected[rejectedStart:]), extractWarnings); err != nil {
-			return fmt.Errorf("write extract checkpoint for lane %q: %w", lane.ID, err)
-		}
-	}
-	sort.SliceStable(values, func(i, j int) bool { return values[i].ChunkIndex < values[j].ChunkIndex })
-	sort.SliceStable(serializedExtracts, func(i, j int) bool { return serializedExtracts[i].ChunkIndex < serializedExtracts[j].ChunkIndex })
-	if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": reportedDecision.Reused, "outputs": debugCheckpointArtifacts(serializedExtracts), "rejected": debugRejectedOutputEnvelopes(output.Rejected[rejectedStart:]), "warnings": debugWarningEnvelopes(extractWarnings)}}); err != nil {
-		return err
-	}
-	if len(values) == 0 {
-		return nil
-	}
-	activeStage = StageMerge
-
-	mergeInputs := make([]contracts.ExtractArtifact[any], len(values))
-	for i, value := range values {
 		mergeInputs[i] = contracts.ExtractArtifact[any]{LaneID: value.LaneID, ExtractorKey: value.ExtractorKey, SourceID: value.SourceID, ChunkID: value.ChunkID, ChunkIndex: value.ChunkIndex, ChunkRef: value.ChunkRef, Value: value.Value}
 	}
-	mergeDeps := artifactCheckpointDigests(serializedExtracts)
+	mergeDeps := artifactCheckpointDigests(extracts.serialized)
 	mergeCP, mergeDecision := loadMerge(loader, lane.ID, lane.Merge.Module, mergeDeps)
 	if mergeDecision.Reused {
 		if _, decodeErr := decodeCheckpointArtifact(typed.codec, mergeCP.Output); decodeErr != nil {
@@ -256,7 +144,7 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C
 		}
 	}
 	recordCheckpointEvent(output, loader, string(StageMerge), lane.ID, lane.Merge.Module, mergeDecision)
-	if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "decision": mergeDecision, "source": debugSourceDocumentEnvelope(doc), "extract_outputs": debugCheckpointArtifacts(serializedExtracts), "options": redactSensitiveMap(lane.Merge.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
+	if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "decision": mergeDecision, "source": debugSourceDocumentEnvelope(doc), "extract_outputs": debugCheckpointArtifacts(extracts.serialized), "options": redactSensitiveMap(lane.Merge.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
 		return err
 	}
 	var merged erasedMergeArtifact
@@ -276,21 +164,48 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C
 			return err
 		}
 		ok, rejection, runErr := runWithRetry(ctx, lane.Merge.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
-			result, callErr := typed.merge(ctx, typed.merger, contracts.TypedMergeRequest[any]{Source: doc, LaneID: lane.ID, ExtractOutputs: mergeInputs, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(lane.MergeReferences.ReferenceSet), LLMProfile: lane.Merge.LLMProfile, Metadata: cloneMetadata(input.Metadata)})
+			started := time.Now().UTC()
+			attemptPath := path.Join("merge", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
+			attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
+			attemptEnvelope := func(payload map[string]any, attemptErr error) error {
+				envelope := debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: attempt, StartedAt: started, Payload: payload}
+				if attemptErr != nil {
+					envelope.Error = attemptErr.Error()
+				}
+				return writeDebugAttempt(input.Debug, attemptPath, envelope, llmScope)
+			}
+			result, callErr := typed.merge(attemptCtx, typed.merger, contracts.TypedMergeRequest[any]{Source: doc, LaneID: lane.ID, ExtractOutputs: mergeInputs, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(lane.MergeReferences.ReferenceSet), LLMProfile: lane.Merge.LLMProfile, Metadata: cloneMetadata(input.Metadata)})
 			if callErr != nil {
-				return false, nil, fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, lane.Merge.Module, callErr)
+				attemptErr := fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, lane.Merge.Module, callErr)
+				if debugErr := attemptEnvelope(nil, attemptErr); debugErr != nil {
+					return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr)
+				}
+				return false, nil, attemptErr
 			}
 			candidate := erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: result.Value}
-			warnings, rejected, validateErr := r.validateTypedArtifact(ctx, typed.codec, typedValidationTarget{stage: StageMerge, laneID: lane.ID, moduleKey: lane.Merge.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.MergeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value}, prepared.mergeValidators, attempt, input.Debug)
+			stored, encodeErr := checkpointArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value)
+			attemptWarnings := cloneWarnings(result.Warnings)
+			if encodeErr != nil {
+				attemptErr := fmt.Errorf("serialize merge candidate for lane %q: %w", lane.ID, encodeErr)
+				if debugErr := attemptEnvelope(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr); debugErr != nil {
+					return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr)
+				}
+				return false, nil, attemptErr
+			}
+			warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageMerge, laneID: lane.ID, moduleKey: lane.Merge.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.MergeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value}, prepared.mergeValidators, attempt, input.Debug)
+			attemptWarnings = append(attemptWarnings, warnings...)
+			payload := map[string]any{"output": debugCheckpointArtifact(stored), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)}
 			if validateErr != nil || rejected != nil {
+				if debugErr := attemptEnvelope(payload, validateErr); debugErr != nil {
+					return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr)
+				}
 				return false, rejected, validateErr
 			}
-			stored, encodeErr := checkpointArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value)
+			if debugErr := attemptEnvelope(payload, nil); debugErr != nil {
-			if encodeErr != nil {
+				return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr)
-				return false, nil, encodeErr
 			}
 			merged, serializedMerge = candidate, stored
-			mergeWarnings = append(cloneWarnings(result.Warnings), warnings...)
+			mergeWarnings = attemptWarnings
 			return true, nil, nil
 		})
 		if runErr != nil {
@@ -339,20 +254,47 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C
 			return err
 		}
 		ok, rejection, runErr := runWithRetry(ctx, lane.Normalize.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
-			result, callErr := typed.normalize(ctx, typed.normalizer, contracts.TypedNormalizeRequest[any]{Source: doc, LaneID: lane.ID, MergeOutput: contracts.MergeArtifact[any]{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: merged.Value}, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(lane.NormalizeReferences.ReferenceSet), LLMProfile: lane.Normalize.LLMProfile, Metadata: cloneMetadata(input.Metadata)})
+			started := time.Now().UTC()
-			if callErr != nil {
+			attemptPath := path.Join("normalize", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
-				return false, nil, fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, lane.Normalize.Module, callErr)
+			attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
+			attemptEnvelope := func(payload map[string]any, attemptErr error) error {
+				envelope := debugTimedEnvelope{Stage: string(StageNormalize), LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: attempt, StartedAt: started, Payload: payload}
+				if attemptErr != nil {
+					envelope.Error = attemptErr.Error()
 				}
-			warnings, rejected, validateErr := r.validateTypedArtifact(ctx, typed.codec, typedValidationTarget{stage: StageNormalize, laneID: lane.ID, moduleKey: lane.Normalize.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.NormalizeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value}, prepared.normalizeValidators, attempt, input.Debug)
+				return writeDebugAttempt(input.Debug, attemptPath, envelope, llmScope)
-			if validateErr != nil || rejected != nil {
+			}
-				return false, rejected, validateErr
+			result, callErr := typed.normalize(attemptCtx, typed.normalizer, contracts.TypedNormalizeRequest[any]{Source: doc, LaneID: lane.ID, MergeOutput: contracts.MergeArtifact[any]{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: merged.Value}, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(lane.NormalizeReferences.ReferenceSet), LLMProfile: lane.Normalize.LLMProfile, Metadata: cloneMetadata(input.Metadata)})
+			if callErr != nil {
+				attemptErr := fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, lane.Normalize.Module, callErr)
+				if debugErr := attemptEnvelope(nil, attemptErr); debugErr != nil {
+					return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr)
+				}
+				return false, nil, attemptErr
 			}
 			stored, encodeErr := checkpointArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value)
+			attemptWarnings := cloneWarnings(result.Warnings)
 			if encodeErr != nil {
-				return false, nil, encodeErr
+				attemptErr := fmt.Errorf("serialize normalize candidate for lane %q: %w", lane.ID, encodeErr)
+				if debugErr := attemptEnvelope(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr); debugErr != nil {
+					return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr)
+				}
+				return false, nil, attemptErr
+			}
+			warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageNormalize, laneID: lane.ID, moduleKey: lane.Normalize.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.NormalizeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value}, prepared.normalizeValidators, attempt, input.Debug)
+			attemptWarnings = append(attemptWarnings, warnings...)
+			payload := map[string]any{"output": debugCheckpointArtifact(stored), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)}
+			if validateErr != nil || rejected != nil {
+				if debugErr := attemptEnvelope(payload, validateErr); debugErr != nil {
+					return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr)
+				}
+				return false, rejected, validateErr
+			}
+			if debugErr := attemptEnvelope(payload, nil); debugErr != nil {
+				return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr)
 			}
 			serializedNormalize = stored
-			normalizeWarnings = append(cloneWarnings(result.Warnings), warnings...)
+			normalizeWarnings = attemptWarnings
 			return true, nil, nil
 		})
 		if runErr != nil {
@@ -410,7 +352,7 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE
 		var err error
 		started := time.Now().UTC()
 		attemptPath := path.Join("validate", debugPathComponent(string(target.stage)), debugPathComponent(target.laneID), debugPathComponent(target.moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d", index+1, debugPathComponent(binding.Module), attempt))
-		validatorCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
+		validatorCtx, llmScope := withIsolatedDebugLLMScope(ctx, attemptPath)
 		switch item.resolved.Target {
 		case ValidatorTargetTyped:
 			target.llmProfile = binding.LLMProfile
@@ -430,7 +372,7 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE
 		if err != nil {
 			debugCall.Error = err.Error()
 		}
-		if debugErr := writeDebugTimed(debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope)); debugErr != nil {
+		if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope); debugErr != nil {
 			return nil, nil, fmt.Errorf("write validation debug artifact: %w", debugErr)
 		}
 		if err != nil {

internal/framework/pipeline/typed_resolution_test.go

@@ -2,6 +2,7 @@ package pipeline
 
 import (
 	"context"
+	"fmt"
 	"reflect"
 	"strings"
 	"testing"
@@ -267,6 +268,101 @@ func TestConstructorRegistrationsRejectUnconfiguredOptions(t *testing.T) {
 	}
 }
 
+func TestTypedRegistrySpecLookupUsesArtifactKindAndStableCatalogOrder(t *testing.T) {
+	type registration struct {
+		kind contracts.ArtifactKind
+		slot string
+	}
+	orders := [][]registration{
+		{{kind: "test/score", slot: "score_notes"}, {kind: "test/notes", slot: "note_notes"}},
+		{{kind: "test/notes", slot: "note_notes"}, {kind: "test/score", slot: "score_notes"}},
+	}
+
+	for orderIndex, order := range orders {
+		t.Run(fmt.Sprintf("registration order %d", orderIndex+1), func(t *testing.T) {
+			mergers := NewMergerRegistry()
+			normalizers := NewNormalizerRegistry()
+			for _, item := range order {
+				base := ModuleSpec{Key: "typed/shared", ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.slot}}}
+				mergeSpec := base
+				mergeSpec.Stage = StageMerge
+				normalizeSpec := base
+				normalizeSpec.Stage = StageNormalize
+				switch item.kind {
+				case "test/notes":
+					if err := RegisterMerger(mergers, mergeSpec, func() (contracts.Merger[codecNotes], error) {
+						return typedTestMerger[codecNotes]{key: "typed/shared"}, nil
+					}); err != nil {
+						t.Fatalf("RegisterMerger(notes): %v", err)
+					}
+					if err := RegisterNormalizer(normalizers, normalizeSpec, func() (contracts.Normalizer[codecNotes], error) {
+						return typedTestNormalizer[codecNotes]{key: "typed/shared"}, nil
+					}); err != nil {
+						t.Fatalf("RegisterNormalizer(notes): %v", err)
+					}
+				case "test/score":
+					if err := RegisterMerger(mergers, mergeSpec, func() (contracts.Merger[codecScore], error) {
+						return typedTestMerger[codecScore]{key: "typed/shared"}, nil
+					}); err != nil {
+						t.Fatalf("RegisterMerger(score): %v", err)
+					}
+					if err := RegisterNormalizer(normalizers, normalizeSpec, func() (contracts.Normalizer[codecScore], error) {
+						return typedTestNormalizer[codecScore]{key: "typed/shared"}, nil
+					}); err != nil {
+						t.Fatalf("RegisterNormalizer(score): %v", err)
+					}
+				}
+			}
+
+			for name, lookup := range map[string]func(contracts.ArtifactKind) (ModuleSpec, bool){
+				"merger": func(kind contracts.ArtifactKind) (ModuleSpec, bool) {
+					return mergers.SpecForArtifact(" typed/shared ", kind)
+				},
+				"normalizer": func(kind contracts.ArtifactKind) (ModuleSpec, bool) {
+					return normalizers.SpecForArtifact(" typed/shared ", kind)
+				},
+			} {
+				t.Run(name, func(t *testing.T) {
+					notes, ok := lookup(" test/notes ")
+					if !ok || len(notes.ReferenceSlots) != 1 || notes.ReferenceSlots[0].Name != "note_notes" {
+						t.Fatalf("notes spec = %#v, ok = %v", notes, ok)
+					}
+					score, ok := lookup("test/score")
+					if !ok || len(score.ReferenceSlots) != 1 || score.ReferenceSlots[0].Name != "score_notes" {
+						t.Fatalf("score spec = %#v, ok = %v", score, ok)
+					}
+					notes.ReferenceSlots[0].Name = "mutated"
+					again, _ := lookup("test/notes")
+					if again.ReferenceSlots[0].Name != "note_notes" {
+						t.Fatalf("registry spec changed after caller mutation: %#v", again)
+					}
+				})
+			}
+
+			for name, lookup := range map[string]func() (ModuleSpec, bool){
+				"merger":     func() (ModuleSpec, bool) { return mergers.Spec("typed/shared") },
+				"normalizer": func() (ModuleSpec, bool) { return normalizers.Spec("typed/shared") },
+			} {
+				t.Run(name+" catalog", func(t *testing.T) {
+					for attempt := 0; attempt < 10; attempt++ {
+						spec, ok := lookup()
+						if !ok || spec.ArtifactKind != "test/notes" || spec.ReferenceSlots[0].Name != "note_notes" {
+							t.Fatalf("Spec() = %#v, ok = %v; want lexicographically first artifact kind", spec, ok)
+						}
+					}
+				})
+			}
+
+			if got := mergers.RegisteredArtifactKinds("typed/shared"); !reflect.DeepEqual(got, []contracts.ArtifactKind{"test/notes", "test/score"}) {
+				t.Fatalf("merger kinds = %#v", got)
+			}
+			if got := normalizers.RegisteredArtifactKinds("typed/shared"); !reflect.DeepEqual(got, []contracts.ArtifactKind{"test/notes", "test/score"}) {
+				t.Fatalf("normalizer kinds = %#v", got)
+			}
+		})
+	}
+}
+
 func TestResolvedPipelineDigestIncludesArtifactSchemaIdentity(t *testing.T) {
 	baseOptions := completeTypedCatalogOptions()
 	base, err := ResolvePipeline(typedResolutionProfile(), ResolveOptions{}, typedResolutionCatalog(t, baseOptions))

internal/modules/import_boundaries_test.go

@@ -41,10 +41,10 @@ func TestImportBoundaryFixtureIsRejected(t *testing.T) {
 	fixture := filepath.Join(repositoryRoot, "internal", "modules", "generic", "testdata", "importboundaries", "imports_dnd.go")
 	err := checkImportBoundaries(repositoryRoot, fixture)
 	if err == nil {
-		t.Fatal("fixture import was accepted, want generic-to-D&D violation")
+		t.Fatal("fixture import was accepted, want generic-to-concrete violation")
 	}
-	if !strings.Contains(err.Error(), "generic packages must not import D&D packages") {
+	if !strings.Contains(err.Error(), "generic family must not import concrete family") {
-		t.Fatalf("fixture error = %q, want generic-to-D&D violation", err)
+		t.Fatalf("fixture error = %q, want generic-to-concrete violation", err)
 	}
 }
 
@@ -52,58 +52,134 @@ func TestImportBoundaryRules(t *testing.T) {
 	tests := []struct {
 		name          string
 		filename      string
+		sourcePackage string
 		importPath    string
 		wantError     bool
 	}{
 		{
-			name:       "D&D implementation cannot import Seriatim",
+			name:          "concrete implementation cannot import peer concrete family",
 			filename:      "internal/modules/dnd/extract/example/extractor.go",
+			sourcePackage: "example",
 			importPath:    moduleImportPrefix + "seriatim/input/transcript",
 			wantError:     true,
 		},
 		{
-			name:       "Seriatim implementation cannot import D&D",
+			name:          "future concrete family cannot import current concrete family",
+			filename:      "internal/modules/almanac/extract/example/extractor.go",
+			sourcePackage: "example",
+			importPath:    moduleImportPrefix + "dnd/shared",
+			wantError:     true,
+		},
+		{
+			name:          "current concrete family cannot import future concrete family",
 			filename:      "internal/modules/seriatim/input/example/adapter.go",
-			importPath: moduleImportPrefix + "dnd/shared",
+			sourcePackage: "example",
+			importPath:    moduleImportPrefix + "almanac/shared",
 			wantError:     true,
 		},
 		{
-			name:       "generic implementation cannot import D&D",
+			name:          "generic implementation cannot import current concrete family",
 			filename:      "internal/modules/generic/merge/example/merger.go",
+			sourcePackage: "example",
 			importPath:    moduleImportPrefix + "dnd/shared",
 			wantError:     true,
 		},
 		{
-			name:       "domain root cannot import child",
+			name:          "generic implementation cannot import future concrete family",
-			filename:   "internal/modules/dnd/types.go",
+			filename:      "internal/modules/generic/merge/example/merger.go",
-			importPath: moduleImportPrefix + "dnd/extract/spells",
+			sourcePackage: "example",
+			importPath:    moduleImportPrefix + "almanac/shared",
 			wantError:     true,
 		},
 		{
-			name:       "D&D implementation may import generic implementation",
+			name:          "concrete implementation cannot import generic implementation",
 			filename:      "internal/modules/dnd/extract/example/extractor.go",
+			sourcePackage: "example",
+			importPath:    moduleImportPrefix + "generic/normalize/noop",
+			wantError:     true,
+		},
+		{
+			name:          "concrete white-box test follows production rules",
+			filename:      "internal/modules/dnd/extract/example/extractor_test.go",
+			sourcePackage: "example",
+			importPath:    moduleImportPrefix + "generic/normalize/noop",
+			wantError:     true,
+		},
+		{
+			name:          "concrete registrar may compose generic implementation",
+			filename:      "internal/modules/almanac/register/register.go",
+			sourcePackage: "register",
 			importPath:    moduleImportPrefix + "generic/normalize/noop",
 		},
 		{
-			name:       "domain registrar may compose child packages",
+			name:          "family root cannot import child implementation",
-			filename:   "internal/modules/dnd/register/register.go",
+			filename:      "internal/modules/almanac/types.go",
-			importPath: moduleImportPrefix + "dnd/extract/spells",
+			sourcePackage: "almanac",
+			importPath:    moduleImportPrefix + "almanac/extract/events",
+			wantError:     true,
 		},
 		{
-			name:       "CLI may compose registrars",
+			name:          "family root cannot import registrar",
+			filename:      "internal/modules/almanac/types.go",
+			sourcePackage: "almanac",
+			importPath:    moduleImportPrefix + "almanac/register",
+			wantError:     true,
+		},
+		{
+			name:          "child may import family root",
+			filename:      "internal/modules/almanac/extract/events/extractor.go",
+			sourcePackage: "events",
+			importPath:    moduleImportPrefix + "almanac",
+		},
+		{
+			name:          "child may import same-family sibling",
+			filename:      "internal/modules/almanac/extract/events/extractor.go",
+			sourcePackage: "events",
+			importPath:    moduleImportPrefix + "almanac/shared",
+		},
+		{
+			name:          "concrete registrar may compose own child",
+			filename:      "internal/modules/almanac/register/register.go",
+			sourcePackage: "register",
+			importPath:    moduleImportPrefix + "almanac/extract/events",
+		},
+		{
+			name:          "generic registrar may compose generic child",
+			filename:      "internal/modules/generic/register/register.go",
+			sourcePackage: "register",
+			importPath:    moduleImportPrefix + "generic/output/json",
+		},
+		{
+			name:          "application composition root may compose registrars",
 			filename:      "internal/cli/catalog.go",
-			importPath: moduleImportPrefix + "dnd/register",
+			sourcePackage: "cli",
+			importPath:    moduleImportPrefix + "almanac/register",
 		},
 		{
-			name:       "external integration test may compose domains",
+			name:          "black-box integration test may compose families",
 			filename:      "internal/modules/integration/example_test.go",
-			importPath: moduleImportPrefix + "dnd/extract/spells",
+			sourcePackage: "integration_test",
+			importPath:    moduleImportPrefix + "almanac/extract/events",
+		},
+		{
+			name:          "white-box integration test is not exempt",
+			filename:      "internal/modules/integration/example_test.go",
+			sourcePackage: "integration",
+			importPath:    moduleImportPrefix + "almanac/extract/events",
+			wantError:     true,
+		},
+		{
+			name:          "integration production file is not exempt",
+			filename:      "internal/modules/integration/example.go",
+			sourcePackage: "integration",
+			importPath:    moduleImportPrefix + "almanac/extract/events",
+			wantError:     true,
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			err := validateImport(tt.filename, tt.importPath)
+			err := validateImport(tt.filename, tt.sourcePackage, tt.importPath)
 			if tt.wantError && err == nil {
 				t.Fatal("validateImport() error = nil, want boundary violation")
 			}
@@ -129,68 +205,77 @@ func checkImportBoundaries(repositoryRoot string, filename string) error {
 		if err != nil {
 			return fmt.Errorf("parse import in %s: %w", relative, err)
 		}
-		if err := validateImport(relative, importPath); err != nil {
+		if err := validateImport(relative, parsed.Name.Name, importPath); err != nil {
 			return fmt.Errorf("%s imports %s: %w", relative, importPath, err)
 		}
 	}
 	return nil
 }
 
-func validateImport(filename string, importPath string) error {
+func validateImport(filename string, sourcePackage string, importPath string) error {
-	if isExternalIntegrationTest(filename) {
+	if isBlackBoxIntegrationTest(filename, sourcePackage) {
 		return nil
 	}
-	sourceDomain, sourceRoot := domainForFile(filename)
+	targetFamily, targetChild := moduleFamilyForImport(importPath)
-	targetDomain, targetChild := domainForImport(importPath)
+	if targetFamily == "" {
-	if sourceDomain == "" || targetDomain == "" {
 		return nil
 	}
-	if sourceRoot && sourceDomain == targetDomain && targetChild {
+	if isIntegrationFile(filename) {
-		return fmt.Errorf("domain root packages must not import child implementations")
+		return fmt.Errorf("module integration composition is allowed only in black-box tests")
-	}
-	if sourceDomain == "generic" && targetDomain == "dnd" {
-		return fmt.Errorf("generic packages must not import D&D packages")
-	}
-	if sourceDomain == "dnd" && targetDomain == "seriatim" {
-		return fmt.Errorf("D&D packages must not import Seriatim packages")
-	}
-	if sourceDomain == "seriatim" && targetDomain == "dnd" {
-		return fmt.Errorf("Seriatim packages must not import D&D packages")
 	}
+	sourceFamily, sourceRoot, sourceRegistrar := moduleFamilyForFile(filename)
+	if sourceFamily == "" {
 		return nil
 	}
+	if sourceRoot && sourceFamily == targetFamily && targetChild {
+		return fmt.Errorf("family root must not import child packages")
+	}
+	if sourceFamily == targetFamily {
+		return nil
+	}
+	if sourceFamily == "generic" {
+		return fmt.Errorf("generic family must not import concrete family %q", targetFamily)
+	}
+	if targetFamily == "generic" {
+		if sourceRegistrar {
+			return nil
+		}
+		return fmt.Errorf("concrete family %q may import generic implementations only from its registrar", sourceFamily)
+	}
+	return fmt.Errorf("concrete family %q must not import concrete family %q", sourceFamily, targetFamily)
+}
 
-func domainForFile(filename string) (domain string, root bool) {
+func moduleFamilyForFile(filename string) (family string, root bool, registrar bool) {
 	const prefix = "internal/modules/"
 	if !strings.HasPrefix(filename, prefix) {
-		return "", false
+		return "", false, false
 	}
 	remainder := strings.TrimPrefix(filename, prefix)
 	parts := strings.Split(remainder, "/")
-	if len(parts) < 2 || !isDomain(parts[0]) {
+	if len(parts) < 2 || parts[0] == "integration" {
-		return "", false
+		return "", false, false
 	}
-	return parts[0], len(parts) == 2
+	return parts[0], len(parts) == 2, len(parts) > 2 && parts[1] == "register"
 }
 
-func domainForImport(importPath string) (domain string, child bool) {
+func moduleFamilyForImport(importPath string) (family string, child bool) {
 	if !strings.HasPrefix(importPath, moduleImportPrefix) {
 		return "", false
 	}
 	remainder := strings.TrimPrefix(importPath, moduleImportPrefix)
 	parts := strings.Split(remainder, "/")
-	if len(parts) == 0 || !isDomain(parts[0]) {
+	if len(parts) == 0 || parts[0] == "" || parts[0] == "integration" {
 		return "", false
 	}
 	return parts[0], len(parts) > 1
 }
 
-func isDomain(name string) bool {
+func isIntegrationFile(filename string) bool {
-	return name == "dnd" || name == "generic" || name == "seriatim"
+	return strings.HasPrefix(filename, "internal/modules/integration/")
 }
 
-func isExternalIntegrationTest(filename string) bool {
+func isBlackBoxIntegrationTest(filename string, sourcePackage string) bool {
-	return strings.HasPrefix(filename, "internal/modules/integration/") && strings.HasSuffix(filename, "_test.go")
+	return isIntegrationFile(filename) && strings.HasSuffix(filename, "_test.go") && sourcePackage == "integration_test"
 }
 
 func testRepositoryRoot(t *testing.T) string {

internal/modules/seriatim/input/transcript/config_test.go

@@ -12,8 +12,6 @@ import (
 	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
-	"gitea.maximumdirect.net/eric/notarius/internal/modules/generic/merge/appendorder"
-	"gitea.maximumdirect.net/eric/notarius/internal/modules/generic/normalize/noop"
 )
 
 func TestPipelineConfigLoadsAndResolvesWithSeriatimInput(t *testing.T) {
@@ -192,12 +190,7 @@ func mustRegisterExtractor(t *testing.T, registry *pipeline.ExtractorRegistry, s
 func mustRegisterMerger(t *testing.T, registry *pipeline.MergerRegistry, spec pipeline.ModuleSpec) {
 	t.Helper()
 	if err := pipeline.RegisterMerger[seriatimArtifact](registry, spec, func() (contracts.Merger[seriatimArtifact], error) {
-		return appendorder.NewTyped(func(values []seriatimArtifact) (seriatimArtifact, error) {
+		return fakeMerger{}, nil
-			if len(values) == 0 {
-				return seriatimArtifact{}, nil
-			}
-			return values[0], nil
-		})
 	}); err != nil {
 		t.Fatalf("register merger: %v", err)
 	}
@@ -206,7 +199,7 @@ func mustRegisterMerger(t *testing.T, registry *pipeline.MergerRegistry, spec pi
 func mustRegisterNormalizer(t *testing.T, registry *pipeline.NormalizerRegistry, spec pipeline.ModuleSpec) {
 	t.Helper()
 	if err := pipeline.RegisterNormalizer[seriatimArtifact](registry, spec, func() (contracts.Normalizer[seriatimArtifact], error) {
-		return noop.NewTyped[seriatimArtifact](), nil
+		return fakeNormalizer{}, nil
 	}); err != nil {
 		t.Fatalf("register normalizer: %v", err)
 	}
@@ -241,6 +234,27 @@ func (fakeExtractor) Extract(ctx context.Context, req contracts.TypedExtractionR
 	return contracts.TypedExtractionResult[seriatimArtifact]{}, nil
 }
 
+type fakeMerger struct{}
+
+func (fakeMerger) Key() string { return pipeline.DefaultMergeModule }
+
+func (fakeMerger) Merge(ctx context.Context, req contracts.TypedMergeRequest[seriatimArtifact]) (contracts.TypedMergeResult[seriatimArtifact], error) {
+	if len(req.ExtractOutputs) == 0 {
+		return contracts.TypedMergeResult[seriatimArtifact]{}, nil
+	}
+	return contracts.TypedMergeResult[seriatimArtifact]{Value: req.ExtractOutputs[0].Value}, nil
+}
+
+type fakeNormalizer struct{}
+
+func (fakeNormalizer) Key() string { return pipeline.DefaultNormalizeModule }
+
+func (fakeNormalizer) ReferenceSlots() []contracts.ReferenceSlot { return nil }
+
+func (fakeNormalizer) Normalize(ctx context.Context, req contracts.TypedNormalizeRequest[seriatimArtifact]) (contracts.TypedNormalizeResult[seriatimArtifact], error) {
+	return contracts.TypedNormalizeResult[seriatimArtifact]{Value: req.MergeOutput.Value}, nil
+}
+
 type fakeOutput struct{}
 
 func (fakeOutput) Key() string { return pipeline.DefaultOutputModule }

internal/modules/seriatim/input/transcript/runner_test.go

@@ -11,8 +11,6 @@ import (
 	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
-	"gitea.maximumdirect.net/eric/notarius/internal/modules/generic/merge/appendorder"
-	"gitea.maximumdirect.net/eric/notarius/internal/modules/generic/normalize/noop"
 )
 
 func runPreparedPipeline(t *testing.T, registries pipeline.Registries, resolved pipeline.ResolvedPipeline, llmClient contracts.StructuredLLMClient, input pipeline.RunInput) (pipeline.RunOutput, error) {
@@ -138,15 +136,14 @@ func seriatimRunnerRegistries(t *testing.T, extractor contracts.Extractor[seriat
 	}); err != nil {
 		t.Fatalf("register extractor: %v", err)
 	}
-	if err := appendorder.RegisterTyped(mergers, seriatimArtifactKind, func(values []seriatimArtifact) (seriatimArtifact, error) {
+	if err := pipeline.RegisterMerger[seriatimArtifact](mergers, pipeline.ModuleSpec{Key: pipeline.DefaultMergeModule, Stage: pipeline.StageMerge, ArtifactKind: seriatimArtifactKind}, func() (contracts.Merger[seriatimArtifact], error) {
-		if len(values) == 0 {
+		return fakeMerger{}, nil
-			return seriatimArtifact{}, nil
-		}
-		return values[0], nil
 	}); err != nil {
 		t.Fatalf("register merger: %v", err)
 	}
-	if err := noop.RegisterTyped[seriatimArtifact](normalizers, seriatimArtifactKind); err != nil {
+	if err := pipeline.RegisterNormalizer[seriatimArtifact](normalizers, pipeline.ModuleSpec{Key: pipeline.DefaultNormalizeModule, Stage: pipeline.StageNormalize, ArtifactKind: seriatimArtifactKind}, func() (contracts.Normalizer[seriatimArtifact], error) {
+		return fakeNormalizer{}, nil
+	}); err != nil {
 		t.Fatalf("register normalizer: %v", err)
 	}
 	if err := outputs.Register(pipeline.DefaultOutputModule, func() (contracts.OutputEncoder, error) {