diff --git a/docs/internal/modules.md b/docs/internal/modules.md index 6329094..c324572 100644 --- a/docs/internal/modules.md +++ b/docs/internal/modules.md @@ -20,11 +20,11 @@ validator chains and prompt asset collection. Production input, chunk, output, and D&D spell-extract packages register strict option decoders and run-local builders. Preparation decodes their options into implementation-owned values and injects dependencies. The spell extractor is -typed over the canonical D&D model; a temporary raw adapter preserves the -current runner path. D&D validators, merge, and normalize use typed variants; -JSON representation validators use serialized requests; and unconditional -validators expose separate chunk and typed variants. Narrow legacy -registrations preserve current raw execution until the runner migrates. +typed over the canonical D&D model. D&D validators, merge, and normalize use +typed variants; JSON representation validators use serialized requests; and +unconditional validators expose separate chunk and typed variants. The D&D +production registrar does not register raw spell-stage implementations in +parallel. Specs expose capability and execution metadata without constructing an implementation. Registry entries separately expose option validation and @@ -111,10 +111,10 @@ The extractor owns its private model-response DTO, embedded prompt, LLM response schema, strict option decoder, injected shared LLM client, and prompt/schema manifest metadata. The separate `internal/modules/dnd/codec/spells` package owns the durable schema and stable JSON representation for artifact kind -`dnd/spell-list`. Production composition currently wraps the typed extractor -with a raw adapter that encodes through this codec, so existing validators and -later stages remain unchanged. Shared D&D helpers keep prompt input names and -source-unit reference conversion consistent with the scene chunker. +`dnd/spell-list`. The runner keeps the result typed through validators and later +stages. Explicit migration-only codec adapters preserve the existing raw +checkpoint, debug, and output envelopes. Shared D&D helpers keep prompt input +names and source-unit reference conversion consistent with the scene chunker. The durable payload and manifest metadata shapes are defined in the [D&D spell artifact contract](../integrations/dnd-spell-artifacts.md). diff --git a/docs/internal/overview.md b/docs/internal/overview.md index 452df8f..ef830c8 100644 --- a/docs/internal/overview.md +++ b/docs/internal/overview.md @@ -62,13 +62,13 @@ run-local construction closures. Preparation injects shared dependencies and constructs input, chunk, validators, ordered lanes, and output before source parsing. Production input, chunk, and output modules use strict construction-time option decoding, and the LLM-backed scene chunker retains the injected shared -client. The D&D family registers the canonical `dnd/spell-list` codec and a -typed spell extractor. A temporary raw adapter serializes that typed result for -the current runner. The D&D family also registers typed spell validators and -kind-specific generic merge and normalize strategies; generic JSON validators -use the serialized-validation contract. Narrow legacy registrations preserve -the existing raw runner path until typed execution lands. The current runner -rejects a typed prepared lane instead of routing it through raw execution. +client. The D&D family registers the canonical `dnd/spell-list` codec, typed +spell extractor and validators, and kind-specific generic merge and normalize +strategies; generic JSON validators use the serialized-validation contract. The +runner executes the production D&D lane through private exact-type-checked +closures and uses migration-only codec adapters for the existing raw +checkpoint, debug, and output envelopes. Legacy raw lanes retain their separate +executor while they migrate. ## Production Extensions diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 1faa2d0..dc6d96d 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -73,13 +73,10 @@ separate target namespaces; serialized registrations declare whether they support chunks, artifacts, or both. Duplicate variants and exact Go-type mismatches are rejected deterministically. -Production composition registers the D&D spell-list codec. The typed spell -extractor also registers a temporary raw adapter, which resolution selects -until runner execution is typed. The D&D family registers matching typed merge, -normalize, and semantic-validator variants, while JSON validators register for -serialized chunk and artifact targets. Narrow legacy registrations preserve the -current raw runner path. A standalone raw registration cannot satisfy a typed -lane. +Production composition registers the D&D spell-list codec and typed extractor, +matching typed merge, normalize, and semantic-validator variants, and +serialized JSON validators. The production D&D lane has no parallel raw stage +registration. A standalone raw registration cannot satisfy a typed lane. A `ModuleSpec` declares its stage plus required and provided capabilities. Chunk, extract, merge, and normalize specs may also declare reference slots. @@ -119,10 +116,10 @@ an LLM client; an LLM-backed chunker receives the shared client during preparation. Their operation requests retain run-specific source, reference, profile, session, and metadata context as applicable. -Typed lanes can be composed, resolved, and prepared but are not executed by the -current raw runner. The runner rejects such input before source work; -production lanes are still resolved and executed exclusively through the -legacy raw path. +Prepared typed lanes retain exact-type-checked erased operation closures. The +runner uses those closures to keep each value typed through extraction, +validation, merge, and normalization. Legacy raw lanes continue through their +existing executor while they migrate independently. Source validation requires every unit to carry a canonical self-reference to its containing document and its own unit ID. Explicit clone, checkpoint, and @@ -153,7 +150,7 @@ and validators while performing these transitions: 1. extract once per accepted chunk and add runner-owned lane, source, and chunk provenance; -2. validate each raw extract result and omit rejected results from merge input; +2. validate each extract result and omit rejected results from merge input; 3. skip the rest of the lane when no extract result is accepted; 4. merge accepted extract results in their existing order; 5. validate the merge result and skip normalization on rejection; @@ -180,10 +177,11 @@ chunks; stricter coverage policy belongs to the chunk implementation. ## Validation And Retries Chunk, extract, merge, and normalize results pass through the resolved validator -chain for their stage and module. Each validator receives the raw payload plus -the relevant source, chunk, prior-stage, schema, reference, session, LLM, option, -and run context. Validators execute in resolved order and stop at the first -error or rejection. An empty chain approves the result. +chain for their stage and module. Chunk validators receive canonical chunks; +typed validators receive the domain value; and serialized validators receive +canonical chunk JSON or artifact codec bytes. Validators execute in resolved +order and stop at the first error or rejection. An empty chain approves the +result. `runWithRetry` applies the effective retry policy around module execution and its complete validation chain. A module or validator error becomes a framework @@ -201,8 +199,10 @@ The runner depends on recorder and loader interfaces, using no-op implementations when collaborators are absent. Each checkpointed workflow boundary records a running, succeeded, or failed transition. Reuse decisions are consulted in workflow order and accepted payloads are cloned before -entering the normal handoff path. Dependency fingerprints connect later -checkpoints to the exact accepted results on which they depend. +entering the normal handoff path. The typed D&D executor uses explicit +migration-only codec adapters to read and write the existing raw artifact +checkpoint envelopes. Dependency fingerprints connect later checkpoints to the +exact accepted results on which they depend. Debug instrumentation wraps run, stage, attempt, validator, and structured LLM boundaries. Context scopes associate nested LLM calls with the module or diff --git a/internal/framework/pipeline/artifact_codec_registry.go b/internal/framework/pipeline/artifact_codec_registry.go index 9d1c2dc..45c0075 100644 --- 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) } diff --git a/internal/framework/pipeline/debug.go b/internal/framework/pipeline/debug.go index cd96a72..1a4fc0c 100644 --- 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))) diff --git a/internal/framework/pipeline/extractor_registry.go b/internal/framework/pipeline/extractor_registry.go index 0a735c7..d682cd8 100644 --- 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 { diff --git a/internal/framework/pipeline/merger_registry.go b/internal/framework/pipeline/merger_registry.go index 8435455..26501e6 100644 --- 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 } diff --git a/internal/framework/pipeline/normalizer_registry.go b/internal/framework/pipeline/normalizer_registry.go index 70113c4..5a73628 100644 --- 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 } diff --git a/internal/framework/pipeline/prepare.go b/internal/framework/pipeline/prepare.go index 2359a4a..839789f 100644 --- 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 { diff --git a/internal/framework/pipeline/profile.go b/internal/framework/pipeline/profile.go index 1c45292..af7079d 100644 --- 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)) } diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index e0220c3..a62d07b 100644 --- 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) } diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go new file mode 100644 index 0000000..251f1ef --- /dev/null +++ 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 +} diff --git a/internal/framework/pipeline/typed_execution.go b/internal/framework/pipeline/typed_execution.go new file mode 100644 index 0000000..ea7862d --- /dev/null +++ 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) diff --git a/internal/framework/pipeline/typed_resolution_test.go b/internal/framework/pipeline/typed_resolution_test.go index ec5c731..332a128 100644 --- 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) } diff --git a/internal/framework/pipeline/validator_registry.go b/internal/framework/pipeline/validator_registry.go index 4409a22..8f067dd 100644 --- 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 } diff --git a/internal/modules/dnd/codec/spells/codec.go b/internal/modules/dnd/codec/spells/codec.go index 7654014..6a4470d 100644 --- a/internal/modules/dnd/codec/spells/codec.go +++ b/internal/modules/dnd/codec/spells/codec.go @@ -40,6 +40,10 @@ func (c *Codec) Schema() contracts.ArtifactSchema { func (c *Codec) MediaType() string { return MediaType } +func (c *Codec) Metadata(value dnd.SpellList) map[string]any { + return map[string]any{"spell_cast_count": len(value.SpellCasts)} +} + func (c *Codec) Encode(value dnd.SpellList) ([]byte, error) { if err := validate(value); err != nil { return nil, fmt.Errorf("encode dnd spell list: %w", err) diff --git a/internal/modules/dnd/register/register.go b/internal/modules/dnd/register/register.go index 73045d3..7a5156b 100644 --- a/internal/modules/dnd/register/register.go +++ b/internal/modules/dnd/register/register.go @@ -33,17 +33,14 @@ func Register(registries pipeline.Registries, assets *llm.AssetRegistry) error { }{ {name: "spells codec", register: func() error { return pipeline.RegisterArtifactCodec(registries.ArtifactCodecs, codec) }}, {name: "scenes chunker", register: func() error { return scenes.Register(registries.Chunkers) }}, - {name: "spells extractor", register: func() error { return spells.RegisterWithRawAdapter(registries.Extractors, codec) }}, + {name: "spells extractor", register: func() error { return spells.Register(registries.Extractors) }}, {name: "spell-list appendorder merger", register: func() error { return appendorder.RegisterTyped(registries.Mergers, dnd.SpellListKind, appendSpellLists) }}, {name: "spell-list noop normalizer", register: func() error { return noop.RegisterTyped[dnd.SpellList](registries.Normalizers, dnd.SpellListKind) }}, {name: "spell shape validator", register: func() error { return spellshape.Register(registries.Validators) }}, - {name: "legacy spell shape validator", register: func() error { return spellshape.RegisterLegacy(registries.Validators, codec) }}, {name: "spell source references validator", register: func() error { return spellsourcerefs.Register(registries.Validators) }}, - {name: "legacy spell source references validator", register: func() error { return spellsourcerefs.RegisterLegacy(registries.Validators, codec) }}, {name: "spell source relatedness validator", register: func() error { return spellrelatedness.Register(registries.Validators) }}, - {name: "legacy spell source relatedness validator", register: func() error { return spellrelatedness.RegisterLegacy(registries.Validators, codec) }}, {name: "spell-list always accept validator", register: func() error { return alwaysaccept.RegisterTyped[dnd.SpellList](registries.Validators, dnd.SpellListKind) }},