"""Drift rules on properties and identifiers: D04 keys, D07 to D11, D13, D14 and D15. A mixin for the Drift class in drift.py, which sets up the plan, the inventory and add(). Thresholds and reasons: references/drift-rules.md. """ import re from collections import OrderedDict, defaultdict from common import (COMPATIBLE, FAMILY, GA4_BUILTIN_PARAMS, ID_RULE_KEYS, REDACTION_TOKENS, ROUTE_TOOLS, TAIL_TOKENS, Plan, first_items, id_rule, is_identifier, is_money, is_pii_name, is_vendor, mask, masked_example, norm_key, num, pct, plan_scope, plural, redact, where) class PropertyRules(object): # ---------- properties, D07 to D11, D13 and D04 keys ---------- def check_properties(self, tool, cls): a, inv, plan = self.args, self.inv, self.plan planned, aliases, mapped = self.resolved[tool] all_planned_props = {Plan.tool_prop_name(p, tool) for p in plan.props} if not any(inv.props[tool].values()): for rid in ("D07", "D08", "D09", "D10", "D11"): self.add(rid, "NOT CHECKED", tool, "", "", "event-level inventory only (names and counts); property " "rules need property rows from a BigQuery or warehouse query or a profiled export.") return names_with_rows = [n for n in inv.names(tool) if inv.total(tool, [n]) > 0] carriers = defaultdict(set) for n in names_with_rows: for pn in inv.props[tool].get(n, {}): carriers[pn].add(n) global_unplanned = set() for pn, events in carriers.items(): if (pn in all_planned_props or is_vendor(pn, tool, self.vendor_params) or pn.endswith(".$") or is_pii_name(pn)): continue if names_with_rows and len(events) / len(names_with_rows) >= a.global_share: global_unplanned.add(pn) self.add("D09", "WARN", tool, "*", pn, f"{pn}: on {len(events)} of {len(names_with_rows)} event names " f"({pct(len(events) / len(names_with_rows))}); one unplanned global property, not one " "finding per event.", "needs decision (add it to data-dictionary.md as a global property, or stop sending)", affected=len(events), denominator=len(names_with_rows), unit="event names carrying it") enum_unchecked, column_only, optional_global = [], [], defaultdict(list) for e in plan.events.values(): fn = mapped[e["name"]][0] if not fn or e["status"] == "retired": continue names = self.obs_names(tool, fn) ev_by_source = inv.by_source(tool, names) ev_total = sum(ev_by_source.values()) if ev_total == 0 and not any(inv.props[tool].get(n) for n in names): continue tier = self.tier(e) label = fn planned_here = set() scope = ("" if any(inv.props[tool].get(n) for n in names) else " The inventory has no property rows for this event; confirm it was included before acting.") self.checked["D09"] += 1 for p in plan.props_for(e["name"], tool): pn = Plan.tool_prop_name(p, tool) planned_here.add(pn) base, alias = f"{label}.{pn}", (p["name"] if pn != p["name"] else None) shown = where(base, alias) rows = [o for n in names for o in inv.props[tool].get(n, {}).get(pn, [])] counted = [o for o in rows if o["rows"] is not None] columns = [o for o in rows if o["rows"] is None] if columns: self.check_column_types(tool, e, p, shown, columns) if rows and not counted and p["required"]: column_only.append(base) by_src = defaultdict(list) for o in counted: by_src[o["source"]].append(o) psrc, esrc = set(by_src), set(ev_by_source) if counted and (psrc == {""} or esrc == {""} or not (psrc & esrc)): # Sources are labelled differently on event and property rows: compare in aggregate. groups = {"": counted} denominators = {"": ev_total or max([o["event_rows"] or 0 for o in counted] or [0])} else: groups = {s: by_src.get(s, []) for s in sorted(esrc | psrc)} denominators = {s: ev_by_source.get(s) or max([o["event_rows"] or 0 for o in by_src.get(s, [])] or [0]) for s in groups} if p["required"] and (counted or not rows): self.check_required(tool, e, p, (base, alias), groups, denominators, tier, scope) if p["required"] is False: self.checked["D08"] += 1 if p["required"] is False and not rows and ev_total >= a.optional_min_rows: if p["event"] == "*": optional_global[pn].append(label) else: self.add("D08", "WARN", tool, e["name"], p["name"], f"{shown}: optional and never sent while the event has {num(ev_total)} rows.{scope}", "needs decision (implement it or drop it from the plan)", affected=ev_total, denominator=ev_total, unit="rows without it") for s, group in groups.items(): if not group: continue self.check_types(tool, e, p, (base, alias), s, group) if p["allowed"] and p["type"] in (None, "string"): values = OrderedDict() for o in group: for v, c in (o["values"] or {}).items(): values[v] = values.get(v, 0) + c if not values: enum_unchecked.append(base) else: self.check_enum(tool, e, p, (base, alias), s, values) for n in names: # D09 unplanned properties on this planned event for pn, rows in inv.props[tool].get(n, {}).items(): if (pn in planned_here or pn in global_unplanned or is_vendor(pn, tool, self.vendor_params) or pn.endswith(".$") or is_pii_name(pn)): continue occurrences = sum(o["rows"] or 0 for o in rows) counted_rows = any(o["rows"] is not None for o in rows) seen = f"{num(occurrences)} occurrences" if counted_rows else "a column (no counts)" self.add("D09", "WARN", tool, e["name"], pn, f"{label}.{pn}: {seen} of a property the contract does not list.", "needs decision (adopt into the plan or stop sending)", affected=occurrences if counted_rows else None, denominator=ev_total or None, unit="rows carrying it") for pn, events in sorted(optional_global.items()): self.add("D08", "WARN", tool, "*", pn, f"{pn} (global, optional): never sent on {plural(len(events), 'event')} with " f"{num(a.optional_min_rows)} or more rows ({first_items(events, 5)}).", "needs decision (implement it or drop it from the plan)", affected=len(events), unit="events without it") if enum_unchecked: noun = "property" if len(enum_unchecked) == 1 else "properties" self.add("D11", "NOT CHECKED", tool, "", "", f"no values column for {len(enum_unchecked)} enumerated {noun} ({first_items(enum_unchecked, 4)}); " "list them with the profiler's --values-for or the SQL top-values part.") if column_only: noun = "property" if len(column_only) == 1 else "properties" self.add("D07", "NOT CHECKED", tool, "", "", f"only column-level types for {len(column_only)} required {noun} ({first_items(column_only, 4)}); " "completeness needs row counts.") self.check_pii_and_keys(tool, all_planned_props) def check_required(self, tool, e, p, label, groups, denominators, tier, scope): """D07: required property present on fewer than all rows, per source.""" a = self.args base, alias = label for s, group in groups.items(): den = denominators.get(s) or 0 if den <= 0: continue self.checked["D07"] += 1 server = s == "server" or (s == "" and self.expected_sources(e, tool) == {"server"}) missing_marks = sum((o["sentinel_hits"] or 0) + (o["empty_hits"] or 0) for o in group) present = max(sum(o["rows"] for o in group if o["type"] != "null") - missing_marks, 0) share = present / den sev = None if share < a.required_fail: sev, reason = "FAIL", f"below {pct(a.required_fail)}" if den < a.min_sample and present > 0: sev, reason = "WARN", f"low sample ({num(den)} rows)" elif share < 1 and tier == 1 and server: sev, reason = "FAIL", "required on a tier 1 server event, so any gap fails" elif share < 1: sev, reason = "WARN", f"between {pct(a.required_fail)} and 100%" if sev: marks = f"; {num(missing_marks)} sentinel or empty strings counted as missing" if missing_marks else "" self.add("D07", sev, tool, e["name"], p["name"], f"{where(base, alias, s)}: present in {num(present)} of {num(den)} rows ({pct(share)}), " f"{reason}{marks}.{scope}", "fix implementation", source=s, affected=den - present, denominator=den, unit="rows without it") def check_pii_and_keys(self, tool, all_planned_props): """D13 personal data on any event, and D04 dynamic property keys.""" inv, a = self.inv, self.args for n in sorted(inv.names(tool)): for pn, rows in sorted(inv.props[tool].get(n, {}).items()): counted = any(o["pii_hits"] is not None for o in rows) self.checked["D13"] += 1 self.pii_rows[tool][0] += 1 self.pii_rows[tool][1] += 1 if counted else 0 hits = sum(o["pii_hits"] or 0 for o in rows) occurrences = sum(o["rows"] or 0 for o in rows) named = is_pii_name(pn) and not all(o["type"] == "boolean" for o in rows) if named or hits: what = "named like personal data" if named else "not named like personal data" found = (f"email, phone or sensitive URL patterns in {num(hits)} of {num(occurrences)} occurrences" if hits else "no such patterns in its values" if counted else "no pattern counts in the inventory") self.add("D13", "FAIL", tool, n, pn, f"{n}.{pn}: {what}; {found}. Values are not printed.", "fix implementation first; the privacy owner decides on deletion through each vendor's " "process", affected=hits or occurrences, denominator=occurrences or None, unit="rows with patterns" if hits else "rows carrying it") keys = defaultdict(set) for n in inv.names(tool): for pn in inv.props[tool].get(n, {}): if pn not in all_planned_props and not is_vendor(pn, tool, self.vendor_params) and mask(pn) != pn: keys[mask(pn)].add(pn) for pattern, members in sorted(keys.items()): if len(members) >= a.dynamic_min: members = sorted(members) self.add("D04", "FAIL", tool, "*", pattern, f"{len(members)} property keys follow {pattern} (for example {masked_example(members[0])}).", "fix implementation (move the variable part into a value)", affected=len(members), unit="property keys") def check_types(self, tool, e, p, label, s, group): a = self.args counts = defaultdict(int) for o in group: if o["type"] and o["type"] != "null": counts[o["type"]] += o["rows"] total = sum(counts.values()) if not total: return self.checked["D10"] += 1 if not p["type"]: families = {FAMILY.get(t, t) for t in counts} if len(families) > 1: self.add("D10", "WARN", tool, e["name"], p["name"], f"{where(label[0], label[1], s)}: inconsistent types " f"({', '.join(f'{t} {num(c)}' for t, c in sorted(counts.items()))}) and the plan has no type " "(Q05).", "amend plan (set the type), then fix the emitter", source=s, affected=total - max(counts.values()), denominator=total, unit="rows outside the main type") return allowed = set(COMPATIBLE[p["type"]]) if p["type"] == "timestamp" and re.search(r"epoch", p["unit"], re.I): allowed.add("integer") bad = {t: c for t, c in counts.items() if t not in allowed} if not bad: return mismatch = sum(bad.values()) share = mismatch / total key = is_identifier(p["name"]) or is_money(p["name"], p["unit"]) if key and self.tier(e) == 1: sev, reason = "FAIL", "an identifier or money property on a tier 1 event, so any mismatch fails" elif share >= a.mismatch_share: sev, reason = "FAIL", f"at or above {pct(a.mismatch_share)}" if total < a.min_sample: sev, reason = "WARN", f"low sample ({num(total)} occurrences)" else: sev, reason = "WARN", f"below {pct(a.mismatch_share)}" parts = [] for t, c in sorted(bad.items(), key=lambda x: -x[1]): parts.append(f"{self.type_message(p, t)} in {num(c)} of {num(total)} occurrences ({pct(c / total)})") self.add("D10", sev, tool, e["name"], p["name"], f"{where(label[0], label[1], s)}: plan {p['type']}; {'; '.join(parts)}; {reason}.", "fix implementation; amend the plan type only when the plan is wrong for the fact", source=s, affected=mismatch, denominator=total, unit="rows with another type") def type_message(self, p, observed): planned = p["type"] if observed == "numeric_string" and planned in ("integer", "number"): return "numbers sent as text" if observed in ("string", "numeric_string") and planned == "boolean": return "booleans sent as text" if observed in ("integer", "number") and planned == "string" and is_identifier(p["name"]): return "IDs sent as numbers" if observed == "number" and planned == "integer": return "decimals in an integer field" if observed in ("integer", "number") and planned == "timestamp": return "epoch number where an ISO timestamp is planned" return f"{observed} values" def check_column_types(self, tool, e, p, shown, columns): if not p["type"]: return self.checked["D10"] += 1 allowed = set(COMPATIBLE[p["type"]]) observed = {o["type"] for o in columns if o["type"]} bad = sorted(t for t in observed if t not in allowed) if not bad: return key = is_identifier(p["name"]) or is_money(p["name"], p["unit"]) kind = "identifier" if is_identifier(p["name"]) else ("money" if key else "") self.add("D10", "FAIL" if key else "WARN", tool, e["name"], p["name"], f"{shown}: column type {', '.join(bad)} where the plan says {p['type']}" + (f" ({kind} property)" if kind else "") + "; the share of bad rows is unknown at column level. Warehouse column types are fixed when the " "column is created.", "fix implementation; changing a warehouse column type needs the vendor's support and a table rebuild") def check_enum(self, tool, e, p, label, s, values): """D11: values outside the allowed list; the untallied tail counts when it cannot hold allowed values.""" a = self.args self.checked["D11"] += 1 allowed = set(p["allowed"]) total = sum(values.values()) unknown_allowed = p["policy"].startswith("send_unknown") and "unknown" not in allowed unknown = values.get("unknown", 0) if unknown_allowed else 0 tail = sum(values.get(t, 0) for t in TAIL_TOKENS) listed = {v for v in values if v not in TAIL_TOKENS} out = OrderedDict() for v, c in values.items(): if v in allowed or v in TAIL_TOKENS or (v == "unknown" and unknown_allowed): continue out[v] = c out_rows = sum(out.values()) tail_out, tail_note = 0, "" if tail and allowed <= listed: tail_out = tail tail_note = (f"; the {num(tail)} rows in the untallied tail count as out of list, because every allowed " "value is already listed") elif tail: tail_note = (f"; {num(tail)} more rows sit in the untallied tail, which may hold allowed values, so the " f"share is between {pct(out_rows / total)} and {pct((out_rows + tail) / total)}") bad_rows = out_rows + tail_out if not bad_rows: if tail and tail / total > a.mismatch_share: self.add("D11", "WARN", tool, e["name"], p["name"], f"{where(label[0], label[1], s)}: no listed value is out of list, but {num(tail)} of " f"{num(total)} rows ({pct(tail / total)}) sit in the untallied tail; list more values " "before ruling out drift.", "needs decision (list more values, then decide)", source=s, affected=tail, denominator=total, unit="rows in the untallied tail") elif unknown: self.add("D11", "INFO", tool, e["name"], p["name"], f"{where(label[0], label[1], s)}: 'unknown' in {num(unknown)} of {num(total)} rows " f"({pct(unknown / total)}) under send_unknown_and_alert; check that the alert fired.") return share = bad_rows / total if p["policy"] == "reject": sev, reason = "FAIL", "the policy is reject, so any out-of-list value fails" elif share > a.mismatch_share: sev, reason = "FAIL", f"above {pct(a.mismatch_share)}" if total < a.min_sample: sev, reason = "WARN", f"low sample ({num(total)} rows)" elif tail and not tail_out and (out_rows + tail) / total > a.mismatch_share: sev, reason = "WARN", f"at or below {pct(a.mismatch_share)} unless the tail is out of list" else: sev, reason = "WARN", f"at or below {pct(a.mismatch_share)}" shown_values = [] for v, c in sorted(out.items(), key=lambda x: -x[1])[:3]: v = v if v in REDACTION_TOKENS else redact(v) hint = "" matches = [x for x in allowed if x.lower() == v.lower()] if matches: hint = f", a case variant of {matches[0]}" shown_values.append(f"'{v}' {num(c)}{hint}") if shown_values: lead = f"{', '.join(shown_values)} not in the allowed list ({', '.join(p['allowed'])})" else: lead = (f"the values in the untallied tail are not in the allowed list ({', '.join(p['allowed'])}), " "because every allowed value is already listed") tail_note = "" expected = f"; policy {p['policy']} expects 'unknown'" if p["policy"].startswith("send_unknown") else "" self.add("D11", sev, tool, e["name"], p["name"], f"{where(label[0], label[1], s)}: {lead}; {num(bad_rows)} of {num(total)} rows ({pct(share)}), " f"{reason}{expected}{tail_note}.", "needs decision (amend the plan as a compatible change, enforcement copies first, or fix the emitter " "mapping)", source=s, affected=bad_rows, denominator=total, unit="rows out of list") # ---------- identifiers and GA4 limits, D13 and D14 ---------- def check_identity(self, tool): """GA4 ceilings and identifier rules. Returns False when the tool has no identifier rows.""" inv, limit = self.inv, self.args.ga4_param_limit if tool == "ga4": planned, aliases, mapped = self.resolved[tool] for e in self.plan.events.values(): fn = mapped[e["name"]][0] if not fn or e["status"] == "retired": continue self.checked["D14"] += 1 params = {Plan.tool_prop_name(p, "ga4") for p in self.plan.props_for(e["name"], "ga4")} if len(params) > limit: self.add("D14", "FAIL", tool, e["name"], "", f"{fn}: the plan maps {len(params)} parameters to one GA4 event, over the limit of " f"{limit}; GA4 does not log parameters past it.", "amend plan (fewer parameters)", affected=len(params), unit="planned parameters") if len(fn) > 40 and inv.total(tool, [fn]) > 0: self.add("D14", "FAIL", tool, e["name"], "", f"{fn}: planned GA4 name is {len(fn)} characters, " "over 40.", "amend plan (a name of 40 characters or fewer)") for n in sorted(inv.names(tool)): top = max([o["max_params"] or 0 for o in inv.events[tool].get(n, [])] or [0]) if top >= limit: self.add("D14", "WARN", tool, n, "", f"{n}: at least one row carries {top} custom parameters, at the limit of {limit}; " "confirm dropped parameters in DebugView.", "fix implementation, or amend the plan", affected=top, unit="custom parameters on the widest row") rows = inv.identity.get(tool, []) if not rows: return False self.checked["D14"] += len(rows) self.checked["id_rows"] += len(rows) totals, per_event = self.identity_groups(rows) to_amplitude = tool in ROUTE_TOOLS and any( self.plan.final_name(e, "amplitude")[0] and re.search(r"amplitude\s+via\s+" + re.escape(tool), e["row"].get("delivery_paths", "").lower()) for e in self.plan.events.values()) for (key, s), (classes, explicit) in totals.items(): self.check_id_classes(tool, key, s, classes, explicit, to_amplitude) for (event, key, s), classes in sorted(per_event.items()): self.check_id_nulls(tool, event, key, s, classes) return True @staticmethod def identity_groups(rows): """Totals per key and source (the '*' rows, or the sum of per-event rows) and per-event classes.""" explicit, summed = defaultdict(lambda: defaultdict(int)), defaultdict(lambda: defaultdict(int)) per_event = defaultdict(lambda: defaultdict(int)) for o in rows: for k, c in (o["values"] or {}).items(): if o["event"] == "*": explicit[(o["property"], o["source"])][k] += c else: per_event[(o["event"], o["property"], o["source"])][k] += c summed[(o["property"], o["source"])][k] += c totals = OrderedDict() for key in sorted(set(explicit) | set(summed)): totals[key] = (explicit[key], True) if key in explicit else (summed[key], False) return totals, per_event def check_id_classes(self, tool, key, s, classes, explicit, to_amplitude): present = sum(c for k, c in classes.items() if k != "null") nulls = classes.get("null", 0) label = f"{key} (source {s or 'all'})" base = norm_key(key.split(".")[-1]) amp_key = to_amplitude and base in ("user_id", "device_id", "anonymous_id") if tool == "amplitude" and explicit and nulls and base == "user_id": self.add("D14", "INFO", tool, "*", key, f"{label}: null on {num(nulls)} of {num(present + nulls)} rows ({pct(nulls / (present + nulls))}). " "Amplitude removes IDs under 5 characters and answers invalid IDs with a 400 error at ingestion, " "so short and placeholder IDs show only in the upstream copy (CDP or warehouse).", source=s, affected=nulls, denominator=present + nulls, unit="rows without it") if not present: return if classes.get("email_like"): self.add("D13", "FAIL", tool, "*", key, f"{label}: {num(classes['email_like'])} of {num(present)} identifiers look like email addresses. " "Values are not printed.", "fix implementation; the privacy owner decides on deletion", source=s, affected=classes["email_like"], denominator=present, unit="identifiers") ph = classes.get("placeholder", 0) if ph: share = ph / present extra = (" Amplitude answers IDs on its invalid list (null, none, 0, -1, the all-zero UUID and others) " "with a 400 error." if amp_key else "") self.add("D14", "FAIL" if share >= 0.01 else "WARN", tool, "*", key, f"{label}: {num(ph)} of {num(present)} identifiers are placeholders such as null, undefined, 0, -1, " f"the all-zero UUID or guest ({pct(share)}); placeholders merge strangers into one user.{extra}", "fix implementation (send no ID until a real one exists)", source=s, affected=ph, denominator=present, unit="identifiers") big = classes.get("numeric_over_2_53", 0) if big: self.add("D14", "FAIL", tool, "*", key, f"{label}: {num(big)} numeric identifiers above 2^53; JSON numbers that large lose digits " "(Mixpanel stores numbers as 64-bit floats). Send IDs as strings.", "fix implementation", source=s, affected=big, denominator=present, unit="identifiers") short = classes.get("short_under_5", 0) if short: if amp_key: sev, why = "FAIL", (f"Amplitude removes IDs under 5 characters, so events routed to Amplitude through " f"{tool} arrive there without this ID and split from the person") elif tool == "amplitude": sev, why = "WARN", ("Amplitude removes IDs under 5 characters at ingestion unless min_id_length is " "lowered, so check that setting and the upstream copy") else: sev, why = "WARN", "short IDs are often counters or placeholders" self.add("D14", sev, tool, "*", key, f"{label}: {num(short)} of {num(present)} identifiers are shorter than 5 characters " f"({pct(short / present)}); {why}.", "fix implementation", source=s, affected=short, denominator=present, unit="identifiers") def check_id_nulls(self, tool, event, key, s, classes): """D14: an identifier the plan names without a null rule, missing on some rows of the event.""" base = norm_key(key.split(".")[-1]) e = self.plan_event_for(tool, event) if base not in ID_RULE_KEYS or not e or e["status"] == "retired": return if id_rule(e["row"].get("identifiers", ""), base) != "required": return total = sum(classes.values()) missing = classes.get("null", 0) + classes.get("placeholder", 0) if not total or not missing: return share = missing / total server = s == "server" or (not s and self.expected_sources(e, tool) == {"server"}) companion = tool in e["overrides"] # a tool-specific route, such as a GA4 browser companion if self.tier(e) == 1 and not companion and share >= 0.01: sev, reason = "FAIL", "1% or more on a tier 1 event" elif self.tier(e) == 1 and not companion and server: sev, reason = "FAIL", "a tier 1 server event, so any gap fails" else: sev, reason = "WARN", "a companion route, so confirm the contract expects it here" if companion else ( f"tier {self.tier(e)}") label = event if event == e["name"] else f"{event} (plan {e['name']})" self.add("D14", sev, tool, e["name"], key, f"{label}.{key} (source {s or 'all'}): null or placeholder on {num(missing)} of {num(total)} rows " f"({pct(share)}), {reason}; the plan names {base} with no null rule, so these rows drop out of " "joins, funnels and group analytics.", "fix implementation (send the ID on every row), or amend the contract to say when it is null", source=s, affected=missing, denominator=total, unit="rows without it") # ---------- GA4 registration, D15 ---------- def check_registration(self, audited): mapped = [e for e in self.plan.events.values() if e["status"] != "retired" and self.plan.final_name(e, "ga4")[0]] if not mapped: return if self.definitions is None: if "ga4" in audited: self.add("D15", "NOT CHECKED", "ga4", "", "", "no --ga4-definitions file; list registrations with get_custom_dimensions_and_metrics from " "Google's GA4 MCP server, or Admin > Data display > Custom definitions, and save " "parameter_name and scope as CSV.") return params = OrderedDict() for e in mapped: fn = self.plan.final_name(e, "ga4")[0] for p in self.plan.props_for(e["name"], "ga4"): pn = Plan.tool_prop_name(p, "ga4") self.checked["D15"] += 1 if (is_identifier(p["name"]) or is_identifier(pn) or pn in GA4_BUILTIN_PARAMS or is_vendor(pn, "ga4", self.vendor_params)): continue # predefined in GA4, or an ID kept for export joins rec = params.setdefault(pn, {"uses": [], "scopes": set(), "types": set()}) rec["uses"].append((e, fn)) rec["scopes"].add(plan_scope(p["scope"])) rec["types"].add(p["type"]) for pn, rec in params.items(): events = first_items(sorted({fn for e, fn in rec["uses"]}), 4) event_field = rec["uses"][0][0]["name"] if len(rec["uses"]) == 1 else "*" regs = self.definitions.get(pn) if not regs: self.add("D15", "WARN", "ga4", event_field, pn, f"{pn} on {events}: not registered as a GA4 custom definition, so reports, explorations and " "audiences cannot use it, and a later registration does not fill past data. Register it if " "a GA4 report or audience needs it; otherwise record it as export-only in the plan.", "needs decision (register it in Admin, or record it as export-only)") continue scopes, kinds = {sc for sc, k in regs}, {k for sc, k in regs} problems = [] wrong = sorted(rec["scopes"] - scopes) if wrong: problems.append(f"the plan scopes it {' and '.join(w.lower() for w in wrong)} but it is registered " f"{' and '.join(sorted(sc.lower() for sc in scopes))}-scoped") if kinds == {"metric"} and rec["types"] & {"string", "boolean"}: problems.append("it is registered only as a custom metric while the plan types it " + " and ".join(sorted(t for t in rec["types"] if t))) if problems: tier1 = any(self.tier(e) == 1 for e, fn in rec["uses"]) self.add("D15", "FAIL" if tier1 else "WARN", "ga4", event_field, pn, f"{pn} on {events}: {'; '.join(problems)}, so GA4 reports read it at the wrong level or " "not at all.", "fix the registration in Admin (a definition with the planned scope; check the live docs " "before archiving the old one)")