eric/notarius
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Adopt location registry durable contract

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-05 19:05:42 +00:00
parent 2f61118e78
commit c006b163d5
41 changed files with 196 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
internal/modules/dnd/normalize/locations/normalizer.go
View File

@@ -40,7 +40,7 @@ const (
var requiredCapabilities = []string{"merged"}
var providedCapabilities = []string{"normalized"}
var _ contracts.Normalizer[dnd.LocationList] = (*Normalizer)(nil)
var _ contracts.Normalizer[dnd.LocationRegistry] = (*Normalizer)(nil)
var _ contracts.ManifestMetadataProvider = (*Normalizer)(nil)
var _ pipeline.CheckpointFingerprintProvider = (*Normalizer)(nil)
@@ -94,18 +94,18 @@ func (n *Normalizer) CheckpointFingerprints() []pipeline.CheckpointFingerprint {
}
}
func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalizeRequest[dnd.LocationList]) (contracts.TypedNormalizeResult[dnd.LocationList], error) {
func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalizeRequest[dnd.LocationRegistry]) (contracts.TypedNormalizeResult[dnd.LocationRegistry], error) {
if n == nil {
return contracts.TypedNormalizeResult[dnd.LocationList]{}, normalizerErrorf("normalizer must not be nil")
return contracts.TypedNormalizeResult[dnd.LocationRegistry]{}, normalizerErrorf("normalizer must not be nil")
}
if n.llm == nil {
return contracts.TypedNormalizeResult[dnd.LocationList]{}, normalizerErrorf("LLM client must not be nil")
return contracts.TypedNormalizeResult[dnd.LocationRegistry]{}, normalizerErrorf("LLM client must not be nil")
}
if ctx == nil {
return contracts.TypedNormalizeResult[dnd.LocationList]{}, normalizerErrorf("context must not be nil")
return contracts.TypedNormalizeResult[dnd.LocationRegistry]{}, normalizerErrorf("context must not be nil")
}
if err := ctx.Err(); err != nil {
return contracts.TypedNormalizeResult[dnd.LocationList]{}, normalizerErrorf("context error before normalize: %w", err)
return contracts.TypedNormalizeResult[dnd.LocationRegistry]{}, normalizerErrorf("context error before normalize: %w", err)
}
order := shared.NewSourceRefOrder(req.Source)
@@ -113,10 +113,10 @@ func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalize
deterministic := recordList(records)
materials, ready, err := entityreconcile.BuildContext(req.Source, reconciliationCandidates(records), semanticContextRadius)
if err != nil {
return contracts.TypedNormalizeResult[dnd.LocationList]{}, normalizerErrorf("build semantic context: %w", err)
return contracts.TypedNormalizeResult[dnd.LocationRegistry]{}, normalizerErrorf("build semantic context: %w", err)
}
if !ready {
return contracts.TypedNormalizeResult[dnd.LocationList]{Value: deterministic, Warnings: limitWarnings(warnings)}, nil
return contracts.TypedNormalizeResult[dnd.LocationRegistry]{Value: deterministic, Warnings: limitWarnings(warnings)}, nil
}
var response entityreconcile.ProposalResponse
@@ -128,26 +128,26 @@ func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalize
if errors.Is(err, contracts.ErrInvalidStructuredOutput) {
return n.invalidStructuredResult(deterministic, warnings), nil
}
return contracts.TypedNormalizeResult[dnd.LocationList]{}, normalizerErrorf("complete structured output: %w", err)
return contracts.TypedNormalizeResult[dnd.LocationRegistry]{}, normalizerErrorf("complete structured output: %w", err)
}
assessment := materials.Assess(response)
applied, semanticWarnings := applySafeGroups(records, reconciliationGroups(assessment, materials.CandidateKeys()), order)
warnings = append(warnings, semanticWarnings...)
if assessment.DiscardedGroups() == 0 {
return contracts.TypedNormalizeResult[dnd.LocationList]{Value: recordList(applied), Warnings: limitWarnings(warnings)}, nil
return contracts.TypedNormalizeResult[dnd.LocationRegistry]{Value: recordList(applied), Warnings: limitWarnings(warnings)}, nil
}
return retryResult(recordList(applied), warnings, assessment), nil
}
func (n *Normalizer) invalidStructuredResult(value dnd.LocationList, warnings []contracts.Warning) contracts.TypedNormalizeResult[dnd.LocationList] {
return contracts.TypedNormalizeResult[dnd.LocationList]{Value: value, Warnings: limitWarningsForRetry(warnings), Retry: &contracts.NormalizeRetry{
func (n *Normalizer) invalidStructuredResult(value dnd.LocationRegistry, warnings []contracts.Warning) contracts.TypedNormalizeResult[dnd.LocationRegistry] {
return contracts.TypedNormalizeResult[dnd.LocationRegistry]{Value: value, Warnings: limitWarningsForRetry(warnings), Retry: &contracts.NormalizeRetry{
ReasonCode: ReasonCodeLocationSemanticProposalInvalid, Message: "semantic proposal requires retry: invalid structured output",
FallbackWarnings: []contracts.Warning{semanticFallbackWarning(-1)},
}}
}
func retryResult(value dnd.LocationList, warnings []contracts.Warning, assessment entityreconcile.Assessment) contracts.TypedNormalizeResult[dnd.LocationList] {
return contracts.TypedNormalizeResult[dnd.LocationList]{Value: value, Warnings: limitWarningsForRetry(warnings), Retry: &contracts.NormalizeRetry{
func retryResult(value dnd.LocationRegistry, warnings []contracts.Warning, assessment entityreconcile.Assessment) contracts.TypedNormalizeResult[dnd.LocationRegistry] {
return contracts.TypedNormalizeResult[dnd.LocationRegistry]{Value: value, Warnings: limitWarningsForRetry(warnings), Retry: &contracts.NormalizeRetry{
ReasonCode: ReasonCodeLocationSemanticProposalInvalid, Message: diagnostics.Aggregate("semantic proposal requires retry", reconciliationIssues(assessment)),
FallbackWarnings: []contracts.Warning{semanticFallbackWarning(assessment.DiscardedGroups())},
}}
@@ -183,7 +183,7 @@ type normalizedRecord struct {
earliest int
}
func preprocessRecords(input dnd.LocationList, order shared.SourceRefOrder) ([]normalizedRecord, []contracts.Warning) {
func preprocessRecords(input dnd.LocationRegistry, order shared.SourceRefOrder) ([]normalizedRecord, []contracts.Warning) {
if input.Locations == nil {
return nil, nil
}
@@ -294,11 +294,11 @@ func recordValues(records []normalizedRecord) []dnd.Location {
}
return values
}
func recordList(records []normalizedRecord) dnd.LocationList {
func recordList(records []normalizedRecord) dnd.LocationRegistry {
if records == nil {
return dnd.LocationList{}
return dnd.LocationRegistry{}
}
return dnd.LocationList{Locations: recordValues(records)}
return dnd.LocationRegistry{Locations: recordValues(records)}
}
func duplicateWarning(retainedIndex int, removed []int) contracts.Warning {
@@ -320,10 +320,10 @@ 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.LocationListKind}
return pipeline.ModuleSpec{Key: Key, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, 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.LocationList], error) {
return pipeline.RegisterNormalizerBuilder(registry, ModuleSpec(), validateOptions, func(request pipeline.BuildRequest) (contracts.Normalizer[dnd.LocationRegistry], error) {
options, err := DecodeOptions(request.Options)
if err != nil {
return nil, err