HTML 77.2%
TypeScript 10.5%
Python 9.6%
JavaScript 2.5%
1"""Canonicalization engine — `aia canonicalize [--apply] [--step …]`.23Turns the flat "everything is a model" corpus into the canonical hierarchy (model_family → model → artifact, effort variants folded4into result configurations), normalises taxonomies, links licences, classifies events (backfill vs live), enforces benchmark result5comparability and flags anomalies. Rules are documented in docs/CANONICALIZATION.md.67Invariants8 * dry-run by default: every step computes a plan from reads only and reports counts + examples; `--apply` executes it9 * nothing is ever deleted; raw snapshots are never touched; every write is idempotent (a second `--apply` is a no-op)10 * merges go through `services.merge.merge_entities` (persisted decision + audit log); relations through `upsert_relation`11 * derived claims are written by a tier-2 `derived` FactWriter (source `ai-atlas.registry`) and never supersede tier-1 statements12"""13from __future__ import annotations1415import json16import logging17import re18from collections import defaultdict19from dataclasses import dataclass, field20from datetime import UTC, datetime, timedelta21from typing import Any2223from sqlalchemy.ext.asyncio import AsyncConnection2425from aiatlas.db import execute, fetch_all, fetch_one, jsonb, transaction26from aiatlas.ids import new_id, normalize_alias, slugify27from aiatlas.ontology import benchmarks as bench_ontology28from aiatlas.ontology.anomalies import Anomaly, check_hardware, check_model, check_price, check_result29from aiatlas.ontology.licenses import LICENSES, normalize_license30from aiatlas.ontology.models import (31 CONVERTER_ORGS,32 analyze_model_name,33 base_name,34 effort_config,35 family_hint,36 family_release_hint,37 official_orgs,38 variant_key,39)40from aiatlas.ontology.openness import derive_openness, normalize_openness, openness_dimensions41from aiatlas.ontology.taxonomy import ORG_TYPE_DEFAULT_KIND, TAXONOMY_PROPERTIES, normalize_property42from aiatlas.sdk.facts import EntityRef, Facts, facts_from_json43from aiatlas.sdk.resolution import EVALUATOR_SCHEMES, Resolver44from aiatlas.sdk.writer import FactWriter, _same45from aiatlas.services.anomalies import record46from aiatlas.services.events import BACKFILL_LAG_DAYS, group_key_for47from aiatlas.services.merge import (48 ORG_TYPES,49 audit,50 enforce_current_results,51 kept_separate,52 merge_entities,53 record_decision,54 registry_source_id,55 upsert_relation,56)5758log = logging.getLogger(__name__)5960CANON_VERSION = "2026.09"61STRUCTURAL_CONNECTORS = {"curation", "canonicalize"}62STEPS = ("duplicates", "variants", "artifacts", "families", "licenses", "taxonomy", "results", "events", "anomalies")63DEDUPE_TYPES = ("model", "provider", "benchmark", "dataset", "framework", "library", "hardware")64LICENSED_TYPES = ("model", "artifact", "dataset", "framework", "library", "repository")65QUANT_ATTR_FORMATS = {"gguf", "awq", "gptq", "exl2", "exl3", "int4", "int8", "fp8", "nvfp4", "mxfp4", "fp4", "bnb", "quantized"}666768@dataclass69class StepReport:70 name: str71 counts: dict[str, int] = field(default_factory=dict)72 examples: list[str] = field(default_factory=list)73 notes: list[str] = field(default_factory=list)7475 def bump(self, key: str, n: int = 1) -> None:76 self.counts[key] = self.counts.get(key, 0) + n7778 def example(self, text: str, *, limit: int = 12) -> None:79 if len(self.examples) < limit:80 self.examples.append(text)8182 @property83 def changes(self) -> int:84 return sum(v for k, v in self.counts.items() if not k.startswith("_"))858687@dataclass88class Report:89 apply: bool90 steps: list[StepReport] = field(default_factory=list)91 started_at: datetime = field(default_factory=lambda: datetime.now(UTC))9293 @property94 def changes(self) -> int:95 return sum(s.changes for s in self.steps)9697 def render(self) -> str:98 mode = "APPLY" if self.apply else "DRY-RUN"99 lines = [f"aia canonicalize — {mode} — {self.started_at:%Y-%m-%d %H:%M:%S} UTC — {self.changes} change(s)"]100 for s in self.steps:101 lines.append(f"\n[{s.name}] {s.changes} change(s)")102 for k, v in sorted(s.counts.items()):103 lines.append(f" {k:<40} {v}")104 for n in s.notes:105 lines.append(f" · {n}")106 for e in s.examples:107 lines.append(f" - {e}")108 return "\n".join(lines)109110 def as_dict(self) -> dict[str, Any]:111 return {"apply": self.apply, "started_at": self.started_at.isoformat(), "changes": self.changes,112 "steps": [{"name": s.name, "counts": s.counts, "examples": s.examples, "notes": s.notes} for s in self.steps]}113114115async def canonicalize(*, apply: bool = False, steps: list[str] | None = None, scope: set[str] | None = None) -> Report:116 """Run the steps in canonical order. `scope` (entity ids) restricts the rows a step *acts on* — lookups (bases, families, licences)117 still see the whole corpus; used by tests and targeted re-runs."""118 wanted = [s for s in STEPS if not steps or s in steps]119 unknown = set(steps or []) - set(STEPS)120 if unknown:121 raise ValueError(f"unknown step(s): {sorted(unknown)}; known: {STEPS}")122 report = Report(apply=apply)123 for name in wanted:124 rep = StepReport(name=name)125 fn = _STEP_FUNCTIONS[name]126 async with transaction() as conn:127 await fn(conn, rep, apply, scope=scope)128 if apply and rep.changes:129 await audit(conn, f"canonicalize.{name}", None, {"counts": rep.counts, "version": CANON_VERSION}, actor="canonicalize")130 report.steps.append(rep)131 log.info("canonicalize step done", extra={"step": name, "apply": apply, **{k: v for k, v in rep.counts.items()}})132 return report133134135# ---------------------------------------------------------------------------------------------- shared helpers136def _derived_writer(conn: AsyncConnection, source_id: str | None) -> FactWriter:137 return FactWriter(conn, source_id=source_id, snapshot_id=None, source_url=None, tier=2, connector_name="canonicalize", extractor="derived",138 extractor_version=CANON_VERSION, run_id=f"canon_{datetime.now(UTC):%Y%m%d}")139140141async def _review(conn: AsyncConnection, kind: str, entity_ids: list[str], reason: str, payload: dict[str, Any], *, apply: bool = True) -> bool:142 """Queue a review item once (dedupe key). Returns True when the item is new (or would be, in dry-run) so reports stay idempotent."""143 dedupe = f"{kind}:{':'.join(entity_ids)}:{normalize_alias(reason)[:80]}"144 if await fetch_one(conn, "select 1 from review_queue where dedupe_key = :d", d=dedupe):145 return False146 if not apply:147 return True148 row = await fetch_one(conn, """insert into review_queue (id, kind, entity_ids, payload, reason, dedupe_key) values (:id, :k, :ids, cast(:p as jsonb), :r, :d)149 on conflict (dedupe_key) do nothing returning id""", id=new_id("review"), k=kind, ids=entity_ids, p=jsonb(payload), r=reason, d=dedupe)150 return row is not None151152153async def _set_attributes(conn: AsyncConnection, entity_id: str, attrs: dict[str, Any], *, source_id: str | None) -> None:154 """Direct attribute write for structural/derived facts (family label, license_key, org_kind default) with `derived` provenance."""155 prov = {k: {"source_id": source_id, "tier": 2, "confidence": "high", "extractor": "derived", "observed_at": datetime.now(UTC).isoformat(timespec="seconds")} for k in attrs}156 await execute(conn, "update entities set attributes = attributes || cast(:a as jsonb), provenance = provenance || cast(:p as jsonb), updated_at = now() where id = :id",157 a=jsonb(attrs), p=jsonb(prov), id=entity_id)158159160async def _claim_counts(conn: AsyncConnection, ids: list[str]) -> dict[str, int]:161 if not ids:162 return {}163 rows = await fetch_all(conn, "select entity_id, count(*) as n from claims where status = 'current' and entity_id = any(cast(:ids as text[])) group by 1", ids=ids)164 return {r["entity_id"]: int(r["n"]) for r in rows}165166167async def _identifiers(conn: AsyncConnection, ids: list[str]) -> dict[str, dict[str, set[str]]]:168 out: dict[str, dict[str, set[str]]] = defaultdict(lambda: defaultdict(set))169 if not ids:170 return out171 rows = await fetch_all(conn, "select entity_id, scheme, value from entity_identifiers where entity_id = any(cast(:ids as text[]))", ids=ids)172 for r in rows:173 out[r["entity_id"]][r["scheme"]].add(r["value"])174 return out175176177_SNAPSHOT_REMAINDER = re.compile(r"^(\d{1,8}|preview\d*|exp\d*|latest|beta|alpha|v\d+)$")178_EFFORT_REMAINDER = {"reasoning", "nonreasoning", "thinking", "nonthinking", "high", "low", "medium", "xhigh", "minimal", "instruct", "it", "chat"}179_VENDOR_PREFIXES = {"meta", "metallama", "google", "openai", "alibaba", "nvidia", "microsoft", "anthropic", "qwen", "deepseek", "mistral", "mistralai", "zai", "moonshotai"}180_VERSION_TOKEN = re.compile(r"^v?\d+(\.\d+)*$")181182183def _identifier_alias(a: str, b: str) -> bool:184 """Two values of one scheme name the same thing when one is a dated/preview snapshot, an effort/instruct suffix, a vendor-prefixed185 or re-versioned spelling of the other (`gemini-2.5-flash-lite` ~ `gemini-2.5-flash-lite-preview-06-17`, `google/gemini-2.5-flash-lite` ~186 `google/gemini-2.5-flash-lite-preview`, `meta-llama/Meta-Llama-3.1-8B-Instruct` ~ `meta-llama/Llama-3.1-8B-Instruct`,187 `gemini-omni-flash` ~ `gemini-omni-1.1-flash`)."""188 na, nb = normalize_alias(a), normalize_alias(b)189 if na == nb:190 return True191 short, long_ = sorted((na, nb), key=len)192 if long_.startswith(short):193 rest = long_[len(short):]194 if _SNAPSHOT_REMAINDER.match(rest) or rest in _EFFORT_REMAINDER or rest.startswith("preview") or rest.startswith("exp"):195 return True196 if long_.endswith(short) and long_[: -len(short)] in _VENDOR_PREFIXES:197 return True198 # re-versioned / vendor-duplicated spellings: same word tokens once version and vendor tokens are removed, and the version tokens of199 # one side are a subset of the other's (`gemini-omni-flash` ⊂ `gemini-omni-1.1-flash`; `gpt-4` vs `gpt-5` stay different)200 ta = [t for t in re.split(r"[-/_.\s]+", a.lower()) if t]201 tb = [t for t in re.split(r"[-/_.\s]+", b.lower()) if t]202 words_a = [t for t in ta if not _VERSION_TOKEN.match(t) and t not in _VENDOR_PREFIXES]203 words_b = [t for t in tb if not _VERSION_TOKEN.match(t) and t not in _VENDOR_PREFIXES]204 va = {t for t in ta if _VERSION_TOKEN.match(t)}205 vb = {t for t in tb if _VERSION_TOKEN.match(t)}206 return bool(words_a) and words_a == words_b and (va <= vb or vb <= va)207208209def _identifiers_conflict(a: dict[str, set[str]], b: dict[str, set[str]]) -> bool:210 """Hard contradiction only: a shared scheme whose values are not aliases of one another."""211 for s in set(a) & set(b):212 if not a[s] or not b[s] or a[s] == b[s]:213 continue214 if not all(_identifier_alias(x, y) for x in a[s] for y in b[s]):215 return True216 return False217218219def _conflict_detail(a: dict[str, set[str]], b: dict[str, set[str]]) -> str:220 parts = []221 for s in sorted(set(a) & set(b)):222 if a[s] and b[s] and a[s] != b[s] and not all(_identifier_alias(x, y) for x in a[s] for y in b[s]):223 parts.append(f"{s}: {'/'.join(sorted(a[s]))} vs {'/'.join(sorted(b[s]))}")224 return "; ".join(parts)225226227def _same_publisher(model_name: str, org_a: dict[str, Any] | None, org_b: dict[str, Any] | None) -> bool:228 """Alibaba/Qwen, Meta/Meta AI…: both organisations publish the model's family officially."""229 if not org_a or not org_b:230 return False231 official = set(official_orgs(model_name))232 return bool(official) and bool(_org_lookup_keys(org_a) & official) and bool(_org_lookup_keys(org_b) & official)233234235def _org_lookup_keys(row: dict[str, Any]) -> set[str]:236 attrs = row.get("attributes") or {}237 keys = {row["slug"], normalize_alias(row["canonical_name"])}238 for k in ("hf_org", "github_org"):239 v = attrs.get(k)240 if isinstance(v, str) and v.strip():241 keys.add(v.strip().lower())242 return {k for k in keys if k}243244245# ---------------------------------------------------------------------------------------------- step: duplicates246def _in_scope(scope: set[str] | None, entity_id: str) -> bool:247 return scope is None or entity_id in scope248249250async def _retype_routers(conn: AsyncConnection, rep: StepReport, apply: bool, scope: set[str] | None) -> None:251 """OpenRouter's own routers (`openrouter/auto`, `openrouter/pareto-code`…) are products, not models: no weights, no developer."""252 rows = await fetch_all(conn, """select distinct e.id, e.slug, e.canonical_name from entities e left join entity_identifiers ei on ei.entity_id = e.id253 where e.entity_type = 'model' and e.merged_into is null254 and ((ei.scheme = 'openrouter' and ei.value like 'openrouter/%') or e.attributes->>'openrouter_id' like 'openrouter/%')""")255 for r in rows:256 if not _in_scope(scope, r["id"]):257 continue258 rep.bump("routers_retyped_as_product")259 rep.example(f"product[router] {r['slug']} ({r['canonical_name']})")260 if apply:261 await execute(conn, """update entities set entity_type = 'product', attributes = attributes || '{"kind": "router"}'::jsonb, updated_at = now() where id = :id""", id=r["id"])262 await execute(conn, "update relations set valid_to = now() where predicate = 'develops' and valid_to is null and (subject_id = :id or object_id = :id)", id=r["id"])263264265async def step_duplicates(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:266 source_id = await registry_source_id(conn)267 await _retype_routers(conn, rep, apply, scope)268 rows = await fetch_all(conn, """select id, entity_type, canonical_name, slug, organization_id, attributes, first_seen_at from entities269 where merged_into is null and entity_type = any(cast(:types as text[]))""", types=list(DEDUPE_TYPES) + sorted(ORG_TYPES))270 if scope is not None:271 rows = [r for r in rows if r["id"] in scope]272 orgs_by_id = {o["id"]: o for o in await fetch_all(conn, "select id, slug, canonical_name, attributes from entities where entity_type = any(cast(:t as text[]))", t=sorted(ORG_TYPES))}273 # group: exact normalised name within a type; organisations across the org group (+ shared hf_org/github_org)274 groups: dict[str, list[dict[str, Any]]] = defaultdict(list)275 for r in rows:276 norm = normalize_alias(r["canonical_name"])277 if r["entity_type"] in ORG_TYPES:278 groups[f"org:{norm}"].append(r)279 for k in ("hf_org", "github_org"):280 v = (r["attributes"] or {}).get(k)281 if isinstance(v, str) and v.strip() and normalize_alias(v) != norm:282 groups[f"org:{normalize_alias(v)}"].append(r)283 else:284 groups[f"{r['entity_type']}:{norm}"].append(r)285 # union overlapping org groups286 parent: dict[str, str] = {}287288 def find(x: str) -> str:289 while parent.setdefault(x, x) != x:290 x = parent[x]291 return x292293 for members in groups.values():294 ids = [m["id"] for m in members]295 for other in ids[1:]:296 parent[find(other)] = find(ids[0])297 clusters: dict[str, list[dict[str, Any]]] = defaultdict(list)298 seen: set[str] = set()299 for members in groups.values():300 for m in members:301 if m["id"] not in seen:302 seen.add(m["id"])303 clusters[find(m["id"])].append(m)304 dup_clusters = [c for c in clusters.values() if len(c) > 1]305 all_ids = [m["id"] for c in dup_clusters for m in c]306 claims = await _claim_counts(conn, all_ids)307 idents = await _identifiers(conn, all_ids)308 type_rank = {"company": 0, "lab": 0, "university": 0, "organization": 1} # curated org types beat the generic hub "organization"309310 def collision_slug(m: dict[str, Any], cluster: list[dict[str, Any]]) -> int:311 """1 when the slug is a collision product — an organisation prefix (`google-gemma-4-31b`, `meta-ai-llama-…`) or a `-2` suffix over312 another member's slug (`minimax-m3-2`): the plain slug survives."""313 org = orgs_by_id.get(m["organization_id"] or "")314 if org and m["slug"].startswith(org["slug"] + "-") and not normalize_alias(m["canonical_name"]).startswith(normalize_alias(org["slug"])):315 return 1316 mm = re.match(r"^(.*)-\d+$", m["slug"])317 return int(bool(mm) and any(o["slug"] == mm.group(1) for o in cluster if o is not m))318319 for cluster in dup_clusters:320 cluster.sort(key=lambda m: (type_rank.get(m["entity_type"], 0), collision_slug(m, cluster), -claims.get(m["id"], 0), m["first_seen_at"]))321 survivor = cluster[0]322 for other in cluster[1:]:323 pair = sorted([survivor["id"], other["id"]])324 if await kept_separate(conn, survivor["id"], other["id"]):325 rep.bump("_kept_separate")326 continue327 if _identifiers_conflict(idents[survivor["id"]], idents[other["id"]]):328 detail = _conflict_detail(idents[survivor["id"]], idents[other["id"]])329 rep.bump("keep_separate_recorded")330 rep.example(f"keep separate: {other['slug']} vs {survivor['slug']} — {detail}")331 if apply:332 await record_decision(conn, other["id"], survivor["id"], "keep_separate", actor="canonicalize",333 note=f"same name, contradictory identifiers ({detail})", payload={"step": "duplicates", "slugs": [other["slug"], survivor["slug"]]})334 await execute(conn, "update review_queue set status = 'rejected', resolved_at = now() where status = 'pending' and kind = 'merge_candidate' and entity_ids @> :ids and entity_ids <@ :ids",335 ids=pair)336 continue337 if (survivor["organization_id"] and other["organization_id"] and survivor["organization_id"] != other["organization_id"] and other["entity_type"] not in ORG_TYPES338 and not _same_publisher(survivor["canonical_name"], orgs_by_id.get(survivor["organization_id"]), orgs_by_id.get(other["organization_id"]))):339 if await _review(conn, "merge_candidate", pair, f"'{other['canonical_name']}' and '{survivor['canonical_name']}' share a name but have different organisations",340 {"slugs": [other["slug"], survivor["slug"]], "step": "duplicates"}, apply=apply):341 rep.bump("review_different_organizations")342 rep.example(f"review: {other['slug']} and {survivor['slug']} share a name but belong to different organisations")343 continue344 rep.bump("merged")345 rep.example(f"merge {other['entity_type']} {other['slug']} → {survivor['slug']}")346 if apply:347 await merge_entities(conn, other["id"], survivor["id"], mode="merge", actor="canonicalize", note="exact normalised-name duplicate",348 payload={"step": "duplicates"})349 # junk organisations (single-letter names) → review only350 for r in rows:351 if r["entity_type"] in ORG_TYPES and len(r["canonical_name"].strip()) <= 1:352 if await _review(conn, "junk_entity", [r["id"]], f"organisation '{r['canonical_name']}' ({r['slug']}) looks like extraction noise", {"slug": r["slug"]}, apply=apply):353 rep.bump("review_junk_organization")354 rep.example(f"review junk organisation {r['slug']!r}")355 # org_kind defaults / normalisation356 for r in rows:357 if r["entity_type"] not in ORG_TYPES:358 continue359 attrs = r["attributes"] or {}360 current = attrs.get("org_kind")361 default = ORG_TYPE_DEFAULT_KIND.get(r["entity_type"])362 canon, _raw, _m = normalize_property(r["entity_type"], "org_kind", current) if current else (None, None, [])363 target = canon if isinstance(canon, str) and canon in ("company", "lab", "university", "nonprofit", "government", "community", "consortium", "individual") else default364 if target and current != target:365 rep.bump("org_kind_set")366 if apply:367 await _set_attributes(conn, r["id"], {"org_kind": target}, source_id=source_id)368369370# ---------------------------------------------------------------------------------------------- step: variants371async def _models(conn: AsyncConnection, *, types: tuple[str, ...] = ("model",)) -> list[dict[str, Any]]:372 return await fetch_all(conn, """select e.id, e.entity_type, e.canonical_name, e.slug, e.organization_id, e.attributes, e.first_seen_at, e.family_id, e.canonical_id,373 e.identity_confidence, o.slug as org_slug, o.canonical_name as org_name, o.attributes->>'hf_org' as org_hf374 from entities e left join entities o on o.id = e.organization_id375 where e.merged_into is null and e.entity_type = any(cast(:t as text[])) order by e.first_seen_at, e.id""", t=list(types))376377378def _variant_index(models: list[dict[str, Any]], aliases: dict[str, list[str]] | None = None) -> tuple[dict[str, str], dict[str, list[dict[str, Any]]]]:379 """variant_key → canonical model id (unique keys only) and the full multi-map; with `aliases` (model id → alias strings) every alias of a380 canonical model contributes its own key (`gpt-3.5-turbo-0613`, `o3-mini-2025-01-31`…)."""381 multi: dict[str, list[dict[str, Any]]] = defaultdict(list)382 for m in models:383 a = analyze_model_name(m["canonical_name"])384 if a.is_effort_variant or a.is_artifact:385 continue386 keys = {variant_key(m["canonical_name"])}387 for al in (aliases or {}).get(m["id"], []):388 aa = analyze_model_name(al)389 if not aa.is_effort_variant and not aa.is_artifact:390 keys.add(variant_key(al))391 for k in keys:392 if k and m not in multi[k]:393 multi[k].append(m)394 return {k: v[0]["id"] for k, v in multi.items() if len(v) == 1}, multi395396397async def _model_aliases(conn: AsyncConnection, ids: list[str]) -> dict[str, list[str]]:398 out: dict[str, list[str]] = defaultdict(list)399 for r in await fetch_all(conn, "select entity_id, alias from entity_aliases where entity_id = any(cast(:ids as text[]))", ids=ids):400 out[r["entity_id"]].append(r["alias"])401 return out402403404async def step_variants(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:405 models = await _models(conn)406 index, _ = _variant_index(models, await _model_aliases(conn, [m["id"] for m in models]))407 resolver = Resolver(conn, source_tier=2, variant_index=index)408 candidates = [m for m in models if _in_scope(scope, m["id"]) and analyze_model_name(m["canonical_name"]).is_effort_variant]409 # names that used to look like variants (before the ontology learnt they are tiers/official releases) drop their hint and regain full identity410 for m in models:411 if _in_scope(scope, m["id"]) and (m["attributes"] or {}).get("evaluation_variant_of_hint") and not analyze_model_name(m["canonical_name"]).is_effort_variant:412 rep.bump("stale_variant_hint_cleared")413 if apply:414 await execute(conn, """update entities set attributes = attributes - 'evaluation_variant_of_hint', provenance = provenance - 'evaluation_variant_of_hint',415 identity_confidence = 'high', updated_at = now() where id = :id""", id=m["id"])416 await execute(conn, "update review_queue set status = 'rejected', resolved_at = now() where status = 'pending' and kind = 'variant_candidate' and :id = any(entity_ids)", id=m["id"])417 idents = await _identifiers(conn, [m["id"] for m in candidates])418 official = {r["entity_id"] for r in await fetch_all(conn, """select distinct entity_id from claims where tier = 1 and status = 'current'419 and entity_id = any(cast(:ids as text[]))""", ids=[m["id"] for m in candidates])} if candidates else set()420 for m in candidates:421 a = analyze_model_name(m["canonical_name"])422 rep.bump("_effort_variants_seen")423 schemes = set(idents.get(m["id"], {}))424 if not schemes <= EVALUATOR_SCHEMES or (m["attributes"] or {}).get("hf_repo") or m["id"] in official:425 # known to an official source, a hub or a provider → a real model that happens to end in "thinking"/"high"; never folded426 rep.bump("_kept_real_model")427 continue428 ref = EntityRef(entity_type="model", name=m["canonical_name"], id=m["id"])429 folded = await resolver.resolve_variant(ref, org_id=m["organization_id"])430 if folded is None:431 hint = base_name(m["canonical_name"])432 attrs = m["attributes"] or {}433 if attrs.get("evaluation_variant_of_hint") == hint and m["identity_confidence"] == "low":434 continue435 rep.bump("unresolved_flagged")436 rep.example(f"unresolved variant {m['slug']} (base '{hint}' not found) → identity low + review")437 if apply:438 await _set_attributes(conn, m["id"], {"evaluation_variant_of_hint": hint}, source_id=await registry_source_id(conn))439 await execute(conn, "update entities set identity_confidence = 'low' where id = :id", id=m["id"])440 await _review(conn, "variant_candidate", [m["id"]], f"'{m['canonical_name']}' looks like an evaluation-effort variant of '{hint}' but no such model exists",441 {"slug": m["slug"], "base": hint, "effort": a.effort}, apply=apply)442 continue443 cid, effort = folded444 rep.bump("folded")445 target = next((x for x in models if x["id"] == cid), None)446 rep.example(f"fold {m['slug']} → {target['slug'] if target else cid} {effort}")447 if apply:448 rows = await fetch_all(conn, "select id, config from benchmark_results where model_id = :m", m=m["id"])449 for r in rows:450 cfg = effort_config(m["canonical_name"], r["config"])451 if not _same(cfg, r["config"]):452 await execute(conn, "update benchmark_results set config = cast(:c as jsonb) where id = :id", c=jsonb(cfg), id=r["id"])453 rep.bump("results_reconfigured", len(rows))454 await merge_entities(conn, m["id"], cid, mode="merge", actor="canonicalize", note="evaluation-effort variant folded into its canonical model",455 payload={"step": "variants", "effort": effort, "variant_slug": m["slug"]})456457458# ---------------------------------------------------------------------------------------------- step: artifacts459async def step_artifacts(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:460 source_id = await registry_source_id(conn)461 models = await _models(conn)462 _, multi = _variant_index(models)463 by_id = {m["id"]: m for m in models}464 # only `quantized_from` names the packaged model; `derived_from`/`fine_tuned_from` point at a *base* model (a fine-tune is a new model)465 rel_rows = await fetch_all(conn, """select r.subject_id, r.object_id from relations r join entities o on o.id = r.object_id466 where r.predicate = 'quantized_from' and r.valid_to is null and o.merged_into is null and o.entity_type = 'model'""")467 base_of: dict[str, str] = {}468 for r in rel_rows:469 base_of.setdefault(r["subject_id"], r["object_id"])470 official_idents = await _identifiers(conn, [m["id"] for m in models if (m["attributes"] or {}).get("hf_repo")])471 for m in models:472 if not _in_scope(scope, m["id"]):473 continue474 attrs = m["attributes"] or {}475 hf_repo = attrs.get("hf_repo") if isinstance(attrs.get("hf_repo"), str) else None476 probe = hf_repo or m["canonical_name"]477 a = analyze_model_name(probe)478 a_name = analyze_model_name(m["canonical_name"]) if hf_repo else a479 repo_org = (a.repo_org or "").lower()480 # repo attributes (is_quantized / quant_format) describe a hub repository: without hf_repo they come from a provider endpoint481 # (OpenRouter lists the serving precision) and say nothing about the model's identity482 quant_attr = str(attrs.get("quant_format") or "").lower() if hf_repo else ""483 attr_quantized = attrs.get("is_quantized") is True and bool(hf_repo)484 has_token = bool(a.quant_formats or a.precision or a_name.quant_formats or a_name.precision)485 flagged = has_token or a.is_artifact or a_name.is_artifact or attr_quantized or bool(quant_attr) or repo_org in CONVERTER_ORGS486 if not flagged:487 continue488 # "official" = the repo belongs to the model's developer: the family's known publisher, or the entity's own organisation when that489 # organisation is not a redistributor (hub-derived entities are attached to the redistributor org, e.g. `bartowski`)490 own_org_keys = {k for k in ((m["org_slug"] or "").lower(), (m["org_hf"] or "").lower(), normalize_alias(m["org_name"] or "")) if k}491 official = bool(repo_org) and repo_org not in CONVERTER_ORGS and (repo_org in set(official_orgs(probe)) or repo_org in own_org_keys)492 vk_candidates = [c for c in multi.get(variant_key(probe), []) if c["id"] != m["id"]]493 cid: str | None = None494 if official:495 # A quant/precision token in the repo name makes it an artifact whatever the organisation (an official FP8/GGUF repo is a496 # conversion of the model). Canonical = the same organisation's plain model with the same variant_key when it exists, else NULL.497 # Two guards, because a model is never invented or erased: (1) a native-dtype attribute without a token in the name498 # (DeepSeek-R1 `quant_format=fp8`) is the model; (2) when evaluators/providers know the tagged entity under its own identity499 # (Nemotron 3 Ultra whose only hub repo is `…-BF16`) it IS the model → merged into the plain sibling when one exists, kept otherwise.500 if not has_token:501 rep.bump("_official_checkpoint_kept_as_model")502 continue503 same_org = [c for c in vk_candidates if c["organization_id"] == m["organization_id"] and not analyze_model_name((c["attributes"] or {}).get("hf_repo") or c["canonical_name"]).is_artifact]504 external = set(official_idents.get(m["id"], {})) - {"hf_repo"}505 if external:506 if len(same_org) == 1 and not await kept_separate(conn, m["id"], same_org[0]["id"]):507 rep.bump("official_checkpoint_merged_into_model")508 rep.example(f"merge official checkpoint {m['slug']} → {same_org[0]['slug']} (known to {sorted(external)})")509 if apply:510 await merge_entities(conn, m["id"], same_org[0]["id"], mode="merge", actor="canonicalize",511 note="official checkpoint repo of the same release (dtype tag in the repo name)", payload={"step": "artifacts"})512 else:513 rep.bump("_official_tagged_repo_is_the_model")514 continue515 cid = same_org[0]["id"] if len(same_org) == 1 else None516 if attr_quantized or a.is_quantized or a_name.is_quantized or quant_attr in QUANT_ATTR_FORMATS:517 kind = "quantization"518 elif a.is_conversion or a_name.is_conversion or a.precision or quant_attr:519 kind = "conversion"520 else:521 kind = "packaging"522 if cid is None:523 # 1) a model with the same variant_key (the name says what was quantised); 2) the hub's `quantized_from` object, but only when it524 # names the same model — a fine-tune's GGUF (Hermes-2-Pro-Mistral-7B-GGUF) is not an artifact of the foundation model (Mistral 7B)525 candidates = vk_candidates526 if len(candidates) > 1:527 same_org = [c for c in candidates if c["organization_id"] == m["organization_id"]]528 official_c = [c for c in candidates if (c["org_slug"] or "").lower() in set(official_orgs(probe)) or (c["org_hf"] or "").lower() in set(official_orgs(probe))]529 candidates = same_org or official_c or candidates530 if len(candidates) == 1:531 cid = candidates[0]["id"]532 if cid is None:533 rel = base_of.get(m["id"])534 if rel and rel in by_id and rel != m["id"]:535 base_probe = (by_id[rel]["attributes"] or {}).get("hf_repo") or by_id[rel]["canonical_name"]536 if variant_key(base_probe) == variant_key(probe) or analyze_model_name(base_probe).base_key in a.base_key:537 cid = rel538 if m["entity_type"] == "artifact" and m["canonical_id"] == cid:539 continue540 rep.bump("artifacts_marked" if cid else "artifacts_unresolved")541 rep.example(f"artifact[{kind}] {m['slug']} → {by_id[cid]['slug'] if cid else 'UNRESOLVED'}")542 if apply:543 await execute(conn, """update entities set entity_type = 'artifact', artifact_kind = coalesce(artifact_kind, :k), canonical_id = :c,544 identity_confidence = :ic, updated_at = now() where id = :id""",545 k=kind, c=cid, ic="high" if cid else "low", id=m["id"])546 if cid:547 await upsert_relation(conn, m["id"], "artifact_of", cid, {"artifact_kind": kind}, source_id=source_id)548 await record_decision(conn, m["id"], cid, "variant_of", actor="canonicalize", payload={"artifact_kind": kind, "step": "artifacts"})549 if not cid:550 await _review(conn, "unresolved_artifact", [m["id"]], f"'{m['canonical_name']}' is a {kind} artifact but its base model is unknown",551 {"slug": m["slug"], "hf_repo": hf_repo, "variant_key": variant_key(probe)}, apply=apply)552 # an organisation develops a model, not a quantisation/conversion of it: live `develops` edges into artifacts (bartowski, unsloth,553 # mlx-community… but also official orgs re-packaging their own weights) are closed and replaced by `published_by`554 scope_sql = "and a.id = any(cast(:ids as text[]))" if scope is not None else ""555 edges = await fetch_all(conn, f"""select r.id, r.subject_id as org_id, r.object_id as artifact_id, r.source_id, r.tier from relations r556 join entities a on a.id = r.object_id where r.predicate = 'develops' and r.valid_to is null and a.entity_type = 'artifact' {scope_sql}""",557 ids=sorted(scope or []))558 for e in edges:559 rep.bump("develops_to_artifact_closed")560 if apply:561 await execute(conn, "update relations set valid_to = now() where id = :id", id=e["id"])562 await upsert_relation(conn, e["artifact_id"], "published_by", e["org_id"], source_id=e["source_id"] or source_id, tier=e["tier"] or 2)563564565# ---------------------------------------------------------------------------------------------- step: families566async def step_families(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:567 source_id = await registry_source_id(conn)568 models = [m for m in await _models(conn) if _in_scope(scope, m["id"])]569 orgs = await fetch_all(conn, "select id, slug, canonical_name, attributes from entities where merged_into is null and entity_type = any(cast(:t as text[]))", t=sorted(ORG_TYPES))570 org_by_key: dict[str, str] = {}571 for o in orgs:572 for k in _org_lookup_keys(o):573 org_by_key.setdefault(k, o["id"])574 fam_rows = await fetch_all(conn, "select id, slug, canonical_name, organization_id from entities where entity_type = 'model_family' and merged_into is null")575 families: dict[str, dict[str, Any]] = {f["canonical_name"].lower(): f for f in fam_rows} # one family per label576 slugs_taken = {r["slug"] for r in await fetch_all(conn, "select slug from entities")}577578 def family_slug(label: str) -> str:579 """slugify(label); on collision with any other entity (the model `gpt-5.5` itself) → `<slug>-family`, then numbered."""580 base = slugify(label)581 for c in [base, f"{base}-family", *(f"{base}-family-{n}" for n in range(2, 20))]:582 if c not in slugs_taken:583 return c584 return f"{base}-family-{new_id('model_family')[-6:].lower()}"585586 # re-slug families created under the former rule (organisation prefix on collision: `qwen-qwen3`, `mistral-mistral`)587 for fam in fam_rows:588 base = slugify(fam["canonical_name"])589 if fam["slug"] in (base, f"{base}-family") or not fam["slug"].endswith(base):590 continue591 slugs_taken.discard(fam["slug"])592 new_slug = family_slug(fam["canonical_name"])593 slugs_taken.add(new_slug)594 rep.bump("families_reslugged")595 rep.example(f"family slug {fam['slug']} → {new_slug}")596 if apply:597 await execute(conn, "update entities set slug = :s, updated_at = now() where id = :id", s=new_slug, id=fam["id"])598 await execute(conn, "insert into entity_aliases (entity_id, alias, alias_norm, kind) values (:e, :a, :n, 'former_name') on conflict (entity_id, alias_norm) do nothing",599 e=fam["id"], a=fam["slug"], n=normalize_alias(fam["slug"]))600 fam["slug"] = new_slug601 for m in models:602 label = family_release_hint(m["canonical_name"])603 if not label:604 rep.bump("_no_family_hint")605 continue606 root = family_hint(m["canonical_name"]) or label607 official = official_orgs(m["canonical_name"])608 org_id = next((org_by_key[k] for k in official if k in org_by_key), None) or m["organization_id"]609 fam = families.get(label.lower())610 if fam is None:611 slug = family_slug(label)612 rep.bump("families_created")613 rep.example(f"family '{label}' ({slug}) root={root}")614 fam = {"id": new_id("model_family"), "slug": slug, "canonical_name": label, "organization_id": org_id, "_new": True}615 families[label.lower()] = fam616 slugs_taken.add(slug)617 if apply:618 await execute(conn, """insert into entities (id, entity_type, canonical_name, slug, status, organization_id, attributes, provenance, first_seen_at, last_seen_at, identity_confidence)619 values (:id, 'model_family', :n, :slug, 'active', :org, cast(:a as jsonb), cast(:p as jsonb), :fs, now(), 'high')""",620 id=fam["id"], n=label, slug=slug, org=org_id, a=jsonb({"family_root": root, "label": label}),621 p=jsonb({"family_root": {"source_id": source_id, "tier": 2, "extractor": "derived"}}), fs=m["first_seen_at"])622 await execute(conn, "insert into entity_aliases (entity_id, alias, alias_norm, kind) values (:e, :a, :n, 'alias') on conflict do nothing",623 e=fam["id"], a=label, n=normalize_alias(label))624 if m["family_id"] != fam["id"]:625 rep.bump("models_linked")626 if apply:627 await execute(conn, "update entities set family_id = :f, first_seen_at = first_seen_at where id = :id", f=fam["id"], id=m["id"])628 await execute(conn, "update entities set first_seen_at = least(first_seen_at, :fs) where id = :f", fs=m["first_seen_at"], f=fam["id"])629 await upsert_relation(conn, m["id"], "member_of_family", fam["id"], source_id=source_id)630 if not (m["attributes"] or {}).get("family"):631 rep.bump("family_attribute_set")632 if apply:633 await _set_attributes(conn, m["id"], {"family": label}, source_id=source_id)634635636# ---------------------------------------------------------------------------------------------- step: licenses637async def step_licenses(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:638 source_id = await registry_source_id(conn)639 rows = await fetch_all(conn, """select id, entity_type, canonical_name, slug, attributes from entities where merged_into is null640 and entity_type = any(cast(:t as text[])) and (attributes ? 'license' or attributes ? 'openness' or attributes ? 'hf_repo')641 order by first_seen_at, id""", t=list(LICENSED_TYPES))642 if scope is not None:643 rows = [r for r in rows if r["id"] in scope]644 lic_rows = await fetch_all(conn, "select id, slug from entities where entity_type = 'license'")645 license_entities: dict[str, str] = {r["slug"]: r["id"] for r in lic_rows}646 existing_rel = {(r["subject_id"], r["object_id"]) for r in await fetch_all(conn, "select subject_id, object_id from relations where predicate = 'uses_license' and valid_to is null")}647 # derived claims that a higher-tier source already contradicts: stored once as `conflicting`, never re-proposed648 contested: dict[tuple[str, str], Any] = {(r["entity_id"], r["property"]): r["value"] for r in await fetch_all(649 conn, "select entity_id, property, value from claims where extractor = 'derived' and status = 'conflicting'")}650 writer = _derived_writer(conn, source_id) if apply else None651 for r in rows:652 attrs = r["attributes"] or {}653 raw = attrs.get("license")654 key = normalize_license(raw) if isinstance(raw, str) else None655 if key is None and isinstance(attrs.get("license_key"), str) and attrs["license_key"] in LICENSES:656 key = attrs["license_key"]657 if isinstance(raw, str) and key is None:658 rep.bump("_license_unclassified")659 rep.example(f"unclassified licence {raw!r} on {r['slug']}") if len(rep.examples) < 4 else None660 if key:661 lslug = key.lower()662 lid = license_entities.get(lslug)663 if lid is None:664 info = LICENSES[key]665 lid = new_id("license")666 license_entities[lslug] = lid667 rep.bump("license_entities_created")668 if apply:669 await execute(conn, """insert into entities (id, entity_type, canonical_name, slug, status, description, attributes, provenance, identity_confidence)670 values (:id, 'license', :n, :slug, 'active', :d, cast(:a as jsonb), cast(:p as jsonb), 'high')""",671 id=lid, n=info.label, slug=lslug, d=f"{info.label} — {info.category} licence" + (f" (SPDX {info.spdx})" if info.spdx else ""),672 a=jsonb({**info.as_dict(), "license_key": key}), p=jsonb({"license_key": {"source_id": source_id, "tier": 2, "extractor": "derived"}}))673 await execute(conn, "insert into entity_aliases (entity_id, alias, alias_norm, kind) values (:e, :a, :n, 'alias') on conflict do nothing",674 e=lid, a=info.label, n=normalize_alias(info.label))675 for alias in (key, info.spdx or key):676 await execute(conn, "insert into entity_aliases (entity_id, alias, alias_norm, kind) values (:e, :a, :n, 'alias') on conflict do nothing",677 e=lid, a=alias, n=normalize_alias(alias))678 if (r["id"], lid) not in existing_rel:679 rep.bump("uses_license_relations")680 existing_rel.add((r["id"], lid))681 if apply:682 await upsert_relation(conn, r["id"], "uses_license", lid, source_id=source_id)683 if attrs.get("license_key") != key:684 rep.bump("license_key_attribute_set")685 if apply:686 await _set_attributes(conn, r["id"], {"license_key": key}, source_id=source_id)687 if r["entity_type"] not in ("model", "artifact"):688 continue689 # openness dimensions (derived claims, tier 2, never supersede a tier-1 statement)690 openness_raw = attrs.get("openness")691 openness_now = normalize_openness(openness_raw) if isinstance(openness_raw, str) else None692 weights: bool | None = None693 if attrs.get("hf_repo") or attrs.get("model_card_url") or attrs.get("weights_url") or (openness_now or "").startswith(("open", "restricted")):694 weights = True695 elif openness_now == "proprietary":696 weights = False697 if weights is None:698 rep.bump("_openness_unknown_skipped")699 continue700 dims = openness_dimensions(weights_available=weights, license_key=key)701 derived = derive_openness(dims, license_key=key)702 wanted: dict[str, Any] = {"weights_available": dims["weights_available"]}703 for k in ("commercial_use_allowed", "redistribution_allowed", "derivatives_allowed"):704 if dims[k] is not None:705 wanted[k] = dims[k]706 if derived != "unknown":707 wanted["openness"] = derived708 changed = {k: v for k, v in wanted.items() if not _same(attrs.get(k), v) and not _same(contested.get((r["id"], k)), v)}709 if not changed:710 continue711 rep.bump("openness_claims_written", len(changed))712 if "openness" in changed:713 rep.bump("openness_category_changed")714 rep.example(f"{r['slug']}: openness {openness_raw!r} → {derived} (licence {key})")715 if writer is not None:716 facts = Facts()717 ref = EntityRef(entity_type=r["entity_type"], name=r["canonical_name"], id=r["id"])718 for k, v in changed.items():719 facts.claim(ref, k, v, confidence="high")720 await writer.write(facts)721 if writer is not None:722 rep.bump("_conflicting_derived_claims", writer.stats.conflicts)723724725# ---------------------------------------------------------------------------------------------- step: taxonomy726async def step_taxonomy(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:727 props = sorted(TAXONOMY_PROPERTIES)728 rows = await fetch_all(conn, """select c.id, c.entity_id, c.property, c.value, c.value_raw, e.entity_type from claims c join entities e on e.id = c.entity_id729 where c.status = 'current' and c.property = any(cast(:p as text[]))""", p=props)730 if scope is not None:731 rows = [r for r in rows if r["entity_id"] in scope]732 mappings: dict[tuple[str, str], str | None] = {}733734 def note(maps: list[tuple[str, str, str | None]]) -> None:735 for d, rw, cn in maps:736 if cn is not None and rw == cn:737 continue # identity mapping: nothing to learn738 mappings[(d, rw)] = cn if cn is not None else mappings.get((d, rw))739740 for r in rows:741 canon, raw, maps = normalize_property(r["entity_type"], r["property"], r["value"])742 note(maps)743 if _same(canon, r["value"]):744 continue745 rep.bump(f"claims_normalized:{r['property']}")746 rep.example(f"{r['entity_type']} {r['property']}: {json.dumps(r['value'], ensure_ascii=False)[:40]} → {json.dumps(canon, ensure_ascii=False)[:40]}")747 if apply:748 keep_raw = r["value_raw"] or (r["value"] if isinstance(r["value"], str) else json.dumps(r["value"], ensure_ascii=False))749 keep_raw = keep_raw if keep_raw != canon else None750 await execute(conn, "update claims set value = cast(:v as jsonb), value_text = :vt, value_raw = :raw where id = :id",751 v=jsonb(canon), vt=canon[:2000] if isinstance(canon, str) else None, raw=keep_raw, id=r["id"])752 attrs = {r["property"]: canon}753 if keep_raw:754 attrs[f"{r['property']}_raw"] = keep_raw755 await execute(conn, "update entities set attributes = attributes || cast(:a as jsonb), updated_at = now() where id = :id", a=jsonb(attrs), id=r["entity_id"])756 if r["property"] == "status" and isinstance(canon, str):757 await execute(conn, "update entities set status = :s where id = :id", s=canon[:40], id=r["entity_id"])758 # attributes without a current claim (seeded/imported values)759 for prop in props:760 ents = await fetch_all(conn, """select e.id, e.entity_type, e.attributes->:p as value from entities e where e.merged_into is null and e.attributes ? :p761 and not exists (select 1 from claims c where c.entity_id = e.id and c.property = :p and c.status = 'current')""", p=prop)762 for e in ents:763 if not _in_scope(scope, e["id"]):764 continue765 canon, raw, maps = normalize_property(e["entity_type"], prop, e["value"])766 note(maps)767 if _same(canon, e["value"]):768 continue769 rep.bump(f"attributes_normalized:{prop}")770 if apply:771 keep_raw = e["value"] if isinstance(e["value"], str) else json.dumps(e["value"], ensure_ascii=False)772 attrs = {prop: canon, f"{prop}_raw": keep_raw}773 await execute(conn, "update entities set attributes = attributes || cast(:a as jsonb), updated_at = now() where id = :id", a=jsonb(attrs), id=e["id"])774 if prop == "status" and isinstance(canon, str):775 await execute(conn, "update entities set status = :s where id = :id", s=canon[:40], id=e["id"])776 existing = {(m["domain"], m["raw"]): m["canonical"] for m in await fetch_all(conn, "select domain, raw, canonical from taxonomy_mappings")}777 for (domain, raw_full), canon in mappings.items():778 raw = raw_full[:300]779 if (domain, raw) in existing and (existing[(domain, raw)] == canon or canon is None):780 continue781 rep.bump("taxonomy_mappings_upserted")782 if apply:783 await execute(conn, """insert into taxonomy_mappings (domain, raw, canonical) values (:d, :r, :c)784 on conflict (domain, raw) do update set canonical = coalesce(excluded.canonical, taxonomy_mappings.canonical), last_seen_at = now()""",785 d=domain, r=raw, c=canon)786 unknown = sorted({f"{d}:{r}" for (d, r), c in mappings.items() if c is None})787 if unknown:788 rep.notes.append(f"{len(unknown)} raw value(s) without canonical mapping kept as-is: {', '.join(unknown[:15])}{'…' if len(unknown) > 15 else ''}")789790791# ---------------------------------------------------------------------------------------------- step: results792async def _rehome_results(conn: AsyncConnection, rep: StepReport, apply: bool, scope: set[str] | None) -> None:793 """Move results from a benchmark *family head* to the registry's variant entity when the row says which variant it measured:794 (a) config.variant / config.board equals (case-insensitively) the variant entity's name, an alias or its `variant` attribute;795 (b) LiveBench `category:<Name>` metrics → `livebench-<slug(Name)>` with the registry metric (`average score`), variant = Name;796 (c) aider `percent_cases_well_formed` → `aider-polyglot-well-formed`.797 Then dedupe/config keys, one-current-row and `evaluated_on` relations are recomputed for every touched model."""798 variants = await fetch_all(conn, """select v.id, v.slug, v.canonical_name, v.attributes, r.object_id as head_id,799 (select array_agg(alias) from entity_aliases a where a.entity_id = v.id) as aliases800 from entities v join relations r on r.subject_id = v.id and r.predicate = 'variant_of' and r.valid_to is null801 where v.entity_type = 'benchmark' and v.merged_into is null""")802 if not variants:803 return804 by_slug: dict[str, dict[str, Any]] = {v["slug"]: v for v in variants}805 by_head: dict[str, list[tuple[set[str], dict[str, Any]]]] = defaultdict(list)806 for v in variants:807 keys = {v["canonical_name"].lower(), v["slug"], *(a.lower() for a in (v["aliases"] or []))}808 var_attr = (v["attributes"] or {}).get("variant")809 if isinstance(var_attr, str):810 keys.add(var_attr.lower())811 by_head[v["head_id"]].append((keys, v))812 heads = list(by_head)813 rows = await fetch_all(conn, """select r.id, r.model_id, r.benchmark_id, r.metric, r.config from benchmark_results r814 where r.benchmark_id = any(cast(:h as text[]))""", h=heads)815 moves: list[tuple[dict[str, Any], dict[str, Any], str | None, dict[str, Any]]] = []816 for r in rows:817 if scope is not None and r["model_id"] not in scope:818 continue819 cfg = dict(r["config"] or {})820 metric = r["metric"] or ""821 target: dict[str, Any] | None = None822 new_metric: str | None = None823 label = next((str(cfg[k]) for k in ("variant", "board") if isinstance(cfg.get(k), str) and cfg[k].strip()), None)824 if label:825 target = next((v for keys, v in by_head[r["benchmark_id"]] if label.lower() in keys), None)826 if target is None and metric.lower().startswith("category:"):827 name = metric.split(":", 1)[1].strip()828 cand = by_slug.get(f"livebench-{slugify(name)}")829 if cand and cand["head_id"] == r["benchmark_id"]:830 target = cand831 new_metric = (cand["attributes"] or {}).get("metric") or "average score"832 cfg["variant"] = name833 if target is None and metric == "percent_cases_well_formed":834 cand = by_slug.get("aider-polyglot-well-formed")835 if cand and cand["head_id"] == r["benchmark_id"]:836 target = cand837 if target is None:838 continue839 moves.append((r, target, new_metric, cfg))840 if not moves:841 return842 touched: set[tuple[str, str, str]] = set()843 for r, target, new_metric, cfg in moves:844 rep.bump(f"results_rehomed:{target['slug']}")845 touched.add((r["model_id"], r["benchmark_id"], target["id"]))846 if apply:847 await execute(conn, "update benchmark_results set benchmark_id = :b, metric = coalesce(:m, metric), config = cast(:c as jsonb), variant = :v where id = :id",848 b=target["id"], m=new_metric, c=jsonb(cfg), v=bench_ontology.variant_from_config(cfg), id=r["id"])849 if not apply:850 return851 source_id = await registry_source_id(conn)852 for model_id in sorted({m for m, _, _ in touched}):853 from aiatlas.services.merge import recompute_result_keys854855 await recompute_result_keys(conn, model_id)856 await enforce_current_results(conn, model_id=model_id)857 for model_id, head_id, target_id in sorted(touched):858 await upsert_relation(conn, model_id, "evaluated_on", target_id, source_id=source_id)859 left = await fetch_one(conn, "select 1 from benchmark_results where model_id = :m and benchmark_id = :b and valid_to is null limit 1", m=model_id, b=head_id)860 if not left:861 await execute(conn, "update relations set valid_to = now() where subject_id = :m and predicate = 'evaluated_on' and object_id = :b and valid_to is null",862 m=model_id, b=head_id)863 rep.bump("evaluated_on_repointed")864865866async def step_results(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:867 await _rehome_results(conn, rep, apply, scope)868 rows = await fetch_all(conn, """select r.id, r.model_id, r.config, r.metric, r.unit, r.score, r.config_key, r.trust_level, r.variant, r.run_group, r.extractor, r.valid_to,869 r.is_current, s.key as source_key from benchmark_results r left join sources s on s.id = r.source_id""")870 if scope is not None:871 rows = [r for r in rows if r["model_id"] in scope]872 for r in rows:873 cfg = r["config"] or {}874 ck = bench_ontology.config_key(cfg, r["metric"])875 trust = r["trust_level"] or bench_ontology.trust_level(r["source_key"], cfg, extractor=r["extractor"] or "deterministic")876 variant = r["variant"] or bench_ontology.variant_from_config(cfg)877 rg = r["run_group"] or bench_ontology.run_group_from_config(cfg)878 current = r["is_current"] and r["valid_to"] is None879 if (ck, trust, variant, rg, current) == (r["config_key"], r["trust_level"], r["variant"], r["run_group"], r["is_current"]):880 continue881 rep.bump("results_backfilled")882 if apply:883 await execute(conn, "update benchmark_results set config_key = :ck, trust_level = :t, variant = :v, run_group = :rg, is_current = :cur where id = :id",884 ck=ck, t=trust, v=variant, rg=rg, cur=current, id=r["id"])885 if scope is None:886 closed = await enforce_current_results(conn, dry_run=not apply)887 else:888 closed = sum([await enforce_current_results(conn, model_id=mid, dry_run=not apply) for mid in sorted(scope)])889 if closed:890 rep.bump("older_run_rows_closed", closed)891 live = await fetch_one(conn, "select count(*) as n from benchmark_results where valid_to is null and is_current")892 rep.notes.append(f"current benchmark results after step: {int(live['n']) if live else 0}")893894895# ---------------------------------------------------------------------------------------------- step: events896async def step_events(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:897 lag = timedelta(days=BACKFILL_LAG_DAYS)898 runs = await fetch_all(conn, """select connector_name, started_at, finished_at from connector_runs where status in ('success', 'unchanged', 'suspect', 'released')899 order by connector_name, started_at""")900 second_start: dict[str, datetime | None] = {}901 per: dict[str, list[dict[str, Any]]] = defaultdict(list)902 for r in runs:903 per[r["connector_name"]].append(r)904 for name, lst in per.items():905 second_start[name] = lst[1]["started_at"] if len(lst) > 1 else None906 before = await fetch_one(conn, """select count(*) as n from change_events where is_backfill = false and event_type <> 'DOCUMENT_CHANGED'907 and observed_at > now() - interval '24 hours'""")908 rows = await fetch_all(conn, """select ev.id, ev.event_type, ev.connector_name, ev.observed_at, ev.effective_at, ev.is_backfill, ev.group_key, ev.entity_id,909 e.attributes->>'release_date' as release_date, e.attributes->>'published_at' as published_at910 from change_events ev left join entities e on e.id = ev.entity_id""")911 from aiatlas.sdk.extract.dates import parse_datetime912913 if scope is not None:914 rows = [r for r in rows if r["entity_id"] in scope]915 for ev in rows:916 obs = ev["observed_at"]917 bf = False918 if ev["connector_name"] in STRUCTURAL_CONNECTORS:919 bf = True # merges, folds, derived corrections: bookkeeping about our own data, never news920 elif ev["connector_name"] and ev["connector_name"] in per:921 s2 = second_start.get(ev["connector_name"])922 if s2 is None or obs < s2:923 bf = True924 elif ev["connector_name"] and ev["connector_name"] not in per:925 bf = True # connector without any successful run yet → initial load926 if not bf and ev["effective_at"] is not None and ev["effective_at"] < obs - lag:927 bf = True928 if not bf and ev["event_type"].startswith("NEW_"):929 hint = parse_datetime(ev["release_date"]) if ev["release_date"] else (parse_datetime(ev["published_at"]) if ev["published_at"] else None)930 if hint is not None:931 hint = hint if hint.tzinfo else hint.replace(tzinfo=UTC)932 if hint < obs - lag:933 bf = True934 gk = group_key_for(ev["event_type"], ev["entity_id"], ev["effective_at"], obs)935 if bf == ev["is_backfill"] and gk == ev["group_key"]:936 continue937 if bf != ev["is_backfill"]:938 rep.bump("backfill_flag_set" if bf else "backfill_flag_cleared")939 if gk != ev["group_key"]:940 rep.bump("group_key_set")941 if apply:942 await execute(conn, "update change_events set is_backfill = :bf, group_key = :gk where id = :id", bf=bf, gk=gk, id=ev["id"])943 after = await fetch_one(conn, """select count(*) as n from change_events where is_backfill = false and event_type <> 'DOCUMENT_CHANGED'944 and observed_at > now() - interval '24 hours'""")945 rep.notes.append(f"live (non-backfill) events in the last 24 h: before {int(before['n']) if before else 0} → after {int(after['n']) if after else 0}"946 + ("" if apply else " (dry-run: after = before)"))947948949# ---------------------------------------------------------------------------------------------- step: anomalies950async def _collect_anomalies(conn: AsyncConnection) -> list[Anomaly]:951 found: list[Anomaly] = []952 for r in await fetch_all(conn, "select id, canonical_name, attributes from entities where entity_type in ('model','artifact') and merged_into is null"):953 found.extend(check_model(r["id"], r["canonical_name"], r["attributes"] or {}))954 for r in await fetch_all(conn, "select id, canonical_name, attributes from entities where entity_type = 'hardware' and merged_into is null"):955 found.extend(check_hardware(r["id"], r["canonical_name"], r["attributes"] or {}))956 for r in await fetch_all(conn, """select p.*, m.canonical_name as model_name, v.canonical_name as provider_name from prices p957 join entities m on m.id = p.model_id join entities v on v.id = p.provider_id where p.valid_to is null"""):958 found.extend(check_price(r))959 for r in await fetch_all(conn, """select r.id, r.model_id, r.benchmark_id, r.score, r.metric, r.unit, r.evaluated_at, m.canonical_name as model_name,960 b.canonical_name as benchmark_name, m.attributes->>'release_date' as model_release_date961 from benchmark_results r join entities m on m.id = r.model_id join entities b on b.id = r.benchmark_id962 where r.valid_to is null and r.is_current"""):963 found.extend(check_result(r))964 return found965966967async def step_anomalies(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:968 found = await _collect_anomalies(conn)969 if scope is not None:970 found = [a for a in found if a.entity_id in scope]971 keys: dict[str, Anomaly] = {}972 for a in found:973 keys.setdefault(a.dedupe_key[:400], a)974 existing = {r["dedupe_key"]: r for r in await fetch_all(conn, "select dedupe_key, status, severity from anomalies")}975 for key, a in keys.items():976 rep.bump(f"_by_severity:{a.severity}")977 prev = existing.get(key)978 if prev is None:979 rep.bump("anomalies_opened")980 rep.example(f"[{a.severity}] {a.message}")981 elif prev["status"] == "resolved":982 rep.bump("anomalies_reopened")983 if apply:984 await record(conn, a)985 stale = [k for k, r in existing.items() if r["status"] == "open" and k not in keys] if scope is None else []986 if stale:987 rep.bump("anomalies_resolved", len(stale))988 if apply:989 await execute(conn, """update anomalies set status = 'resolved', resolved_at = now(), resolution = 'check no longer fires'990 where status = 'open' and dedupe_key = any(cast(:k as text[]))""", k=stale)991992993_STEP_FUNCTIONS = {994 "duplicates": step_duplicates, "variants": step_variants, "artifacts": step_artifacts, "families": step_families, "licenses": step_licenses,995 "taxonomy": step_taxonomy, "results": step_results, "events": step_events, "anomalies": step_anomalies,996}997998999# ---------------------------------------------------------------------------------------------- quarantine release / discard1000async def release_quarantine(conn: AsyncConnection, quarantine_id: str, *, actor: str = "admin") -> dict[str, Any]:1001 """Write the held facts of a quarantined run exactly as the connector would have (same source, snapshot, tier, run id)."""1002 from aiatlas.connectors import get as get_connector10031004 q = await fetch_one(conn, "select * from quarantined_runs where id = :id", id=quarantine_id)1005 if not q:1006 raise LookupError(f"quarantined run {quarantine_id} not found")1007 if q["status"] != "pending":1008 raise ValueError(f"quarantined run {quarantine_id} is already {q['status']}")1009 connector = get_connector(q["connector_name"])1010 state = await fetch_one(conn, """select c.source_id, s.key as source_key from connectors c left join sources s on s.id = c.source_id where c.name = :n""", n=q["connector_name"])1011 totals: dict[str, int] = defaultdict(int)1012 for item in q["facts"] or []:1013 facts = facts_from_json(item["facts"])1014 fetched_at = datetime.fromisoformat(item["fetched_at"])1015 writer = FactWriter(conn, source_id=state["source_id"] if state else None, snapshot_id=item.get("snapshot_id"), source_url=item.get("source_url"),1016 tier=connector.tier, connector_name=q["connector_name"], extractor="deterministic", extractor_version=connector.parser_version,1017 observed_at=fetched_at, run_id=q["run_id"], source_key=state["source_key"] if state else None)1018 ws = await writer.write(facts)1019 for k, v in ws.as_dict().items():1020 totals[k] += v1021 main = facts.document_entity1022 if main and main.id is None:1023 await writer.resolver.resolve(main)1024 if main and main.id and item.get("doc_id"):1025 await execute(conn, "update documents set entity_id = coalesce(entity_id, :e), title = coalesce(:t, title) where id = :id", e=main.id, t=facts.document_title, id=item["doc_id"])1026 if item.get("snapshot_id"):1027 await execute(conn, "update snapshots set processing_status = 'extracted' where id = :id", id=item["snapshot_id"])1028 await execute(conn, "update quarantined_runs set status = 'released', resolved_at = now(), resolved_by = :who where id = :id", who=actor, id=quarantine_id)1029 await execute(conn, "update connector_runs set status = 'released' where id = :r", r=q["run_id"])1030 await execute(conn, "update review_queue set status = 'approved', resolved_at = now() where dedupe_key = :d", d=f"quarantine:{quarantine_id}")1031 observed = (q["stats"] or {}).get("observed")1032 if observed and (q["stats"] or {}).get("full_extraction"):1033 baseline = (await fetch_one(conn, "select baseline from connectors where name = :n", n=q["connector_name"]) or {}).get("baseline")1034 nb = connector._next_baseline(baseline, observed) # noqa: SLF0011035 await execute(conn, "update connectors set baseline = cast(:b as jsonb) where name = :n", b=jsonb(nb), n=q["connector_name"])1036 await audit(conn, "quarantine.release", quarantine_id, {"connector": q["connector_name"], "run_id": q["run_id"], **totals}, actor=actor)1037 return {"id": quarantine_id, "connector": q["connector_name"], **totals}103810391040async def discard_quarantine(conn: AsyncConnection, quarantine_id: str, *, actor: str = "admin", note: str | None = None) -> dict[str, Any]:1041 q = await fetch_one(conn, "select id, run_id, connector_name, status, facts from quarantined_runs where id = :id", id=quarantine_id)1042 if not q:1043 raise LookupError(f"quarantined run {quarantine_id} not found")1044 if q["status"] != "pending":1045 raise ValueError(f"quarantined run {quarantine_id} is already {q['status']}")1046 await execute(conn, "update quarantined_runs set status = 'discarded', resolved_at = now(), resolved_by = :who where id = :id", who=actor, id=quarantine_id)1047 await execute(conn, "update connector_runs set status = 'discarded' where id = :r", r=q["run_id"])1048 await execute(conn, "update review_queue set status = 'rejected', resolved_at = now() where dedupe_key = :d", d=f"quarantine:{quarantine_id}")1049 for item in q["facts"] or []:1050 if item.get("snapshot_id"):1051 await execute(conn, "update snapshots set processing_status = 'discarded' where id = :id", id=item["snapshot_id"])1052 await audit(conn, "quarantine.discard", quarantine_id, {"connector": q["connector_name"], "run_id": q["run_id"], "note": note}, actor=actor)1053 return {"id": quarantine_id, "connector": q["connector_name"], "status": "discarded"}105410551056async def list_quarantine(conn: AsyncConnection, *, status: str = "pending", limit: int = 50) -> list[dict[str, Any]]:1057 return await fetch_all(conn, """select id, run_id, connector_name, reason, stats, status, created_at, resolved_at, resolved_by, jsonb_array_length(facts) as documents1058 from quarantined_runs where (cast(:s as text) = '' or status = :s) order by created_at desc limit :n""", s=status or "", n=limit)105910601061__all__ = ["CANON_VERSION", "STEPS", "Report", "StepReport", "canonicalize", "discard_quarantine", "list_quarantine", "release_quarantine"]1062