diff --git a/docs/internal/dnd.md b/docs/internal/dnd.md index 53f79be9..af53933a 100644 --- a/docs/internal/dnd.md +++ b/docs/internal/dnd.md @@ -33,7 +33,9 @@ typed builder. Scene chunking, every extractor, and NPC, location, and item-regi normalization are registered as `llm_backed`; the remaining current D&D mergers and normalizers are `deterministic`. The metadata is available to catalog inspection and resolved-pipeline debug data and determines which selected bindings inherit the -pipeline profile. Configuration remains the canonical owner of the exact keys, +pipeline profile. The registry normalizers use `single_response_v1`, forwarding +corrections to their reconciliation completion and retaining the accepted raw +proposal only as an owned model candidate. Configuration remains the canonical owner of the exact keys, profile precedence, and validator order. Private structured-LLM response schemas are deliberately minimal. They reject diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index e4b9f3e6..2664d29e 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -35,6 +35,9 @@ module capabilities and typed artifact compatibility, validates options, and assigns a deterministic resolved-composition digest. A correction protocol is selected from each eligible LLM-backed producer specification and becomes part of that resolved identity; only `single_response_v1` is currently supported. +Preparation rejects an LLM-backed producer that combines a non-empty validator +chain with positive producer retries unless it declares that protocol. Producers +without validators or without retries remain valid without correction support. The resolved pipeline contains bindings and declared reference targets, not external reference bytes. After selection, the resolver applies command, binding, and pipeline profile diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 796d20e5..05f2c5e5 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -379,7 +379,7 @@ producers. Every direct production D&D LLM extractor implements the same correction protocol without changing artifact semantics. This stage is one Terra prompt. -## Stage 8 — Migrate Semantic Reconciliation Normalizers +## Stage 8 — Migrate Semantic Reconciliation Normalizers ✅ ### Goal diff --git a/internal/framework/pipeline/prepare.go b/internal/framework/pipeline/prepare.go index 31c39705..dcd02bcd 100644 --- a/internal/framework/pipeline/prepare.go +++ b/internal/framework/pipeline/prepare.go @@ -91,6 +91,9 @@ func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDepend if err := validateResolvedPipeline(resolved); err != nil { return nil, err } + if err := validateCorrectionRetryCapabilities(resolved); err != nil { + return nil, err + } if err := validateRegistrySet(resolved, registries); err != nil { return nil, err } @@ -209,6 +212,40 @@ func prepareEvidencePlan(resolved ResolvedPipeline, registries Registries, outpu return plan, nil } +func validateCorrectionRetryCapabilities(pipeline ResolvedPipeline) error { + validate := func(stage ModuleStage, laneID string, binding ModuleBinding, executionClass contracts.ExecutionClass, protocol contracts.CorrectionProtocol) error { + if executionClass != contracts.ExecutionClassLLMBacked || binding.Retries == 0 { + return nil + } + chain := resolvedValidatorChain(stage, laneID, binding.Module, pipeline.ValidatorChains) + if len(chain.Validators) == 0 || protocol == contracts.CorrectionProtocolSingleResponseV1 { + return nil + } + if laneID == "" { + return fmt.Errorf("pipeline %q %s module %q configures validators and retries but does not declare correction protocol %q", pipeline.ID, stage, binding.Module, contracts.CorrectionProtocolSingleResponseV1) + } + return fmt.Errorf("pipeline %q lane %q %s module %q configures validators and retries but does not declare correction protocol %q", pipeline.ID, laneID, stage, binding.Module, contracts.CorrectionProtocolSingleResponseV1) + } + + if err := validate(StageChunk, "", pipeline.Chunk, pipeline.ChunkExecutionClass, pipeline.ChunkCorrectionProtocol); err != nil { + return err + } + for _, step := range pipeline.Steps { + for _, lane := range step.ArtifactLanes { + if err := validate(StageExtract, lane.ID, lane.Extract, lane.ExtractExecutionClass, lane.ExtractCorrectionProtocol); err != nil { + return err + } + if err := validate(StageMerge, lane.ID, lane.Merge, lane.MergeExecutionClass, lane.MergeCorrectionProtocol); err != nil { + return err + } + if err := validate(StageNormalize, lane.ID, lane.Normalize, lane.NormalizeExecutionClass, lane.NormalizeCorrectionProtocol); err != nil { + return err + } + } + } + return nil +} + func prepareLane(pipeline ResolvedPipeline, lane ResolvedArtifactLane, registries Registries, deps ModuleDependencies) (preparedLaneExecutor, error) { executor := preparedLaneExecutor{resolved: cloneResolvedArtifactLane(lane)} request := func(binding ModuleBinding, references contracts.ReferenceSet) BuildRequest { diff --git a/internal/framework/pipeline/profile_test.go b/internal/framework/pipeline/profile_test.go index 435ba505..000531a7 100644 --- a/internal/framework/pipeline/profile_test.go +++ b/internal/framework/pipeline/profile_test.go @@ -173,29 +173,68 @@ func TestResolvePipelineCarriesCorrectionProtocolsIntoPreparedMetadata(t *testin } } -func TestPrepareAllowsLLMProducerWithoutCorrectionCapability(t *testing.T) { - catalog := newProfileCatalogWithOverrides(t, - ModuleSpec{Key: "llm-input", Stage: StageInput, ExecutionClass: contracts.ExecutionClassLLMBacked, Provides: []string{"source"}}, - ModuleSpec{Key: "llm-chunk", Stage: StageChunk, ExecutionClass: contracts.ExecutionClassLLMBacked, Requires: []string{"source"}, Provides: []string{"chunk"}}, - ModuleSpec{Key: "llm-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassLLMBacked, ArtifactKind: "test/notes", Requires: []string{"chunk"}, Provides: []string{"candidate"}}, - ModuleSpec{Key: "llm-merge", Stage: StageMerge, ExecutionClass: contracts.ExecutionClassLLMBacked, ArtifactKind: "test/notes", Requires: []string{"candidate"}, Provides: []string{"merged"}}, - ModuleSpec{Key: "llm-normalize", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, ArtifactKind: "test/notes", Requires: []string{"merged"}, Provides: []string{"normalized"}}, - ModuleSpec{Key: "llm-output", Stage: StageOutput, ExecutionClass: contracts.ExecutionClassLLMBacked, Requires: []string{"normalized"}, Provides: []string{"encoded"}}, - ) - profile := llmProfilePipeline() - profile.Chunk.Retries = 1 - resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) - if err != nil { - t.Fatalf("ResolvePipeline() error = %v, want nil", err) - } - if resolved.ChunkCorrectionProtocol != "" { - t.Fatalf("ChunkCorrectionProtocol = %q, want empty unsupported value", resolved.ChunkCorrectionProtocol) - } - if _, err := Prepare(resolved, registriesFromModuleCatalog(catalog), ModuleDependencies{}); err != nil { - t.Fatalf("Prepare() error = %v, want nil", err) +func TestPrepareRequiresCorrectionCapabilityForValidatorRetries(t *testing.T) { + for _, test := range []struct { + name string + protocol contracts.CorrectionProtocol + retries int + validators bool + want string + }{ + {name: "supported", protocol: contracts.CorrectionProtocolSingleResponseV1, retries: 1, validators: true}, + {name: "unsupported", retries: 1, validators: true, want: "does not declare correction protocol"}, + {name: "no validators", retries: 1}, + {name: "no retries", validators: true}, + } { + t.Run(test.name, func(t *testing.T) { + catalog := newProfileCatalogWithOverrides(t, + ModuleSpec{Key: "llm-input", Stage: StageInput, ExecutionClass: contracts.ExecutionClassLLMBacked, Provides: []string{"source"}}, + ModuleSpec{Key: "llm-chunk", Stage: StageChunk, ExecutionClass: contracts.ExecutionClassLLMBacked, CorrectionProtocol: test.protocol, Requires: []string{"source"}, Provides: []string{"chunk"}}, + ModuleSpec{Key: "llm-extractor", Stage: StageExtract, ExecutionClass: contracts.ExecutionClassLLMBacked, ArtifactKind: "test/notes", Requires: []string{"chunk"}, Provides: []string{"candidate"}}, + ModuleSpec{Key: "llm-merge", Stage: StageMerge, ExecutionClass: contracts.ExecutionClassLLMBacked, ArtifactKind: "test/notes", Requires: []string{"candidate"}, Provides: []string{"merged"}}, + ModuleSpec{Key: "llm-normalize", Stage: StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, ArtifactKind: "test/notes", Requires: []string{"merged"}, Provides: []string{"normalized"}}, + ModuleSpec{Key: "llm-output", Stage: StageOutput, ExecutionClass: contracts.ExecutionClassLLMBacked, Requires: []string{"normalized"}, Provides: []string{"encoded"}}, + ) + if err := RegisterChunkValidator(catalog.Validators, ValidatorSpec{Key: "chunk-validator", ExecutionClass: contracts.ExecutionClassLLMBacked}, func() (contracts.ChunkValidator, error) { + return llmProfileTestChunkValidator{key: "chunk-validator"}, nil + }); err != nil { + t.Fatal(err) + } + profile := llmProfilePipeline() + profile.Chunk.Retries = test.retries + if test.validators { + profile.Chunk.Validators = ValidatorOverride{Set: true, Validators: []ModuleBinding{{Module: "chunk-validator"}}} + } + resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + _, err = Prepare(resolved, registriesFromModuleCatalog(catalog), ModuleDependencies{}) + if test.want != "" { + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("Prepare() error = %v, want %q", err, test.want) + } + return + } + if err != nil { + t.Fatalf("Prepare() error = %v, want nil", err) + } + }) } } +type llmProfileTestChunkValidator struct{ key string } + +func (validator llmProfileTestChunkValidator) Name() string { return validator.key } + +func (llmProfileTestChunkValidator) ExecutionClass() contracts.ExecutionClass { + return contracts.ExecutionClassLLMBacked +} + +func (llmProfileTestChunkValidator) Validate(context.Context, contracts.ChunkValidationRequest) (contracts.ValidationResult, error) { + return contracts.ValidationResult{Approved: true}, nil +} + func TestResolvePipelineValidatorRetriesRequireLLMBackedValidator(t *testing.T) { for _, test := range []struct { name string diff --git a/internal/framework/semanticreconcile/engine.go b/internal/framework/semanticreconcile/engine.go index 68bef3d5..0134c1a0 100644 --- a/internal/framework/semanticreconcile/engine.go +++ b/internal/framework/semanticreconcile/engine.go @@ -49,6 +49,7 @@ type Request struct { ProfileID string StructuredOutputRepairAttempts *int SessionID string + Correction *contracts.SemanticCorrection } // ResultDisposition classifies a provider-neutral reconciliation outcome. @@ -69,6 +70,7 @@ type Result struct { issues []Issue discardedGroupCount int candidateMappings []CandidateMapping + modelCandidate *contracts.ModelCandidate } // Disposition returns the classified outcome. @@ -88,6 +90,16 @@ func (result Result) CandidateMappings() []CandidateMapping { return append([]CandidateMapping(nil), result.candidateMappings...) } +// ModelCandidate returns an owned copy of the proposal response when a model +// completion produced this result. +func (result Result) ModelCandidate() *contracts.ModelCandidate { + candidate, err := contracts.CloneModelCandidate(result.modelCandidate) + if err != nil { + return nil + } + return candidate +} + func (result Result) planCopy() Plan { return Plan{groups: result.plan.Groups()} } @@ -162,8 +174,12 @@ func (engine *Engine) Reconcile(ctx context.Context, request Request) (Result, e if err := ctx.Err(); err != nil { return Result{}, fmt.Errorf("semantic reconciliation %q: context error before completion: %w", request.StageName, err) } + correction, err := contracts.CloneSemanticCorrection(request.Correction) + if err != nil { + return Result{}, fmt.Errorf("semantic reconciliation %q: clone correction: %w", request.StageName, err) + } var response ProposalResponse - _, err = engine.client.CompleteStructured(ctx, contracts.StructuredCompletionRequest{ + completion, err := engine.client.CompleteStructured(ctx, contracts.StructuredCompletionRequest{ StageName: request.StageName, PromptID: engine.prompt.ID, PromptVersion: engine.prompt.Version, @@ -171,6 +187,7 @@ func (engine *Engine) Reconcile(ctx context.Context, request Request) (Result, e SessionID: request.SessionID, StructuredOutputRepairAttempts: request.StructuredOutputRepairAttempts, Inputs: preparation.Materials(), + Correction: correction, }, &response) if err != nil { if errors.Is(err, contracts.ErrInvalidStructuredOutput) { @@ -179,6 +196,11 @@ func (engine *Engine) Reconcile(ctx context.Context, request Request) (Result, e } return Result{}, fmt.Errorf("semantic reconciliation %q: complete structured output: %w", request.StageName, err) } + candidate, err := contracts.NewModelCandidate(completion.Content, contracts.CorrectionProtocolSingleResponseV1) + if err != nil { + return Result{}, fmt.Errorf("semantic reconciliation %q: own model candidate: %w", request.StageName, err) + } + result.modelCandidate = candidate assessment := preparation.Assess(response) result.plan = assessment.Plan() diff --git a/internal/framework/semanticreconcile/engine_test.go b/internal/framework/semanticreconcile/engine_test.go index a138f40b..884cd840 100644 --- a/internal/framework/semanticreconcile/engine_test.go +++ b/internal/framework/semanticreconcile/engine_test.go @@ -56,6 +56,11 @@ func TestEnginePropagatesRequestAndAssessesResponse(t *testing.T) { request := readyEngineRequest() request.ProfileID = " profile-as-resolved " request.SessionID = " session-as-supplied " + correction, err := contracts.NewSemanticCorrection([]byte(`{"duplicate_groups":[]}`), "retain distinct candidates") + if err != nil { + t.Fatal(err) + } + request.Correction = correction result, err := engine.Reconcile(context.Background(), request) if err != nil { @@ -75,24 +80,32 @@ func TestEnginePropagatesRequestAndAssessesResponse(t *testing.T) { if got.StageName != request.StageName || got.PromptID != engine.prompt.ID || got.PromptVersion != engine.prompt.Version || got.ProfileID != request.ProfileID || got.SessionID != request.SessionID { t.Fatalf("structured request = %#v, want exact routing values", got) } + if !reflect.DeepEqual(got.Correction, correction) { + t.Fatalf("structured request correction = %#v, want %#v", got.Correction, correction) + } if len(got.Inputs) != 2 || got.Inputs["candidates"].Name != "candidates" || got.Inputs["transcript"].Name != "transcript" || len(got.Vars) != 0 { t.Fatalf("structured request inputs = %#v vars = %#v, want only candidate and transcript materials", got.Inputs, got.Vars) } + candidate := result.ModelCandidate() + if candidate == nil || candidate.Protocol != contracts.CorrectionProtocolSingleResponseV1 || string(candidate.Response) != `{"duplicate_groups":[]}` { + t.Fatalf("model candidate = %#v, want exact owned completion response", candidate) + } } func TestEngineClassifiesSemanticAndTransportOutcomes(t *testing.T) { transportErr := errors.New("provider unavailable") tests := []struct { - name string - response ProposalResponse - completion error - want ResultDisposition - wantDiscard int - wantIssues bool - wantError error + name string + response ProposalResponse + completion error + want ResultDisposition + wantDiscard int + wantIssues bool + wantCandidate bool + wantError error }{ - {name: "empty groups complete", response: ProposalResponse{DuplicateGroups: []DuplicateGroup{}}, want: Complete}, - {name: "discarded proposal retryable", response: ProposalResponse{DuplicateGroups: []DuplicateGroup{{CandidateIDs: []int{1, 99}, CanonicalCandidateID: 1}}}, want: RetryableDiscardedProposalGroups, wantDiscard: 1, wantIssues: true}, + {name: "empty groups complete", response: ProposalResponse{DuplicateGroups: []DuplicateGroup{}}, want: Complete, wantCandidate: true}, + {name: "discarded proposal retryable", response: ProposalResponse{DuplicateGroups: []DuplicateGroup{{CandidateIDs: []int{1, 99}, CanonicalCandidateID: 1}}}, want: RetryableDiscardedProposalGroups, wantDiscard: 1, wantIssues: true, wantCandidate: true}, {name: "invalid structured output retryable", completion: fmt.Errorf("decode response: %w", contracts.ErrInvalidStructuredOutput), want: RetryableInvalidStructuredOutput}, {name: "transport failure", completion: transportErr, wantError: transportErr}, } @@ -112,6 +125,9 @@ func TestEngineClassifiesSemanticAndTransportOutcomes(t *testing.T) { if result.Disposition() != test.want || result.DiscardedGroupCount() != test.wantDiscard || (len(result.Issues()) > 0) != test.wantIssues { t.Fatalf("result = disposition %v discarded %d issues %#v", result.Disposition(), result.DiscardedGroupCount(), result.Issues()) } + if (result.ModelCandidate() != nil) != test.wantCandidate { + t.Fatalf("model candidate = %#v, want presence %t", result.ModelCandidate(), test.wantCandidate) + } if len(client.requests) != 1 { t.Fatalf("completion calls = %d, want one", len(client.requests)) } @@ -141,6 +157,9 @@ func TestEngineSkipsDeterministicOutcomesWithoutCompletion(t *testing.T) { if result.Disposition() != test.want || len(result.CandidateMappings()) != test.mappingLen { t.Fatalf("result disposition = %v mappings = %#v", result.Disposition(), result.CandidateMappings()) } + if result.ModelCandidate() != nil { + t.Fatalf("model candidate = %#v, want nil for no-call outcome", result.ModelCandidate()) + } if len(client.requests) != 0 { t.Fatalf("completion calls = %d, want zero", len(client.requests)) } @@ -209,7 +228,9 @@ func TestEngineCallsAreIndependentAndResultsAreOwned(t *testing.T) { firstIssues[0].Category = "changed" firstMappings := first.CandidateMappings() firstMappings[0].CandidatePosition = 99 - if first.Plan().Groups()[0].MemberPositions()[0] != 0 || first.Issues()[0].Category == "changed" || first.CandidateMappings()[0].CandidatePosition != 0 { + firstCandidate := first.ModelCandidate() + firstCandidate.Response[0] = 'x' + if first.Plan().Groups()[0].MemberPositions()[0] != 0 || first.Issues()[0].Category == "changed" || first.CandidateMappings()[0].CandidatePosition != 0 || string(first.ModelCandidate().Response) != `{"duplicate_groups":[]}` { t.Fatal("result accessors exposed retained state") } @@ -217,7 +238,7 @@ func TestEngineCallsAreIndependentAndResultsAreOwned(t *testing.T) { if err != nil { t.Fatal(err) } - if second.Disposition() != Complete || len(second.Plan().Groups()) != 0 || len(second.Issues()) != 0 || second.DiscardedGroupCount() != 0 || len(second.CandidateMappings()) != 2 { + if second.Disposition() != Complete || len(second.Plan().Groups()) != 0 || len(second.Issues()) != 0 || second.DiscardedGroupCount() != 0 || len(second.CandidateMappings()) != 2 || second.ModelCandidate() == nil { t.Fatalf("second result retained prior call state: disposition %v plan %#v issues %#v discarded %d mappings %#v", second.Disposition(), second.Plan().Groups(), second.Issues(), second.DiscardedGroupCount(), second.CandidateMappings()) } } @@ -226,6 +247,7 @@ type recordingReconciliationClient struct { requests []contracts.StructuredCompletionRequest responses []ProposalResponse errors []error + content []byte } func (client *recordingReconciliationClient) CompleteStructured(_ context.Context, request contracts.StructuredCompletionRequest, output any) (contracts.StructuredCompletionResponse, error) { @@ -247,7 +269,11 @@ func (client *recordingReconciliationClient) CompleteStructured(_ context.Contex return contracts.StructuredCompletionResponse{}, fmt.Errorf("output type = %T", output) } *target = response - return contracts.StructuredCompletionResponse{}, nil + content := append([]byte(nil), client.content...) + if len(content) == 0 { + content = []byte(`{"duplicate_groups":[]}`) + } + return contracts.StructuredCompletionResponse{Content: content}, nil } func cloneProposalResponse(response ProposalResponse) ProposalResponse { diff --git a/internal/modules/dnd/normalize/itemregistry/normalizer.go b/internal/modules/dnd/normalize/itemregistry/normalizer.go index 86ac4bc8..1f21d02d 100644 --- a/internal/modules/dnd/normalize/itemregistry/normalizer.go +++ b/internal/modules/dnd/normalize/itemregistry/normalizer.go @@ -119,7 +119,7 @@ func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalize } reconciliation, err := n.engine.Reconcile(ctx, semanticreconcile.Request{ StageName: Key, Source: req.Source, Candidates: candidates, - ProfileID: req.LLMProfile, SessionID: req.SessionID, StructuredOutputRepairAttempts: req.StructuredOutputRepairAttempts, + ProfileID: req.LLMProfile, SessionID: req.SessionID, StructuredOutputRepairAttempts: req.StructuredOutputRepairAttempts, Correction: req.Correction, }) if err != nil { return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("reconcile semantic duplicates: %w", err) @@ -144,7 +144,7 @@ func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalize warnings = append(warnings, semanticWarnings...) discardedGroups := reconciliation.DiscardedGroupCount() + rejectedGroups if discardedGroups == 0 { - return contracts.TypedNormalizeResult[dnd.ItemRegistry]{Value: recordList(applied), Warnings: limitWarnings(warnings)}, nil + return contracts.TypedNormalizeResult[dnd.ItemRegistry]{Value: recordList(applied), Warnings: limitWarnings(warnings), ModelCandidate: reconciliation.ModelCandidate()}, nil } return retryResult(recordList(applied), warnings, reconciliation, rejectedGroups), nil } @@ -162,7 +162,7 @@ func retryResult(value dnd.ItemRegistry, warnings []contracts.Warning, reconcili details = append(details, "currency may only be consolidated with aliases of one denomination") } discardedGroups := reconciliation.DiscardedGroupCount() + rejectedGroups - return contracts.TypedNormalizeResult[dnd.ItemRegistry]{Value: value, Warnings: limitWarningsForRetry(warnings), Retry: &contracts.NormalizeRetry{ + return contracts.TypedNormalizeResult[dnd.ItemRegistry]{Value: value, Warnings: limitWarningsForRetry(warnings), ModelCandidate: reconciliation.ModelCandidate(), Retry: &contracts.NormalizeRetry{ ReasonCode: ReasonCodeItemSemanticProposalInvalid, Message: diagnostics.Aggregate("semantic proposal requires retry", details), FallbackWarnings: []contracts.Warning{semanticFallbackWarning(discardedGroups)}, }} @@ -334,7 +334,7 @@ func duplicateWarning(retainedIndex int, removed []int) contracts.Warning { func itemScope(index int) string { return fmt.Sprintf("items[%d]", index) } func ModuleSpec() pipeline.ModuleSpec { - return pipeline.ModuleSpec{Key: Key, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, Requires: append([]string(nil), requiredCapabilities...), Provides: append([]string(nil), providedCapabilities...), ArtifactKind: dnd.ItemRegistryKind} + return pipeline.ModuleSpec{Key: Key, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, CorrectionProtocol: contracts.CorrectionProtocolSingleResponseV1, Requires: append([]string(nil), requiredCapabilities...), Provides: append([]string(nil), providedCapabilities...), ArtifactKind: dnd.ItemRegistryKind} } func Register(registry *pipeline.NormalizerRegistry) error { return pipeline.RegisterNormalizerBuilder(registry, ModuleSpec(), validateOptions, func(request pipeline.BuildRequest) (contracts.Normalizer[dnd.ItemRegistry], error) { diff --git a/internal/modules/dnd/normalize/itemregistry/normalizer_test.go b/internal/modules/dnd/normalize/itemregistry/normalizer_test.go index f3306adf..28f1fa81 100644 --- a/internal/modules/dnd/normalize/itemregistry/normalizer_test.go +++ b/internal/modules/dnd/normalize/itemregistry/normalizer_test.go @@ -24,7 +24,7 @@ import ( ) func TestModuleContractAndMetadata(t *testing.T) { - want := pipeline.ModuleSpec{Key: Key, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, Requires: []string{"merged"}, Provides: []string{"normalized"}, ArtifactKind: dnd.ItemRegistryKind} + want := pipeline.ModuleSpec{Key: Key, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, CorrectionProtocol: contracts.CorrectionProtocolSingleResponseV1, Requires: []string{"merged"}, Provides: []string{"normalized"}, ArtifactKind: dnd.ItemRegistryKind} if got := ModuleSpec(); !reflect.DeepEqual(got, want) { t.Fatalf("ModuleSpec() = %#v, want %#v", got, want) } diff --git a/internal/modules/dnd/normalize/locationregistry/normalizer.go b/internal/modules/dnd/normalize/locationregistry/normalizer.go index 83c547c6..2c7f1762 100644 --- a/internal/modules/dnd/normalize/locationregistry/normalizer.go +++ b/internal/modules/dnd/normalize/locationregistry/normalizer.go @@ -120,7 +120,7 @@ func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalize } reconciliation, err := n.engine.Reconcile(ctx, semanticreconcile.Request{ StageName: Key, Source: req.Source, Candidates: candidates, - ProfileID: req.LLMProfile, SessionID: req.SessionID, StructuredOutputRepairAttempts: req.StructuredOutputRepairAttempts, + ProfileID: req.LLMProfile, SessionID: req.SessionID, StructuredOutputRepairAttempts: req.StructuredOutputRepairAttempts, Correction: req.Correction, }) if err != nil { return contracts.TypedNormalizeResult[dnd.LocationRegistry]{}, normalizerErrorf("reconcile semantic duplicates: %w", err) @@ -144,7 +144,7 @@ func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalize } warnings = append(warnings, semanticWarnings...) if reconciliation.Disposition() == semanticreconcile.Complete { - return contracts.TypedNormalizeResult[dnd.LocationRegistry]{Value: recordList(applied), Warnings: limitWarnings(warnings)}, nil + return contracts.TypedNormalizeResult[dnd.LocationRegistry]{Value: recordList(applied), Warnings: limitWarnings(warnings), ModelCandidate: reconciliation.ModelCandidate()}, nil } return retryResult(recordList(applied), warnings, reconciliation), nil } @@ -157,7 +157,7 @@ func (n *Normalizer) invalidStructuredResult(value dnd.LocationRegistry, warning } func retryResult(value dnd.LocationRegistry, warnings []contracts.Warning, reconciliation semanticreconcile.Result) contracts.TypedNormalizeResult[dnd.LocationRegistry] { - return contracts.TypedNormalizeResult[dnd.LocationRegistry]{Value: value, Warnings: limitWarningsForRetry(warnings), Retry: &contracts.NormalizeRetry{ + return contracts.TypedNormalizeResult[dnd.LocationRegistry]{Value: value, Warnings: limitWarningsForRetry(warnings), ModelCandidate: reconciliation.ModelCandidate(), Retry: &contracts.NormalizeRetry{ ReasonCode: ReasonCodeLocationSemanticProposalInvalid, Message: diagnostics.Aggregate("semantic proposal requires retry", semanticreconcile.IssueDetails(reconciliation.Issues())), FallbackWarnings: []contracts.Warning{semanticFallbackWarning(reconciliation.DiscardedGroupCount())}, }} @@ -354,7 +354,7 @@ func duplicateWarning(retainedIndex int, removed []int) contracts.Warning { func locationScope(index int) string { return fmt.Sprintf("locations[%d]", index) } func ModuleSpec() pipeline.ModuleSpec { - return pipeline.ModuleSpec{Key: Key, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, Requires: append([]string(nil), requiredCapabilities...), Provides: append([]string(nil), providedCapabilities...), ArtifactKind: dnd.LocationRegistryKind} + return pipeline.ModuleSpec{Key: Key, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, CorrectionProtocol: contracts.CorrectionProtocolSingleResponseV1, Requires: append([]string(nil), requiredCapabilities...), Provides: append([]string(nil), providedCapabilities...), ArtifactKind: dnd.LocationRegistryKind} } func Register(registry *pipeline.NormalizerRegistry) error { return pipeline.RegisterNormalizerBuilder(registry, ModuleSpec(), validateOptions, func(request pipeline.BuildRequest) (contracts.Normalizer[dnd.LocationRegistry], error) { diff --git a/internal/modules/dnd/normalize/locationregistry/normalizer_test.go b/internal/modules/dnd/normalize/locationregistry/normalizer_test.go index 4b66a17d..c002a59c 100644 --- a/internal/modules/dnd/normalize/locationregistry/normalizer_test.go +++ b/internal/modules/dnd/normalize/locationregistry/normalizer_test.go @@ -20,7 +20,7 @@ import ( ) func TestModuleContractAndMetadata(t *testing.T) { - want := pipeline.ModuleSpec{Key: Key, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, Requires: []string{"merged"}, Provides: []string{"normalized"}, ArtifactKind: dnd.LocationRegistryKind} + want := pipeline.ModuleSpec{Key: Key, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, CorrectionProtocol: contracts.CorrectionProtocolSingleResponseV1, Requires: []string{"merged"}, Provides: []string{"normalized"}, ArtifactKind: dnd.LocationRegistryKind} if got := ModuleSpec(); !reflect.DeepEqual(got, want) { t.Fatalf("ModuleSpec() = %#v, want %#v", got, want) } diff --git a/internal/modules/dnd/normalize/npcregistry/normalizer.go b/internal/modules/dnd/normalize/npcregistry/normalizer.go index 56e40fa3..d9a29bd1 100644 --- a/internal/modules/dnd/normalize/npcregistry/normalizer.go +++ b/internal/modules/dnd/normalize/npcregistry/normalizer.go @@ -119,7 +119,7 @@ func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalize } reconciliation, err := n.engine.Reconcile(ctx, semanticreconcile.Request{ StageName: Key, Source: req.Source, Candidates: candidates, - ProfileID: req.LLMProfile, SessionID: req.SessionID, StructuredOutputRepairAttempts: req.StructuredOutputRepairAttempts, + ProfileID: req.LLMProfile, SessionID: req.SessionID, StructuredOutputRepairAttempts: req.StructuredOutputRepairAttempts, Correction: req.Correction, }) if err != nil { return contracts.TypedNormalizeResult[dnd.NPCRegistry]{}, normalizerErrorf("reconcile semantic duplicates: %w", err) @@ -143,7 +143,7 @@ func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalize } warnings = append(warnings, semanticWarnings...) if reconciliation.Disposition() == semanticreconcile.Complete { - return contracts.TypedNormalizeResult[dnd.NPCRegistry]{Value: recordList(applied), Warnings: limitWarnings(warnings)}, nil + return contracts.TypedNormalizeResult[dnd.NPCRegistry]{Value: recordList(applied), Warnings: limitWarnings(warnings), ModelCandidate: reconciliation.ModelCandidate()}, nil } return retryResult(recordList(applied), warnings, reconciliation), nil } @@ -162,8 +162,9 @@ func (n *Normalizer) invalidStructuredResult(value dnd.NPCRegistry, warnings []c func retryResult(value dnd.NPCRegistry, warnings []contracts.Warning, reconciliation semanticreconcile.Result) contracts.TypedNormalizeResult[dnd.NPCRegistry] { return contracts.TypedNormalizeResult[dnd.NPCRegistry]{ - Value: value, - Warnings: limitWarningsForRetry(warnings), + Value: value, + Warnings: limitWarningsForRetry(warnings), + ModelCandidate: reconciliation.ModelCandidate(), Retry: &contracts.NormalizeRetry{ ReasonCode: ReasonCodeNPCSemanticProposalInvalid, Message: diagnostics.Aggregate("semantic proposal requires retry", semanticreconcile.IssueDetails(reconciliation.Issues())), @@ -366,7 +367,7 @@ func duplicateWarning(retainedIndex int, removed []int) contracts.Warning { func npcScope(index int) string { return fmt.Sprintf("npcs[%d]", index) } func ModuleSpec() pipeline.ModuleSpec { - return pipeline.ModuleSpec{Key: Key, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, Requires: append([]string(nil), requiredCapabilities...), Provides: append([]string(nil), providedCapabilities...), ArtifactKind: dnd.NPCRegistryKind} + return pipeline.ModuleSpec{Key: Key, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, CorrectionProtocol: contracts.CorrectionProtocolSingleResponseV1, Requires: append([]string(nil), requiredCapabilities...), Provides: append([]string(nil), providedCapabilities...), ArtifactKind: dnd.NPCRegistryKind} } func Register(registry *pipeline.NormalizerRegistry) error { diff --git a/internal/modules/dnd/normalize/npcregistry/normalizer_test.go b/internal/modules/dnd/normalize/npcregistry/normalizer_test.go index 4ae86258..002fffe5 100644 --- a/internal/modules/dnd/normalize/npcregistry/normalizer_test.go +++ b/internal/modules/dnd/normalize/npcregistry/normalizer_test.go @@ -23,7 +23,7 @@ func TestModuleContractAndIdentity(t *testing.T) { if _, err := DecodeOptions(map[string]any{"unexpected": true}); err == nil { t.Fatal("DecodeOptions() accepted unknown option") } - want := pipeline.ModuleSpec{Key: Key, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, Requires: []string{"merged"}, Provides: []string{"normalized"}, ArtifactKind: dnd.NPCRegistryKind} + want := pipeline.ModuleSpec{Key: Key, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, CorrectionProtocol: contracts.CorrectionProtocolSingleResponseV1, Requires: []string{"merged"}, Provides: []string{"normalized"}, ArtifactKind: dnd.NPCRegistryKind} if got := ModuleSpec(); !reflect.DeepEqual(got, want) { t.Fatalf("ModuleSpec() = %#v, want %#v", got, want) } diff --git a/internal/modules/dnd/normalize/npcregistry/semantic_normalizer_test.go b/internal/modules/dnd/normalize/npcregistry/semantic_normalizer_test.go index ca90d77d..bb1df7ce 100644 --- a/internal/modules/dnd/normalize/npcregistry/semantic_normalizer_test.go +++ b/internal/modules/dnd/normalize/npcregistry/semantic_normalizer_test.go @@ -27,6 +27,9 @@ func TestNormalizeSkipsSemanticCompletionWithoutTwoEligibleCandidates(t *testing if err != nil || len(client.requests) != 0 || result.Retry != nil { t.Fatalf("Normalize() = %#v, %v; calls = %d, want deterministic no-call result", result, err, len(client.requests)) } + if result.ModelCandidate != nil { + t.Fatalf("model candidate = %#v, want nil for no-call result", result.ModelCandidate) + } if result.Value.NPCs[0].Name != "Mira Thorn" { t.Fatalf("NPCs = %#v, want deterministic record", result.Value.NPCs) } @@ -45,6 +48,11 @@ func TestNormalizeAppliesSafeProposalAndUsesPrivateInputs(t *testing.T) { request := normalizeRequestWithSource(input, doc) request.LLMProfile = "normalizer-profile" request.SessionID = "normalizer-session" + correction, err := contracts.NewSemanticCorrection([]byte(`{"duplicate_groups":[]}`), "keep the distinct captain") + if err != nil { + t.Fatal(err) + } + request.Correction = correction result, err := normalizer.Normalize(context.Background(), request) if err != nil || result.Retry != nil { t.Fatalf("Normalize() = %#v, %v; want accepted semantic result", result, err) @@ -66,6 +74,12 @@ func TestNormalizeAppliesSafeProposalAndUsesPrivateInputs(t *testing.T) { t.Fatalf("completion calls = %d, want one", len(client.requests)) } completion := client.requests[0] + if !reflect.DeepEqual(completion.Correction, correction) { + t.Fatalf("completion correction = %#v, want %#v", completion.Correction, correction) + } + if result.ModelCandidate == nil || result.ModelCandidate.Protocol != contracts.CorrectionProtocolSingleResponseV1 || string(result.ModelCandidate.Response) != client.response { + t.Fatalf("model candidate = %#v, want exact semantic response", result.ModelCandidate) + } if completion.StageName != Key || completion.PromptID != PromptID || completion.PromptVersion != PromptVersion || completion.ProfileID != request.LLMProfile || completion.SessionID != request.SessionID || len(completion.Inputs) != 2 { t.Fatalf("completion request = %#v, want normalize request identity and exactly two inputs", completion) }