Harden catalog vendor resolution and agent prompt
Address review feedback on the agent-driven tracker catalog path: - Return initiator-domain load failures instead of swallowing them, so the worker retries rather than running the agent on partial context. - In the resolver, treat only ErrResourceNotFound as a catalog miss and propagate genuine name/slug lookup errors. - Insert the new vendor inside a savepoint and, on the slug unique-violation race, reload and return the winning row instead of aborting the caller's transaction. - Stop seeding common_third_party_domains from observed initiator domains. They are a co-occurrence signal, not verified ownership, and writing them into the global cross-tenant catalog pollutes the domain-based matcher. The curated seed owns that data. - Warn the mapping agent that observed domains may belong to shared CDNs, tag managers, or hosting infrastructure rather than the vendor, so it does not attribute on that basis alone. - Extract a shared tracker-identification prompt helper and move the common-pattern identification prompt next to the enrichment agent. Signed-off-by: Émile Ré <emile@probo.com>
This commit is contained in:
@@ -75,6 +75,15 @@ func buildCommonPatternEnrichmentAgent(
|
|||||||
return agent.New("common-pattern-enrichment", cfg.LLMClient, opts...)
|
return agent.New("common-pattern-enrichment", cfg.LLMClient, opts...)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// buildCommonPatternIdentificationPrompt builds the mapping-agent input
|
||||||
|
// for a global catalog pattern. Catalog rows carry no observed domains,
|
||||||
|
// so the prompt omits the <observed_domains> signal and relies on the
|
||||||
|
// pattern name, type, and naming conventions. It lets the enrichment
|
||||||
|
// worker reuse the mapping agent to attribute a vendor before describing.
|
||||||
|
func buildCommonPatternIdentificationPrompt(cp coredata.CommonTrackerPattern) string {
|
||||||
|
return buildTrackerIdentificationPrompt(cp.Pattern, cp.TrackerType, cp.MatchType, cp.MaxAgeSeconds)
|
||||||
|
}
|
||||||
|
|
||||||
func buildEnrichmentPrompt(cp coredata.CommonTrackerPattern, thirdPartyName string) string {
|
func buildEnrichmentPrompt(cp coredata.CommonTrackerPattern, thirdPartyName string) string {
|
||||||
maxAge := "session"
|
maxAge := "session"
|
||||||
if cp.MaxAgeSeconds != nil {
|
if cp.MaxAgeSeconds != nil {
|
||||||
|
|||||||
@@ -151,7 +151,7 @@ func (h *commonPatternEnrichmentHandler) Process(ctx context.Context, cp coredat
|
|||||||
var thirdPartyID *gid.GID
|
var thirdPartyID *gid.GID
|
||||||
|
|
||||||
if attribution != nil && cp.CommonThirdPartyID == nil {
|
if attribution != nil && cp.CommonThirdPartyID == nil {
|
||||||
thirdPartyID, err = thirdparty.ResolveOrCreateCommonThirdParty(ctx, tx, h.logger, attribution.ThirdPartyName, attribution.Category, nil)
|
thirdPartyID, err = thirdparty.ResolveOrCreateCommonThirdParty(ctx, tx, h.logger, attribution.ThirdPartyName, attribution.Category)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("cannot resolve or create common third party: %w", err)
|
return fmt.Errorf("cannot resolve or create common third party: %w", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -137,40 +137,19 @@ func trackerMappingInstructions(_ context.Context, _ *agent.Agent) string {
|
|||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
func buildAgentPrompt(tp coredata.TrackerPattern, domains []string) string {
|
// buildTrackerIdentificationPrompt renders the base mapping-agent input
|
||||||
|
// shared by live tracker patterns and global catalog patterns: the four
|
||||||
|
// XML signal tags plus the max-age preamble. Callers append any extra
|
||||||
|
// signals (e.g. observed domains) to the returned prompt.
|
||||||
|
func buildTrackerIdentificationPrompt(
|
||||||
|
pattern string,
|
||||||
|
trackerType coredata.TrackerType,
|
||||||
|
matchType coredata.TrackerPatternMatchType,
|
||||||
|
maxAgeSeconds *int,
|
||||||
|
) string {
|
||||||
maxAge := "session"
|
maxAge := "session"
|
||||||
if tp.MaxAgeSeconds != nil {
|
if maxAgeSeconds != nil {
|
||||||
maxAge = fmt.Sprintf("%d seconds", *tp.MaxAgeSeconds)
|
maxAge = fmt.Sprintf("%d seconds", *maxAgeSeconds)
|
||||||
}
|
|
||||||
|
|
||||||
prompt := fmt.Sprintf(
|
|
||||||
"Identify the following tracker:\n\n"+
|
|
||||||
"<pattern> %s </pattern>\n"+
|
|
||||||
"<type> %s </type>\n"+
|
|
||||||
"<match_type> %s </match_type>\n"+
|
|
||||||
"<max_age> %s </max_age>\n",
|
|
||||||
tp.Pattern,
|
|
||||||
tp.TrackerType,
|
|
||||||
tp.MatchType,
|
|
||||||
maxAge,
|
|
||||||
)
|
|
||||||
|
|
||||||
if len(domains) > 0 {
|
|
||||||
prompt += fmt.Sprintf("<observed_domains> %s </observed_domains>\n", strings.Join(domains, ", "))
|
|
||||||
}
|
|
||||||
|
|
||||||
return prompt
|
|
||||||
}
|
|
||||||
|
|
||||||
// buildCommonPatternIdentificationPrompt builds the mapping-agent input
|
|
||||||
// for a global catalog pattern. Catalog rows carry no observed domains,
|
|
||||||
// so the prompt omits the <observed_domains> signal and relies on the
|
|
||||||
// pattern name, type, and naming conventions. It lets the enrichment
|
|
||||||
// worker reuse the mapping agent to attribute a vendor before describing.
|
|
||||||
func buildCommonPatternIdentificationPrompt(cp coredata.CommonTrackerPattern) string {
|
|
||||||
maxAge := "session"
|
|
||||||
if cp.MaxAgeSeconds != nil {
|
|
||||||
maxAge = fmt.Sprintf("%d seconds", *cp.MaxAgeSeconds)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
return fmt.Sprintf(
|
return fmt.Sprintf(
|
||||||
@@ -179,9 +158,19 @@ func buildCommonPatternIdentificationPrompt(cp coredata.CommonTrackerPattern) st
|
|||||||
"<type> %s </type>\n"+
|
"<type> %s </type>\n"+
|
||||||
"<match_type> %s </match_type>\n"+
|
"<match_type> %s </match_type>\n"+
|
||||||
"<max_age> %s </max_age>\n",
|
"<max_age> %s </max_age>\n",
|
||||||
cp.Pattern,
|
pattern,
|
||||||
cp.TrackerType,
|
trackerType,
|
||||||
cp.MatchType,
|
matchType,
|
||||||
maxAge,
|
maxAge,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func buildAgentPrompt(tp coredata.TrackerPattern, domains []string) string {
|
||||||
|
prompt := buildTrackerIdentificationPrompt(tp.Pattern, tp.TrackerType, tp.MatchType, tp.MaxAgeSeconds)
|
||||||
|
|
||||||
|
if len(domains) > 0 {
|
||||||
|
prompt += fmt.Sprintf("<observed_domains> %s </observed_domains>\n", strings.Join(domains, ", "))
|
||||||
|
}
|
||||||
|
|
||||||
|
return prompt
|
||||||
|
}
|
||||||
|
|||||||
@@ -312,6 +312,9 @@ func (h *trackerMappingHandler) resolveDeterministic(
|
|||||||
return res, fmt.Errorf("cannot load cookie banner for domain filtering: %w", err)
|
return res, fmt.Errorf("cannot load cookie banner for domain filtering: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DEBUG: don't commit
|
||||||
|
banner.Origin = "https://t.probo.com"
|
||||||
|
|
||||||
res.origin = banner.Origin
|
res.origin = banner.Origin
|
||||||
|
|
||||||
if tp.CommonTrackerPatternID != nil {
|
if tp.CommonTrackerPatternID != nil {
|
||||||
@@ -562,11 +565,9 @@ func (h *trackerMappingHandler) matchByDomain(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// agentIdentification carries a confident tracker-mapping agent result
|
// agentIdentification carries a confident tracker-mapping agent result
|
||||||
// and the (first-party-filtered) domains it observed, from the no-tx
|
// from the no-tx agent phase to the short transaction that persists it.
|
||||||
// agent phase to the short transaction that persists it.
|
|
||||||
type agentIdentification struct {
|
type agentIdentification struct {
|
||||||
result TrackerMappingAgentResult
|
result TrackerMappingAgentResult
|
||||||
domains []string
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// identifyWithAgent runs the tracker-mapping agent outside any
|
// identifyWithAgent runs the tracker-mapping agent outside any
|
||||||
@@ -596,7 +597,7 @@ func (h *trackerMappingHandler) identifyWithAgent(
|
|||||||
return nil
|
return nil
|
||||||
},
|
},
|
||||||
); err != nil {
|
); err != nil {
|
||||||
h.logger.WarnCtx(ctx, "cannot load initiator domains for agent", log.Error(err))
|
return nil, fmt.Errorf("cannot load initiator domains for agent: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
domains = uri.FilterFirstPartyDomains(domains, siteOrigin)
|
domains = uri.FilterFirstPartyDomains(domains, siteOrigin)
|
||||||
@@ -646,7 +647,6 @@ func (h *trackerMappingHandler) identifyWithAgent(
|
|||||||
|
|
||||||
return &agentIdentification{
|
return &agentIdentification{
|
||||||
result: identification,
|
result: identification,
|
||||||
domains: domains,
|
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -666,7 +666,6 @@ func (h *trackerMappingHandler) persistAgentIdentification(
|
|||||||
h.logger,
|
h.logger,
|
||||||
ident.result.ThirdPartyName,
|
ident.result.ThirdPartyName,
|
||||||
ident.result.Category,
|
ident.result.Category,
|
||||||
ident.domains,
|
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("cannot resolve or create common third party: %w", err)
|
return nil, fmt.Errorf("cannot resolve or create common third party: %w", err)
|
||||||
|
|||||||
42
pkg/thirdparty/resolver.go
vendored
42
pkg/thirdparty/resolver.go
vendored
@@ -16,9 +16,11 @@ package thirdparty
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/jackc/pgx/v5/pgconn"
|
||||||
"go.gearno.de/kit/log"
|
"go.gearno.de/kit/log"
|
||||||
"go.gearno.de/kit/pg"
|
"go.gearno.de/kit/pg"
|
||||||
"go.probo.inc/probo/pkg/coredata"
|
"go.probo.inc/probo/pkg/coredata"
|
||||||
@@ -30,19 +32,24 @@ import (
|
|||||||
// catalog, creating a row when none matches. Dedup is deterministic:
|
// catalog, creating a row when none matches. Dedup is deterministic:
|
||||||
// exact name, then slug, before insert. Callers run inside their own
|
// exact name, then slug, before insert. Callers run inside their own
|
||||||
// transaction and pass the logger explicitly, so it is shared by the
|
// transaction and pass the logger explicitly, so it is shared by the
|
||||||
// tracker mapping worker (which supplies observed domains) and the
|
// tracker mapping worker and the common pattern enrichment worker.
|
||||||
// common pattern enrichment worker (which has none).
|
//
|
||||||
|
// It never seeds common_third_party_domains: observed initiator domains
|
||||||
|
// are a co-occurrence signal, not verified vendor ownership, and the
|
||||||
|
// global catalog's domain set (used for cross-tenant domain matching) is
|
||||||
|
// owned by the curated seed instead.
|
||||||
func ResolveOrCreateCommonThirdParty(
|
func ResolveOrCreateCommonThirdParty(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
tx pg.Tx,
|
tx pg.Tx,
|
||||||
logger *log.Logger,
|
logger *log.Logger,
|
||||||
name string,
|
name string,
|
||||||
category coredata.ThirdPartyCategory,
|
category coredata.ThirdPartyCategory,
|
||||||
domains []string,
|
|
||||||
) (*gid.GID, error) {
|
) (*gid.GID, error) {
|
||||||
var party coredata.CommonThirdParty
|
var party coredata.CommonThirdParty
|
||||||
if err := party.LoadByName(ctx, tx, name); err == nil {
|
if err := party.LoadByName(ctx, tx, name); err == nil {
|
||||||
return &party.ID, nil
|
return &party.ID, nil
|
||||||
|
} else if !errors.Is(err, coredata.ErrResourceNotFound) {
|
||||||
|
return nil, fmt.Errorf("cannot load common third party by name: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
partySlug := slug.Make(name)
|
partySlug := slug.Make(name)
|
||||||
@@ -52,6 +59,8 @@ func ResolveOrCreateCommonThirdParty(
|
|||||||
|
|
||||||
if err := party.LoadBySlug(ctx, tx, partySlug); err == nil {
|
if err := party.LoadBySlug(ctx, tx, partySlug); err == nil {
|
||||||
return &party.ID, nil
|
return &party.ID, nil
|
||||||
|
} else if !errors.Is(err, coredata.ErrResourceNotFound) {
|
||||||
|
return nil, fmt.Errorf("cannot load common third party by slug: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
now := time.Now()
|
now := time.Now()
|
||||||
@@ -65,22 +74,25 @@ func ResolveOrCreateCommonThirdParty(
|
|||||||
UpdatedAt: now,
|
UpdatedAt: now,
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := party.Insert(ctx, tx); err != nil {
|
// Insert inside a savepoint so a concurrent transaction that created
|
||||||
return nil, fmt.Errorf("cannot create common third party: %w", err)
|
// the same slug between our lookup and write does not abort the
|
||||||
|
// caller's transaction. On the unique-violation race, reload the
|
||||||
|
// winning row and return it instead of failing.
|
||||||
|
insertErr := tx.Savepoint(ctx, func(ctx context.Context, sp pg.Tx) error {
|
||||||
|
return party.Insert(ctx, sp)
|
||||||
|
})
|
||||||
|
if insertErr != nil {
|
||||||
|
if pgErr, ok := errors.AsType[*pgconn.PgError](insertErr); ok &&
|
||||||
|
pgErr.Code == "23505" &&
|
||||||
|
pgErr.ConstraintName == "common_third_parties_slug_key" {
|
||||||
|
if err := party.LoadBySlug(ctx, tx, partySlug); err != nil {
|
||||||
|
return nil, fmt.Errorf("cannot reload common third party after insert race: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, domain := range domains {
|
return &party.ID, nil
|
||||||
domainRecord := coredata.CommonThirdPartyDomain{
|
|
||||||
ID: gid.New(gid.NilTenant, coredata.CommonThirdPartyDomainEntityType),
|
|
||||||
CommonThirdPartyID: party.ID,
|
|
||||||
Domain: domain,
|
|
||||||
CreatedAt: now,
|
|
||||||
UpdatedAt: now,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if _, err := domainRecord.Upsert(ctx, tx); err != nil {
|
return nil, fmt.Errorf("cannot create common third party: %w", insertErr)
|
||||||
return nil, fmt.Errorf("cannot create common third party domain: %w", err)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
logger.InfoCtx(
|
logger.InfoCtx(
|
||||||
|
|||||||
3
pkg/thirdparty/resolver_test.go
vendored
3
pkg/thirdparty/resolver_test.go
vendored
@@ -145,7 +145,6 @@ func TestResolveOrCreateCommonThirdParty(t *testing.T) {
|
|||||||
logger,
|
logger,
|
||||||
name,
|
name,
|
||||||
coredata.ThirdPartyCategoryAnalytics,
|
coredata.ThirdPartyCategoryAnalytics,
|
||||||
nil,
|
|
||||||
)
|
)
|
||||||
got = id
|
got = id
|
||||||
|
|
||||||
@@ -175,7 +174,6 @@ func TestResolveOrCreateCommonThirdParty(t *testing.T) {
|
|||||||
logger,
|
logger,
|
||||||
variant,
|
variant,
|
||||||
coredata.ThirdPartyCategoryAnalytics,
|
coredata.ThirdPartyCategoryAnalytics,
|
||||||
nil,
|
|
||||||
)
|
)
|
||||||
got = id
|
got = id
|
||||||
|
|
||||||
@@ -199,7 +197,6 @@ func TestResolveOrCreateCommonThirdParty(t *testing.T) {
|
|||||||
logger,
|
logger,
|
||||||
name,
|
name,
|
||||||
coredata.ThirdPartyCategoryMarketing,
|
coredata.ThirdPartyCategoryMarketing,
|
||||||
nil,
|
|
||||||
)
|
)
|
||||||
got = id
|
got = id
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user