From 3826f115ecc7f409a31b11d41b31bb3dd2d78ba2 Mon Sep 17 00:00:00 2001 From: Simon Axelsson Date: Mon, 7 Sep 2026 17:58:06 +0200 Subject: [PATCH] fix: enforce canonical hard non-match constraints across ER clusters --- services/api/workers/resolve.py | 117 ++++++++++++++++++++++---------- 1 file changed, 81 insertions(+), 36 deletions(-) diff --git a/services/api/workers/resolve.py b/services/api/workers/resolve.py index edb6161..99ac22b 100644 --- a/services/api/workers/resolve.py +++ b/services/api/workers/resolve.py @@ -4,6 +4,10 @@ Python no longer scores identity. It sends opaque records to /api/0pt/internal/er/candidates, then only applies the returned canonical `match` decisions. If the canonical engine is unavailable, records are returned UNMERGED (fail closed against false identity merges). + +A canonical hard non-match is also a cluster-wide constraint: transitive +matches are never allowed to pull two records with conflicting stable identity +evidence into the same entity. """ from __future__ import annotations @@ -37,21 +41,36 @@ def _record_id(record: dict[str, Any]) -> str: return "erp_" + hashlib.sha256(raw).hexdigest()[:32] +def _unique_record_ids(records: list[dict[str, Any]]) -> list[str]: + seen: dict[str, int] = {} + unique_ids: list[str] = [] + for record in records: + rid = _record_id(record) + n = seen.get(rid, 0) + seen[rid] = n + 1 + unique_ids.append(rid if n == 0 else f"{rid}_{n}") + return unique_ids + + def _apply_matches(records: list[dict[str, Any]], pairs: list[dict[str, Any]]) -> list[dict[str, Any]]: if not records: return [] - ids = [_record_id(r) for r in records] - # Exact duplicate payloads are indistinguishable; deterministic suffixes - # make request ids unique without changing merge semantics. - seen: dict[str, int] = {} - unique_ids: list[str] = [] - for rid in ids: - n = seen.get(rid, 0) - seen[rid] = n + 1 - unique_ids.append(rid if n == 0 else f"{rid}_{n}") - + unique_ids = _unique_record_ids(records) + known_ids = set(unique_ids) parent = {rid: rid for rid in unique_ids} + members = {rid: {rid} for rid in unique_ids} + + # Only explicit hard-rule non-matches are cluster-wide negative evidence. + # A normal score-based non_match merely means "insufficient similarity" and + # must not become a universal cannot-link rule. + hard_non_matches: set[frozenset[str]] = set() + for pair in pairs: + a, b = pair.get("record_a"), pair.get("record_b") + if a not in known_ids or b not in known_ids: + continue + if pair.get("status") == "non_match" and pair.get("decided_by") not in (None, "score"): + hard_non_matches.add(frozenset((a, b))) def find(x: str) -> str: while parent[x] != x: @@ -59,42 +78,72 @@ def _apply_matches(records: list[dict[str, Any]], pairs: list[dict[str, Any]]) - x = parent[x] return x - def union(a: str, b: str) -> None: - ra, rb = find(a), find(b) - if ra != rb: - parent[max(ra, rb)] = min(ra, rb) + def components_conflict(left: set[str], right: set[str]) -> bool: + return any(frozenset((a, b)) in hard_non_matches for a in left for b in right) - match_pairs = [p for p in pairs if p.get("status") == "match"] + def union(a: str, b: str) -> bool: + ra, rb = find(a), find(b) + if ra == rb: + return True + left, right = members[ra], members[rb] + if components_conflict(left, right): + return False + keep, drop = (ra, rb) if ra < rb else (rb, ra) + parent[drop] = keep + members[keep] = members[ra] | members[rb] + members.pop(drop, None) + return True + + # Hard-rule matches are applied before probabilistic/score matches; within + # each class, stronger scores go first. No later edge may violate a hard + # negative constraint established above. + match_pairs = [ + p for p in pairs + if p.get("status") == "match" + and p.get("record_a") in known_ids + and p.get("record_b") in known_ids + ] + match_pairs.sort( + key=lambda p: ( + 0 if p.get("decided_by") not in (None, "score") else 1, + -float(p.get("score") or 0.0), + str(p.get("candidate_id") or ""), + ) + ) + accepted_edges: list[dict[str, Any]] = [] for pair in match_pairs: - a, b = pair.get("record_a"), pair.get("record_b") - if a in parent and b in parent: - union(a, b) + a, b = str(pair["record_a"]), str(pair["record_b"]) + if union(a, b): + accepted_edges.append(pair) groups: dict[str, list[tuple[str, dict[str, Any]]]] = {} for rid, record in zip(unique_ids, records): groups.setdefault(find(rid), []).append((rid, copy.deepcopy(record))) out: list[dict[str, Any]] = [] - for members in groups.values(): - members.sort(key=lambda item: (-float(item[1].get("confidence", 0.0)), item[0])) - anchor_id, anchor = members[0] - sources = sorted({str(r.get("source")) for _, r in members if r.get("source")}) + for group in groups.values(): + group.sort(key=lambda item: (-float(item[1].get("confidence", 0.0)), item[0])) + _, anchor = group[0] + sources = sorted({str(r.get("source")) for _, r in group if r.get("source")}) anchor["agreed_sources"] = len(sources) or 1 - anchor["confidence"] = min(0.99, float(anchor.get("confidence", 0.0)) + 0.1 * max(0, len(sources) - 1)) - anchor["merged_from"] = [r.get("source") for _, r in members[1:] if r.get("source")] + anchor["confidence"] = min( + 0.99, + float(anchor.get("confidence", 0.0)) + 0.1 * max(0, len(sources) - 1), + ) + anchor["merged_from"] = [r.get("source") for _, r in group[1:] if r.get("source")] provenance = list(anchor.get("provenance", [])) keys = {(p.get("field_name"), p.get("source_id")) for p in provenance} - for _, record in members[1:]: + for _, record in group[1:]: for p in record.get("provenance", []): key = (p.get("field_name"), p.get("source_id")) if key not in keys: provenance.append(copy.deepcopy(p)) keys.add(key) - member_ids = {rid for rid, _ in members} + member_ids = {rid for rid, _ in group} edges = [ - p for p in match_pairs + p for p in accepted_edges if p.get("record_a") in member_ids and p.get("record_b") in member_ids ] er_info = { @@ -109,7 +158,7 @@ def _apply_matches(records: list[dict[str, Any]], pairs: list[dict[str, Any]]) - for p in edges ], } - if len(members) > 1: + if len(group) > 1: for p in provenance: p["entity_resolution"] = er_info anchor["provenance"] = provenance @@ -131,19 +180,15 @@ async def resolve( url = _url() token = os.getenv("INTERNAL_TOKEN", "") - if not url or not token: + if not url or not token or not tenant_id or not project_id: return [copy.deepcopy(r) for r in records] - ids = [_record_id(r) for r in records] - seen: dict[str, int] = {} + unique_ids = _unique_record_ids(records) er_records = [] - for rid, record in zip(ids, records): - n = seen.get(rid, 0) - seen[rid] = n + 1 - unique = rid if n == 0 else f"{rid}_{n}" + for rid, record in zip(unique_ids, records): name = str(record.get("name") or "") er_records.append({ - "record_id": unique, + "record_id": rid, "fields": { "market": record.get("market"), "block_initial": name[:1].upper(),