diff --git a/router/config.go b/router/config.go index 01343696..52a5d3fc 100644 --- a/router/config.go +++ b/router/config.go @@ -63,14 +63,9 @@ type ProviderConfig struct { Timeout time.Duration `json:"timeout"` - // Priority is documented in the schema (tmp/provider-registration.json) and - // the spec config sample. Accepted on the wire so spec-aligned configs - // parse, but has no effect on routing today: the upstream spec contradicts - // itself between provider-registration.json ("router keeps the offer from - // the higher-priority provider") and router-architecture.mdx (first-received - // wins on duplicate package_id) — tracked at - // https://github.com/adcontextprotocol/adcp/issues/5722. Wiring priority - // through dedup/conflict-resolution waits on that resolution. + // Priority resolves duplicate package_id offers returned by different + // Context Match providers. Lower values have higher priority; equal values + // are broken by response arrival order. Priority int `json:"priority,omitempty"` } @@ -138,10 +133,6 @@ func ValidateProviderConfig(p *ProviderConfig, latencyBudget time.Duration) erro // ProviderConfigFromRegistration converts a schema-generated ProviderRegistration // (the wire format from discovery endpoints) into a router ProviderConfig. -// -// Note: Priority is not currently used by the router. It is captured on -// ProviderRegistration in the spec for future use (merge conflict resolution, -// adaptive timeout allocation) but has no effect on routing today. func ProviderConfigFromRegistration(r *tmproto.ProviderRegistration) ProviderConfig { uidTypes := make([]string, len(r.UIDTypes)) for i, u := range r.UIDTypes { @@ -157,6 +148,7 @@ func ProviderConfigFromRegistration(r *tmproto.ProviderRegistration) ProviderCon UIDTypes: uidTypes, PropertyRIDs: r.Properties, // Properties in the spec are registry RIDs (UUIDs), not slugs. Timeout: time.Duration(r.TimeoutMs) * time.Millisecond, + Priority: r.Priority, TmpxSlots: append([]string(nil), r.TmpxSlots...), } } diff --git a/router/config_test.go b/router/config_test.go index 93d58ca4..d02d16e1 100644 --- a/router/config_test.go +++ b/router/config_test.go @@ -4,6 +4,7 @@ import ( "testing" "time" + "github.com/adcontextprotocol/adcp-go/tmproto" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -156,10 +157,22 @@ func TestEffectiveTimeout(t *testing.T) { }) } +func TestProviderConfigFromRegistrationCarriesPriority(t *testing.T) { + registration := &tmproto.ProviderRegistration{ + ProviderID: "priority-provider", + Endpoint: "https://provider.example.com", + Priority: 7, + } + + config := ProviderConfigFromRegistration(registration) + + assert.Equal(t, 7, config.Priority) +} + func TestProviderSet_ActiveFiltersByStatus(t *testing.T) { ps := NewProviderSet([]ProviderConfig{ {ID: "active", Status: ProviderStatusActive, ContextMatch: true}, - {ID: "empty-status", ContextMatch: true}, // defaults to active + {ID: "empty-status", ContextMatch: true}, // defaults to active {ID: "inactive", Status: ProviderStatusInactive, ContextMatch: true}, {ID: "draining", Status: ProviderStatusDraining, ContextMatch: true}, }) diff --git a/router/router.go b/router/router.go index b8a1dfd8..18a26cbe 100644 --- a/router/router.go +++ b/router/router.go @@ -379,6 +379,7 @@ func (r *Router) effectiveTimeout(providerTimeout time.Duration) time.Duration { type contextResult struct { providerID string + priority int response *tmproto.ContextMatchResponse } @@ -421,7 +422,7 @@ func (r *Router) fanOutContext(ctx context.Context, providers []ProviderConfig, // don't touch cached.RequestID here — any assignment // would be dead. mu.Lock() - results = append(results, contextResult{providerID: p.ID, response: cached}) + results = append(results, contextResult{providerID: p.ID, priority: p.Priority, response: cached}) mu.Unlock() return } @@ -505,7 +506,7 @@ func (r *Router) fanOutContext(ctx context.Context, providers []ProviderConfig, } mu.Lock() - results = append(results, contextResult{providerID: p.ID, response: &cmResp}) + results = append(results, contextResult{providerID: p.ID, priority: p.Priority, response: &cmResp}) mu.Unlock() }) } @@ -690,10 +691,10 @@ var providerHopForbiddenFields = []string{ // // Packages are provider-specific per docs/trusted-match/router-architecture.mdx // §"Response Aggregation": duplicate `package_id` across providers is a -// configuration error. The router keeps the first response received for a -// duplicated package and SHOULD log a warning, so we dedup by package_id and -// emit a warning naming both providers when the same package_id appears in -// more than one response. +// configuration error. The router keeps the offer from the provider with the +// lower numeric priority. Equal priorities are broken by response arrival +// order. Every cross-provider duplicate emits a warning naming the providers +// and the selected winner. func mergeContextResponses(requestID string, responses []contextResult, logger *slog.Logger) *tmproto.ContextMatchResponse { merged := &tmproto.ContextMatchResponse{ Type: tmproto.TypeContextMatchResponse, @@ -702,33 +703,58 @@ func mergeContextResponses(requestID string, responses []contextResult, logger * } mergedSignals := make(map[string]any) - seenPkg := make(map[string]string) // package_id -> first provider that returned it + type offerWinner struct { + providerID string + priority int + index int + } + seenPkg := make(map[string]offerWinner) for _, res := range responses { if res.response == nil { continue } for _, offer := range res.response.Offers { - if first, dup := seenPkg[offer.PackageID]; dup { + if current, dup := seenPkg[offer.PackageID]; dup { if logger != nil { - if first == res.providerID { + if current.providerID == res.providerID { logger.Warn("repeated package_id within a single provider's response — keeping first offer", "request_id", requestID, "package_id", offer.PackageID, "provider", res.providerID, ) } else { - logger.Warn("duplicate package_id across providers — keeping first response (configuration error)", + winnerID := current.providerID + winnerPriority := current.priority + if res.priority < current.priority { + winnerID = res.providerID + winnerPriority = res.priority + } + logger.Warn("duplicate package_id across providers — keeping higher-priority offer (configuration error)", "request_id", requestID, "package_id", offer.PackageID, - "first_provider", first, + "first_provider", current.providerID, "duplicate_provider", res.providerID, + "winner_provider", winnerID, + "winner_priority", winnerPriority, ) } } + if current.providerID != res.providerID && res.priority < current.priority { + merged.Offers[current.index] = offer + seenPkg[offer.PackageID] = offerWinner{ + providerID: res.providerID, + priority: res.priority, + index: current.index, + } + } continue } - seenPkg[offer.PackageID] = res.providerID + seenPkg[offer.PackageID] = offerWinner{ + providerID: res.providerID, + priority: res.priority, + index: len(merged.Offers), + } merged.Offers = append(merged.Offers, offer) } maps.Copy(mergedSignals, res.response.Signals) diff --git a/router/router_test.go b/router/router_test.go index 51052bc4..afb05bdc 100644 --- a/router/router_test.go +++ b/router/router_test.go @@ -359,7 +359,7 @@ func TestMergeIdentityResponses_MixedEmission(t *testing.T) { // TestMergeContextResponses_DuplicatePackageID covers the dedup-warn path the // router-architecture spec calls out: same package_id from two providers MUST -// keep the first response and SHOULD log a warning naming both providers. +// keep the higher-priority offer and SHOULD log a warning naming the winner. func TestMergeContextResponses_DuplicatePackageID(t *testing.T) { r1 := &tmproto.ContextMatchResponse{ Offers: []tmproto.Offer{{PackageID: "pkg-dup", Summary: "first"}}, @@ -372,20 +372,35 @@ func TestMergeContextResponses_DuplicatePackageID(t *testing.T) { logger := slog.New(slog.NewJSONHandler(&logs, nil)) merged := mergeContextResponses("ctx-dup", []contextResult{ - {providerID: "alpha", response: r1}, - {providerID: "beta", response: r2}, + {providerID: "alpha", priority: 20, response: r1}, + {providerID: "beta", priority: 10, response: r2}, }, logger) require.Len(t, merged.Offers, 2, "duplicate package_id should be deduped, unique one kept") - assert.Equal(t, "first", merged.Offers[0].Summary, "first provider's offer wins on dup") + assert.Equal(t, "second", merged.Offers[0].Summary, "lower numeric priority wins even when it responds later") logText := logs.String() assert.Contains(t, logText, "duplicate package_id across providers") assert.Contains(t, logText, `"first_provider":"alpha"`) assert.Contains(t, logText, `"duplicate_provider":"beta"`) + assert.Contains(t, logText, `"winner_provider":"beta"`) + assert.Contains(t, logText, `"winner_priority":10`) assert.Contains(t, logText, `"package_id":"pkg-dup"`) } +func TestMergeContextResponses_EqualPriorityKeepsFirstResponse(t *testing.T) { + first := &tmproto.ContextMatchResponse{Offers: []tmproto.Offer{{PackageID: "pkg-dup", Summary: "first"}}} + second := &tmproto.ContextMatchResponse{Offers: []tmproto.Offer{{PackageID: "pkg-dup", Summary: "second"}}} + + merged := mergeContextResponses("ctx-priority-tie", []contextResult{ + {providerID: "first", priority: 10, response: first}, + {providerID: "second", priority: 10, response: second}, + }, nil) + + require.Len(t, merged.Offers, 1) + assert.Equal(t, "first", merged.Offers[0].Summary, "arrival order breaks equal-priority ties") +} + // TestMergeContextResponses_SingleProviderRepeat covers the within-response // repeat case: a single provider that returns the same package_id twice in // its own offers list. The warning names the provider once rather than