rohit/probo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Stop auto-creating org third parties on mapping

Browse Source 
The tracker-mapping worker materialized a per-org ThirdParty for every
categorized tracker, linking or creating one through heuristic and
disambiguation-agent matching. Concurrent mapping of two patterns for
the same common third party raced the load-then-create check and left
duplicate org third parties with the same name.

Reduce the worker to catalog resolution only: it resolves the shared
common_tracker_pattern_id / common_third_party_id link and leaves
third_party_id untouched, preserving any link set elsewhere. Org third
parties will instead be created through an explicit per-vendor import
action added in a later commit.

Remove resolveOrgThirdParty, prepareOrgThirdParty, the sibling
direct-link signal, and the disambiguation-agent wiring (including its
constructor parameter and buildTrackerAgents return), and update the
worker tests to assert the catalog link is resolved while third_party_id
is preserved.

Signed-off-by: Émile Ré <emile@probo.com>
This commit is contained in:
Émile Ré
2026-06-10 14:32:52 +02:00
parent 04a34c9757
commit 0f5ac03e70
4 changed files with 69 additions and 664 deletions
Download Patch File Download Diff File Expand all files Collapse all files

394
pkg/cookiebanner/tracker_mapping_worker.go
View File

@@ -39,20 +39,17 @@ import (
const defaultMappingStaleAfter = 10 * time.Minute const defaultMappingStaleAfter = 10 * time.Minute
type trackerMappingHandler struct { type trackerMappingHandler struct {
pg *pg.Client pg *pg.Client
logger *log.Logger logger *log.Logger
mappingAgent *agent.Agent mappingAgent *agent.Agent
disambiguationAgent *agent.Agent agentTimeout time.Duration
agentTimeout time.Duration staleAfter time.Duration
disambiguationTimeout time.Duration
staleAfter time.Duration
} }
func NewTrackerMappingWorker( func NewTrackerMappingWorker(
pgClient *pg.Client, pgClient *pg.Client,
logger *log.Logger, logger *log.Logger,
mappingCfg TrackerMappingAgentConfig, mappingCfg TrackerMappingAgentConfig,
disambiguationCfg thirdparty.DisambiguationAgentConfig,
staleAfter time.Duration, staleAfter time.Duration,
opts ...worker.Option, opts ...worker.Option,
) *worker.Worker[coredata.TrackerPattern] { ) *worker.Worker[coredata.TrackerPattern] {
@@ -66,21 +63,16 @@ func NewTrackerMappingWorker(
} }
h := &trackerMappingHandler{ h := &trackerMappingHandler{
pg: pgClient, pg: pgClient,
logger: logger, logger: logger,
agentTimeout: agentTimeout, agentTimeout: agentTimeout,
disambiguationTimeout: disambiguationCfg.Timeout, staleAfter: staleAfter,
staleAfter: staleAfter,
} }
if mappingCfg.LLMClient != nil { if mappingCfg.LLMClient != nil {
h.mappingAgent = buildTrackerMappingAgent(mappingCfg, pgClient, logger) h.mappingAgent = buildTrackerMappingAgent(mappingCfg, pgClient, logger)
} }
if disambiguationCfg.LLMClient != nil {
h.disambiguationAgent = thirdparty.BuildDisambiguationAgent(disambiguationCfg, logger)
}
return worker.New( return worker.New(
"tracker-mapping-worker", "tracker-mapping-worker",
h, h,
@@ -132,14 +124,11 @@ func (h *trackerMappingHandler) RecoverStale(ctx context.Context) error {
// catalogMatch is the result of a single catalog signal. commonPatternID // catalogMatch is the result of a single catalog signal. commonPatternID
// is the catalog row the signal resolved (or backfilled); commonThirdPartyID // is the catalog row the signal resolved (or backfilled); commonThirdPartyID
// is the catalog third party the signal discovered, when any; thirdPartyID // is the catalog third party the signal discovered, when any. A nil
// is an existing org ThirdParty the signal knows directly (e.g. a sibling // *catalogMatch means the signal produced nothing.
// pattern already promoted in the same organization). A nil *catalogMatch
// means the signal produced nothing.
type catalogMatch struct { type catalogMatch struct {
commonPatternID *gid.GID commonPatternID *gid.GID
commonThirdPartyID *gid.GID commonThirdPartyID *gid.GID
thirdPartyID *gid.GID
} }
// Process resolves the catalog mapping for a tracker pattern and links it // Process resolves the catalog mapping for a tracker pattern and links it
@@ -156,10 +145,11 @@ type catalogMatch struct {
// common_tracker_pattern_id but its catalog row has no common third // common_tracker_pattern_id but its catalog row has no common third
// party yet. // party yet.
// //
// Org ThirdParty resolution links to an existing party freely (even for // The worker no longer materializes per-org ThirdParty rows: it resolves
// uncategorised or extension-sourced patterns); only the creation of a // the shared catalog link only. An org ThirdParty is created exclusively
// brand new org ThirdParty stays gated behind categorisation and a // through the explicit per-vendor import action, which also backfills
// non-extension source. // tracker_patterns.third_party_id; an already-set third_party_id is
// preserved here untouched.
func (h *trackerMappingHandler) Process(ctx context.Context, tp coredata.TrackerPattern) error { func (h *trackerMappingHandler) Process(ctx context.Context, tp coredata.TrackerPattern) error {
scope := coredata.NewScopeFromObjectID(tp.ID) scope := coredata.NewScopeFromObjectID(tp.ID)
@@ -184,7 +174,6 @@ func (h *trackerMappingHandler) Process(ctx context.Context, tp coredata.Tracker
commonPatternID := det.commonPatternID commonPatternID := det.commonPatternID
commonThirdPartyID := det.commonThirdPartyID commonThirdPartyID := det.commonThirdPartyID
directThirdPartyID := det.directThirdPartyID
// Phase 2: tracker-mapping agent (no transaction). It runs only when // Phase 2: tracker-mapping agent (no transaction). It runs only when
// the deterministic signals could not resolve a catalog third party. // the deterministic signals could not resolve a catalog third party.
@@ -216,25 +205,6 @@ func (h *trackerMappingHandler) Process(ctx context.Context, tp coredata.Tracker
} }
} }
// Phase 3: org ThirdParty resolution. The heuristic ranking and the
// disambiguation agent run without a transaction; only the final link
// or create touches the database (in a short transaction).
thirdPartyID := tp.ThirdPartyID
if thirdPartyID == nil {
switch {
case directThirdPartyID != nil:
thirdPartyID = directThirdPartyID
case commonThirdPartyID != nil:
resolved, err := h.resolveOrgThirdParty(ctx, tp, *commonThirdPartyID)
if err != nil {
return fmt.Errorf("cannot resolve org third party: %w", err)
}
thirdPartyID = resolved
}
}
// Phase 4: persist the pattern mapping in a short transaction. The // Phase 4: persist the pattern mapping in a short transaction. The
// unmatched fallback keeps catalog coverage complete even when no // unmatched fallback keeps catalog coverage complete even when no
// vendor was resolved. // vendor was resolved.
@@ -251,7 +221,6 @@ func (h *trackerMappingHandler) Process(ctx context.Context, tp coredata.Tracker
} }
tp.CommonTrackerPatternID = commonPatternID tp.CommonTrackerPatternID = commonPatternID
tp.ThirdPartyID = thirdPartyID
tp.UpdatedAt = time.Now() tp.UpdatedAt = time.Now()
// Descriptions are owned by the common-pattern enrichment // Descriptions are owned by the common-pattern enrichment
@@ -321,7 +290,6 @@ type deterministicResult struct {
origin string origin string
commonPatternID *gid.GID commonPatternID *gid.GID
commonThirdPartyID *gid.GID commonThirdPartyID *gid.GID
directThirdPartyID *gid.GID
domains []string domains []string
commonThirdPartyPreexisted bool commonThirdPartyPreexisted bool
} }
@@ -403,7 +371,6 @@ func (h *trackerMappingHandler) resolveDeterministic(
if siblingMatch != nil { if siblingMatch != nil {
res.commonPatternID = firstNonNil(res.commonPatternID, siblingMatch.commonPatternID) res.commonPatternID = firstNonNil(res.commonPatternID, siblingMatch.commonPatternID)
res.commonThirdPartyID = siblingMatch.commonThirdPartyID res.commonThirdPartyID = siblingMatch.commonThirdPartyID
res.directThirdPartyID = siblingMatch.thirdPartyID
} }
if res.commonThirdPartyID != nil { if res.commonThirdPartyID != nil {
@@ -498,27 +465,6 @@ func (h *trackerMappingHandler) loadInitiatorDomains(
return domains, nil return domains, nil
} }
// creationAllowed reports whether the pattern is eligible for creating a
// brand new org ThirdParty. Extension-sourced patterns are never allowed
// to create one, and a pattern must be categorized first.
func (h *trackerMappingHandler) creationAllowed(
ctx context.Context,
conn pg.Querier,
scope coredata.Scoper,
tp coredata.TrackerPattern,
) (bool, error) {
if tp.Source != nil && *tp.Source == coredata.CookieSourceExtension {
return false, nil
}
var category coredata.CookieCategory
if err := category.LoadByID(ctx, conn, scope, tp.CookieCategoryID); err != nil {
return false, fmt.Errorf("cannot load cookie category: %w", err)
}
return category.Kind != coredata.CookieCategoryKindUncategorised, nil
}
// matchByPattern looks for a catalog row with the same pattern and // matchByPattern looks for a catalog row with the same pattern and
// surfaces both the row id and the common third party it points at (when // surfaces both the row id and the common third party it points at (when
// set), so the caller can short-circuit promotion or keep probing for a // set), so the caller can short-circuit promotion or keep probing for a
@@ -859,10 +805,8 @@ func (h *trackerMappingHandler) persistAgentIdentification(
// matchBySiblingOrigin finds other tracker patterns on the same banner // matchBySiblingOrigin finds other tracker patterns on the same banner
// that share initiator domains with the current pattern. Sharing an // that share initiator domains with the current pattern. Sharing an
// origin across multiple detected patterns is a strong indicator of the // origin across multiple detected patterns is a strong indicator of the
// same third party. When the siblings resolve to a single existing org // same third party, so the common third party the siblings resolve to is
// ThirdParty, that id is returned directly so promotion can link to it // upserted onto the catalog row.
// without re-running heuristics; otherwise the resolved common third
// party is upserted onto the catalog row.
func (h *trackerMappingHandler) matchBySiblingOrigin( func (h *trackerMappingHandler) matchBySiblingOrigin(
ctx context.Context, ctx context.Context,
tx pg.Tx, tx pg.Tx,
@@ -893,19 +837,14 @@ func (h *trackerMappingHandler) matchBySiblingOrigin(
scope := coredata.NewScopeFromObjectID(tp.ID) scope := coredata.NewScopeFromObjectID(tp.ID)
commonThirdPartyID, thirdPartyID, err := h.resolveThirdPartyFromSiblings(ctx, tx, scope, siblingIDs) commonThirdPartyID, err := h.resolveThirdPartyFromSiblings(ctx, tx, scope, siblingIDs)
if err != nil { if err != nil {
return nil, fmt.Errorf("cannot resolve third party from siblings: %w", err) return nil, fmt.Errorf("cannot resolve third party from siblings: %w", err)
} }
// No catalog third party to record: surface a directly-known org // No catalog third party to record: leave catalog creation to a later
// third party (if any) so promotion can still link to it, and leave // signal or the unmatched fallback.
// catalog creation to a later signal or the unmatched fallback.
if commonThirdPartyID == nil { if commonThirdPartyID == nil {
if thirdPartyID != nil {
return &catalogMatch{thirdPartyID: thirdPartyID}, nil
}
return nil, nil return nil, nil
} }
@@ -937,37 +876,25 @@ func (h *trackerMappingHandler) matchBySiblingOrigin(
return &catalogMatch{ return &catalogMatch{
commonPatternID: &commonPattern.ID, commonPatternID: &commonPattern.ID,
commonThirdPartyID: commonPattern.CommonThirdPartyID, commonThirdPartyID: commonPattern.CommonThirdPartyID,
thirdPartyID: thirdPartyID,
}, nil }, nil
} }
// resolveThirdPartyFromSiblings inspects sibling patterns to resolve a // resolveThirdPartyFromSiblings inspects sibling patterns to resolve a
// third party. It returns two independent signals: a direct org // single unambiguous catalog third party for backfill. It is resolved
// ThirdParty (set only when the siblings share a single one — the // first from the siblings' org ThirdParties, then, when those carry none,
// strongest, same-org signal), and a single unambiguous catalog third // from siblings' common_tracker_pattern rows. It returns nil when the
// party for backfill. The catalog third party is resolved first from the // siblings carry no catalog third party or disagree on one.
// siblings' org ThirdParties, then, when those carry none, from siblings'
// common_tracker_pattern rows. Either signal may be nil; siblings that
// disagree on the catalog third party resolve it to nothing.
func (h *trackerMappingHandler) resolveThirdPartyFromSiblings( func (h *trackerMappingHandler) resolveThirdPartyFromSiblings(
ctx context.Context, ctx context.Context,
conn pg.Querier, conn pg.Querier,
scope coredata.Scoper, scope coredata.Scoper,
siblingIDs []gid.GID, siblingIDs []gid.GID,
) (commonThirdPartyID *gid.GID, thirdPartyID *gid.GID, err error) { ) (commonThirdPartyID *gid.GID, err error) {
var patterns coredata.TrackerPatterns var patterns coredata.TrackerPatterns
thirdPartyIDs, err := patterns.LoadDistinctThirdPartyIDsByIDs(ctx, conn, scope, siblingIDs) thirdPartyIDs, err := patterns.LoadDistinctThirdPartyIDsByIDs(ctx, conn, scope, siblingIDs)
if err != nil { if err != nil {
return nil, nil, fmt.Errorf("cannot load distinct third party ids from siblings: %w", err) return nil, fmt.Errorf("cannot load distinct third party ids from siblings: %w", err)
}
// A single org third party shared across the siblings is the
// strongest, same-org signal: link to it directly. This is resolved
// independently from the catalog third party used for backfill.
if len(thirdPartyIDs) == 1 {
directID := thirdPartyIDs[0]
thirdPartyID = &directID
} }
if len(thirdPartyIDs) > 0 { if len(thirdPartyIDs) > 0 {
@@ -986,29 +913,26 @@ func (h *trackerMappingHandler) resolveThirdPartyFromSiblings(
if len(commonIDs) == 1 { if len(commonIDs) == 1 {
for id := range commonIDs { for id := range commonIDs {
return &id, thirdPartyID, nil return &id, nil
} }
} }
// Siblings are promoted to several different catalog third // Siblings are linked to several different catalog third
// parties: do not guess one. A single shared org third party (if // parties: do not guess one.
// any) is still a safe direct link.
if len(commonIDs) > 1 { if len(commonIDs) > 1 {
return nil, thirdPartyID, nil return nil, nil
} }
} }
// Fall back to siblings carrying only a common_tracker_pattern_id, or // Fall back to siblings carrying only a common_tracker_pattern_id, or
// whose org ThirdParty is not itself linked to the catalog. This is // whose org ThirdParty is not itself linked to the catalog.
// reached when the org-third-party scan above found no catalog third
// party, so it must not be short-circuited by a direct match.
commonPatternIDs, err := patterns.LoadDistinctCommonTrackerPatternIDsByIDs(ctx, conn, scope, siblingIDs) commonPatternIDs, err := patterns.LoadDistinctCommonTrackerPatternIDsByIDs(ctx, conn, scope, siblingIDs)
if err != nil { if err != nil {
return nil, nil, fmt.Errorf("cannot load distinct common tracker pattern ids from siblings: %w", err) return nil, fmt.Errorf("cannot load distinct common tracker pattern ids from siblings: %w", err)
} }
if len(commonPatternIDs) == 0 { if len(commonPatternIDs) == 0 {
return nil, thirdPartyID, nil return nil, nil
} }
commonIDs := make(map[gid.GID]struct{}) commonIDs := make(map[gid.GID]struct{})
@@ -1026,11 +950,11 @@ func (h *trackerMappingHandler) resolveThirdPartyFromSiblings(
if len(commonIDs) == 1 { if len(commonIDs) == 1 {
for id := range commonIDs { for id := range commonIDs {
return &id, thirdPartyID, nil return &id, nil
} }
} }
return nil, thirdPartyID, nil return nil, nil
} }
func (h *trackerMappingHandler) createUnmatchedPattern( func (h *trackerMappingHandler) createUnmatchedPattern(
@@ -1056,249 +980,3 @@ func (h *trackerMappingHandler) createUnmatchedPattern(
return &commonPattern.ID, nil return &commonPattern.ID, nil
} }
// resolveOrgThirdParty resolves an org ThirdParty for the given pattern
// from a known catalog third party. The resolution order is:
//
// 1. Exact link by common_third_party_id (O(1)).
// 2. Heuristic match against the org's existing ThirdParty rows
// (lowercased name, suffix-stripped name, slug, website host,
// CommonThirdPartyDomain overlap).
// 3. Agent disambiguation when the heuristic is ambiguous.
// 4. Fallback create from CommonThirdParty — only when allowCreate.
//
// Linking to an existing org ThirdParty (steps 1-3) is always allowed.
// Creating a brand new org ThirdParty (step 4) is gated by allowCreate:
// when false, the function returns (nil, nil) rather than creating one.
// A confident heuristic/agent match is auto-tagged with
// common_third_party_id so subsequent resolutions hit the exact-link
// path in O(1).
func (h *trackerMappingHandler) resolveOrgThirdParty(
ctx context.Context,
tp coredata.TrackerPattern,
commonThirdPartyID gid.GID,
) (*gid.GID, error) {
scope := coredata.NewScopeFromObjectID(tp.ID)
// Read phase: exact link, candidate ranking, eligibility, and
// creation gating. No write or LLM call happens here.
var prep orgThirdPartyPrep
if err := h.pg.WithConn(
ctx,
func(ctx context.Context, conn pg.Querier) error {
var err error
prep, err = h.prepareOrgThirdParty(ctx, conn, scope, tp, commonThirdPartyID)
return err
},
); err != nil {
return nil, err
}
if prep.existingID != nil {
return prep.existingID, nil
}
picked := prep.highConfidence
viaAgent := false
// Agent phase (no transaction): disambiguate among the heuristic
// candidates when none scored high enough on its own.
if picked == nil && prep.eligibleForAgent && h.disambiguationAgent != nil {
matchedID, err := thirdparty.Disambiguate(
ctx,
h.disambiguationAgent,
h.logger,
prep.commonParty,
prep.commonDomains,
prep.agentSet,
h.disambiguationTimeout,
)
if err != nil {
h.logger.WarnCtx(
ctx,
"third-party disambiguation agent failed",
log.Error(err),
log.String("tracker_pattern_id", tp.ID.String()),
)
}
if matchedID != nil {
for _, c := range prep.agentSet {
if c.ThirdParty.ID == *matchedID {
picked = c.ThirdParty
viaAgent = true
break
}
}
}
}
// Nothing to link and creation is not allowed: leave the pattern
// without an org third party.
if picked == nil && !prep.allowCreate {
return nil, nil
}
// Write phase: link the picked candidate or create a new org third
// party from the catalog entry, in a short transaction.
var resolved *gid.GID
if err := h.pg.WithTx(
ctx,
func(ctx context.Context, tx pg.Tx) error {
if picked != nil {
if err := thirdparty.LinkToCommon(ctx, tx, scope, picked, commonThirdPartyID); err != nil {
return fmt.Errorf("cannot link third party to common: %w", err)
}
if viaAgent {
h.logger.InfoCtx(
ctx,
"promoted tracker pattern via disambiguation agent",
log.String("tracker_pattern_id", tp.ID.String()),
log.String("third_party_id", picked.ID.String()),
)
} else {
h.logger.InfoCtx(
ctx,
"promoted tracker pattern via heuristic match",
log.String("tracker_pattern_id", tp.ID.String()),
log.String("third_party_id", picked.ID.String()),
log.Float64("score", prep.highScore),
)
}
resolved = &picked.ID
return nil
}
created, err := thirdparty.CreateFromCommon(ctx, tx, scope, tp.OrganizationID, prep.commonParty)
if err != nil {
return fmt.Errorf("cannot create third party from common: %w", err)
}
h.logger.InfoCtx(
ctx,
"promoted tracker pattern by creating org third party from catalog",
log.String("tracker_pattern_id", tp.ID.String()),
log.String("third_party_id", created.ID.String()),
log.String("common_third_party_id", commonThirdPartyID.String()),
)
resolved = &created.ID
return nil
},
); err != nil {
return nil, err
}
return resolved, nil
}
// orgThirdPartyPrep is the read-phase outcome for org ThirdParty
// resolution. existingID is set when an exact common-id link already
// exists (the other fields are then unused). Otherwise highConfidence
// holds a heuristic match at or above HighConfidenceScore (with
// highScore), or agentSet/eligibleForAgent describe the disambiguation
// candidates. allowCreate gates falling back to creating a new org
// ThirdParty from the catalog entry.
type orgThirdPartyPrep struct {
existingID *gid.GID
commonParty coredata.CommonThirdParty
commonDomains coredata.CommonThirdPartyDomains
agentSet []thirdparty.ScoredCandidate
highConfidence *coredata.ThirdParty
highScore float64
eligibleForAgent bool
allowCreate bool
}
// prepareOrgThirdParty performs the read-only work for org ThirdParty
// resolution: it checks for an exact common-id link, loads the catalog
// entry and the org's existing third parties, ranks the candidates, and
// computes creation eligibility. It makes no writes and no LLM call.
func (h *trackerMappingHandler) prepareOrgThirdParty(
ctx context.Context,
conn pg.Querier,
scope coredata.Scoper,
tp coredata.TrackerPattern,
commonThirdPartyID gid.GID,
) (orgThirdPartyPrep, error) {
var prep orgThirdPartyPrep
var existing coredata.ThirdParty
err := existing.LoadByOrganizationIDAndCommonThirdPartyID(
ctx,
conn,
scope,
tp.OrganizationID,
commonThirdPartyID,
)
if err == nil {
id := existing.ID
prep.existingID = &id
return prep, nil
}
if !errors.Is(err, coredata.ErrResourceNotFound) {
return prep, fmt.Errorf("cannot load org third party by common id: %w", err)
}
if err := prep.commonParty.LoadByID(ctx, conn, commonThirdPartyID); err != nil {
return prep, fmt.Errorf("cannot load common third party: %w", err)
}
if err := prep.commonDomains.LoadByCommonThirdPartyID(ctx, conn, commonThirdPartyID); err != nil {
return prep, fmt.Errorf("cannot load common third party domains: %w", err)
}
firstLevel := 1
var orgThirdParties coredata.ThirdParties
if err := orgThirdParties.LoadAllByOrganizationID(
ctx,
conn,
scope,
tp.OrganizationID,
coredata.NewThirdPartyFilter(nil, &firstLevel, nil),
); err != nil {
return prep, fmt.Errorf("cannot load org third parties: %w", err)
}
ranked := thirdparty.RankCandidates(prep.commonParty, prep.commonDomains, orgThirdParties)
if len(ranked) > 0 && ranked[0].Score >= thirdparty.HighConfidenceScore {
prep.highConfidence = ranked[0].ThirdParty
prep.highScore = ranked[0].Score
} else {
prep.agentSet = ranked
if len(prep.agentSet) > thirdparty.MaxAgentCandidates {
prep.agentSet = prep.agentSet[:thirdparty.MaxAgentCandidates]
}
for _, c := range prep.agentSet {
if c.Score >= thirdparty.MinAgentScore {
prep.eligibleForAgent = true
break
}
}
}
allowCreate, err := h.creationAllowed(ctx, conn, scope, tp)
if err != nil {
return prep, err
}
prep.allowCreate = allowCreate
return prep, nil
}

284
pkg/cookiebanner/tracker_mapping_worker_test.go
View File

@@ -31,10 +31,10 @@ import (
// promotionFixture extends workerFixture with a CommonThirdParty and a // promotionFixture extends workerFixture with a CommonThirdParty and a
// CommonTrackerPattern linking the catalog to the test pattern. It is // CommonTrackerPattern linking the catalog to the test pattern. It is
// the minimum scaffolding resolveOrgThirdParty needs to run end-to-end. // the minimum scaffolding the catalog-resolution paths need to run
// end-to-end.
type promotionFixture struct { type promotionFixture struct {
workerFixture workerFixture
commonThirdParty coredata.CommonThirdParty
commonPatternID gid.GID commonPatternID gid.GID
trackerPattern coredata.TrackerPattern trackerPattern coredata.TrackerPattern
commonThirdPartyID gid.GID commonThirdPartyID gid.GID
@@ -130,7 +130,6 @@ func seedPromotionFixture(t *testing.T, ctx context.Context, client *pg.Client)
return promotionFixture{ return promotionFixture{
workerFixture: fx, workerFixture: fx,
commonThirdParty: commonThirdParty,
commonPatternID: commonPattern.ID, commonPatternID: commonPattern.ID,
commonThirdPartyID: commonThirdPartyID, commonThirdPartyID: commonThirdPartyID,
trackerPattern: pattern, trackerPattern: pattern,
@@ -144,142 +143,6 @@ func newMappingHandler(client *pg.Client) *trackerMappingHandler {
} }
} }
// promote runs resolveOrgThirdParty, which manages its own short
// transactions internally (creation gating is derived from the
// pattern's category, not passed in).
func promote(
t *testing.T,
ctx context.Context,
h *trackerMappingHandler,
tp coredata.TrackerPattern,
commonThirdPartyID gid.GID,
) *gid.GID {
t.Helper()
got, err := h.resolveOrgThirdParty(ctx, tp, commonThirdPartyID)
require.NoError(t, err)
return got
}
func TestPromoteThirdParty_ExactCommonLink(t *testing.T) {
t.Parallel()
client := test.PGClient(t)
ctx := context.Background()
fx := seedPromotionFixture(t, ctx, client)
now := time.Now().UTC().Truncate(time.Microsecond)
existing := coredata.ThirdParty{
ID: gid.New(fx.scope.GetTenantID(), coredata.ThirdPartyEntityType),
OrganizationID: fx.organizationID,
CommonThirdPartyID: &fx.commonThirdPartyID,
Name: "Google LLC",
Category: coredata.ThirdPartyCategoryAnalytics,
Certifications: []string{},
Countries: coredata.CountryCodes{},
CreatedAt: now,
UpdatedAt: now,
}
require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error {
return existing.Insert(ctx, tx, fx.scope)
}))
got := promote(t, ctx, newMappingHandler(client), fx.trackerPattern, fx.commonThirdPartyID)
require.NotNil(t, got)
assert.Equal(t, existing.ID, *got, "should return the existing org ThirdParty linked by common id")
}
func TestPromoteThirdParty_HeuristicMatch(t *testing.T) {
t.Parallel()
client := test.PGClient(t)
ctx := context.Background()
fx := seedPromotionFixture(t, ctx, client)
now := time.Now().UTC().Truncate(time.Microsecond)
// Append a corporate suffix to the catalog name so the heuristic
// matches on the suffix-stripped name (score 0.9) rather than an
// exact link.
manualEntry := coredata.ThirdParty{
ID: gid.New(fx.scope.GetTenantID(), coredata.ThirdPartyEntityType),
OrganizationID: fx.organizationID,
Name: fx.commonThirdParty.Name + " LLC",
Category: coredata.ThirdPartyCategoryAnalytics,
Certifications: []string{},
Countries: coredata.CountryCodes{},
CreatedAt: now,
UpdatedAt: now,
}
require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error {
return manualEntry.Insert(ctx, tx, fx.scope)
}))
got := promote(t, ctx, newMappingHandler(client), fx.trackerPattern, fx.commonThirdPartyID)
require.NotNil(t, got)
assert.Equal(t, manualEntry.ID, *got, "heuristic match should return the manually-entered ThirdParty")
var reloaded coredata.ThirdParty
require.NoError(t, client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error {
return reloaded.LoadByID(ctx, conn, fx.scope, manualEntry.ID)
}))
require.NotNil(t, reloaded.CommonThirdPartyID, "matched row must be tagged with common_third_party_id")
assert.Equal(t, fx.commonThirdPartyID, *reloaded.CommonThirdPartyID)
}
func TestPromoteThirdParty_FallbackCreate(t *testing.T) {
t.Parallel()
client := test.PGClient(t)
ctx := context.Background()
fx := seedPromotionFixture(t, ctx, client)
got := promote(t, ctx, newMappingHandler(client), fx.trackerPattern, fx.commonThirdPartyID)
require.NotNil(t, got, "fallback should create a new ThirdParty")
var reloaded coredata.ThirdParty
require.NoError(t, client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error {
return reloaded.LoadByID(ctx, conn, fx.scope, *got)
}))
assert.Equal(t, fx.organizationID, reloaded.OrganizationID)
assert.Equal(t, fx.commonThirdParty.Name, reloaded.Name)
require.NotNil(t, reloaded.CommonThirdPartyID)
assert.Equal(t, fx.commonThirdPartyID, *reloaded.CommonThirdPartyID)
assert.Equal(t, coredata.ThirdPartyCategoryAnalytics, reloaded.Category)
assert.Equal(t, 1, reloaded.Level)
assert.False(t, reloaded.ShowOnTrustCenter)
}
// TestResolveOrgThirdParty_CreationGated asserts that when no existing
// org ThirdParty matches the catalog third party, creating a new one is
// suppressed for an uncategorised pattern (creation gating is derived
// from the pattern's category) and proceeds for a categorised one.
func TestResolveOrgThirdParty_CreationGated(t *testing.T) {
t.Parallel()
client := test.PGClient(t)
ctx := context.Background()
fx := seedPromotionFixture(t, ctx, client)
gatedPattern := fx.trackerPattern
gatedPattern.CookieCategoryID = fx.uncategorisedID
gated := promote(t, ctx, newMappingHandler(client), gatedPattern, fx.commonThirdPartyID)
assert.Nil(t, gated, "creation must be suppressed for an uncategorised pattern with nothing to link")
allowed := promote(t, ctx, newMappingHandler(client), fx.trackerPattern, fx.commonThirdPartyID)
require.NotNil(t, allowed, "creation must proceed for a categorised pattern")
}
// TestProcess_PreservesCatalogMappingOnReTrigger asserts that when // TestProcess_PreservesCatalogMappingOnReTrigger asserts that when
// Process is called for a pattern that already carries a // Process is called for a pattern that already carries a
// common_tracker_pattern_id, the catalog pipeline is skipped and the // common_tracker_pattern_id, the catalog pipeline is skipped and the
@@ -306,7 +169,7 @@ func TestProcess_PreservesCatalogMappingOnReTrigger(t *testing.T) {
require.NotNil(t, reloaded.CommonTrackerPatternID, "common tracker pattern link must be preserved") require.NotNil(t, reloaded.CommonTrackerPatternID, "common tracker pattern link must be preserved")
assert.Equal(t, fx.commonPatternID, *reloaded.CommonTrackerPatternID) assert.Equal(t, fx.commonPatternID, *reloaded.CommonTrackerPatternID)
require.NotNil(t, reloaded.ThirdPartyID, "the worker should have promoted to an org ThirdParty") assert.Nil(t, reloaded.ThirdPartyID, "the worker must not auto-create or link an org ThirdParty")
} }
// TestProcess_UncategorisedPatternIsNotPromoted asserts that a pattern // TestProcess_UncategorisedPatternIsNotPromoted asserts that a pattern
@@ -579,8 +442,6 @@ func TestMatchBySiblingOrigin_SiblingWithThirdPartyID(t *testing.T) {
require.NotNil(t, got, "sibling origin match should return a catalog match") require.NotNil(t, got, "sibling origin match should return a catalog match")
require.NotNil(t, got.commonPatternID, "sibling origin match should return a common tracker pattern ID") require.NotNil(t, got.commonPatternID, "sibling origin match should return a common tracker pattern ID")
require.NotNil(t, got.thirdPartyID, "sibling origin match should surface the sibling's org third party directly")
assert.Equal(t, orgThirdParty.ID, *got.thirdPartyID)
var commonPattern coredata.CommonTrackerPattern var commonPattern coredata.CommonTrackerPattern
@@ -1013,56 +874,11 @@ func TestMatchBySiblingOrigin_ConvergentSiblings(t *testing.T) {
assert.Equal(t, fx.commonThirdPartyID, *commonPattern.CommonThirdPartyID) assert.Equal(t, fx.commonThirdPartyID, *commonPattern.CommonThirdPartyID)
} }
func TestPromoteThirdParty_ExactCommonLinkIgnoresSimilarUnlinked(t *testing.T) {
t.Parallel()
client := test.PGClient(t)
ctx := context.Background()
fx := seedPromotionFixture(t, ctx, client)
now := time.Now().UTC().Truncate(time.Microsecond)
manualEntry := coredata.ThirdParty{
ID: gid.New(fx.scope.GetTenantID(), coredata.ThirdPartyEntityType),
OrganizationID: fx.organizationID,
Name: "Google LLC",
Category: coredata.ThirdPartyCategoryAnalytics,
Certifications: []string{},
Countries: coredata.CountryCodes{},
CreatedAt: now,
UpdatedAt: now,
}
linked := coredata.ThirdParty{
ID: gid.New(fx.scope.GetTenantID(), coredata.ThirdPartyEntityType),
OrganizationID: fx.organizationID,
CommonThirdPartyID: &fx.commonThirdPartyID,
Name: "Google",
Category: coredata.ThirdPartyCategoryAnalytics,
Certifications: []string{},
Countries: coredata.CountryCodes{},
CreatedAt: now,
UpdatedAt: now,
}
require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error {
if err := manualEntry.Insert(ctx, tx, fx.scope); err != nil {
return err
}
return linked.Insert(ctx, tx, fx.scope)
}))
got := promote(t, ctx, newMappingHandler(client), fx.trackerPattern, fx.commonThirdPartyID)
require.NotNil(t, got)
assert.Equal(t, linked.ID, *got, "exact-link path must short-circuit before the heuristic fires")
}
// TestProcess_BackfillsCommonThirdPartyFromSibling asserts that a pattern // TestProcess_BackfillsCommonThirdPartyFromSibling asserts that a pattern
// linked to an unlinked catalog row (no common_third_party_id) gets its // linked to an unlinked catalog row (no common_third_party_id) gets its
// catalog row backfilled from a sibling signal, and is promoted directly // catalog row backfilled from a sibling signal. The worker resolves the
// to the sibling's existing org ThirdParty. // catalog link only; it must not promote the pattern to an org
// ThirdParty.
func TestProcess_BackfillsCommonThirdPartyFromSibling(t *testing.T) { func TestProcess_BackfillsCommonThirdPartyFromSibling(t *testing.T) {
t.Parallel() t.Parallel()
@@ -1206,81 +1022,18 @@ func TestProcess_BackfillsCommonThirdPartyFromSibling(t *testing.T) {
return reloadedTarget.LoadByID(ctx, conn, fx.scope, target.ID) return reloadedTarget.LoadByID(ctx, conn, fx.scope, target.ID)
})) }))
require.NotNil(t, reloadedTarget.ThirdPartyID, "target must be promoted to the sibling's org third party") assert.Nil(t, reloadedTarget.ThirdPartyID, "target must not be auto-promoted to an org third party")
assert.Equal(t, orgThirdParty.ID, *reloadedTarget.ThirdPartyID)
require.NotNil(t, reloadedTarget.CommonTrackerPatternID) require.NotNil(t, reloadedTarget.CommonTrackerPatternID)
assert.Equal(t, unlinkedCommon.ID, *reloadedTarget.CommonTrackerPatternID, "the existing catalog link must be preserved") assert.Equal(t, unlinkedCommon.ID, *reloadedTarget.CommonTrackerPatternID, "the existing catalog link must be preserved")
} }
// TestProcess_UncategorisedLinksExistingThirdParty asserts that an // TestProcess_SiblingCatalogResolutionOnFirstPartyOrigin asserts that a
// uncategorised pattern is still linked to an already-existing matching // pattern detected on the banner's own (first-party) origin is still
// org ThirdParty (linking to an existing party is ungated); only the // grouped with its siblings sharing that origin for catalog resolution.
// creation of a new party stays gated, as covered by // Sibling matching is an org-local co-occurrence signal and must not be
// TestProcess_UncategorisedPatternIsNotPromoted. // defeated by the first-party domain filter that only protects the global
func TestProcess_UncategorisedLinksExistingThirdParty(t *testing.T) { // catalog (domain) match.
t.Parallel() func TestProcess_SiblingCatalogResolutionOnFirstPartyOrigin(t *testing.T) {
client := test.PGClient(t)
ctx := context.Background()
fx := seedPromotionFixture(t, ctx, client)
now := time.Now().UTC().Truncate(time.Microsecond)
existing := coredata.ThirdParty{
ID: gid.New(fx.scope.GetTenantID(), coredata.ThirdPartyEntityType),
OrganizationID: fx.organizationID,
CommonThirdPartyID: &fx.commonThirdPartyID,
Name: "Google LLC",
Category: coredata.ThirdPartyCategoryAnalytics,
Certifications: []string{},
Countries: coredata.CountryCodes{},
CreatedAt: now,
UpdatedAt: now,
}
require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error {
if err := existing.Insert(ctx, tx, fx.scope); err != nil {
return err
}
_, err := tx.Exec(
ctx,
`UPDATE tracker_patterns
SET cookie_category_id = $1,
mapping_requested_at = $2
WHERE id = $3`,
fx.uncategorisedID,
now,
fx.trackerPattern.ID,
)
return err
}))
var reloadedBefore coredata.TrackerPattern
require.NoError(t, client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error {
return reloadedBefore.LoadByID(ctx, conn, fx.scope, fx.trackerPattern.ID)
}))
h := newMappingHandler(client)
require.NoError(t, h.Process(ctx, reloadedBefore))
var reloaded coredata.TrackerPattern
require.NoError(t, client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error {
return reloaded.LoadByID(ctx, conn, fx.scope, fx.trackerPattern.ID)
}))
require.NotNil(t, reloaded.ThirdPartyID, "uncategorised pattern must still link to an existing org third party")
assert.Equal(t, existing.ID, *reloaded.ThirdPartyID)
}
// TestProcess_SiblingPromotionOnFirstPartyOrigin asserts that a pattern
// detected on the banner's own (first-party) origin is still grouped with
// its siblings sharing that origin. Sibling matching is an org-local
// co-occurrence signal and must not be defeated by the first-party domain
// filter that only protects the global catalog (domain) match.
func TestProcess_SiblingPromotionOnFirstPartyOrigin(t *testing.T) {
t.Parallel() t.Parallel()
client := test.PGClient(t) client := test.PGClient(t)
@@ -1424,8 +1177,7 @@ func TestProcess_SiblingPromotionOnFirstPartyOrigin(t *testing.T) {
return reloadedTarget.LoadByID(ctx, conn, fx.scope, target.ID) return reloadedTarget.LoadByID(ctx, conn, fx.scope, target.ID)
})) }))
require.NotNil(t, reloadedTarget.ThirdPartyID, "target sharing a first-party origin must be promoted via its sibling") assert.Nil(t, reloadedTarget.ThirdPartyID, "target must not be auto-promoted to an org third party")
assert.Equal(t, orgThirdParty.ID, *reloadedTarget.ThirdPartyID)
} }
// TestProcess_ReenqueuesUnmappedSiblingOnResolve asserts that when a // TestProcess_ReenqueuesUnmappedSiblingOnResolve asserts that when a
@@ -1543,7 +1295,8 @@ func TestProcess_ReenqueuesUnmappedSiblingOnResolve(t *testing.T) {
return reloadedTarget.LoadByID(ctx, conn, fx.scope, target.ID) return reloadedTarget.LoadByID(ctx, conn, fx.scope, target.ID)
})) }))
require.NotNil(t, reloadedTarget.ThirdPartyID, "target must resolve via its promoted sibling") require.NotNil(t, reloadedTarget.CommonTrackerPatternID, "target must resolve a catalog link via its sibling")
assert.Nil(t, reloadedTarget.ThirdPartyID, "target must not be auto-promoted to an org third party")
var reloadedUnmapped coredata.TrackerPattern var reloadedUnmapped coredata.TrackerPattern
@@ -1697,7 +1450,8 @@ func TestProcess_DoesNotReenqueuePromotedOrExtensionSiblings(t *testing.T) {
return p return p
} }
require.NotNil(t, reload(target.ID).ThirdPartyID, "target must resolve via its promoted sibling") require.NotNil(t, reload(target.ID).CommonTrackerPatternID, "target must resolve a catalog link via its sibling")
assert.Nil(t, reload(target.ID).ThirdPartyID, "target must not be auto-promoted to an org third party")
require.NotNil(t, reload(plainSibling.ID).MappingRequestedAt, "plain unmapped sibling must be re-enqueued") require.NotNil(t, reload(plainSibling.ID).MappingRequestedAt, "plain unmapped sibling must be re-enqueued")
assert.Nil(t, reload(mappedSibling.ID).MappingRequestedAt, "promoted sibling must not be re-enqueued") assert.Nil(t, reload(mappedSibling.ID).MappingRequestedAt, "promoted sibling must not be re-enqueued")
assert.Nil(t, reload(extensionSibling.ID).MappingRequestedAt, "EXTENSION-sourced sibling must not be re-enqueued") assert.Nil(t, reload(extensionSibling.ID).MappingRequestedAt, "EXTENSION-sourced sibling must not be re-enqueued")

3
pkg/probod/probod.go
View File

@@ -321,7 +321,7 @@ func (impl *Implm) Run(
return err return err
} }
trackerMappingCfg, trackerEnrichmentCfg, thirdPartyDisambiguationCfg, err := impl.buildTrackerAgents(l, tp, r) trackerMappingCfg, trackerEnrichmentCfg, err := impl.buildTrackerAgents(l, tp, r)
if err != nil { if err != nil {
return err return err
} }
@@ -775,7 +775,6 @@ func (impl *Implm) Run(
pgClient, pgClient,
l, l,
trackerMappingCfg, trackerMappingCfg,
thirdPartyDisambiguationCfg,
time.Duration(impl.cfg.TrackerMappingWorker.StaleAfter)*time.Second, time.Duration(impl.cfg.TrackerMappingWorker.StaleAfter)*time.Second,
worker.WithInterval(time.Duration(impl.cfg.TrackerMappingWorker.Interval)*time.Second), worker.WithInterval(time.Duration(impl.cfg.TrackerMappingWorker.Interval)*time.Second),
worker.WithMaxConcurrency(impl.cfg.TrackerMappingWorker.MaxConcurrency), worker.WithMaxConcurrency(impl.cfg.TrackerMappingWorker.MaxConcurrency),

52
pkg/probod/tracker_agents.go
View File

@@ -22,27 +22,25 @@ import (
"go.gearno.de/kit/log" "go.gearno.de/kit/log"
"go.opentelemetry.io/otel/trace" "go.opentelemetry.io/otel/trace"
"go.probo.inc/probo/pkg/cookiebanner" "go.probo.inc/probo/pkg/cookiebanner"
"go.probo.inc/probo/pkg/thirdparty"
) )
// buildTrackerAgents wires the three tracker agents from the probod // buildTrackerAgents wires the two tracker agents from the probod
// config, each with its own LLM client and tuning: the tracker-mapping // config, each with its own LLM client and tuning: the tracker-mapping
// agent (catalog identification), the common-pattern enrichment agent // agent (catalog identification) and the common-pattern enrichment agent
// (description research), and the third-party disambiguation agent. All // (description research). Both are opt-in: when
// are opt-in: when `llm.tracker-mapping.provider` is empty it returns // `llm.tracker-mapping.provider` is empty it returns zero configs (nil
// zero configs (nil LLM clients) so callers run without agent fallback. // LLM clients) so callers run without agent fallback.
// //
// The enrichment and disambiguation agents fall back to the // The enrichment agent falls back to the tracker-mapping config when its
// tracker-mapping config when their own provider slot is empty, so a // own provider slot is empty, so a deployment that configures only
// deployment that configures only `tracker-mapping` keeps wiring all // `tracker-mapping` keeps wiring both agents.
// three agents.
func (impl *Implm) buildTrackerAgents( func (impl *Implm) buildTrackerAgents(
l *log.Logger, l *log.Logger,
tp trace.TracerProvider, tp trace.TracerProvider,
r prometheus.Registerer, r prometheus.Registerer,
) (cookiebanner.TrackerMappingAgentConfig, cookiebanner.TrackerEnrichmentAgentConfig, thirdparty.DisambiguationAgentConfig, error) { ) (cookiebanner.TrackerMappingAgentConfig, cookiebanner.TrackerEnrichmentAgentConfig, error) {
if impl.cfg.Agents.TrackerMapping.Provider == "" { if impl.cfg.Agents.TrackerMapping.Provider == "" {
return cookiebanner.TrackerMappingAgentConfig{}, cookiebanner.TrackerEnrichmentAgentConfig{}, thirdparty.DisambiguationAgentConfig{}, nil return cookiebanner.TrackerMappingAgentConfig{}, cookiebanner.TrackerEnrichmentAgentConfig{}, nil
} }
firecrawlAPIKey := impl.cfg.Agents.Tools.FirecrawlAPIKey firecrawlAPIKey := impl.cfg.Agents.Tools.FirecrawlAPIKey
@@ -55,7 +53,7 @@ func (impl *Implm) buildTrackerAgents(
r, r,
) )
if err != nil { if err != nil {
return cookiebanner.TrackerMappingAgentConfig{}, cookiebanner.TrackerEnrichmentAgentConfig{}, thirdparty.DisambiguationAgentConfig{}, fmt.Errorf("cannot resolve tracker mapping agent client: %w", err) return cookiebanner.TrackerMappingAgentConfig{}, cookiebanner.TrackerEnrichmentAgentConfig{}, fmt.Errorf("cannot resolve tracker mapping agent client: %w", err)
} }
mappingCfg := cookiebanner.TrackerMappingAgentConfig{ mappingCfg := cookiebanner.TrackerMappingAgentConfig{
@@ -81,7 +79,7 @@ func (impl *Implm) buildTrackerAgents(
r, r,
) )
if err != nil { if err != nil {
return cookiebanner.TrackerMappingAgentConfig{}, cookiebanner.TrackerEnrichmentAgentConfig{}, thirdparty.DisambiguationAgentConfig{}, fmt.Errorf("cannot resolve tracker enrichment agent client: %w", err) return cookiebanner.TrackerMappingAgentConfig{}, cookiebanner.TrackerEnrichmentAgentConfig{}, fmt.Errorf("cannot resolve tracker enrichment agent client: %w", err)
} }
enrichmentCfg := cookiebanner.TrackerEnrichmentAgentConfig{ enrichmentCfg := cookiebanner.TrackerEnrichmentAgentConfig{
@@ -94,29 +92,5 @@ func (impl *Implm) buildTrackerAgents(
MaxTurns: impl.cfg.CommonPatternEnrichmentWorker.AgentMaxTurns, MaxTurns: impl.cfg.CommonPatternEnrichmentWorker.AgentMaxTurns,
} }
disambiguationSlot := impl.cfg.Agents.ThirdPartyDisambiguation return mappingCfg, enrichmentCfg, nil
if disambiguationSlot.Provider == "" {
disambiguationSlot = impl.cfg.Agents.TrackerMapping
}
disambiguationAgentCfg, disambiguationClient, err := impl.resolveAgentClient(
"third-party-disambiguation",
disambiguationSlot,
l,
tp,
r,
)
if err != nil {
return cookiebanner.TrackerMappingAgentConfig{}, cookiebanner.TrackerEnrichmentAgentConfig{}, thirdparty.DisambiguationAgentConfig{}, fmt.Errorf("cannot resolve third party disambiguation agent client: %w", err)
}
disambiguationCfg := thirdparty.DisambiguationAgentConfig{
LLMClient: disambiguationClient,
Model: disambiguationAgentCfg.ModelName,
MaxTokens: disambiguationAgentCfg.MaxTokens,
Temperature: disambiguationAgentCfg.Temperature,
Timeout: time.Duration(impl.cfg.TrackerMappingWorker.DisambiguationAgentTimeout) * time.Second,
}
return mappingCfg, enrichmentCfg, disambiguationCfg, nil
} }