Migrate NPC reconciliation to shared engine

This commit is contained in:
2026-08-09 16:38:43 +00:00
parent 569e12c6f4
commit 8c071800cf
14 changed files with 270 additions and 304 deletions

View File

@@ -1,3 +0,0 @@
NPC candidates for identity comparison:
{{ input "candidates" }}

View File

@@ -1,6 +1,7 @@
Use candidate aliases and their cited transcript windows to determine whether
candidates refer to the same individual. Preserve distinct individuals even
when their names are similar.
Use the supplied positive integer candidate IDs and their cited transcript
windows to determine whether candidates refer to the same individual. Preserve
distinct individuals even when their names are similar or their contextual
descriptions are identical.
When selecting a canonical display name, prefer a complete, stable proper name
over an abbreviation. Prefer an unadorned proper name over that name plus a

View File

@@ -12,19 +12,19 @@ messages:
- role: system
content_file: ./sharedassets/common-dnd-system.md
- role: user
content_file: ./instructions.md
content_file: ./sharedassets/protocol.md
- role: user
content_file: ./sharedassets/common-dnd-entity-reconciliation.md
content_file: ./instructions.md
cache_control:
type: ephemeral
- role: user
content_file: ./candidates.md
content_file: ./sharedassets/candidates.md
- role: user
content_file: ./sharedassets/common-dnd-transcript-windows.md
content_file: ./sharedassets/transcript-windows.md
cache_control:
type: ephemeral
output:
format: json
validation_mode: json_schema
schema_path: dnd_entity_reconcile_llm.v1.json
schema_path: semantic_reconciliation_llm.v1.json
repair_attempts: 0

View File

@@ -213,7 +213,7 @@ func Prepare(document *source.SourceDocument, candidates []Candidate, limits Lim
}
windows, err := buildContextWindows(document.Units, coalesceIntervals(intervals), cited)
if err != nil {
return Preparation{}, fmt.Errorf("prepare semantic reconciliation: build transcript material: %w", err)
return Preparation{}, fmt.Errorf("prepare semantic reconciliation: build transcript material: invalid source metadata")
}
transcriptContent, err := json.Marshal(transcriptInput{Windows: windows})
if err != nil {

View File

@@ -4,6 +4,7 @@ import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"math"
"reflect"
"strings"
"testing"
@@ -148,6 +149,27 @@ func TestPrepareBuildsContiguousCandidatesAndOwnedSourceContext(t *testing.T) {
}
}
func TestPrepareRedactsInvalidSourceMetadata(t *testing.T) {
const sensitiveKey = "sensitive-metadata-key"
document := &source.SourceDocument{ID: "private-source", Units: []source.SourceUnit{
{ID: 1, Text: "private transcript", Metadata: map[string]any{sensitiveKey: math.NaN()}},
{ID: 2, Text: "other private transcript"},
}}
candidates := []Candidate{
{Label: "Private One", SourceRefs: []source.SourceRef{{SourceID: document.ID, StartUnitID: 1, EndUnitID: 1}}},
{Label: "Private Two", SourceRefs: []source.SourceRef{{SourceID: document.ID, StartUnitID: 2, EndUnitID: 2}}},
}
_, err := Prepare(document, candidates, DefaultLimits())
if err == nil || !strings.Contains(err.Error(), "invalid source metadata") {
t.Fatalf("Prepare() error = %v, want redacted metadata failure", err)
}
for _, forbidden := range []string{sensitiveKey, document.ID, document.Units[0].Text, candidates[0].Label, "non-finite", "float64"} {
if strings.Contains(err.Error(), forbidden) {
t.Fatalf("Prepare() error leaked %q: %v", forbidden, err)
}
}
}
func TestPrepareFiltersUnsafeCandidatesAndCoalescesAdjacentWindows(t *testing.T) {
document := &source.SourceDocument{ID: "session", Units: []source.SourceUnit{
{ID: 9}, {ID: 3}, {ID: 8}, {ID: 1}, {ID: 7},

View File

@@ -3,7 +3,6 @@ package npcregistry
import (
"context"
"errors"
"fmt"
"reflect"
"sort"
@@ -13,19 +12,19 @@ import (
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
"gitea.maximumdirect.net/eric/notarius/internal/framework/semanticreconcile"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/npcs/identity"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared/diagnostics"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared/entityreconcile"
)
const (
Key = "dnd/npc-registry"
PromptID = "dnd.npc_registry.normalize"
normalizationPolicy = "dnd.npc_registry.normalize.v4"
semanticContextPolicy = "dnd.entity_reconcile.context.v1"
NormalizationPolicy = normalizationPolicy
Key = "dnd/npc-registry"
PromptID = "dnd.npc_registry.normalize"
PromptVersion = "v1"
normalizationPolicy = "dnd.npc_registry.normalize.v5"
NormalizationPolicy = normalizationPolicy
ReasonCodeNPCFieldsNormalized = "npc_fields_normalized"
ReasonCodeNPCIDRecomputed = "npc_id_recomputed"
@@ -46,9 +45,7 @@ var _ pipeline.CheckpointFingerprintProvider = (*Normalizer)(nil)
type Options struct{}
type Normalizer struct {
llm contracts.StructuredLLMClient
promptSHA string
responseSchemaSHA string
engine *semanticreconcile.Engine
}
func New(llmClient contracts.StructuredLLMClient, _ Options) (*Normalizer, error) {
@@ -59,55 +56,45 @@ func New(llmClient contracts.StructuredLLMClient, _ Options) (*Normalizer, error
if err != nil {
return nil, normalizerErrorf("load prompt metadata: %w", err)
}
responseSchema, err := entityreconcile.LoadResponseSchema()
engine, err := semanticreconcile.NewEngine(llmClient, semanticreconcile.PromptSpec{
ID: PromptID, Version: PromptVersion, SHA256: promptSHA,
}, semanticreconcile.DefaultLimits())
if err != nil {
return nil, normalizerErrorf("load response schema: %w", err)
return nil, normalizerErrorf("construct semantic reconciliation engine: %w", err)
}
return &Normalizer{llm: llmClient, promptSHA: promptSHA, responseSchemaSHA: responseSchema.SHA256}, nil
return &Normalizer{engine: engine}, nil
}
func (n *Normalizer) Key() string { return Key }
func (n *Normalizer) ReferenceSlots() []contracts.ReferenceSlot { return nil }
func (n *Normalizer) ManifestMetadata() map[string]any {
if n == nil {
if n == nil || n.engine == nil {
return nil
}
return map[string]any{
"prompt_id": PromptID,
"prompt_version": entityreconcile.SchemaVersion,
"prompt_sha256": n.promptSHA,
"response_schema_key": string(entityreconcile.ResponseSchemaKey),
"response_schema_id": entityreconcile.ResponseSchemaID,
"response_schema_name": entityreconcile.ResponseSchemaName,
"response_schema_version": entityreconcile.SchemaVersion,
"response_schema_sha256": n.responseSchemaSHA,
"identity_policy": identity.Policy,
"normalization_policy": normalizationPolicy,
"semantic_context_policy": semanticContextPolicy,
"semantic_context_radius": semanticContextRadius,
}
metadata := n.engine.ManifestMetadata()
metadata["identity_policy"] = identity.Policy
metadata["normalization_policy"] = normalizationPolicy
return metadata
}
func (n *Normalizer) CheckpointFingerprints() []pipeline.CheckpointFingerprint {
if n == nil {
if n == nil || n.engine == nil {
return nil
}
return []pipeline.CheckpointFingerprint{
{Name: "prompt", Value: n.promptSHA},
{Name: "response_schema", Value: n.responseSchemaSHA},
{Name: "identity_policy", Value: identity.Policy},
{Name: "normalization_policy", Value: normalizationPolicy},
{Name: "semantic_context_policy", Value: fmt.Sprintf("%s:%d", semanticContextPolicy, semanticContextRadius)},
}
fingerprints := n.engine.CheckpointFingerprints()
return append(fingerprints,
pipeline.CheckpointFingerprint{Name: "identity_policy", Value: identity.Policy},
pipeline.CheckpointFingerprint{Name: "normalization_policy", Value: normalizationPolicy},
)
}
func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalizeRequest[dnd.NPCRegistry]) (contracts.TypedNormalizeResult[dnd.NPCRegistry], error) {
if n == nil {
return contracts.TypedNormalizeResult[dnd.NPCRegistry]{}, normalizerErrorf("normalizer must not be nil")
}
if n.llm == nil {
return contracts.TypedNormalizeResult[dnd.NPCRegistry]{}, normalizerErrorf("LLM client must not be nil")
if n.engine == nil {
return contracts.TypedNormalizeResult[dnd.NPCRegistry]{}, normalizerErrorf("semantic reconciliation engine must not be nil")
}
if ctx == nil {
return contracts.TypedNormalizeResult[dnd.NPCRegistry]{}, normalizerErrorf("context must not be nil")
@@ -119,33 +106,43 @@ func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalize
order := shared.NewSourceRefOrder(req.Source)
records, warnings := preprocessRecords(req.MergeOutput.Value, order)
deterministic := recordList(records)
materials, ready, err := entityreconcile.BuildContext(req.Source, reconciliationCandidates(records), semanticContextRadius)
if err != nil {
return contracts.TypedNormalizeResult[dnd.NPCRegistry]{}, normalizerErrorf("build semantic context: %w", err)
}
if !ready {
if len(records) < 2 || req.Source == nil {
return contracts.TypedNormalizeResult[dnd.NPCRegistry]{Value: deterministic, Warnings: limitWarnings(warnings)}, nil
}
var response entityreconcile.ProposalResponse
if _, err := n.llm.CompleteStructured(ctx, contracts.StructuredCompletionRequest{
StageName: Key, PromptID: PromptID, PromptVersion: entityreconcile.SchemaVersion,
candidates, envelopes, err := reconciliationInputs(records)
if err != nil {
return contracts.TypedNormalizeResult[dnd.NPCRegistry]{}, normalizerErrorf("prepare semantic reconciliation inputs: %w", err)
}
reconciliation, err := n.engine.Reconcile(ctx, semanticreconcile.Request{
StageName: Key, Source: req.Source, Candidates: candidates,
ProfileID: req.LLMProfile, SessionID: req.SessionID,
Inputs: contracts.LLMInputSet{"candidates": materials.Candidates, "transcript": materials.Transcript},
}, &response); err != nil {
if errors.Is(err, contracts.ErrInvalidStructuredOutput) {
return n.invalidStructuredResult(deterministic, warnings), nil
}
return contracts.TypedNormalizeResult[dnd.NPCRegistry]{}, normalizerErrorf("complete structured output: %w", err)
})
if err != nil {
return contracts.TypedNormalizeResult[dnd.NPCRegistry]{}, normalizerErrorf("reconcile semantic duplicates: %w", err)
}
assessment := materials.Assess(response)
applied, semanticWarnings := applySafeGroups(records, reconciliationGroups(assessment, materials.CandidateKeys()), order)
switch reconciliation.Disposition() {
case semanticreconcile.SkippedInsufficientCandidates:
return contracts.TypedNormalizeResult[dnd.NPCRegistry]{Value: deterministic, Warnings: limitWarnings(warnings)}, nil
case semanticreconcile.SkippedLimitExceeded:
return contracts.TypedNormalizeResult[dnd.NPCRegistry]{Value: deterministic, Warnings: limitWarningsWithSemanticFallback(warnings)}, nil
case semanticreconcile.RetryableInvalidStructuredOutput:
return n.invalidStructuredResult(deterministic, warnings), nil
case semanticreconcile.Complete, semanticreconcile.RetryableDiscardedProposalGroups:
default:
return contracts.TypedNormalizeResult[dnd.NPCRegistry]{}, normalizerErrorf("unknown semantic reconciliation disposition %d", reconciliation.Disposition())
}
applied, semanticWarnings, err := applyReconciliationPlan(reconciliation.Plan(), records, envelopes, order)
if err != nil {
return contracts.TypedNormalizeResult[dnd.NPCRegistry]{}, normalizerErrorf("apply semantic reconciliation plan: %w", err)
}
warnings = append(warnings, semanticWarnings...)
if assessment.DiscardedGroups() == 0 {
if reconciliation.Disposition() == semanticreconcile.Complete {
return contracts.TypedNormalizeResult[dnd.NPCRegistry]{Value: recordList(applied), Warnings: limitWarnings(warnings)}, nil
}
return retryResult(recordList(applied), warnings, assessment), nil
return retryResult(recordList(applied), warnings, reconciliation), nil
}
func (n *Normalizer) invalidStructuredResult(value dnd.NPCRegistry, warnings []contracts.Warning) contracts.TypedNormalizeResult[dnd.NPCRegistry] {
@@ -160,14 +157,14 @@ func (n *Normalizer) invalidStructuredResult(value dnd.NPCRegistry, warnings []c
}
}
func retryResult(value dnd.NPCRegistry, warnings []contracts.Warning, assessment entityreconcile.Assessment) contracts.TypedNormalizeResult[dnd.NPCRegistry] {
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),
Retry: &contracts.NormalizeRetry{
ReasonCode: ReasonCodeNPCSemanticProposalInvalid,
Message: diagnostics.Aggregate("semantic proposal requires retry", reconciliationIssues(assessment)),
FallbackWarnings: []contracts.Warning{semanticFallbackWarning(assessment.DiscardedGroups())},
Message: diagnostics.Aggregate("semantic proposal requires retry", reconciliationIssues(reconciliation.Issues())),
FallbackWarnings: []contracts.Warning{semanticFallbackWarning(reconciliation.DiscardedGroupCount())},
},
}
}
@@ -199,6 +196,10 @@ func limitWarningsForRetry(warnings []contracts.Warning) []contracts.Warning {
})
}
func limitWarningsWithSemanticFallback(warnings []contracts.Warning) []contracts.Warning {
return append(limitWarningsForRetry(warnings), semanticFallbackWarning(-1))
}
type normalizedRecord struct {
npc dnd.NPC
inputIndexes []int

View File

@@ -4,16 +4,15 @@ import (
"context"
"encoding/json"
"reflect"
"strconv"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
"gitea.maximumdirect.net/eric/notarius/internal/framework/semanticreconcile"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/npcs/identity"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared/entityreconcile"
)
func TestModuleContractAndIdentity(t *testing.T) {
@@ -35,12 +34,16 @@ func TestModuleContractAndIdentity(t *testing.T) {
t.Fatal("New(nil, Options{}) error = nil, want nil client rejection")
}
normalizer := newNormalizer(t, &recordingNPCNormalizerClient{})
if metadata := normalizer.ManifestMetadata(); metadata["identity_policy"] != identity.Policy || metadata["normalization_policy"] != normalizationPolicy || metadata["prompt_id"] != PromptID || metadata["response_schema_key"] != string(entityreconcile.ResponseSchemaKey) || metadata["response_schema_id"] != entityreconcile.ResponseSchemaID || metadata["response_schema_name"] != entityreconcile.ResponseSchemaName || metadata["semantic_context_policy"] != semanticContextPolicy || metadata["semantic_context_radius"] != semanticContextRadius {
metadata := normalizer.ManifestMetadata()
limits, ok := metadata["semantic_reconciliation_limits"].(map[string]any)
if !ok || metadata["identity_policy"] != identity.Policy || metadata["normalization_policy"] != normalizationPolicy || metadata["prompt_id"] != PromptID || metadata["prompt_version"] != PromptVersion || metadata["response_schema_key"] != string(semanticreconcile.ResponseSchemaKey) || metadata["response_schema_id"] != semanticreconcile.ResponseSchemaID || metadata["response_schema_name"] != semanticreconcile.ResponseSchemaName || metadata["semantic_reconciliation_policy"] != semanticreconcile.Policy || len(limits) != 3 {
t.Fatalf("metadata = %#v", metadata)
}
wantFingerprints := []pipeline.CheckpointFingerprint{{Name: "prompt", Value: normalizer.promptSHA}, {Name: "response_schema", Value: normalizer.responseSchemaSHA}, {Name: "identity_policy", Value: identity.Policy}, {Name: "normalization_policy", Value: normalizationPolicy}, {Name: "semantic_context_policy", Value: semanticContextPolicy + ":2"}}
if got := normalizer.CheckpointFingerprints(); !reflect.DeepEqual(got, wantFingerprints) {
t.Fatalf("fingerprints = %#v, want %#v", got, wantFingerprints)
fingerprints := normalizer.CheckpointFingerprints()
for _, name := range []string{"prompt", "response_schema", "semantic_reconciliation_policy", "semantic_reconciliation_limits", "identity_policy", "normalization_policy"} {
if !hasFingerprint(fingerprints, name) {
t.Fatalf("fingerprints = %#v, want %q", fingerprints, name)
}
}
}
@@ -157,53 +160,13 @@ func (c *recordingNPCNormalizerClient) CompleteStructured(_ context.Context, req
if response == "" {
response = `{"duplicate_groups":[]}`
}
content, err := contextualProposalResponse(response, request.Inputs["candidates"].Content)
if err != nil {
return contracts.StructuredCompletionResponse{}, err
}
content := []byte(response)
if err := json.Unmarshal(content, output); err != nil {
return contracts.StructuredCompletionResponse{}, err
}
return contracts.StructuredCompletionResponse{Content: content}, nil
}
func contextualProposalResponse(response string, candidateContent []byte) ([]byte, error) {
if !strings.Contains(response, "candidate-") {
return []byte(response), nil
}
var selection struct {
DuplicateGroups []struct {
Members []string `json:"members"`
Canonical string `json:"canonical"`
} `json:"duplicate_groups"`
}
if err := json.Unmarshal([]byte(response), &selection); err != nil {
return nil, err
}
var candidates struct {
Candidates []entityreconcile.Selector `json:"candidates"`
}
if err := json.Unmarshal(candidateContent, &candidates); err != nil {
return nil, err
}
selector := func(key string) entityreconcile.Selector {
index, err := strconv.Atoi(strings.TrimPrefix(key, "candidate-"))
if err != nil || index < 1 || index > len(candidates.Candidates) {
return entityreconcile.Selector{Name: key, SourceRefs: []entityreconcile.SourceRange{}}
}
return candidates.Candidates[index-1].Clone()
}
proposal := entityreconcile.ProposalResponse{DuplicateGroups: make([]entityreconcile.DuplicateGroup, len(selection.DuplicateGroups))}
for index, group := range selection.DuplicateGroups {
members := make([]entityreconcile.Selector, len(group.Members))
for memberIndex, key := range group.Members {
members[memberIndex] = selector(key)
}
proposal.DuplicateGroups[index] = entityreconcile.DuplicateGroup{Members: members, Canonical: selector(group.Canonical)}
}
return json.Marshal(proposal)
}
func newNormalizer(t *testing.T, client contracts.StructuredLLMClient) *Normalizer {
t.Helper()
normalizer, err := New(client, Options{})
@@ -231,3 +194,12 @@ func hasWarning(warnings []contracts.Warning, reason, scope string) bool {
}
return false
}
func hasFingerprint(fingerprints []pipeline.CheckpointFingerprint, name string) bool {
for _, fingerprint := range fingerprints {
if fingerprint.Name == name && fingerprint.Value != "" {
return true
}
}
return false
}

View File

@@ -8,23 +8,26 @@ import (
rootassets "gitea.maximumdirect.net/eric/notarius/assets"
"gitea.maximumdirect.net/eric/notarius/internal/framework/llm"
"gitea.maximumdirect.net/eric/notarius/internal/framework/promptfs"
"gitea.maximumdirect.net/eric/notarius/internal/framework/semanticreconcile"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared"
)
const promptAssetRoot = "assets/prompts"
var promptAssetManifest = shared.PromptAssetManifest{
ModuleDir: PromptID,
ModuleFiles: []promptfs.ModulePromptFile{
{Name: "prompt.yaml", Path: "prompts/prompt.yaml"},
{Name: "instructions.md", Path: "prompts/instructions.md"},
{Name: "candidates.md", Path: "prompts/candidates.md"},
},
SharedFiles: []string{
"common-dnd-system.md",
"common-dnd-entity-reconciliation.md",
"common-dnd-transcript-windows.md",
},
func promptAssetManifest() (shared.PromptAssetManifest, error) {
sharedFiles, err := semanticreconcile.SharedPromptFiles()
if err != nil {
return shared.PromptAssetManifest{}, fmt.Errorf("load shared semantic reconciliation prompt assets: %w", err)
}
return shared.PromptAssetManifest{
ModuleDir: PromptID,
ModuleFiles: []promptfs.ModulePromptFile{
{Name: "prompt.yaml", Path: "prompts/prompt.yaml"},
{Name: "instructions.md", Path: "prompts/instructions.md"},
},
SharedFiles: []string{"common-dnd-system.md"},
ExternalSharedFiles: sharedFiles,
}, nil
}
func moduleAssetFS() (fs.FS, error) {
@@ -40,7 +43,11 @@ func RegisterPromptAssets(registry *llm.AssetRegistry) error {
if err != nil {
return err
}
promptFS, err := promptAssetManifest.PromptFS(assets)
manifest, err := promptAssetManifest()
if err != nil {
return err
}
promptFS, err := manifest.PromptFS(assets)
if err != nil {
return fmt.Errorf("prepare NPC normalization prompt assets: %w", err)
}
@@ -54,7 +61,12 @@ func promptAssetMetadata() (string, error) {
promptAssetHashErr = err
return
}
promptAssetHash, promptAssetHashErr = promptAssetManifest.Hash(assets)
manifest, err := promptAssetManifest()
if err != nil {
promptAssetHashErr = err
return
}
promptAssetHash, promptAssetHashErr = manifest.Hash(assets)
})
return promptAssetHash, promptAssetHashErr
}

View File

@@ -7,7 +7,7 @@ import (
"time"
"gitea.maximumdirect.net/eric/notarius/internal/framework/llm"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared/entityreconcile"
"gitea.maximumdirect.net/eric/notarius/internal/framework/semanticreconcile"
"gitea.maximumdirect.net/eric/promptkit"
)
@@ -16,8 +16,8 @@ func TestRegisterPromptAssetsPreparesNormalizationPrompt(t *testing.T) {
t.Fatalf("promptAssetMetadata() = %q, %v; want prompt fingerprint", promptHash, err)
}
registry := llm.NewAssetRegistry()
if err := entityreconcile.RegisterSchemaAssets(registry); err != nil {
t.Fatalf("RegisterSchemaAssets() error = %v", err)
if err := semanticreconcile.RegisterAssets(registry); err != nil {
t.Fatalf("RegisterAssets() error = %v", err)
}
if err := RegisterPromptAssets(registry); err != nil {
t.Fatalf("RegisterPromptAssets() error = %v", err)
@@ -34,21 +34,27 @@ func TestRegisterPromptAssetsPreparesNormalizationPrompt(t *testing.T) {
t.Fatalf("NewEngine() error = %v", err)
}
prepared, err := engine.Prepare(context.Background(), promptkit.RunRequest{
PromptID: PromptID, PromptVersion: entityreconcile.SchemaVersion, ProfileID: "normalize-test-profile",
PromptID: PromptID, PromptVersion: PromptVersion, ProfileID: "normalize-test-profile",
Inputs: map[string]promptkit.ArtifactRef{
"candidates": promptkit.Inline(`{"candidates":[{"name":"Mira","source_refs":[{"start_unit_id":1,"end_unit_id":1}]}]}`),
"candidates": promptkit.Inline(`{"candidates":[{"candidate_id":1,"label":"Mira","source_refs":[{"start_unit_id":1,"end_unit_id":1}]}]}`),
"transcript": promptkit.Inline(`{"windows":[{"units":[]}]}`),
},
})
if err != nil {
t.Fatalf("Prepare() error = %v", err)
}
if prepared.PromptID != PromptID || prepared.OutputContract.SchemaPath != "dnd_entity_reconcile_llm.v1.json" {
if prepared.PromptID != PromptID || prepared.OutputContract.SchemaPath != "semantic_reconciliation_llm.v1.json" {
t.Fatalf("prepared prompt = %#v, want normalization prompt identity and schema", prepared)
}
if prepared.Messages[0].Role != "system" {
t.Fatalf("initial message role = %q, want system", prepared.Messages[0].Role)
}
if !strings.Contains(prepared.Messages[1].Content, "candidate_id") || !strings.Contains(prepared.Messages[1].Content, "integer") {
t.Fatalf("protocol message = %q, want shared integer-handle protocol", prepared.Messages[1].Content)
}
if !strings.Contains(prepared.Messages[2].Content, "same individual") || strings.Contains(prepared.Messages[2].Content, "source ranges") {
t.Fatalf("NPC policy message = %q, want domain distinctions without copied ranges", prepared.Messages[2].Content)
}
for _, index := range []int{2, 4} {
if cache := prepared.Messages[index].CacheControl; cache == nil || cache.Type != promptkit.CacheControlEphemeral {
t.Errorf("message %d cache control = %#v, want ephemeral", index, cache)

View File

@@ -4,60 +4,66 @@ import (
"fmt"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/semanticreconcile"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/npcs/identity"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared/diagnostics"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared/entityreconcile"
)
const semanticContextRadius = 2
type safeReconciliationGroup struct {
members []int
canonical int
}
func reconciliationCandidates(records []normalizedRecord) []entityreconcile.Candidate {
candidates := make([]entityreconcile.Candidate, len(records))
func reconciliationInputs(records []normalizedRecord) ([]semanticreconcile.Candidate, []semanticreconcile.Record[dnd.NPC], error) {
candidates := make([]semanticreconcile.Candidate, len(records))
envelopes := make([]semanticreconcile.Record[dnd.NPC], len(records))
for index, record := range records {
candidates[index] = entityreconcile.Candidate{
Name: record.npc.Name,
candidates[index] = semanticreconcile.Candidate{
Label: record.npc.Name,
SourceRefs: cloneSourceRefs(record.npc.SourceRefs),
}
envelope, err := semanticreconcile.NewRecord(record.npc, record.inputIndexes, record.earliest, cloneNPC)
if err != nil {
return nil, nil, fmt.Errorf("record %d: %w", index, err)
}
envelopes[index] = envelope
}
return candidates
return candidates, envelopes, nil
}
func reconciliationGroups(assessment entityreconcile.Assessment, candidateKeys []string) []safeReconciliationGroup {
positions := make(map[string]int, len(candidateKeys))
for index, key := range candidateKeys {
positions[key] = index
}
safeGroups := assessment.SafeGroups()
groups := make([]safeReconciliationGroup, 0, len(safeGroups))
for _, group := range safeGroups {
members := group.Members()
memberPositions := make([]int, len(members))
valid := true
for index, key := range members {
position, ok := positions[key]
if !ok {
valid = false
break
func applyReconciliationPlan(plan semanticreconcile.Plan, records []normalizedRecord, envelopes []semanticreconcile.Record[dnd.NPC], order shared.SourceRefOrder) ([]normalizedRecord, []contracts.Warning, error) {
application, err := semanticreconcile.ApplyPlan(plan, envelopes, semanticreconcile.ApplicationPolicy[dnd.NPC]{
CloneValue: cloneNPC,
ConsolidateGroup: func(members []dnd.NPC, canonical dnd.NPC) (dnd.NPC, error) {
output := cloneNPC(canonical)
output.SourceRefs = nil
for _, member := range members {
output.SourceRefs = append(output.SourceRefs, member.SourceRefs...)
}
memberPositions[index] = position
}
canonical, ok := positions[group.Canonical()]
if !valid || !ok {
continue
}
groups = append(groups, safeReconciliationGroup{members: memberPositions, canonical: canonical})
output.SourceRefs = order.Canonicalize(output.SourceRefs)
output.ID = identity.DeriveID(output.Name)
return output, nil
},
})
if err != nil {
return nil, nil, err
}
return groups
applied := application.Records()
output := make([]normalizedRecord, len(applied))
for index, record := range applied {
output[index] = normalizedRecord{
npc: record.Value(),
inputIndexes: record.OriginalInputIndexes(),
earliest: record.EarliestInputPosition(),
}
}
warnings := make([]contracts.Warning, 0, len(application.AppliedGroups()))
for _, event := range application.AppliedGroups() {
provenance := event.Provenance()
warnings = append(warnings, semanticDuplicateWarning(provenance, records[provenance.CanonicalPosition()]))
}
return output, warnings, nil
}
func reconciliationIssues(assessment entityreconcile.Assessment) []string {
issues := assessment.Issues()
func reconciliationIssues(issues []semanticreconcile.Issue) []string {
details := make([]string, len(issues))
for index, issue := range issues {
details[index] = fmt.Sprintf("group %d: %s", issue.GroupIndex, issue.Category)
@@ -65,57 +71,17 @@ func reconciliationIssues(assessment entityreconcile.Assessment) []string {
return details
}
func applySafeGroups(records []normalizedRecord, groups []safeReconciliationGroup, order shared.SourceRefOrder) ([]normalizedRecord, []contracts.Warning) {
byMember := make(map[int]safeReconciliationGroup, len(groups)*2)
for _, group := range groups {
for _, member := range group.members {
byMember[member] = group
}
}
output := make([]normalizedRecord, 0, len(records)-len(groups))
warnings := make([]contracts.Warning, 0, len(groups))
for index, record := range records {
group, grouped := byMember[index]
if !grouped {
output = append(output, cloneRecord(record))
continue
}
if group.members[0] != index {
continue
}
consolidated := consolidateSemanticGroup(records, group, order)
output = append(output, consolidated)
warnings = append(warnings, semanticDuplicateWarning(consolidated, records[group.canonical]))
}
return output, warnings
}
func consolidateSemanticGroup(records []normalizedRecord, group safeReconciliationGroup, order shared.SourceRefOrder) normalizedRecord {
output := cloneRecord(records[group.members[0]])
output.npc.Name = records[group.canonical].npc.Name
for _, member := range group.members[1:] {
output.npc.SourceRefs = append(output.npc.SourceRefs, records[member].npc.SourceRefs...)
output.inputIndexes = append(output.inputIndexes, records[member].inputIndexes...)
if records[member].earliest < output.earliest {
output.earliest = records[member].earliest
}
}
output.inputIndexes = sortedUniqueIndexes(output.inputIndexes)
output.npc.SourceRefs = order.Canonicalize(output.npc.SourceRefs)
output.npc.ID = identity.DeriveID(output.npc.Name)
return output
}
func semanticDuplicateWarning(record normalizedRecord, canonical normalizedRecord) contracts.Warning {
details := make([]string, 0, len(record.inputIndexes)+1)
for _, inputIndex := range record.inputIndexes {
func semanticDuplicateWarning(provenance semanticreconcile.GroupProvenance, canonical normalizedRecord) contracts.Warning {
inputIndexes := provenance.OriginalInputIndexes()
details := make([]string, 0, len(inputIndexes)+1)
for _, inputIndex := range inputIndexes {
details = append(details, fmt.Sprintf("input index %d", inputIndex))
}
if canonical.earliest != record.earliest {
if canonical.earliest != provenance.EarliestInputPosition() {
details = append(details, fmt.Sprintf("canonical display name from input index %d", canonical.earliest))
}
return contracts.Warning{
Scope: npcScope(record.earliest),
Scope: npcScope(provenance.EarliestInputPosition()),
ReasonCode: ReasonCodeDuplicateNPCCollapsed,
Message: diagnostics.Aggregate("semantic duplicate consolidation", details),
}

View File

@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"math"
"reflect"
"strings"
@@ -11,9 +12,10 @@ import (
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/semanticreconcile"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/npcs/identity"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared/entityreconcile"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared/diagnostics"
)
func TestNormalizeSkipsSemanticCompletionWithoutTwoEligibleCandidates(t *testing.T) {
@@ -31,7 +33,7 @@ func TestNormalizeSkipsSemanticCompletionWithoutTwoEligibleCandidates(t *testing
}
func TestNormalizeAppliesSafeProposalAndUsesPrivateInputs(t *testing.T) {
client := &recordingNPCNormalizerClient{response: `{"duplicate_groups":[{"members":["candidate-000001","candidate-000002"],"canonical":"candidate-000002"}]}`}
client := &recordingNPCNormalizerClient{response: `{"duplicate_groups":[{"candidate_ids":[1,2],"canonical_candidate_id":2}]}`}
normalizer := newNormalizer(t, client)
doc := semanticDocument()
input := dnd.NPCRegistry{NPCs: []dnd.NPC{
@@ -64,17 +66,28 @@ func TestNormalizeAppliesSafeProposalAndUsesPrivateInputs(t *testing.T) {
t.Fatalf("completion calls = %d, want one", len(client.requests))
}
completion := client.requests[0]
if completion.StageName != Key || completion.PromptID != PromptID || completion.PromptVersion != entityreconcile.SchemaVersion || completion.ProfileID != request.LLMProfile || completion.SessionID != request.SessionID || len(completion.Inputs) != 2 {
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)
}
encoded := string(completion.Inputs["candidates"].Content) + string(completion.Inputs["transcript"].Content)
if strings.Contains(encoded, "npc:sha256:") || strings.Contains(encoded, doc.ID) {
t.Fatalf("completion inputs leaked private identifiers: %s", encoded)
}
var visible struct {
Candidates []struct {
CandidateID int `json:"candidate_id"`
} `json:"candidates"`
}
if err := json.Unmarshal(completion.Inputs["candidates"].Content, &visible); err != nil {
t.Fatalf("decode candidates: %v", err)
}
if got := []int{visible.Candidates[0].CandidateID, visible.Candidates[1].CandidateID, visible.Candidates[2].CandidateID}; !reflect.DeepEqual(got, []int{1, 2, 3}) {
t.Fatalf("candidate IDs = %v, want contiguous request-local handles", got)
}
}
func TestNormalizeUnsafeProposalReturnsSafeRetryFallback(t *testing.T) {
client := &recordingNPCNormalizerClient{response: `{"duplicate_groups":[{"members":["candidate-000001","candidate-000002"],"canonical":"candidate-000002"},{"members":["candidate-000003","unknown"],"canonical":"candidate-000003"}]}`}
client := &recordingNPCNormalizerClient{response: `{"duplicate_groups":[{"candidate_ids":[1,2],"canonical_candidate_id":2},{"candidate_ids":[3,99],"canonical_candidate_id":3}]}`}
normalizer := newNormalizer(t, client)
doc := semanticDocument()
input := dnd.NPCRegistry{NPCs: []dnd.NPC{
@@ -137,7 +150,7 @@ func TestNormalizeRedactsContextMaterialFailures(t *testing.T) {
request := normalizeRequestWithSource(input, doc)
request.SourceInput = contracts.NewLLMInputMaterial("source", "application/json", []byte(transcript), "sha256:test", originPath)
_, err := normalizer.Normalize(context.Background(), request)
if err == nil || !strings.Contains(err.Error(), "build entity reconciliation context: invalid source metadata") {
if err == nil || !strings.Contains(err.Error(), "build transcript material: invalid source metadata") {
t.Fatalf("Normalize() error = %v; want content-safe context-material failure", err)
}
for _, forbidden := range []string{metadataKey, metadataValue, transcript, firstName, secondName, sourceID, originPath, "float64", "non-finite"} {
@@ -152,8 +165,8 @@ func TestNormalizeRedactsContextMaterialFailures(t *testing.T) {
func TestNormalizeDoesNotAccumulateSafeGroupsAcrossAttempts(t *testing.T) {
client := &recordingNPCNormalizerClient{responses: []string{
`{"duplicate_groups":[{"members":["candidate-000001","candidate-000002"],"canonical":"candidate-000002"},{"members":["candidate-000003","unknown"],"canonical":"candidate-000003"}]}`,
`{"duplicate_groups":[{"members":["candidate-000001","candidate-000003"],"canonical":"candidate-000003"}]}`,
`{"duplicate_groups":[{"candidate_ids":[1,2],"canonical_candidate_id":2},{"candidate_ids":[3,99],"canonical_candidate_id":3}]}`,
`{"duplicate_groups":[{"candidate_ids":[1,3],"canonical_candidate_id":3}]}`,
}}
normalizer := newNormalizer(t, client)
doc := semanticDocument()
@@ -175,8 +188,8 @@ func TestNormalizeDoesNotAccumulateSafeGroupsAcrossAttempts(t *testing.T) {
}
}
func TestNormalizeRetainsRecordsWhenAProposalUsesAnUnknownDescriptor(t *testing.T) {
client := &recordingNPCNormalizerClient{response: `{"duplicate_groups":[{"members":["candidate-000001","unknown"],"canonical":"candidate-000001"}]}`}
func TestNormalizeRetainsRecordsWhenAProposalUsesAnUnknownHandle(t *testing.T) {
client := &recordingNPCNormalizerClient{response: `{"duplicate_groups":[{"candidate_ids":[1,99],"canonical_candidate_id":1}]}`}
normalizer := newNormalizer(t, client)
doc := semanticDocument()
input := dnd.NPCRegistry{NPCs: []dnd.NPC{
@@ -193,32 +206,52 @@ func TestNormalizeRetainsRecordsWhenAProposalUsesAnUnknownDescriptor(t *testing.
}
}
func TestReconciliationCandidatesKeepEqualDisplayNamesDistinct(t *testing.T) {
func TestReconciliationCandidatesKeepEqualContextualDescriptorsDistinct(t *testing.T) {
doc := semanticDocument()
records := []normalizedRecord{
{npc: dnd.NPC{Name: "The Guard", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 10, EndUnitID: 10}}}},
{npc: dnd.NPC{Name: "The Guard", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 20, EndUnitID: 20}}}},
}
materials, ready, err := entityreconcile.BuildContext(doc, reconciliationCandidates(records), semanticContextRadius)
if err != nil || !ready {
t.Fatalf("BuildContext() = %#v, %t, %v; want ready keyed candidates", materials, ready, err)
candidates, _, err := reconciliationInputs(records)
if err != nil {
t.Fatalf("reconciliationInputs() error = %v", err)
}
keys := materials.CandidateKeys()
if !reflect.DeepEqual(keys, []string{"candidate-000001", "candidate-000002"}) || strings.Count(string(materials.Candidates.Content), `"The Guard"`) != 2 {
t.Fatalf("candidate keys and inputs = %#v, %s; want distinct equal-display candidates", keys, materials.Candidates.Content)
preparation, err := semanticreconcile.Prepare(doc, candidates, semanticreconcile.DefaultLimits())
if err != nil || preparation.Disposition() != semanticreconcile.Ready {
t.Fatalf("Prepare() = %#v, %v; want ready candidates", preparation, err)
}
var candidateInput struct {
Candidates []entityreconcile.Selector `json:"candidates"`
Candidates []struct {
CandidateID int `json:"candidate_id"`
Label string `json:"label"`
} `json:"candidates"`
}
if err := json.Unmarshal(materials.Candidates.Content, &candidateInput); err != nil {
if err := json.Unmarshal(preparation.Materials()["candidates"].Content, &candidateInput); err != nil {
t.Fatal(err)
}
assessment := materials.Assess(entityreconcile.ProposalResponse{DuplicateGroups: []entityreconcile.DuplicateGroup{{
Members: candidateInput.Candidates, Canonical: candidateInput.Candidates[1],
}}})
groups := reconciliationGroups(assessment, keys)
if assessment.DiscardedGroups() != 0 || len(groups) != 1 || groups[0].canonical != 1 {
t.Fatalf("reconciliation = %#v, %#v; want distinct keyed group", assessment, groups)
if len(candidateInput.Candidates) != 2 || candidateInput.Candidates[0].CandidateID != 1 || candidateInput.Candidates[1].CandidateID != 2 || candidateInput.Candidates[0].Label != "The Guard" || candidateInput.Candidates[1].Label != "The Guard" {
t.Fatalf("candidate input = %#v, want distinct integer handles for equal descriptors", candidateInput)
}
}
func TestNormalizeLimitSkipDoesNotCallLLMAndAddsBoundedFallbackWarning(t *testing.T) {
client := &recordingNPCNormalizerClient{}
normalizer := newNormalizer(t, client)
doc := &source.SourceDocument{ID: "session", Units: []source.SourceUnit{{ID: 1, Kind: "speech", Text: "A crowded hall"}}}
limit := semanticreconcile.DefaultLimits().MaximumCandidates
input := dnd.NPCRegistry{NPCs: make([]dnd.NPC, limit+1)}
for index := range input.NPCs {
input.NPCs[index] = dnd.NPC{Name: fmt.Sprintf("Person %d", index), SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 1, EndUnitID: 1}}}
}
result, err := normalizer.Normalize(context.Background(), normalizeRequestWithSource(input, doc))
if err != nil || result.Retry != nil {
t.Fatalf("Normalize() = %#v, %v; want deterministic limit fallback", result, err)
}
if len(client.requests) != 0 || len(result.Value.NPCs) != limit+1 {
t.Fatalf("completion calls = %d, NPCs = %d; want no call and all records", len(client.requests), len(result.Value.NPCs))
}
if !hasWarning(result.Warnings, ReasonCodeNPCSemanticReconciliationExhausted, "npcs") || len(result.Warnings) > diagnostics.MaxWarnings {
t.Fatalf("warnings = %#v, want bounded reconciliation fallback", result.Warnings)
}
}

View File

@@ -17,6 +17,7 @@ import (
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/evidencecontext"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
"gitea.maximumdirect.net/eric/notarius/internal/framework/semanticreconcile"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd"
combatcodec "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/codec/combatturns"
npccodec "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/codec/npcregistry"
@@ -49,12 +50,12 @@ func TestNPCOutputGroundsSpellAndCombatConsumersThroughOneOperation(t *testing.T
t.Fatalf("Prepare() error = %v", err)
}
for name, value := range map[string]string{
"extract:npc_registry:dnd/npc-registry:mapping_policy": "dnd.npc_registry.extract_mapping.v2",
"normalize:npc_registry:dnd/npc-registry:identity_policy": "dnd.npc_registry.identity.v1",
"normalize:npc_registry:dnd/npc-registry:normalization_policy": "dnd.npc_registry.normalize.v4",
"normalize:npc_registry:dnd/npc-registry:semantic_context_policy": "dnd.entity_reconcile.context.v1:2",
"extract:spells:dnd/spells:mapping_policy": "dnd.spells.extract_mapping.v2",
"extract:combat:dnd/combat-turns:scene_gate_policy": "dnd.combat_turns.scene_gate.v1",
"extract:npc_registry:dnd/npc-registry:mapping_policy": "dnd.npc_registry.extract_mapping.v2",
"normalize:npc_registry:dnd/npc-registry:identity_policy": "dnd.npc_registry.identity.v1",
"normalize:npc_registry:dnd/npc-registry:normalization_policy": npcnormalize.NormalizationPolicy,
"normalize:npc_registry:dnd/npc-registry:semantic_reconciliation_policy": semanticreconcile.Policy,
"extract:spells:dnd/spells:mapping_policy": "dnd.spells.extract_mapping.v2",
"extract:combat:dnd/combat-turns:scene_gate_policy": "dnd.combat_turns.scene_gate.v1",
} {
assertFingerprintValue(t, prepared.CheckpointFingerprints(), name, value)
}
@@ -63,6 +64,7 @@ func TestNPCOutputGroundsSpellAndCombatConsumersThroughOneOperation(t *testing.T
"extract:npc_registry:dnd/npc-registry:response_schema",
"normalize:npc_registry:dnd/npc-registry:prompt",
"normalize:npc_registry:dnd/npc-registry:response_schema",
"normalize:npc_registry:dnd/npc-registry:semantic_reconciliation_limits",
"extract:spells:dnd/spells:prompt",
"extract:spells:dnd/spells:response_schema",
"extract:spells:dnd/spells:npc_registry",

View File

@@ -292,10 +292,7 @@ func (client *semanticNPCOccurrenceClient) CompleteStructured(_ context.Context,
}
payload = map[string]any{"npcs": []any{map[string]any{"name": name, "source_refs": []any{map[string]int{"start_unit_id": client.npcCalls, "end_unit_id": client.npcCalls}}}}}
case npcnormalize.PromptID:
content, err := contextualReconciliationContent([]byte(`{"duplicate_groups":[{"members":["candidate-000001","candidate-000002"],"canonical":"candidate-000001"}]}`), request.Inputs["candidates"].Content)
if err != nil {
return contracts.StructuredCompletionResponse{}, err
}
content := []byte(`{"duplicate_groups":[{"candidate_ids":[1,2],"canonical_candidate_id":1}]}`)
if err := json.Unmarshal(content, out); err != nil {
return contracts.StructuredCompletionResponse{}, err
}

View File

@@ -5,7 +5,6 @@ import (
"encoding/json"
"fmt"
"os"
"strconv"
"strings"
"testing"
@@ -13,12 +12,12 @@ import (
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/llm"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
"gitea.maximumdirect.net/eric/notarius/internal/framework/semanticreconcile"
npccodec "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/codec/npcregistry"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/extract/npcregistry"
npcnormalize "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/normalize/npcregistry"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/npcs/identity"
dndregister "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/register"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared/entityreconcile"
npcshape "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/validate/npcregistry/shape"
npcregistrysourcerefs "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/validate/npcregistry/source_refs"
genericregister "gitea.maximumdirect.net/eric/notarius/internal/modules/generic/register"
@@ -98,7 +97,7 @@ func TestRunnerProcessesSeriatimInputWithProductionDNDNPCPipeline(t *testing.T)
t.Fatalf("manifest lane = %#v, want NPC production composition", lane)
}
normalizerMetadata, ok := lane.Metadata["normalizer"].(map[string]any)
if !ok || normalizerMetadata["identity_policy"] != identity.Policy || normalizerMetadata["normalization_policy"] != npcnormalize.NormalizationPolicy || normalizerMetadata["prompt_id"] != npcnormalize.PromptID || normalizerMetadata["response_schema_id"] != entityreconcile.ResponseSchemaID {
if !ok || normalizerMetadata["identity_policy"] != identity.Policy || normalizerMetadata["normalization_policy"] != npcnormalize.NormalizationPolicy || normalizerMetadata["prompt_id"] != npcnormalize.PromptID || normalizerMetadata["response_schema_id"] != semanticreconcile.ResponseSchemaID {
t.Fatalf("normalizer metadata = %#v, want identity and normalization policies", lane.Metadata)
}
var npcOutputFile *contracts.OutputFile
@@ -180,8 +179,8 @@ func TestProductionNPCNormalizationRetryUsesFinalSafeProposal(t *testing.T) {
{Name: "Mira Thorn", SourceRefs: []npcProductionSourceRef{{StartUnitID: 2, EndUnitID: 2}}},
{Name: "Hooded Guard", SourceRefs: []npcProductionSourceRef{{StartUnitID: 3, EndUnitID: 3}}},
}}
partial := []byte(`{"duplicate_groups":[{"members":["candidate-000001","candidate-000002"],"canonical":"candidate-000002"},{"members":["candidate-000003","unknown"],"canonical":"candidate-000003"}]}`)
safe := []byte(`{"duplicate_groups":[{"members":["candidate-000001","candidate-000002"],"canonical":"candidate-000002"}]}`)
partial := []byte(`{"duplicate_groups":[{"candidate_ids":[1,2],"canonical_candidate_id":2},{"candidate_ids":[3,99],"canonical_candidate_id":3}]}`)
safe := []byte(`{"duplicate_groups":[{"candidate_ids":[1,2],"canonical_candidate_id":2}]}`)
for _, test := range []struct {
name string
@@ -267,11 +266,6 @@ func (client *fakeNPCProductionLLMClient) CompleteStructured(_ context.Context,
}
content = append([]byte(nil), client.normalizeResponses[index]...)
}
var err error
content, err = contextualReconciliationContent(content, req.Inputs["candidates"].Content)
if err != nil {
return contracts.StructuredCompletionResponse{}, err
}
default:
return contracts.StructuredCompletionResponse{}, fmt.Errorf("unexpected fake NPC prompt %q", req.PromptID)
}
@@ -281,43 +275,6 @@ func (client *fakeNPCProductionLLMClient) CompleteStructured(_ context.Context,
return contracts.StructuredCompletionResponse{Content: content}, nil
}
func contextualReconciliationContent(content, candidateContent []byte) ([]byte, error) {
if !strings.Contains(string(content), "candidate-") {
return content, nil
}
var selection struct {
DuplicateGroups []struct {
Members []string `json:"members"`
Canonical string `json:"canonical"`
} `json:"duplicate_groups"`
}
if err := json.Unmarshal(content, &selection); err != nil {
return nil, err
}
var candidates struct {
Candidates []entityreconcile.Selector `json:"candidates"`
}
if err := json.Unmarshal(candidateContent, &candidates); err != nil {
return nil, err
}
selector := func(key string) entityreconcile.Selector {
index, err := strconv.Atoi(strings.TrimPrefix(key, "candidate-"))
if err != nil || index < 1 || index > len(candidates.Candidates) {
return entityreconcile.Selector{Name: key, SourceRefs: []entityreconcile.SourceRange{}}
}
return candidates.Candidates[index-1].Clone()
}
proposal := entityreconcile.ProposalResponse{DuplicateGroups: make([]entityreconcile.DuplicateGroup, len(selection.DuplicateGroups))}
for index, group := range selection.DuplicateGroups {
members := make([]entityreconcile.Selector, len(group.Members))
for memberIndex, key := range group.Members {
members[memberIndex] = selector(key)
}
proposal.DuplicateGroups[index] = entityreconcile.DuplicateGroup{Members: members, Canonical: selector(group.Canonical)}
}
return json.Marshal(proposal)
}
func (client *fakeNPCProductionLLMClient) requestCount(promptID string) int {
count := 0
for _, request := range client.requests {