"""Entity resolution — deterministic first, never merge blindly. 1. identifiers (scheme, value) → exact entity 2. normalized alias within the same entity type (disambiguated by organization when several match) 3. slug collision within the same type 4. evaluation-effort variants of an existing model ("gpt-5-4-mini-medium") fold onto the canonical model — they are a result configuration, never an entity of their own 5. otherwise create, and park ambiguous cases in the review queue as merge candidates Guards: persisted `resolution_decisions` (keep_separate) are honoured — two entities an operator kept apart are never re-fused by an alias match; aliases whose normalisation collapses digits and dots ("Qwen3-8B" ≡ "Qwen 38B") additionally require the same `variant_key`; `model` and `artifact` are compatible types for lookups (an artifact used to be typed model — the same hub repo must keep resolving to the same row). """ from __future__ import annotations import logging import re from datetime import UTC, datetime from typing import Any from sqlalchemy.ext.asyncio import AsyncConnection from aiatlas.db import execute, fetch_all, fetch_one, jsonb from aiatlas.ids import ENTITY_TYPES, new_id, normalize_alias, slugify from aiatlas.ontology.models import analyze_model_name, base_name, variant_key from aiatlas.sdk.facts import EntityRef log = logging.getLogger(__name__) # Types whose slug should be prefixed by the organization slug to stay unique and readable (models: `qwen-qwen3-8b` is ugly → # we keep the model name and only prefix on collision). GENERIC_NAMES = {"model", "models", "api", "pricing", "docs", "blog", "news", "research", "overview"} # lookup-compatible type groups: a ref of one type may resolve to a stored entity of another type in the same group _COMPATIBLE: dict[str, tuple[str, ...]] = {"model": ("model", "artifact"), "artifact": ("artifact", "model")} # names whose normalised alias is ambiguous: a digit, a separator, a digit ("Qwen3-8B" / "Qwen 38B" → "qwen38b") _DIGIT_SEP_DIGIT = re.compile(r"\d[.\-\s_]\d") # identifier schemes issued by evaluators: an entity known ONLY through these may be an evaluation configuration rather than a model EVALUATOR_SCHEMES = frozenset({"artificial_analysis", "livebench_model_id", "aider_model", "lmarena", "openrouter"}) # OpenRouter lists `o3-mini-high` as an endpoint of o3-mini def compatible_types(entity_type: str) -> tuple[str, ...]: return _COMPATIBLE.get(entity_type, (entity_type,)) class Resolver: def __init__(self, conn: AsyncConnection, *, snapshot_id: str | None = None, source_tier: int = 2, variant_index: dict[str, str] | None = None): self.conn = conn self.snapshot_id = snapshot_id self.tier = source_tier self.variant_index = variant_index # optional precomputed variant_key → canonical model id (canonicalization) self._cache: dict[str, str] = {} self.created: list[str] = [] self.updated: set[str] = set() self.folded: dict[str, tuple[str, dict[str, str]]] = {} # ref key → (canonical id, effort config) for folded variants async def resolve(self, ref: EntityRef, *, create: bool = True) -> str | None: if ref.id: return ref.id if ref.entity_type not in ENTITY_TYPES: raise ValueError(f"unknown entity type {ref.entity_type!r}") key = ref.key() if key in self._cache: ref.id = self._cache[key] await self._refresh_links(ref) return ref.id org_id = await self.resolve(ref.organization) if ref.organization else None found = await self._by_identifiers(ref) if found is None: found = await self._by_alias(ref, org_id) if found is None: found = await self._by_slug(ref) if found is None and ref.entity_type == "model" and self.tier >= 2 and set(ref.identifiers) <= EVALUATOR_SCHEMES and not ref.attributes.get("hf_repo"): # only evaluator-only references fold: a ref carrying official/hub identifiers (or a tier-1 source) names a real model variant = await self.resolve_variant(ref, org_id=org_id) if variant: found = variant[0] self.folded[key] = variant if found is None and not create: return None if found is None: found = await self._create(ref, org_id) else: await self._touch(found, ref, org_id) ref.id = found self._cache[key] = found await self._refresh_links(ref) return found # ---------------------------------------------------------------------------------------------- lookups async def _by_identifiers(self, ref: EntityRef) -> str | None: types = compatible_types(ref.entity_type) for scheme, value in ref.identifiers.items(): row = await fetch_one(self.conn, """select ei.entity_id, e.entity_type, e.merged_into from entity_identifiers ei join entities e on e.id = ei.entity_id where ei.scheme = :s and ei.value = :v""", s=scheme, v=str(value)) if row: if row["entity_type"] not in types: log.warning("identifier type mismatch", extra={"scheme": scheme, "value": value, "have": row["entity_type"], "want": ref.entity_type}) continue return row["merged_into"] or row["entity_id"] return None async def _conflicting_identifier(self, eid: str, ref: EntityRef) -> bool: """True when `eid` already carries an identifier of a scheme the ref also has, with a different value (e.g. two Codestral versions sharing a display name but different API ids) — never fuse those.""" if not ref.identifiers: return False rows = await fetch_all(self.conn, "select scheme, value from entity_identifiers where entity_id = :e and scheme = any(cast(:s as text[]))", e=eid, s=list(ref.identifiers)) return any(str(ref.identifiers[r["scheme"]]) != r["value"] for r in rows) async def _by_alias(self, ref: EntityRef, org_id: str | None) -> str | None: names = [ref.name, *ref.aliases] norms = {normalize_alias(n) for n in names if n and normalize_alias(n)} if not norms: return None rows = await fetch_all(self.conn, """select distinct e.id, e.organization_id, e.canonical_name, e.merged_into, a.alias from entity_aliases a join entities e on e.id = a.entity_id where a.alias_norm = any(cast(:norms as text[])) and e.entity_type = any(cast(:types as text[]))""", norms=list(norms), types=list(compatible_types(ref.entity_type))) if not rows: return None rows = [{**r, "id": r["merged_into"] or r["id"]} for r in rows] # collision safety: when the normalised alias erases digit separators, the candidate must share the ref's variant key # (or match one of the ref's names textually) — "Qwen 38B" must not land on "Qwen3-8B" if ref.entity_type in ("model", "artifact") and any(_DIGIT_SEP_DIGIT.search(n) for n in names if n): lower_names = {n.strip().lower() for n in names if n} ref_vk = variant_key(ref.name) safe = [] for r in rows: if (r["alias"] or "").strip().lower() in lower_names or variant_key(r["canonical_name"]) == ref_vk: safe.append(r) else: log.info("alias collision refused", extra={"ref_name": ref.name, "candidate": r["canonical_name"]}) # `name` is reserved by LogRecord rows = safe if not rows: return None kept = [] for r in rows: if await self._conflicting_identifier(r["id"], ref): await self._review("merge_candidate", sorted({r["id"]}), f"'{ref.name}' shares a name with {r['canonical_name']} but has different identifiers", {"name": ref.name, "identifiers": ref.identifiers}) else: kept.append(r) rows = kept if not rows: return None ids = {r["id"] for r in rows} # persisted decisions: a candidate an operator kept separate from another entity is only accepted with a matching organisation separate = await self._kept_separate(sorted(ids)) if separate: rows = [r for r in rows if r["id"] not in separate or (org_id and r["organization_id"] == org_id)] ids = {r["id"] for r in rows} if not rows: return None if len(ids) == 1: return rows[0]["id"] if org_id: same_org = [r for r in rows if r["organization_id"] == org_id] if len({r["id"] for r in same_org}) == 1: return same_org[0]["id"] # ambiguous: do not guess — create and flag await self._review("merge_candidate", sorted(ids), f"alias '{ref.name}' matches {len(ids)} {ref.entity_type} entities", {"name": ref.name, "identifiers": ref.identifiers}) return None async def _kept_separate(self, ids: list[str]) -> set[str]: """Ids among `ids` that carry a `keep_separate` decision with any entity.""" if not ids: return set() rows = await fetch_all(self.conn, """select a_id, b_id from resolution_decisions where decision = 'keep_separate' and (a_id = any(cast(:ids as text[])) or b_id = any(cast(:ids as text[])))""", ids=ids) out: set[str] = set() for r in rows: out.update(x for x in (r["a_id"], r["b_id"]) if x in ids) return out async def kept_separate(self, a: str, b: str) -> bool: row = await fetch_one(self.conn, """select 1 from resolution_decisions where decision = 'keep_separate' and ((a_id = :a and b_id = :b) or (a_id = :b and b_id = :a)) limit 1""", a=a, b=b) return row is not None async def _by_slug(self, ref: EntityRef) -> str | None: slug = ref.slug_hint or slugify(ref.name) row = await fetch_one(self.conn, "select id, entity_type, organization_id, merged_into from entities where slug = :s", s=slug) if row and row["entity_type"] in compatible_types(ref.entity_type) and not await self._conflicting_identifier(row["merged_into"] or row["id"], ref): return row["merged_into"] or row["id"] return None async def resolve_variant(self, ref: EntityRef, *, org_id: str | None = None) -> tuple[str, dict[str, str]] | None: """Evaluation-effort variant ("claude-opus-5-medium", "qwen3-6-27b-non-reasoning") → (canonical model id, effort config) when a canonical model exists: same `artificial_analysis` identifier or alias as the base name, or the same `variant_key` (through the optional precomputed index). Returns None when the base cannot be resolved or is ambiguous.""" a = analyze_model_name(ref.name) if not a.is_effort_variant: return None base = base_name(ref.name) bases = {base, base.lower(), a.base_key, slugify(base)} # a dated snapshot in the base ("gpt-5.2-2025-12-11-high") also names the undated model undated = re.sub(r"[-_ ]?(20\d{2}[-_.]?\d{2}[-_.]?\d{2}|\d{4})$", "", base).strip("-_ ") if undated and undated != base and re.search(r"[a-z]", undated, re.I): bases |= {undated, undated.lower(), slugify(undated)} bases = sorted(bases) base_norms = sorted({normalize_alias(b) for b in bases} - {""}) # the base may be known by any alias, any identifier value (AA slug, api model id, provider id…), its slug or its api_model_id rows = await fetch_all(self.conn, """select distinct coalesce(e.merged_into, e.id) as id, e.canonical_name, e.organization_id from entities e left join entity_identifiers ei on ei.entity_id = e.id left join entity_aliases al on al.entity_id = e.id where e.entity_type = 'model' and (ei.value = any(cast(:bases as text[])) or al.alias_norm = any(cast(:norms as text[])) or e.slug = any(cast(:bases as text[])) or lower(e.attributes->>'api_model_id') = any(cast(:bases as text[])))""", bases=bases, norms=base_norms) candidates = {r["id"]: r for r in rows if r["id"] != ref.id} if self.variant_index: vk = variant_key(base) cid = self.variant_index.get(vk) if cid and cid != ref.id and cid not in candidates: row = await fetch_one(self.conn, "select id, canonical_name, organization_id, merged_into from entities where id = :id and merged_into is null", id=cid) if row: candidates[cid] = row # candidates must be live models (a merged row resolved to its survivor above) and never another effort variant live = {r["id"] for r in await fetch_all(self.conn, "select id from entities where id = any(cast(:ids as text[])) and merged_into is null and entity_type = 'model'", ids=sorted(candidates))} if candidates else set() candidates = {k: v for k, v in candidates.items() if k in live and not analyze_model_name(v["canonical_name"]).is_effort_variant} if not candidates: return None if len(candidates) > 1: # prefer the candidate an evaluator/vendor identifies by the exact base name, then the variant's organisation exact = await fetch_all(self.conn, "select distinct entity_id from entity_identifiers where value = any(cast(:b as text[])) and entity_id = any(cast(:ids as text[]))", b=bases, ids=sorted(candidates)) exact_ids = {r["entity_id"] for r in exact} if len(exact_ids) == 1: candidates = {k: v for k, v in candidates.items() if k in exact_ids} if len(candidates) > 1 and org_id: same = {k: v for k, v in candidates.items() if v["organization_id"] == org_id} if same: candidates = same if len(candidates) != 1: await self._review("merge_candidate", sorted(candidates), f"effort variant '{ref.name}' matches {len(candidates)} base models", {"name": ref.name, "base": base}) return None cid = next(iter(candidates)) return cid, dict(a.effort) # ---------------------------------------------------------------------------------------------- writes async def _create(self, ref: EntityRef, org_id: str | None) -> str: eid = new_id(ref.entity_type) slug = await self._unique_slug(ref, org_id) now = datetime.now(UTC) first_seen = now if ref.first_seen_hint is not None: hint = ref.first_seen_hint if ref.first_seen_hint.tzinfo else ref.first_seen_hint.replace(tzinfo=UTC) first_seen = min(now, hint) await execute(self.conn, """insert into entities (id, entity_type, canonical_name, slug, description, status, organization_id, first_seen_at, last_seen_at, identity_confidence, artifact_kind) values (:id, :t, :n, :slug, :d, :status, :org, :fs, now(), :ic, :ak)""", id=eid, t=ref.entity_type, n=ref.name.strip()[:300], slug=slug, d=(ref.description or None), status=ref.status or "active", org=org_id, fs=first_seen, ic=ref.identity_confidence or "high", ak=ref.artifact_kind if ref.entity_type == "artifact" else None) self.created.append(eid) return eid async def _touch(self, eid: str, ref: EntityRef, org_id: str | None) -> None: sets = ["last_seen_at = now()"] params: dict[str, Any] = {"id": eid} if org_id: sets.append("organization_id = coalesce(organization_id, :org)") params["org"] = org_id if ref.description: sets.append("description = case when description is null or length(description) < 40 then :d else description end") params["d"] = ref.description if ref.first_seen_hint is not None: sets.append("first_seen_at = least(first_seen_at, :fs)") params["fs"] = ref.first_seen_hint if ref.first_seen_hint.tzinfo else ref.first_seen_hint.replace(tzinfo=UTC) await execute(self.conn, f"update entities set {', '.join(sets)} where id = :id", **params) self.updated.add(eid) async def _refresh_links(self, ref: EntityRef) -> None: assert ref.id for alias in {ref.name, *ref.aliases}: norm = normalize_alias(alias) if not norm or len(norm) < 2: continue await execute(self.conn, """insert into entity_aliases (entity_id, alias, alias_norm, kind, snapshot_id) values (:e, :a, :n, 'alias', :s) on conflict (entity_id, alias_norm) do nothing""", e=ref.id, a=alias.strip()[:300], n=norm, s=self.snapshot_id) for scheme, value in ref.identifiers.items(): await execute(self.conn, """insert into entity_identifiers (entity_id, scheme, value, snapshot_id) values (:e, :s, :v, :snap) on conflict (scheme, value) do nothing""", e=ref.id, s=scheme, v=str(value)[:500], snap=self.snapshot_id) async def _unique_slug(self, ref: EntityRef, org_id: str | None) -> str: base = ref.slug_hint or slugify(ref.name) if base in GENERIC_NAMES or len(base) < 2: base = f"{ref.entity_type}-{base}" candidates = [base] if org_id: org_slug = await fetch_one(self.conn, "select slug from entities where id = :id", id=org_id) if org_slug and not base.startswith(org_slug["slug"]): candidates.append(f"{org_slug['slug']}-{base}") for c in candidates: if not await fetch_one(self.conn, "select 1 from entities where slug = :s", s=c): return c n = 2 while True: c = f"{base}-{n}" if not await fetch_one(self.conn, "select 1 from entities where slug = :s", s=c): return c n += 1 async def _review(self, kind: str, entity_ids: list[str], reason: str, payload: dict[str, Any]) -> None: dedupe = f"{kind}:{':'.join(entity_ids)}:{normalize_alias(reason)[:80]}" await execute(self.conn, """insert into review_queue (id, kind, entity_ids, payload, reason, dedupe_key) values (:id, :k, :ids, cast(:p as jsonb), :r, :d) on conflict (dedupe_key) do nothing""", id=new_id("review"), k=kind, ids=entity_ids, p=jsonb(payload), r=reason, d=dedupe) __all__ = ["EVALUATOR_SCHEMES", "Resolver", "compatible_types"]