Run D&D spell lanes through typed artifacts

2026-07-17 07:19:58 +00:00
parent 52e6b31408
commit 66de1a5520
17 changed files with 699 additions and 56 deletions
--- a/internal/framework/pipeline/artifact_codec_registry.go
+++ b/internal/framework/pipeline/artifact_codec_registry.go
@@ -63,10 +63,12 @@ type ArtifactCodecRegistry struct {
 }

 type artifactCodecEntry struct {
-	spec      ArtifactCodecSpec
-	valueType reflect.Type
-	encode    func(any) ([]byte, error)
-	decode    func([]byte) (any, error)
+	spec            ArtifactCodecSpec
+	valueType       reflect.Type
+	encode          func(any) ([]byte, error)
+	encodeCandidate func(any) ([]byte, error)
+	metadata        func(any) map[string]any
+	decode          func([]byte) (any, error)
 }

 func NewArtifactCodecRegistry() *ArtifactCodecRegistry {
@@ -118,6 +120,29 @@ func RegisterArtifactCodec[T any](registry *ArtifactCodecRegistry, codec contrac
 			return decoded, nil
 		},
 	}
+	entry.encodeCandidate = entry.encode
+	if candidate, ok := any(codec).(interface{ EncodeCandidate(T) ([]byte, error) }); ok {
+		entry.encodeCandidate = func(value any) ([]byte, error) {
+			typed, err := exactTypedValue[T]("encode candidate artifact", value)
+			if err != nil {
+				return nil, err
+			}
+			content, err := candidate.EncodeCandidate(typed)
+			if err != nil {
+				return nil, &ArtifactCodecOperationError{Operation: "encode", Kind: spec.Kind, Err: err}
+			}
+			return append([]byte(nil), content...), nil
+		}
+	}
+	if provider, ok := any(codec).(interface{ Metadata(T) map[string]any }); ok {
+		entry.metadata = func(value any) map[string]any {
+			typed, err := exactTypedValue[T]("artifact metadata", value)
+			if err != nil {
+				return nil
+			}
+			return cloneMetadata(provider.Metadata(typed))
+		}
+	}
 	if registry.entries == nil {
 		registry.entries = make(map[contracts.ArtifactKind]artifactCodecEntry)
 	}
--- a/internal/framework/pipeline/debug.go
+++ b/internal/framework/pipeline/debug.go
@@ -719,6 +719,14 @@ func debugValidationResultEnvelope(result contracts.ValidationResult) contracts.
 	return result
 }

+func debugWarningEnvelopes(warnings []contracts.Warning) []contracts.Warning {
+	out := cloneWarnings(warnings)
+	for i := range out {
+		out[i].Message = string(redactSecretBytes([]byte(out[i].Message)))
+	}
+	return out
+}
+
 func debugRejectedOutputEnvelope(rejected contracts.RejectedOutput) contracts.RejectedOutput {
 	rejected.Message = string(redactSecretBytes([]byte(rejected.Message)))
 	rejected.DiagnosticArtifactPath = string(redactSecretBytes([]byte(rejected.DiagnosticArtifactPath)))
--- a/internal/framework/pipeline/extractor_registry.go
+++ b/internal/framework/pipeline/extractor_registry.go
@@ -1,6 +1,7 @@
 package pipeline

 import (
+	"context"
 	"fmt"
 	"reflect"
 	"strings"
@@ -23,6 +24,7 @@ type typedExtractorEntry struct {
 	valueType       reflect.Type
 	validateOptions OptionValidator
 	builder         func(BuildRequest) (any, error)
+	extract         typedExtractOperation
 	rawBuilder      LegacyRawExtractorBuilder
 }

@@ -133,6 +135,17 @@ func registerExtractorBuilder[T any](registry *ExtractorRegistry, spec ModuleSpe
 		builder: func(request BuildRequest) (any, error) {
 			return builder(cloneBuildRequest(request))
 		},
+		extract: func(ctx context.Context, implementation any, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
+			extractor, ok := implementation.(contracts.Extractor[T])
+			if !ok {
+				return erasedTypedResult{}, fmt.Errorf("extractor %q has incompatible implementation %T", normalizedSpec.Key, implementation)
+			}
+			result, err := extractor.Extract(ctx, request)
+			if err != nil {
+				return erasedTypedResult{}, err
+			}
+			return erasedTypedResult{Value: result.Value, Warnings: result.Warnings}, nil
+		},
 		rawBuilder: rawBuilder,
 	}
 	if registry.typedEntries == nil {
--- a/internal/framework/pipeline/merger_registry.go
+++ b/internal/framework/pipeline/merger_registry.go
@@ -1,6 +1,7 @@
 package pipeline

 import (
+	"context"
 	"fmt"
 	"reflect"
 	"strings"
@@ -28,6 +29,7 @@ type typedMergerEntry struct {
 	valueType       reflect.Type
 	validateOptions OptionValidator
 	builder         func(BuildRequest) (any, error)
+	merge           typedMergeOperation
 }

 func NewMergerRegistry() *MergerRegistry {
@@ -127,6 +129,25 @@ func RegisterMergerBuilder[T any](registry *MergerRegistry, spec ModuleSpec, val
 		builder: func(request BuildRequest) (any, error) {
 			return builder(cloneBuildRequest(request))
 		},
+		merge: func(ctx context.Context, implementation any, request contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
+			merger, ok := implementation.(contracts.Merger[T])
+			if !ok {
+				return erasedTypedResult{}, fmt.Errorf("merger %q has incompatible implementation %T", normalizedSpec.Key, implementation)
+			}
+			outputs := make([]contracts.ExtractArtifact[T], len(request.ExtractOutputs))
+			for i, output := range request.ExtractOutputs {
+				value, err := exactTypedValue[T]("merge extract value", output.Value)
+				if err != nil {
+					return erasedTypedResult{}, err
+				}
+				outputs[i] = contracts.ExtractArtifact[T]{LaneID: output.LaneID, ExtractorKey: output.ExtractorKey, SourceID: output.SourceID, ChunkID: output.ChunkID, ChunkIndex: output.ChunkIndex, ChunkRef: output.ChunkRef, Value: value}
+			}
+			result, err := merger.Merge(ctx, contracts.TypedMergeRequest[T]{Source: request.Source, LaneID: request.LaneID, ExtractOutputs: outputs, SourceInput: request.SourceInput, SessionID: request.SessionID, References: request.References, LLMProfile: request.LLMProfile, Metadata: request.Metadata})
+			if err != nil {
+				return erasedTypedResult{}, err
+			}
+			return erasedTypedResult{Value: result.Value, Warnings: result.Warnings}, nil
+		},
 	}
 	return nil
 }
--- a/internal/framework/pipeline/normalizer_registry.go
+++ b/internal/framework/pipeline/normalizer_registry.go
@@ -1,6 +1,7 @@
 package pipeline

 import (
+	"context"
 	"fmt"
 	"reflect"
 	"strings"
@@ -23,6 +24,7 @@ type typedNormalizerEntry struct {
 	valueType       reflect.Type
 	validateOptions OptionValidator
 	builder         func(BuildRequest) (any, error)
+	normalize       typedNormalizeOperation
 }

 func NewNormalizerRegistry() *NormalizerRegistry {
@@ -122,6 +124,21 @@ func RegisterNormalizerBuilder[T any](registry *NormalizerRegistry, spec ModuleS
 		builder: func(request BuildRequest) (any, error) {
 			return builder(cloneBuildRequest(request))
 		},
+		normalize: func(ctx context.Context, implementation any, request contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
+			normalizer, ok := implementation.(contracts.Normalizer[T])
+			if !ok {
+				return erasedTypedResult{}, fmt.Errorf("normalizer %q has incompatible implementation %T", normalizedSpec.Key, implementation)
+			}
+			value, err := exactTypedValue[T]("normalize merge value", request.MergeOutput.Value)
+			if err != nil {
+				return erasedTypedResult{}, err
+			}
+			result, err := normalizer.Normalize(ctx, contracts.TypedNormalizeRequest[T]{Source: request.Source, LaneID: request.LaneID, MergeOutput: contracts.MergeArtifact[T]{LaneID: request.MergeOutput.LaneID, MergerKey: request.MergeOutput.MergerKey, SourceID: request.MergeOutput.SourceID, Value: value}, SourceInput: request.SourceInput, SessionID: request.SessionID, References: request.References, LLMProfile: request.LLMProfile, Metadata: request.Metadata})
+			if err != nil {
+				return erasedTypedResult{}, err
+			}
+			return erasedTypedResult{Value: result.Value, Warnings: result.Warnings}, nil
+		},
 	}
 	return nil
 }
--- a/internal/framework/pipeline/prepare.go
+++ b/internal/framework/pipeline/prepare.go
@@ -49,6 +49,10 @@ type preparedTypedLane struct {
 	extractor  any
 	merger     any
 	normalizer any
+	extract    typedExtractOperation
+	merge      typedMergeOperation
+	normalize  typedNormalizeOperation
+	codec      artifactCodecEntry
 }

 type preparedValidatorChain struct {
@@ -57,17 +61,18 @@ type preparedValidatorChain struct {
 }

 type preparedValidator struct {
-	resolved   ResolvedValidator
-	legacy     contracts.LegacyRawValidator
-	typed      any
-	chunk      contracts.ChunkValidator
-	serialized contracts.SerializedValidator
+	resolved      ResolvedValidator
+	legacy        contracts.LegacyRawValidator
+	typed         any
+	typedValidate typedValidateOperation
+	chunk         contracts.ChunkValidator
+	serialized    contracts.SerializedValidator
 }

 // Prepare validates all configured options and constructs every selected
 // module and validator before any operation method can run.
 func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDependencies) (*PreparedPipeline, error) {
-	if err := validateResolvedPipeline(resolved, false); err != nil {
+	if err := validateResolvedPipeline(resolved); err != nil {
 		return nil, err
 	}
 	if err := validateRegistrySet(resolved, registries); err != nil {
@@ -140,7 +145,11 @@ func prepareLane(pipeline ResolvedPipeline, lane ResolvedArtifactLane, registrie
 		if err != nil {
 			return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageExtract, lane.Extract.Module, "", err)
 		}
-		executor.typed = &preparedTypedLane{extractor: module}
+		codec, _, codecErr := registries.ArtifactCodecs.entry(lane.ArtifactKind)
+		if codecErr != nil {
+			return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageExtract, lane.Extract.Module, "", codecErr)
+		}
+		executor.typed = &preparedTypedLane{extractor: module, extract: entry.extract, codec: codec}
 	}

 	var err error
@@ -165,6 +174,7 @@ func prepareLane(pipeline ResolvedPipeline, lane ResolvedArtifactLane, registrie
 			return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageMerge, lane.Merge.Module, "", err)
 		}
 		executor.typed.merger = module
+		executor.typed.merge = entry.merge
 	}
 	executor.mergeValidators, err = prepareValidatorChain(pipeline, registries, deps, StageMerge, lane.ID, lane.Merge.Module)
 	if err != nil {
@@ -187,6 +197,7 @@ func prepareLane(pipeline ResolvedPipeline, lane ResolvedArtifactLane, registrie
 			return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageNormalize, lane.Normalize.Module, "", err)
 		}
 		executor.typed.normalizer = module
+		executor.typed.normalize = entry.normalize
 	}
 	executor.normalizeValidators, err = prepareValidatorChain(pipeline, registries, deps, StageNormalize, lane.ID, lane.Normalize.Module)
 	if err != nil {
@@ -222,6 +233,7 @@ func buildPreparedValidator(registry *ValidatorRegistry, resolved ResolvedValida
 		}
 		implementation, err = entry.builder(cloneBuildRequest(request))
 		prepared.typed = implementation
+		prepared.typedValidate = entry.validate
 	case ValidatorTargetChunk:
 		entry, ok := registry.chunkEntry(key)
 		if !ok {
--- a/internal/framework/pipeline/profile.go
+++ b/internal/framework/pipeline/profile.go
@@ -481,6 +481,9 @@ func validatorSpecForTarget(registry *ValidatorRegistry, stage ModuleStage, key
 	if entry, ok := registry.serializedEntry(key); ok && entry.spec.SupportsArtifacts {
 		return entry.spec.ValidatorSpec, ValidatorTargetSerialized, nil
 	}
+	if _, ok := registry.Spec(key); !ok {
+		return ValidatorSpec{}, "", fmt.Errorf("references unknown validator %q for artifact kind %q", key, kind)
+	}
 	return ValidatorSpec{}, "", missingArtifactVariantError("validator", key, kind, registry.registeredTypedKinds(key))
 }

--- a/internal/framework/pipeline/runner.go
+++ b/internal/framework/pipeline/runner.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"encoding/json"
 	"fmt"
 	"mime"
 	"path"
@@ -244,7 +245,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
 				}, llmScope))
 				return false, nil, err
 			}
-			validationWarnings, rejection, err := r.validateChunksRaw(attemptCtx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.llmClient, input.Metadata, input.Prepared.chunkValidators, attempt, input.Debug)
+			validationWarnings, rejection, err := r.validateChunks(attemptCtx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.llmClient, input.Metadata, input.Prepared.chunkValidators, attempt, input.Debug)
 			if err != nil || rejection != nil {
 				_ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{
 					Stage:     string(StageChunk),
@@ -376,6 +377,13 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
 }

 func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, checkpointLoader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, prepared preparedLaneExecutor, output *RunOutput) error {
+	if prepared.typed != nil {
+		return r.runTypedLane(ctx, input, checkpoints, checkpointLoader, doc, sourceInput, sessionID, chunks, prepared, output)
+	}
+	return r.runLegacyLane(ctx, input, checkpoints, checkpointLoader, doc, sourceInput, sessionID, chunks, prepared, output)
+}
+
+func (r *Runner) runLegacyLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, checkpointLoader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, prepared preparedLaneExecutor, output *RunOutput) error {
 	lane := prepared.resolved
 	if prepared.legacy == nil {
 		return fmt.Errorf("resolved pipeline lane %q uses typed artifact kind %q, which the legacy raw runner cannot execute", lane.ID, lane.ArtifactKind)
@@ -950,6 +958,64 @@ func (r *Runner) validateChunksRaw(ctx context.Context, doc *source.SourceDocume
 	})
 }

+func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument, moduleKey string, chunks []source.Chunk, sourceInput contracts.LLMInputMaterial, sessionID string, references contracts.ReferenceSet, llmClient contracts.StructuredLLMClient, metadata map[string]any, prepared preparedValidatorChain, attempt int, debug DebugRecorder) ([]contracts.Warning, *contracts.RejectedOutput, error) {
+	allLegacy := true
+	for _, item := range prepared.validators {
+		if item.resolved.Target != ValidatorTargetLegacyRaw && item.resolved.Target != "" {
+			allLegacy = false
+			break
+		}
+	}
+	if allLegacy {
+		return r.validateChunksRaw(ctx, doc, moduleKey, chunks, sourceInput, sessionID, references, llmClient, metadata, prepared, attempt, debug)
+	}
+	content, err := json.Marshal(chunks)
+	if err != nil {
+		return nil, nil, fmt.Errorf("encode canonical chunks for validation: %w", err)
+	}
+	schema := contracts.ArtifactSchema{ID: "notarius.source.chunks", Name: "notarius_source_chunks", Version: "v1", JSONSchema: []byte(`{"$schema":"https://json-schema.org/draft/2020-12/schema","type":"array"}`)}
+	var warnings []contracts.Warning
+	for index, item := range prepared.validators {
+		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)
+		var result contracts.ValidationResult
+		switch item.resolved.Target {
+		case ValidatorTargetChunk:
+			result, err = item.chunk.Validate(validatorCtx, contracts.ChunkValidationRequest{ModuleKey: moduleKey, Source: doc, SourceID: doc.ID, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(references), LLMProfile: binding.LLMProfile, Metadata: cloneMetadata(metadata), Chunks: cloneSourceChunks(chunks)})
+		case ValidatorTargetSerialized:
+			result, err = item.serialized.Validate(validatorCtx, contracts.SerializedValidationRequest{Stage: string(StageChunk), ModuleKey: moduleKey, Source: doc, SourceID: doc.ID, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(references), LLMProfile: binding.LLMProfile, Metadata: cloneMetadata(metadata), Chunks: cloneSourceChunks(chunks), Schema: contracts.CloneArtifactSchema(schema), MediaType: "application/json", Content: append([]byte(nil), content...)})
+		default:
+			return nil, nil, fmt.Errorf("validator %q is incompatible with chunk validation", binding.Module)
+		}
+		debugRequest := contracts.ValidationRequest{Stage: string(StageChunk), ModuleKey: moduleKey, Source: doc, SourceID: doc.ID, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(references), LLMProfile: binding.LLMProfile, Metadata: cloneMetadata(metadata), Chunks: cloneSourceChunks(chunks), Schema: contracts.ResponseSchema{ID: schema.ID, Name: schema.Name, Version: schema.Version, JSONSchema: append([]byte(nil), schema.JSONSchema...)}, Payload: contracts.RawPayload{Content: content, MediaType: "application/json"}}
+		debugCall := debugValidationCall{ValidatorName: binding.Module, Request: debugValidationRequestEnvelope(debugRequest), Result: debugValidationResultEnvelope(result)}
+		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 {
+			return nil, nil, debugErr
+		}
+		if err != nil {
+			return nil, nil, fmt.Errorf("validate chunks with validator %q: %w", binding.Module, err)
+		}
+		if !result.Approved {
+			reason := result.ReasonCode
+			if reason == "" {
+				reason = "raw_output_rejected"
+			}
+			message := result.Message
+			if message == "" {
+				message = "raw output rejected"
+			}
+			return nil, &contracts.RejectedOutput{Stage: string(StageChunk), ModuleKey: moduleKey, ValidatorName: binding.Module, ReasonCode: reason, Message: message, AttemptCount: attempt, DiagnosticArtifactPath: result.DiagnosticArtifactPath}, nil
+		}
+		warnings = append(warnings, result.Warnings...)
+	}
+	return warnings, nil, nil
+}
+
 func (r *Runner) validateRaw(ctx context.Context, target rawValidationTarget) ([]contracts.Warning, *contracts.RejectedOutput, error) {
 	if len(target.prepared.validators) == 0 {
 		return nil, nil, nil
@@ -1070,10 +1136,10 @@ func validateRunInput(input RunInput) error {
 	if input.Prepared == nil {
 		return fmt.Errorf("prepared pipeline must not be nil")
 	}
-	return validateResolvedPipeline(input.Prepared.resolved, true)
+	return validateResolvedPipeline(input.Prepared.resolved)
 }

-func validateResolvedPipeline(pipeline ResolvedPipeline, rejectTyped bool) error {
+func validateResolvedPipeline(pipeline ResolvedPipeline) error {
 	if pipeline.ID == "" {
 		return fmt.Errorf("resolved pipeline id must not be empty")
 	}
@@ -1096,9 +1162,6 @@ func validateResolvedPipeline(pipeline ResolvedPipeline, rejectTyped bool) error
 		if lane.ID == "" {
 			return fmt.Errorf("resolved pipeline artifact lane id must not be empty")
 		}
-		if rejectTyped && lane.ArtifactKind != "" {
-			return fmt.Errorf("resolved pipeline lane %q uses typed artifact kind %q, which the legacy raw runner cannot execute", lane.ID, lane.ArtifactKind)
-		}
 		if lane.Extract.Module == "" {
 			return fmt.Errorf("resolved pipeline lane %q extract module must not be empty", lane.ID)
 		}
--- a/internal/framework/pipeline/runner_typed.go
+++ b/internal/framework/pipeline/runner_typed.go
@@ -0,0 +1,366 @@
+package pipeline
+
+import (
+	"context"
+	"fmt"
+	"path"
+	"sort"
+	"time"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+)
+
+// migrationRawArtifact serializes a typed value into the existing raw
+// checkpoint, debug, and output envelopes. It is removed when those boundaries
+// consume SerializedArtifact directly.
+func migrationRawArtifact(codec artifactCodecEntry, value any) (contracts.ResponseSchema, contracts.RawPayload, error) {
+	content, err := codec.encodeCandidate(value)
+	if err != nil {
+		return contracts.ResponseSchema{}, contracts.RawPayload{}, err
+	}
+	schema := codec.spec.Schema
+	metadata := map[string]any(nil)
+	if codec.metadata != nil {
+		metadata = codec.metadata(value)
+	}
+	return contracts.ResponseSchema{ID: schema.ID, Name: schema.Name, Version: schema.Version, JSONSchema: append([]byte(nil), schema.JSONSchema...)}, contracts.RawPayload{Content: content, MediaType: codec.spec.MediaType, Metadata: metadata}, nil
+}
+
+// migrationDecodeArtifact restores a typed value from an existing raw
+// checkpoint envelope through the registered codec.
+func migrationDecodeArtifact(codec artifactCodecEntry, payload contracts.RawPayload) (any, error) {
+	return codec.decode(append([]byte(nil), payload.Content...))
+}
+
+func migrationExtractOutput(codec artifactCodecEntry, artifact erasedExtractArtifact) (contracts.ExtractOutput, error) {
+	schema, payload, err := migrationRawArtifact(codec, artifact.Value)
+	if err != nil {
+		return contracts.ExtractOutput{}, err
+	}
+	return contracts.ExtractOutput{LaneID: artifact.LaneID, ExtractorKey: artifact.ExtractorKey, SourceID: artifact.SourceID, ChunkID: artifact.ChunkID, ChunkIndex: artifact.ChunkIndex, Schema: schema, Payload: payload}, nil
+}
+
+func migrationMergeOutput(codec artifactCodecEntry, artifact erasedMergeArtifact) (contracts.MergeOutput, error) {
+	schema, payload, err := migrationRawArtifact(codec, artifact.Value)
+	if err != nil {
+		return contracts.MergeOutput{}, err
+	}
+	return contracts.MergeOutput{LaneID: artifact.LaneID, MergerKey: artifact.MergerKey, SourceID: artifact.SourceID, Schema: schema, Payload: payload}, nil
+}
+
+func migrationNormalizeOutput(codec artifactCodecEntry, laneID, key, sourceID string, value any) (contracts.NormalizeOutput, error) {
+	schema, payload, err := migrationRawArtifact(codec, value)
+	if err != nil {
+		return contracts.NormalizeOutput{}, err
+	}
+	return contracts.NormalizeOutput{LaneID: laneID, NormalizerKey: key, SourceID: sourceID, Schema: schema, Payload: payload}, nil
+}
+
+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) error {
+	lane, typed := prepared.resolved, prepared.typed
+	if typed == nil {
+		return fmt.Errorf("typed lane %q executor is not prepared", lane.ID)
+	}
+	setTypedLaneManifestMetadata(output, lane.ID, typed.extractor, typed.merger, typed.normalizer)
+
+	values := make([]erasedExtractArtifact, 0, len(chunks))
+	rawExtracts := make([]contracts.ExtractOutput, 0, len(chunks))
+	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 := loader.Extract(lane.ID, lane.Extract.Module, extractDeps)
+	recordCheckpointEvent(output, loader, string(StageExtract), lane.ID, lane.Extract.Module, 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": decision.Reused, "decision": decision, "source": debugSourceDocumentEnvelope(doc), "chunks": debugSourceChunkEnvelopes(chunks), "options": redactSensitiveMap(lane.Extract.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
+		return err
+	}
+	if decision.Reused {
+		for _, raw := range cp.Outputs {
+			value, decodeErr := migrationDecodeArtifact(typed.codec, raw.Payload)
+			if decodeErr != nil {
+				return fmt.Errorf("decode extract checkpoint for lane %q: %w", lane.ID, decodeErr)
+			}
+			artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: raw.ChunkID, ChunkIndex: raw.ChunkIndex, Value: value}
+			if raw.ChunkIndex >= 0 && raw.ChunkIndex < len(chunks) {
+				artifact.ChunkRef = chunks[raw.ChunkIndex].Ref
+			}
+			values = append(values, artifact)
+			rawExtracts = append(rawExtracts, cloneExtractOutput(raw))
+		}
+		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 rawAccepted contracts.ExtractOutput
+			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
+				}
+				raw, encodeErr := migrationExtractOutput(typed.codec, artifact)
+				if encodeErr != nil {
+					return false, nil, encodeErr
+				}
+				raw.Payload.Warnings = append(raw.Payload.Warnings, cloneWarnings(result.Warnings)...)
+				accepted, rawAccepted = artifact, raw
+				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": debugExtractOutputEnvelope(raw), "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)
+			rawExtracts = append(rawExtracts, rawAccepted)
+			extractWarnings = append(extractWarnings, acceptedWarnings...)
+			output.Warnings = append(output.Warnings, acceptedWarnings...)
+		}
+		if err := checkpoints.ExtractSucceeded(lane.ID, lane.Extract.Module, extractDeps, rawExtracts, 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(rawExtracts, func(i, j int) bool { return rawExtracts[i].ChunkIndex < rawExtracts[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": decision.Reused, "outputs": debugExtractOutputEnvelopes(rawExtracts), "rejected": debugRejectedOutputEnvelopes(output.Rejected[rejectedStart:]), "warnings": debugWarningEnvelopes(extractWarnings)}}); err != nil {
+		return err
+	}
+	if len(values) == 0 {
+		return nil
+	}
+
+	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 := rawOutputDigests(extractPayloads(rawExtracts))
+	mergeCP, mergeDecision := loader.Merge(lane.ID, lane.Merge.Module, mergeDeps)
+	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": debugExtractOutputEnvelopes(rawExtracts), "options": redactSensitiveMap(lane.Merge.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
+		return err
+	}
+	var merged erasedMergeArtifact
+	var rawMerge contracts.MergeOutput
+	var mergeWarnings []contracts.Warning
+	if mergeDecision.Reused {
+		value, decodeErr := migrationDecodeArtifact(typed.codec, mergeCP.Output.Payload)
+		if decodeErr != nil {
+			return fmt.Errorf("decode merge checkpoint for lane %q: %w", lane.ID, decodeErr)
+		}
+		merged = erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: value}
+		rawMerge = cloneMergeOutput(mergeCP.Output)
+		mergeWarnings = cloneWarnings(mergeCP.Warnings)
+		output.Warnings = append(output.Warnings, mergeWarnings...)
+	} else {
+		if err := checkpoints.MergeRunning(lane.ID, lane.Merge.Module, mergeDeps); err != nil {
+			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)})
+			if callErr != nil {
+				return false, nil, fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, lane.Merge.Module, callErr)
+			}
+			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)
+			if validateErr != nil || rejected != nil {
+				return false, rejected, validateErr
+			}
+			raw, encodeErr := migrationMergeOutput(typed.codec, candidate)
+			if encodeErr != nil {
+				return false, nil, encodeErr
+			}
+			raw.Payload.Warnings = append(raw.Payload.Warnings, cloneWarnings(result.Warnings)...)
+			merged, rawMerge = candidate, raw
+			mergeWarnings = append(cloneWarnings(result.Warnings), warnings...)
+			return true, nil, nil
+		})
+		if runErr != nil {
+			_ = checkpoints.MergeFailed(lane.ID, lane.Merge.Module, mergeDeps, runErr)
+			return runErr
+		}
+		if !ok {
+			output.Rejected = append(output.Rejected, *rejection)
+			if err := checkpoints.MergeRejected(lane.ID, lane.Merge.Module, mergeDeps, *rejection); err != nil {
+				return err
+			}
+			return nil
+		}
+		output.Warnings = append(output.Warnings, mergeWarnings...)
+		if err := checkpoints.MergeSucceeded(lane.ID, lane.Merge.Module, mergeDeps, rawMerge, mergeWarnings); err != nil {
+			return err
+		}
+	}
+	if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "accepted": true, "output": debugMergeOutputEnvelope(rawMerge), "warnings": debugWarningEnvelopes(mergeWarnings)}}); err != nil {
+		return err
+	}
+
+	normalizeDeps := rawOutputDigests([]contracts.RawPayload{rawMerge.Payload})
+	normalizeCP, normalizeDecision := loader.Normalize(lane.ID, lane.Normalize.Module, normalizeDeps)
+	recordCheckpointEvent(output, loader, string(StageNormalize), lane.ID, lane.Normalize.Module, normalizeDecision)
+	if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageNormalize), LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "decision": normalizeDecision, "source": debugSourceDocumentEnvelope(doc), "merge_output": debugMergeOutputEnvelope(rawMerge), "options": redactSensitiveMap(lane.Normalize.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
+		return err
+	}
+	var rawNormalize contracts.NormalizeOutput
+	var normalizeWarnings []contracts.Warning
+	if normalizeDecision.Reused {
+		_, decodeErr := migrationDecodeArtifact(typed.codec, normalizeCP.Output.Payload)
+		if decodeErr != nil {
+			return fmt.Errorf("decode normalize checkpoint for lane %q: %w", lane.ID, decodeErr)
+		}
+		rawNormalize, normalizeWarnings = cloneNormalizeOutput(normalizeCP.Output), cloneWarnings(normalizeCP.Warnings)
+		output.Warnings = append(output.Warnings, normalizeWarnings...)
+	} else {
+		if err := checkpoints.NormalizeRunning(lane.ID, lane.Normalize.Module, normalizeDeps); err != nil {
+			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)})
+			if callErr != nil {
+				return false, nil, fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, lane.Normalize.Module, callErr)
+			}
+			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)
+			if validateErr != nil || rejected != nil {
+				return false, rejected, validateErr
+			}
+			raw, encodeErr := migrationNormalizeOutput(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value)
+			if encodeErr != nil {
+				return false, nil, encodeErr
+			}
+			raw.Payload.Warnings = append(raw.Payload.Warnings, cloneWarnings(result.Warnings)...)
+			rawNormalize = raw
+			normalizeWarnings = append(cloneWarnings(result.Warnings), warnings...)
+			return true, nil, nil
+		})
+		if runErr != nil {
+			_ = checkpoints.NormalizeFailed(lane.ID, lane.Normalize.Module, normalizeDeps, runErr)
+			return runErr
+		}
+		if !ok {
+			output.Rejected = append(output.Rejected, *rejection)
+			if err := checkpoints.NormalizeRejected(lane.ID, lane.Normalize.Module, normalizeDeps, *rejection); err != nil {
+				return err
+			}
+			return nil
+		}
+		output.Warnings = append(output.Warnings, normalizeWarnings...)
+		if err := checkpoints.NormalizeSucceeded(lane.ID, lane.Normalize.Module, normalizeDeps, rawNormalize, normalizeWarnings); err != nil {
+			return err
+		}
+	}
+	if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageNormalize), LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "accepted": true, "output": debugNormalizeOutputEnvelope(rawNormalize), "warnings": debugWarningEnvelopes(normalizeWarnings)}}); err != nil {
+		return err
+	}
+	output.NormalizeOutputs = append(output.NormalizeOutputs, rawNormalize)
+	return nil
+}
+
+func setTypedLaneManifestMetadata(output *RunOutput, laneID string, extractor, merger, normalizer any) {
+	if output == nil {
+		return
+	}
+	for i := range output.Manifest.ArtifactLanes {
+		if output.Manifest.ArtifactLanes[i].ID != laneID {
+			continue
+		}
+		metadata := make(map[string]any)
+		for _, item := range []struct {
+			name   string
+			module any
+		}{{"extractor", extractor}, {"merger", merger}, {"normalizer", normalizer}} {
+			if value, ok := moduleManifestMetadata(item.module); ok {
+				metadata[item.name] = value
+			}
+		}
+		if len(metadata) > 0 {
+			output.Manifest.ArtifactLanes[i].Metadata = metadata
+		}
+		return
+	}
+}
+
+func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecEntry, target typedValidationTarget, chain preparedValidatorChain, attempt int, debug DebugRecorder) ([]contracts.Warning, *contracts.RejectedOutput, error) {
+	var warnings []contracts.Warning
+	for index, item := range chain.validators {
+		binding := item.resolved.Binding
+		var result contracts.ValidationResult
+		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)
+		switch item.resolved.Target {
+		case ValidatorTargetTyped:
+			target.llmProfile = binding.LLMProfile
+			result, err = item.typedValidate(validatorCtx, item.typed, target)
+		case ValidatorTargetSerialized:
+			schema, payload, encodeErr := migrationRawArtifact(codec, target.value)
+			if encodeErr != nil {
+				err = encodeErr
+				break
+			}
+			result, err = item.serialized.Validate(validatorCtx, contracts.SerializedValidationRequest{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Source: target.source, SourceID: target.sourceID, SourceInput: target.sourceInput.Clone(), SessionID: target.sessionID, References: CloneReferenceSet(target.references), LLMProfile: binding.LLMProfile, Metadata: cloneMetadata(target.metadata), Chunk: cloneSourceChunkPtr(target.chunk), Chunks: cloneSourceChunks(target.chunks), Schema: contracts.ArtifactSchema{ID: schema.ID, Name: schema.Name, Version: schema.Version, JSONSchema: append([]byte(nil), schema.JSONSchema...)}, MediaType: payload.MediaType, Content: append([]byte(nil), payload.Content...)})
+		default:
+			return nil, nil, fmt.Errorf("validator %q is incompatible with typed artifact validation", binding.Module)
+		}
+		schema, payload, _ := migrationRawArtifact(codec, target.value)
+		debugRequest := contracts.ValidationRequest{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Source: target.source, SourceID: target.sourceID, SourceInput: target.sourceInput.Clone(), SessionID: target.sessionID, References: CloneReferenceSet(target.references), LLMProfile: binding.LLMProfile, Metadata: cloneMetadata(target.metadata), Chunk: cloneSourceChunkPtr(target.chunk), Chunks: cloneSourceChunks(target.chunks), Schema: schema, Payload: payload}
+		debugCall := debugValidationCall{ValidatorName: binding.Module, Request: debugValidationRequestEnvelope(debugRequest), Result: debugValidationResultEnvelope(result)}
+		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 {
+			return nil, nil, fmt.Errorf("write validation debug artifact: %w", debugErr)
+		}
+		if err != nil {
+			return nil, nil, fmt.Errorf("validate typed %s output with validator %q: %w", target.stage, binding.Module, err)
+		}
+		if !result.Approved {
+			reason := result.ReasonCode
+			if reason == "" {
+				reason = "raw_output_rejected"
+			}
+			message := result.Message
+			if message == "" {
+				message = "raw output rejected"
+			}
+			return nil, &contracts.RejectedOutput{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, ChunkID: func() string {
+				if target.chunk != nil {
+					return target.chunk.ID
+				}
+				return ""
+			}(), ChunkIndex: func() int {
+				if target.chunk != nil {
+					return target.chunk.Index
+				}
+				return 0
+			}(), ValidatorName: binding.Module, ReasonCode: reason, Message: message, AttemptCount: attempt, DiagnosticArtifactPath: result.DiagnosticArtifactPath}, nil
+		}
+		warnings = append(warnings, result.Warnings...)
+	}
+	return warnings, nil, nil
+}
--- a/internal/framework/pipeline/typed_execution.go
+++ b/internal/framework/pipeline/typed_execution.go
@@ -0,0 +1,63 @@
+package pipeline
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+)
+
+type erasedExtractArtifact struct {
+	LaneID, ExtractorKey, SourceID, ChunkID string
+	ChunkIndex                              int
+	ChunkRef                                source.SourceRef
+	Value                                   any
+}
+
+type erasedMergeArtifact struct {
+	LaneID, MergerKey, SourceID string
+	Value                       any
+}
+
+type erasedTypedResult struct {
+	Value    any
+	Warnings []contracts.Warning
+}
+
+type typedValidationTarget struct {
+	stage       ModuleStage
+	laneID      string
+	moduleKey   string
+	source      *source.SourceDocument
+	sourceID    string
+	sourceInput contracts.LLMInputMaterial
+	sessionID   string
+	references  contracts.ReferenceSet
+	llmProfile  string
+	metadata    map[string]any
+	chunk       *source.Chunk
+	chunks      []source.Chunk
+	ref         source.SourceRef
+	value       any
+}
+
+func exactTypedValue[T any](operation string, value any) (T, error) {
+	want := reflect.TypeFor[T]()
+	if reflect.TypeOf(value) != want {
+		var zero T
+		return zero, fmt.Errorf("%s: expected exact Go type %s, got %T", operation, want, value)
+	}
+	typed, ok := value.(T)
+	if !ok {
+		var zero T
+		return zero, fmt.Errorf("%s: expected exact Go type %s, got %T", operation, want, value)
+	}
+	return typed, nil
+}
+
+type typedExtractOperation func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error)
+type typedMergeOperation func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error)
+type typedNormalizeOperation func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error)
+type typedValidateOperation func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error)
--- a/internal/framework/pipeline/typed_resolution_test.go
+++ b/internal/framework/pipeline/typed_resolution_test.go
@@ -6,6 +6,7 @@ import (
 	"strings"
 	"testing"

+	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
 )

@@ -130,6 +131,28 @@ func TestPrepareConstructsHeterogeneousTypedLanes(t *testing.T) {
 	}
 }

+func TestRunExecutesHeterogeneousTypedLanesWithCheckpointsDisabled(t *testing.T) {
+	catalog := typedResolutionCatalog(t, completeTypedCatalogOptions())
+	resolved, err := ResolvePipeline(typedResolutionProfile(), ResolveOptions{}, catalog)
+	if err != nil {
+		t.Fatalf("ResolvePipeline() error = %v", err)
+	}
+	prepared, err := Prepare(resolved, registriesFromModuleCatalog(catalog), ModuleDependencies{})
+	if err != nil {
+		t.Fatalf("Prepare() error = %v", err)
+	}
+	output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte(`{"source":true}`), RunID: "typed-run"})
+	if err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	if len(output.NormalizeOutputs) != 2 {
+		t.Fatalf("normalize outputs = %#v, want two typed lane results", output.NormalizeOutputs)
+	}
+	if len(output.CheckpointEvents) != 0 {
+		t.Fatalf("checkpoint events = %#v, want none with checkpoint loading disabled", output.CheckpointEvents)
+	}
+}
+
 func TestResolveTypedLaneRejectsIncompatibleComposition(t *testing.T) {
 	tests := []struct {
 		name   string
@@ -284,7 +307,8 @@ func mustRegisterTypedTestBase(t *testing.T, catalog ModuleCatalog) {
 		t.Fatalf("register input: %v", err)
 	}
 	if err := catalog.Chunkers.RegisterWithSpec(ModuleSpec{Key: "typed/chunk", Stage: StageChunk}, func() (contracts.Chunker, error) {
-		return &runnerChunker{key: "typed/chunk"}, nil
+		doc := validSourceDocument()
+		return &runnerChunker{key: "typed/chunk", chunks: []source.Chunk{{ID: "chunk-1", SourceID: doc.ID, Index: 0, Ref: doc.Units[0].Ref, Content: []byte(`{"chunk":1}`), MediaType: "application/json", Units: []source.SourceUnit{doc.Units[0]}}}}, nil
 	}); err != nil {
 		t.Fatalf("register chunker: %v", err)
 	}
--- a/internal/framework/pipeline/validator_registry.go
+++ b/internal/framework/pipeline/validator_registry.go
@@ -1,6 +1,7 @@
 package pipeline

 import (
+	"context"
 	"fmt"
 	"reflect"
 	"sort"
@@ -47,6 +48,7 @@ type typedValidatorEntry struct {
 	valueType       reflect.Type
 	validateOptions OptionValidator
 	builder         func(BuildRequest) (any, error)
+	validate        typedValidateOperation
 }

 type chunkValidatorEntry struct {
@@ -159,6 +161,17 @@ func RegisterTypedValidatorBuilder[T any](registry *ValidatorRegistry, kind cont
 		builder: func(request BuildRequest) (any, error) {
 			return builder(cloneBuildRequest(request))
 		},
+		validate: func(ctx context.Context, implementation any, target typedValidationTarget) (contracts.ValidationResult, error) {
+			validator, ok := implementation.(contracts.TypedValidator[T])
+			if !ok {
+				return contracts.ValidationResult{}, fmt.Errorf("validator %q has incompatible implementation %T", normalizedSpec.Key, implementation)
+			}
+			value, err := exactTypedValue[T]("validate artifact value", target.value)
+			if err != nil {
+				return contracts.ValidationResult{}, err
+			}
+			return validator.Validate(ctx, contracts.TypedValidationRequest[T]{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Source: target.source, SourceID: target.sourceID, SourceInput: target.sourceInput, SessionID: target.sessionID, References: target.references, LLMProfile: target.llmProfile, Metadata: target.metadata, Chunk: target.chunk, Chunks: target.chunks, Ref: target.ref, Value: value})
+		},
 	}
 	return nil
 }
@@ -302,6 +315,20 @@ func (r *ValidatorRegistry) Spec(key string) (ValidatorSpec, bool) {
 		return ValidatorSpec{}, false
 	}
 	spec, ok := r.legacySpecs[strings.TrimSpace(key)]
+	if ok {
+		return spec, true
+	}
+	normalized := strings.TrimSpace(key)
+	if entry, found := r.chunkEntries[normalized]; found {
+		return entry.spec, true
+	}
+	if entry, found := r.serializedEntries[normalized]; found {
+		return entry.spec.ValidatorSpec, true
+	}
+	if kinds := r.registeredTypedKinds(normalized); len(kinds) > 0 {
+		entry, found := r.typedEntry(normalized, kinds[0])
+		return entry.spec, found
+	}
 	return spec, ok
 }