"""Canonicalization engine — `aia canonicalize [--apply] [--step …]`. Turns the flat "everything is a model" corpus into the canonical hierarchy (model_family → model → artifact, effort variants folded into result configurations), normalises taxonomies, links licences, classifies events (backfill vs live), enforces benchmark result comparability and flags anomalies. Rules are documented in docs/CANONICALIZATION.md. Invariants * dry-run by default: every step computes a plan from reads only and reports counts + examples; `--apply` executes it * nothing is ever deleted; raw snapshots are never touched; every write is idempotent (a second `--apply` is a no-op) * merges go through `services.merge.merge_entities` (persisted decision + audit log); relations through `upsert_relation` * derived claims are written by a tier-2 `derived` FactWriter (source `ai-atlas.registry`) and never supersede tier-1 statements """ from __future__ import annotations import json import logging import re from collections import defaultdict from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy.ext.asyncio import AsyncConnection from aiatlas.db import execute, fetch_all, fetch_one, jsonb, transaction from aiatlas.ids import new_id, normalize_alias, slugify from aiatlas.ontology import benchmarks as bench_ontology from aiatlas.ontology.anomalies import Anomaly, check_hardware, check_model, check_price, check_result from aiatlas.ontology.licenses import LICENSES, normalize_license from aiatlas.ontology.models import ( CONVERTER_ORGS, analyze_model_name, base_name, effort_config, family_hint, family_release_hint, official_orgs, variant_key, ) from aiatlas.ontology.openness import derive_openness, normalize_openness, openness_dimensions from aiatlas.ontology.taxonomy import ORG_TYPE_DEFAULT_KIND, TAXONOMY_PROPERTIES, normalize_property from aiatlas.sdk.facts import EntityRef, Facts, facts_from_json from aiatlas.sdk.resolution import EVALUATOR_SCHEMES, Resolver from aiatlas.sdk.writer import FactWriter, _same from aiatlas.services.anomalies import record from aiatlas.services.events import BACKFILL_LAG_DAYS, group_key_for from aiatlas.services.merge import ( ORG_TYPES, audit, enforce_current_results, kept_separate, merge_entities, record_decision, registry_source_id, upsert_relation, ) log = logging.getLogger(__name__) CANON_VERSION = "2026.09" STRUCTURAL_CONNECTORS = {"curation", "canonicalize"} STEPS = ("duplicates", "variants", "artifacts", "families", "licenses", "taxonomy", "results", "events", "anomalies") DEDUPE_TYPES = ("model", "provider", "benchmark", "dataset", "framework", "library", "hardware") LICENSED_TYPES = ("model", "artifact", "dataset", "framework", "library", "repository") QUANT_ATTR_FORMATS = {"gguf", "awq", "gptq", "exl2", "exl3", "int4", "int8", "fp8", "nvfp4", "mxfp4", "fp4", "bnb", "quantized"} @dataclass class StepReport: name: str counts: dict[str, int] = field(default_factory=dict) examples: list[str] = field(default_factory=list) notes: list[str] = field(default_factory=list) def bump(self, key: str, n: int = 1) -> None: self.counts[key] = self.counts.get(key, 0) + n def example(self, text: str, *, limit: int = 12) -> None: if len(self.examples) < limit: self.examples.append(text) @property def changes(self) -> int: return sum(v for k, v in self.counts.items() if not k.startswith("_")) @dataclass class Report: apply: bool steps: list[StepReport] = field(default_factory=list) started_at: datetime = field(default_factory=lambda: datetime.now(UTC)) @property def changes(self) -> int: return sum(s.changes for s in self.steps) def render(self) -> str: mode = "APPLY" if self.apply else "DRY-RUN" lines = [f"aia canonicalize — {mode} — {self.started_at:%Y-%m-%d %H:%M:%S} UTC — {self.changes} change(s)"] for s in self.steps: lines.append(f"\n[{s.name}] {s.changes} change(s)") for k, v in sorted(s.counts.items()): lines.append(f" {k:<40} {v}") for n in s.notes: lines.append(f" · {n}") for e in s.examples: lines.append(f" - {e}") return "\n".join(lines) def as_dict(self) -> dict[str, Any]: return {"apply": self.apply, "started_at": self.started_at.isoformat(), "changes": self.changes, "steps": [{"name": s.name, "counts": s.counts, "examples": s.examples, "notes": s.notes} for s in self.steps]} async def canonicalize(*, apply: bool = False, steps: list[str] | None = None, scope: set[str] | None = None) -> Report: """Run the steps in canonical order. `scope` (entity ids) restricts the rows a step *acts on* — lookups (bases, families, licences) still see the whole corpus; used by tests and targeted re-runs.""" wanted = [s for s in STEPS if not steps or s in steps] unknown = set(steps or []) - set(STEPS) if unknown: raise ValueError(f"unknown step(s): {sorted(unknown)}; known: {STEPS}") report = Report(apply=apply) for name in wanted: rep = StepReport(name=name) fn = _STEP_FUNCTIONS[name] async with transaction() as conn: await fn(conn, rep, apply, scope=scope) if apply and rep.changes: await audit(conn, f"canonicalize.{name}", None, {"counts": rep.counts, "version": CANON_VERSION}, actor="canonicalize") report.steps.append(rep) log.info("canonicalize step done", extra={"step": name, "apply": apply, **{k: v for k, v in rep.counts.items()}}) return report # ---------------------------------------------------------------------------------------------- shared helpers def _derived_writer(conn: AsyncConnection, source_id: str | None) -> FactWriter: return FactWriter(conn, source_id=source_id, snapshot_id=None, source_url=None, tier=2, connector_name="canonicalize", extractor="derived", extractor_version=CANON_VERSION, run_id=f"canon_{datetime.now(UTC):%Y%m%d}") async def _review(conn: AsyncConnection, kind: str, entity_ids: list[str], reason: str, payload: dict[str, Any], *, apply: bool = True) -> bool: """Queue a review item once (dedupe key). Returns True when the item is new (or would be, in dry-run) so reports stay idempotent.""" dedupe = f"{kind}:{':'.join(entity_ids)}:{normalize_alias(reason)[:80]}" if await fetch_one(conn, "select 1 from review_queue where dedupe_key = :d", d=dedupe): return False if not apply: return True 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) on conflict (dedupe_key) do nothing returning id""", id=new_id("review"), k=kind, ids=entity_ids, p=jsonb(payload), r=reason, d=dedupe) return row is not None async def _set_attributes(conn: AsyncConnection, entity_id: str, attrs: dict[str, Any], *, source_id: str | None) -> None: """Direct attribute write for structural/derived facts (family label, license_key, org_kind default) with `derived` provenance.""" prov = {k: {"source_id": source_id, "tier": 2, "confidence": "high", "extractor": "derived", "observed_at": datetime.now(UTC).isoformat(timespec="seconds")} for k in attrs} await execute(conn, "update entities set attributes = attributes || cast(:a as jsonb), provenance = provenance || cast(:p as jsonb), updated_at = now() where id = :id", a=jsonb(attrs), p=jsonb(prov), id=entity_id) async def _claim_counts(conn: AsyncConnection, ids: list[str]) -> dict[str, int]: if not ids: return {} 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) return {r["entity_id"]: int(r["n"]) for r in rows} async def _identifiers(conn: AsyncConnection, ids: list[str]) -> dict[str, dict[str, set[str]]]: out: dict[str, dict[str, set[str]]] = defaultdict(lambda: defaultdict(set)) if not ids: return out rows = await fetch_all(conn, "select entity_id, scheme, value from entity_identifiers where entity_id = any(cast(:ids as text[]))", ids=ids) for r in rows: out[r["entity_id"]][r["scheme"]].add(r["value"]) return out _SNAPSHOT_REMAINDER = re.compile(r"^(\d{1,8}|preview\d*|exp\d*|latest|beta|alpha|v\d+)$") _EFFORT_REMAINDER = {"reasoning", "nonreasoning", "thinking", "nonthinking", "high", "low", "medium", "xhigh", "minimal", "instruct", "it", "chat"} _VENDOR_PREFIXES = {"meta", "metallama", "google", "openai", "alibaba", "nvidia", "microsoft", "anthropic", "qwen", "deepseek", "mistral", "mistralai", "zai", "moonshotai"} _VERSION_TOKEN = re.compile(r"^v?\d+(\.\d+)*$") def _identifier_alias(a: str, b: str) -> bool: """Two values of one scheme name the same thing when one is a dated/preview snapshot, an effort/instruct suffix, a vendor-prefixed 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` ~ `google/gemini-2.5-flash-lite-preview`, `meta-llama/Meta-Llama-3.1-8B-Instruct` ~ `meta-llama/Llama-3.1-8B-Instruct`, `gemini-omni-flash` ~ `gemini-omni-1.1-flash`).""" na, nb = normalize_alias(a), normalize_alias(b) if na == nb: return True short, long_ = sorted((na, nb), key=len) if long_.startswith(short): rest = long_[len(short):] if _SNAPSHOT_REMAINDER.match(rest) or rest in _EFFORT_REMAINDER or rest.startswith("preview") or rest.startswith("exp"): return True if long_.endswith(short) and long_[: -len(short)] in _VENDOR_PREFIXES: return True # re-versioned / vendor-duplicated spellings: same word tokens once version and vendor tokens are removed, and the version tokens of # one side are a subset of the other's (`gemini-omni-flash` ⊂ `gemini-omni-1.1-flash`; `gpt-4` vs `gpt-5` stay different) ta = [t for t in re.split(r"[-/_.\s]+", a.lower()) if t] tb = [t for t in re.split(r"[-/_.\s]+", b.lower()) if t] words_a = [t for t in ta if not _VERSION_TOKEN.match(t) and t not in _VENDOR_PREFIXES] words_b = [t for t in tb if not _VERSION_TOKEN.match(t) and t not in _VENDOR_PREFIXES] va = {t for t in ta if _VERSION_TOKEN.match(t)} vb = {t for t in tb if _VERSION_TOKEN.match(t)} return bool(words_a) and words_a == words_b and (va <= vb or vb <= va) def _identifiers_conflict(a: dict[str, set[str]], b: dict[str, set[str]]) -> bool: """Hard contradiction only: a shared scheme whose values are not aliases of one another.""" for s in set(a) & set(b): if not a[s] or not b[s] or a[s] == b[s]: continue if not all(_identifier_alias(x, y) for x in a[s] for y in b[s]): return True return False def _conflict_detail(a: dict[str, set[str]], b: dict[str, set[str]]) -> str: parts = [] for s in sorted(set(a) & set(b)): 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]): parts.append(f"{s}: {'/'.join(sorted(a[s]))} vs {'/'.join(sorted(b[s]))}") return "; ".join(parts) def _same_publisher(model_name: str, org_a: dict[str, Any] | None, org_b: dict[str, Any] | None) -> bool: """Alibaba/Qwen, Meta/Meta AI…: both organisations publish the model's family officially.""" if not org_a or not org_b: return False official = set(official_orgs(model_name)) return bool(official) and bool(_org_lookup_keys(org_a) & official) and bool(_org_lookup_keys(org_b) & official) def _org_lookup_keys(row: dict[str, Any]) -> set[str]: attrs = row.get("attributes") or {} keys = {row["slug"], normalize_alias(row["canonical_name"])} for k in ("hf_org", "github_org"): v = attrs.get(k) if isinstance(v, str) and v.strip(): keys.add(v.strip().lower()) return {k for k in keys if k} # ---------------------------------------------------------------------------------------------- step: duplicates def _in_scope(scope: set[str] | None, entity_id: str) -> bool: return scope is None or entity_id in scope async def _retype_routers(conn: AsyncConnection, rep: StepReport, apply: bool, scope: set[str] | None) -> None: """OpenRouter's own routers (`openrouter/auto`, `openrouter/pareto-code`…) are products, not models: no weights, no developer.""" 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.id where e.entity_type = 'model' and e.merged_into is null and ((ei.scheme = 'openrouter' and ei.value like 'openrouter/%') or e.attributes->>'openrouter_id' like 'openrouter/%')""") for r in rows: if not _in_scope(scope, r["id"]): continue rep.bump("routers_retyped_as_product") rep.example(f"product[router] {r['slug']} ({r['canonical_name']})") if apply: await execute(conn, """update entities set entity_type = 'product', attributes = attributes || '{"kind": "router"}'::jsonb, updated_at = now() where id = :id""", id=r["id"]) 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"]) async def step_duplicates(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None: source_id = await registry_source_id(conn) await _retype_routers(conn, rep, apply, scope) rows = await fetch_all(conn, """select id, entity_type, canonical_name, slug, organization_id, attributes, first_seen_at from entities where merged_into is null and entity_type = any(cast(:types as text[]))""", types=list(DEDUPE_TYPES) + sorted(ORG_TYPES)) if scope is not None: rows = [r for r in rows if r["id"] in scope] 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))} # group: exact normalised name within a type; organisations across the org group (+ shared hf_org/github_org) groups: dict[str, list[dict[str, Any]]] = defaultdict(list) for r in rows: norm = normalize_alias(r["canonical_name"]) if r["entity_type"] in ORG_TYPES: groups[f"org:{norm}"].append(r) for k in ("hf_org", "github_org"): v = (r["attributes"] or {}).get(k) if isinstance(v, str) and v.strip() and normalize_alias(v) != norm: groups[f"org:{normalize_alias(v)}"].append(r) else: groups[f"{r['entity_type']}:{norm}"].append(r) # union overlapping org groups parent: dict[str, str] = {} def find(x: str) -> str: while parent.setdefault(x, x) != x: x = parent[x] return x for members in groups.values(): ids = [m["id"] for m in members] for other in ids[1:]: parent[find(other)] = find(ids[0]) clusters: dict[str, list[dict[str, Any]]] = defaultdict(list) seen: set[str] = set() for members in groups.values(): for m in members: if m["id"] not in seen: seen.add(m["id"]) clusters[find(m["id"])].append(m) dup_clusters = [c for c in clusters.values() if len(c) > 1] all_ids = [m["id"] for c in dup_clusters for m in c] claims = await _claim_counts(conn, all_ids) idents = await _identifiers(conn, all_ids) type_rank = {"company": 0, "lab": 0, "university": 0, "organization": 1} # curated org types beat the generic hub "organization" def collision_slug(m: dict[str, Any], cluster: list[dict[str, Any]]) -> int: """1 when the slug is a collision product — an organisation prefix (`google-gemma-4-31b`, `meta-ai-llama-…`) or a `-2` suffix over another member's slug (`minimax-m3-2`): the plain slug survives.""" org = orgs_by_id.get(m["organization_id"] or "") if org and m["slug"].startswith(org["slug"] + "-") and not normalize_alias(m["canonical_name"]).startswith(normalize_alias(org["slug"])): return 1 mm = re.match(r"^(.*)-\d+$", m["slug"]) return int(bool(mm) and any(o["slug"] == mm.group(1) for o in cluster if o is not m)) for cluster in dup_clusters: 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"])) survivor = cluster[0] for other in cluster[1:]: pair = sorted([survivor["id"], other["id"]]) if await kept_separate(conn, survivor["id"], other["id"]): rep.bump("_kept_separate") continue if _identifiers_conflict(idents[survivor["id"]], idents[other["id"]]): detail = _conflict_detail(idents[survivor["id"]], idents[other["id"]]) rep.bump("keep_separate_recorded") rep.example(f"keep separate: {other['slug']} vs {survivor['slug']} — {detail}") if apply: await record_decision(conn, other["id"], survivor["id"], "keep_separate", actor="canonicalize", note=f"same name, contradictory identifiers ({detail})", payload={"step": "duplicates", "slugs": [other["slug"], survivor["slug"]]}) 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", ids=pair) continue if (survivor["organization_id"] and other["organization_id"] and survivor["organization_id"] != other["organization_id"] and other["entity_type"] not in ORG_TYPES and not _same_publisher(survivor["canonical_name"], orgs_by_id.get(survivor["organization_id"]), orgs_by_id.get(other["organization_id"]))): if await _review(conn, "merge_candidate", pair, f"'{other['canonical_name']}' and '{survivor['canonical_name']}' share a name but have different organisations", {"slugs": [other["slug"], survivor["slug"]], "step": "duplicates"}, apply=apply): rep.bump("review_different_organizations") rep.example(f"review: {other['slug']} and {survivor['slug']} share a name but belong to different organisations") continue rep.bump("merged") rep.example(f"merge {other['entity_type']} {other['slug']} → {survivor['slug']}") if apply: await merge_entities(conn, other["id"], survivor["id"], mode="merge", actor="canonicalize", note="exact normalised-name duplicate", payload={"step": "duplicates"}) # junk organisations (single-letter names) → review only for r in rows: if r["entity_type"] in ORG_TYPES and len(r["canonical_name"].strip()) <= 1: if await _review(conn, "junk_entity", [r["id"]], f"organisation '{r['canonical_name']}' ({r['slug']}) looks like extraction noise", {"slug": r["slug"]}, apply=apply): rep.bump("review_junk_organization") rep.example(f"review junk organisation {r['slug']!r}") # org_kind defaults / normalisation for r in rows: if r["entity_type"] not in ORG_TYPES: continue attrs = r["attributes"] or {} current = attrs.get("org_kind") default = ORG_TYPE_DEFAULT_KIND.get(r["entity_type"]) canon, _raw, _m = normalize_property(r["entity_type"], "org_kind", current) if current else (None, None, []) target = canon if isinstance(canon, str) and canon in ("company", "lab", "university", "nonprofit", "government", "community", "consortium", "individual") else default if target and current != target: rep.bump("org_kind_set") if apply: await _set_attributes(conn, r["id"], {"org_kind": target}, source_id=source_id) # ---------------------------------------------------------------------------------------------- step: variants async def _models(conn: AsyncConnection, *, types: tuple[str, ...] = ("model",)) -> list[dict[str, Any]]: 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, e.identity_confidence, o.slug as org_slug, o.canonical_name as org_name, o.attributes->>'hf_org' as org_hf from entities e left join entities o on o.id = e.organization_id 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)) def _variant_index(models: list[dict[str, Any]], aliases: dict[str, list[str]] | None = None) -> tuple[dict[str, str], dict[str, list[dict[str, Any]]]]: """variant_key → canonical model id (unique keys only) and the full multi-map; with `aliases` (model id → alias strings) every alias of a canonical model contributes its own key (`gpt-3.5-turbo-0613`, `o3-mini-2025-01-31`…).""" multi: dict[str, list[dict[str, Any]]] = defaultdict(list) for m in models: a = analyze_model_name(m["canonical_name"]) if a.is_effort_variant or a.is_artifact: continue keys = {variant_key(m["canonical_name"])} for al in (aliases or {}).get(m["id"], []): aa = analyze_model_name(al) if not aa.is_effort_variant and not aa.is_artifact: keys.add(variant_key(al)) for k in keys: if k and m not in multi[k]: multi[k].append(m) return {k: v[0]["id"] for k, v in multi.items() if len(v) == 1}, multi async def _model_aliases(conn: AsyncConnection, ids: list[str]) -> dict[str, list[str]]: out: dict[str, list[str]] = defaultdict(list) for r in await fetch_all(conn, "select entity_id, alias from entity_aliases where entity_id = any(cast(:ids as text[]))", ids=ids): out[r["entity_id"]].append(r["alias"]) return out async def step_variants(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None: models = await _models(conn) index, _ = _variant_index(models, await _model_aliases(conn, [m["id"] for m in models])) resolver = Resolver(conn, source_tier=2, variant_index=index) candidates = [m for m in models if _in_scope(scope, m["id"]) and analyze_model_name(m["canonical_name"]).is_effort_variant] # names that used to look like variants (before the ontology learnt they are tiers/official releases) drop their hint and regain full identity for m in models: 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: rep.bump("stale_variant_hint_cleared") if apply: await execute(conn, """update entities set attributes = attributes - 'evaluation_variant_of_hint', provenance = provenance - 'evaluation_variant_of_hint', identity_confidence = 'high', updated_at = now() where id = :id""", id=m["id"]) 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"]) idents = await _identifiers(conn, [m["id"] for m in candidates]) official = {r["entity_id"] for r in await fetch_all(conn, """select distinct entity_id from claims where tier = 1 and status = 'current' and entity_id = any(cast(:ids as text[]))""", ids=[m["id"] for m in candidates])} if candidates else set() for m in candidates: a = analyze_model_name(m["canonical_name"]) rep.bump("_effort_variants_seen") schemes = set(idents.get(m["id"], {})) if not schemes <= EVALUATOR_SCHEMES or (m["attributes"] or {}).get("hf_repo") or m["id"] in official: # known to an official source, a hub or a provider → a real model that happens to end in "thinking"/"high"; never folded rep.bump("_kept_real_model") continue ref = EntityRef(entity_type="model", name=m["canonical_name"], id=m["id"]) folded = await resolver.resolve_variant(ref, org_id=m["organization_id"]) if folded is None: hint = base_name(m["canonical_name"]) attrs = m["attributes"] or {} if attrs.get("evaluation_variant_of_hint") == hint and m["identity_confidence"] == "low": continue rep.bump("unresolved_flagged") rep.example(f"unresolved variant {m['slug']} (base '{hint}' not found) → identity low + review") if apply: await _set_attributes(conn, m["id"], {"evaluation_variant_of_hint": hint}, source_id=await registry_source_id(conn)) await execute(conn, "update entities set identity_confidence = 'low' where id = :id", id=m["id"]) await _review(conn, "variant_candidate", [m["id"]], f"'{m['canonical_name']}' looks like an evaluation-effort variant of '{hint}' but no such model exists", {"slug": m["slug"], "base": hint, "effort": a.effort}, apply=apply) continue cid, effort = folded rep.bump("folded") target = next((x for x in models if x["id"] == cid), None) rep.example(f"fold {m['slug']} → {target['slug'] if target else cid} {effort}") if apply: rows = await fetch_all(conn, "select id, config from benchmark_results where model_id = :m", m=m["id"]) for r in rows: cfg = effort_config(m["canonical_name"], r["config"]) if not _same(cfg, r["config"]): await execute(conn, "update benchmark_results set config = cast(:c as jsonb) where id = :id", c=jsonb(cfg), id=r["id"]) rep.bump("results_reconfigured", len(rows)) await merge_entities(conn, m["id"], cid, mode="merge", actor="canonicalize", note="evaluation-effort variant folded into its canonical model", payload={"step": "variants", "effort": effort, "variant_slug": m["slug"]}) # ---------------------------------------------------------------------------------------------- step: artifacts async def step_artifacts(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None: source_id = await registry_source_id(conn) models = await _models(conn) _, multi = _variant_index(models) by_id = {m["id"]: m for m in models} # only `quantized_from` names the packaged model; `derived_from`/`fine_tuned_from` point at a *base* model (a fine-tune is a new model) rel_rows = await fetch_all(conn, """select r.subject_id, r.object_id from relations r join entities o on o.id = r.object_id where r.predicate = 'quantized_from' and r.valid_to is null and o.merged_into is null and o.entity_type = 'model'""") base_of: dict[str, str] = {} for r in rel_rows: base_of.setdefault(r["subject_id"], r["object_id"]) official_idents = await _identifiers(conn, [m["id"] for m in models if (m["attributes"] or {}).get("hf_repo")]) for m in models: if not _in_scope(scope, m["id"]): continue attrs = m["attributes"] or {} hf_repo = attrs.get("hf_repo") if isinstance(attrs.get("hf_repo"), str) else None probe = hf_repo or m["canonical_name"] a = analyze_model_name(probe) a_name = analyze_model_name(m["canonical_name"]) if hf_repo else a repo_org = (a.repo_org or "").lower() # repo attributes (is_quantized / quant_format) describe a hub repository: without hf_repo they come from a provider endpoint # (OpenRouter lists the serving precision) and say nothing about the model's identity quant_attr = str(attrs.get("quant_format") or "").lower() if hf_repo else "" attr_quantized = attrs.get("is_quantized") is True and bool(hf_repo) has_token = bool(a.quant_formats or a.precision or a_name.quant_formats or a_name.precision) flagged = has_token or a.is_artifact or a_name.is_artifact or attr_quantized or bool(quant_attr) or repo_org in CONVERTER_ORGS if not flagged: continue # "official" = the repo belongs to the model's developer: the family's known publisher, or the entity's own organisation when that # organisation is not a redistributor (hub-derived entities are attached to the redistributor org, e.g. `bartowski`) 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} 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) vk_candidates = [c for c in multi.get(variant_key(probe), []) if c["id"] != m["id"]] cid: str | None = None if official: # A quant/precision token in the repo name makes it an artifact whatever the organisation (an official FP8/GGUF repo is a # conversion of the model). Canonical = the same organisation's plain model with the same variant_key when it exists, else NULL. # Two guards, because a model is never invented or erased: (1) a native-dtype attribute without a token in the name # (DeepSeek-R1 `quant_format=fp8`) is the model; (2) when evaluators/providers know the tagged entity under its own identity # (Nemotron 3 Ultra whose only hub repo is `…-BF16`) it IS the model → merged into the plain sibling when one exists, kept otherwise. if not has_token: rep.bump("_official_checkpoint_kept_as_model") continue 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] external = set(official_idents.get(m["id"], {})) - {"hf_repo"} if external: if len(same_org) == 1 and not await kept_separate(conn, m["id"], same_org[0]["id"]): rep.bump("official_checkpoint_merged_into_model") rep.example(f"merge official checkpoint {m['slug']} → {same_org[0]['slug']} (known to {sorted(external)})") if apply: await merge_entities(conn, m["id"], same_org[0]["id"], mode="merge", actor="canonicalize", note="official checkpoint repo of the same release (dtype tag in the repo name)", payload={"step": "artifacts"}) else: rep.bump("_official_tagged_repo_is_the_model") continue cid = same_org[0]["id"] if len(same_org) == 1 else None if attr_quantized or a.is_quantized or a_name.is_quantized or quant_attr in QUANT_ATTR_FORMATS: kind = "quantization" elif a.is_conversion or a_name.is_conversion or a.precision or quant_attr: kind = "conversion" else: kind = "packaging" if cid is None: # 1) a model with the same variant_key (the name says what was quantised); 2) the hub's `quantized_from` object, but only when it # 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) candidates = vk_candidates if len(candidates) > 1: same_org = [c for c in candidates if c["organization_id"] == m["organization_id"]] 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))] candidates = same_org or official_c or candidates if len(candidates) == 1: cid = candidates[0]["id"] if cid is None: rel = base_of.get(m["id"]) if rel and rel in by_id and rel != m["id"]: base_probe = (by_id[rel]["attributes"] or {}).get("hf_repo") or by_id[rel]["canonical_name"] if variant_key(base_probe) == variant_key(probe) or analyze_model_name(base_probe).base_key in a.base_key: cid = rel if m["entity_type"] == "artifact" and m["canonical_id"] == cid: continue rep.bump("artifacts_marked" if cid else "artifacts_unresolved") rep.example(f"artifact[{kind}] {m['slug']} → {by_id[cid]['slug'] if cid else 'UNRESOLVED'}") if apply: await execute(conn, """update entities set entity_type = 'artifact', artifact_kind = coalesce(artifact_kind, :k), canonical_id = :c, identity_confidence = :ic, updated_at = now() where id = :id""", k=kind, c=cid, ic="high" if cid else "low", id=m["id"]) if cid: await upsert_relation(conn, m["id"], "artifact_of", cid, {"artifact_kind": kind}, source_id=source_id) await record_decision(conn, m["id"], cid, "variant_of", actor="canonicalize", payload={"artifact_kind": kind, "step": "artifacts"}) if not cid: await _review(conn, "unresolved_artifact", [m["id"]], f"'{m['canonical_name']}' is a {kind} artifact but its base model is unknown", {"slug": m["slug"], "hf_repo": hf_repo, "variant_key": variant_key(probe)}, apply=apply) # an organisation develops a model, not a quantisation/conversion of it: live `develops` edges into artifacts (bartowski, unsloth, # mlx-community… but also official orgs re-packaging their own weights) are closed and replaced by `published_by` scope_sql = "and a.id = any(cast(:ids as text[]))" if scope is not None else "" 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 r 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}""", ids=sorted(scope or [])) for e in edges: rep.bump("develops_to_artifact_closed") if apply: await execute(conn, "update relations set valid_to = now() where id = :id", id=e["id"]) 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) # ---------------------------------------------------------------------------------------------- step: families async def step_families(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None: source_id = await registry_source_id(conn) models = [m for m in await _models(conn) if _in_scope(scope, m["id"])] 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)) org_by_key: dict[str, str] = {} for o in orgs: for k in _org_lookup_keys(o): org_by_key.setdefault(k, o["id"]) 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") families: dict[str, dict[str, Any]] = {f["canonical_name"].lower(): f for f in fam_rows} # one family per label slugs_taken = {r["slug"] for r in await fetch_all(conn, "select slug from entities")} def family_slug(label: str) -> str: """slugify(label); on collision with any other entity (the model `gpt-5.5` itself) → `-family`, then numbered.""" base = slugify(label) for c in [base, f"{base}-family", *(f"{base}-family-{n}" for n in range(2, 20))]: if c not in slugs_taken: return c return f"{base}-family-{new_id('model_family')[-6:].lower()}" # re-slug families created under the former rule (organisation prefix on collision: `qwen-qwen3`, `mistral-mistral`) for fam in fam_rows: base = slugify(fam["canonical_name"]) if fam["slug"] in (base, f"{base}-family") or not fam["slug"].endswith(base): continue slugs_taken.discard(fam["slug"]) new_slug = family_slug(fam["canonical_name"]) slugs_taken.add(new_slug) rep.bump("families_reslugged") rep.example(f"family slug {fam['slug']} → {new_slug}") if apply: await execute(conn, "update entities set slug = :s, updated_at = now() where id = :id", s=new_slug, id=fam["id"]) 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", e=fam["id"], a=fam["slug"], n=normalize_alias(fam["slug"])) fam["slug"] = new_slug for m in models: label = family_release_hint(m["canonical_name"]) if not label: rep.bump("_no_family_hint") continue root = family_hint(m["canonical_name"]) or label official = official_orgs(m["canonical_name"]) org_id = next((org_by_key[k] for k in official if k in org_by_key), None) or m["organization_id"] fam = families.get(label.lower()) if fam is None: slug = family_slug(label) rep.bump("families_created") rep.example(f"family '{label}' ({slug}) root={root}") fam = {"id": new_id("model_family"), "slug": slug, "canonical_name": label, "organization_id": org_id, "_new": True} families[label.lower()] = fam slugs_taken.add(slug) if apply: 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) values (:id, 'model_family', :n, :slug, 'active', :org, cast(:a as jsonb), cast(:p as jsonb), :fs, now(), 'high')""", id=fam["id"], n=label, slug=slug, org=org_id, a=jsonb({"family_root": root, "label": label}), p=jsonb({"family_root": {"source_id": source_id, "tier": 2, "extractor": "derived"}}), fs=m["first_seen_at"]) await execute(conn, "insert into entity_aliases (entity_id, alias, alias_norm, kind) values (:e, :a, :n, 'alias') on conflict do nothing", e=fam["id"], a=label, n=normalize_alias(label)) if m["family_id"] != fam["id"]: rep.bump("models_linked") if apply: await execute(conn, "update entities set family_id = :f, first_seen_at = first_seen_at where id = :id", f=fam["id"], id=m["id"]) 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"]) await upsert_relation(conn, m["id"], "member_of_family", fam["id"], source_id=source_id) if not (m["attributes"] or {}).get("family"): rep.bump("family_attribute_set") if apply: await _set_attributes(conn, m["id"], {"family": label}, source_id=source_id) # ---------------------------------------------------------------------------------------------- step: licenses async def step_licenses(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None: source_id = await registry_source_id(conn) rows = await fetch_all(conn, """select id, entity_type, canonical_name, slug, attributes from entities where merged_into is null and entity_type = any(cast(:t as text[])) and (attributes ? 'license' or attributes ? 'openness' or attributes ? 'hf_repo') order by first_seen_at, id""", t=list(LICENSED_TYPES)) if scope is not None: rows = [r for r in rows if r["id"] in scope] lic_rows = await fetch_all(conn, "select id, slug from entities where entity_type = 'license'") license_entities: dict[str, str] = {r["slug"]: r["id"] for r in lic_rows} 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")} # derived claims that a higher-tier source already contradicts: stored once as `conflicting`, never re-proposed contested: dict[tuple[str, str], Any] = {(r["entity_id"], r["property"]): r["value"] for r in await fetch_all( conn, "select entity_id, property, value from claims where extractor = 'derived' and status = 'conflicting'")} writer = _derived_writer(conn, source_id) if apply else None for r in rows: attrs = r["attributes"] or {} raw = attrs.get("license") key = normalize_license(raw) if isinstance(raw, str) else None if key is None and isinstance(attrs.get("license_key"), str) and attrs["license_key"] in LICENSES: key = attrs["license_key"] if isinstance(raw, str) and key is None: rep.bump("_license_unclassified") rep.example(f"unclassified licence {raw!r} on {r['slug']}") if len(rep.examples) < 4 else None if key: lslug = key.lower() lid = license_entities.get(lslug) if lid is None: info = LICENSES[key] lid = new_id("license") license_entities[lslug] = lid rep.bump("license_entities_created") if apply: await execute(conn, """insert into entities (id, entity_type, canonical_name, slug, status, description, attributes, provenance, identity_confidence) values (:id, 'license', :n, :slug, 'active', :d, cast(:a as jsonb), cast(:p as jsonb), 'high')""", id=lid, n=info.label, slug=lslug, d=f"{info.label} — {info.category} licence" + (f" (SPDX {info.spdx})" if info.spdx else ""), a=jsonb({**info.as_dict(), "license_key": key}), p=jsonb({"license_key": {"source_id": source_id, "tier": 2, "extractor": "derived"}})) await execute(conn, "insert into entity_aliases (entity_id, alias, alias_norm, kind) values (:e, :a, :n, 'alias') on conflict do nothing", e=lid, a=info.label, n=normalize_alias(info.label)) for alias in (key, info.spdx or key): await execute(conn, "insert into entity_aliases (entity_id, alias, alias_norm, kind) values (:e, :a, :n, 'alias') on conflict do nothing", e=lid, a=alias, n=normalize_alias(alias)) if (r["id"], lid) not in existing_rel: rep.bump("uses_license_relations") existing_rel.add((r["id"], lid)) if apply: await upsert_relation(conn, r["id"], "uses_license", lid, source_id=source_id) if attrs.get("license_key") != key: rep.bump("license_key_attribute_set") if apply: await _set_attributes(conn, r["id"], {"license_key": key}, source_id=source_id) if r["entity_type"] not in ("model", "artifact"): continue # openness dimensions (derived claims, tier 2, never supersede a tier-1 statement) openness_raw = attrs.get("openness") openness_now = normalize_openness(openness_raw) if isinstance(openness_raw, str) else None weights: bool | None = None if attrs.get("hf_repo") or attrs.get("model_card_url") or attrs.get("weights_url") or (openness_now or "").startswith(("open", "restricted")): weights = True elif openness_now == "proprietary": weights = False if weights is None: rep.bump("_openness_unknown_skipped") continue dims = openness_dimensions(weights_available=weights, license_key=key) derived = derive_openness(dims, license_key=key) wanted: dict[str, Any] = {"weights_available": dims["weights_available"]} for k in ("commercial_use_allowed", "redistribution_allowed", "derivatives_allowed"): if dims[k] is not None: wanted[k] = dims[k] if derived != "unknown": wanted["openness"] = derived 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)} if not changed: continue rep.bump("openness_claims_written", len(changed)) if "openness" in changed: rep.bump("openness_category_changed") rep.example(f"{r['slug']}: openness {openness_raw!r} → {derived} (licence {key})") if writer is not None: facts = Facts() ref = EntityRef(entity_type=r["entity_type"], name=r["canonical_name"], id=r["id"]) for k, v in changed.items(): facts.claim(ref, k, v, confidence="high") await writer.write(facts) if writer is not None: rep.bump("_conflicting_derived_claims", writer.stats.conflicts) # ---------------------------------------------------------------------------------------------- step: taxonomy async def step_taxonomy(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None: props = sorted(TAXONOMY_PROPERTIES) 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_id where c.status = 'current' and c.property = any(cast(:p as text[]))""", p=props) if scope is not None: rows = [r for r in rows if r["entity_id"] in scope] mappings: dict[tuple[str, str], str | None] = {} def note(maps: list[tuple[str, str, str | None]]) -> None: for d, rw, cn in maps: if cn is not None and rw == cn: continue # identity mapping: nothing to learn mappings[(d, rw)] = cn if cn is not None else mappings.get((d, rw)) for r in rows: canon, raw, maps = normalize_property(r["entity_type"], r["property"], r["value"]) note(maps) if _same(canon, r["value"]): continue rep.bump(f"claims_normalized:{r['property']}") rep.example(f"{r['entity_type']} {r['property']}: {json.dumps(r['value'], ensure_ascii=False)[:40]} → {json.dumps(canon, ensure_ascii=False)[:40]}") if apply: keep_raw = r["value_raw"] or (r["value"] if isinstance(r["value"], str) else json.dumps(r["value"], ensure_ascii=False)) keep_raw = keep_raw if keep_raw != canon else None await execute(conn, "update claims set value = cast(:v as jsonb), value_text = :vt, value_raw = :raw where id = :id", v=jsonb(canon), vt=canon[:2000] if isinstance(canon, str) else None, raw=keep_raw, id=r["id"]) attrs = {r["property"]: canon} if keep_raw: attrs[f"{r['property']}_raw"] = keep_raw 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"]) if r["property"] == "status" and isinstance(canon, str): await execute(conn, "update entities set status = :s where id = :id", s=canon[:40], id=r["entity_id"]) # attributes without a current claim (seeded/imported values) for prop in props: 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 ? :p and not exists (select 1 from claims c where c.entity_id = e.id and c.property = :p and c.status = 'current')""", p=prop) for e in ents: if not _in_scope(scope, e["id"]): continue canon, raw, maps = normalize_property(e["entity_type"], prop, e["value"]) note(maps) if _same(canon, e["value"]): continue rep.bump(f"attributes_normalized:{prop}") if apply: keep_raw = e["value"] if isinstance(e["value"], str) else json.dumps(e["value"], ensure_ascii=False) attrs = {prop: canon, f"{prop}_raw": keep_raw} await execute(conn, "update entities set attributes = attributes || cast(:a as jsonb), updated_at = now() where id = :id", a=jsonb(attrs), id=e["id"]) if prop == "status" and isinstance(canon, str): await execute(conn, "update entities set status = :s where id = :id", s=canon[:40], id=e["id"]) existing = {(m["domain"], m["raw"]): m["canonical"] for m in await fetch_all(conn, "select domain, raw, canonical from taxonomy_mappings")} for (domain, raw_full), canon in mappings.items(): raw = raw_full[:300] if (domain, raw) in existing and (existing[(domain, raw)] == canon or canon is None): continue rep.bump("taxonomy_mappings_upserted") if apply: await execute(conn, """insert into taxonomy_mappings (domain, raw, canonical) values (:d, :r, :c) on conflict (domain, raw) do update set canonical = coalesce(excluded.canonical, taxonomy_mappings.canonical), last_seen_at = now()""", d=domain, r=raw, c=canon) unknown = sorted({f"{d}:{r}" for (d, r), c in mappings.items() if c is None}) if unknown: rep.notes.append(f"{len(unknown)} raw value(s) without canonical mapping kept as-is: {', '.join(unknown[:15])}{'…' if len(unknown) > 15 else ''}") # ---------------------------------------------------------------------------------------------- step: results async def _rehome_results(conn: AsyncConnection, rep: StepReport, apply: bool, scope: set[str] | None) -> None: """Move results from a benchmark *family head* to the registry's variant entity when the row says which variant it measured: (a) config.variant / config.board equals (case-insensitively) the variant entity's name, an alias or its `variant` attribute; (b) LiveBench `category:` metrics → `livebench-` with the registry metric (`average score`), variant = Name; (c) aider `percent_cases_well_formed` → `aider-polyglot-well-formed`. Then dedupe/config keys, one-current-row and `evaluated_on` relations are recomputed for every touched model.""" variants = await fetch_all(conn, """select v.id, v.slug, v.canonical_name, v.attributes, r.object_id as head_id, (select array_agg(alias) from entity_aliases a where a.entity_id = v.id) as aliases from entities v join relations r on r.subject_id = v.id and r.predicate = 'variant_of' and r.valid_to is null where v.entity_type = 'benchmark' and v.merged_into is null""") if not variants: return by_slug: dict[str, dict[str, Any]] = {v["slug"]: v for v in variants} by_head: dict[str, list[tuple[set[str], dict[str, Any]]]] = defaultdict(list) for v in variants: keys = {v["canonical_name"].lower(), v["slug"], *(a.lower() for a in (v["aliases"] or []))} var_attr = (v["attributes"] or {}).get("variant") if isinstance(var_attr, str): keys.add(var_attr.lower()) by_head[v["head_id"]].append((keys, v)) heads = list(by_head) rows = await fetch_all(conn, """select r.id, r.model_id, r.benchmark_id, r.metric, r.config from benchmark_results r where r.benchmark_id = any(cast(:h as text[]))""", h=heads) moves: list[tuple[dict[str, Any], dict[str, Any], str | None, dict[str, Any]]] = [] for r in rows: if scope is not None and r["model_id"] not in scope: continue cfg = dict(r["config"] or {}) metric = r["metric"] or "" target: dict[str, Any] | None = None new_metric: str | None = None label = next((str(cfg[k]) for k in ("variant", "board") if isinstance(cfg.get(k), str) and cfg[k].strip()), None) if label: target = next((v for keys, v in by_head[r["benchmark_id"]] if label.lower() in keys), None) if target is None and metric.lower().startswith("category:"): name = metric.split(":", 1)[1].strip() cand = by_slug.get(f"livebench-{slugify(name)}") if cand and cand["head_id"] == r["benchmark_id"]: target = cand new_metric = (cand["attributes"] or {}).get("metric") or "average score" cfg["variant"] = name if target is None and metric == "percent_cases_well_formed": cand = by_slug.get("aider-polyglot-well-formed") if cand and cand["head_id"] == r["benchmark_id"]: target = cand if target is None: continue moves.append((r, target, new_metric, cfg)) if not moves: return touched: set[tuple[str, str, str]] = set() for r, target, new_metric, cfg in moves: rep.bump(f"results_rehomed:{target['slug']}") touched.add((r["model_id"], r["benchmark_id"], target["id"])) if apply: await execute(conn, "update benchmark_results set benchmark_id = :b, metric = coalesce(:m, metric), config = cast(:c as jsonb), variant = :v where id = :id", b=target["id"], m=new_metric, c=jsonb(cfg), v=bench_ontology.variant_from_config(cfg), id=r["id"]) if not apply: return source_id = await registry_source_id(conn) for model_id in sorted({m for m, _, _ in touched}): from aiatlas.services.merge import recompute_result_keys await recompute_result_keys(conn, model_id) await enforce_current_results(conn, model_id=model_id) for model_id, head_id, target_id in sorted(touched): await upsert_relation(conn, model_id, "evaluated_on", target_id, source_id=source_id) 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) if not left: 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", m=model_id, b=head_id) rep.bump("evaluated_on_repointed") async def step_results(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None: await _rehome_results(conn, rep, apply, scope) 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, r.is_current, s.key as source_key from benchmark_results r left join sources s on s.id = r.source_id""") if scope is not None: rows = [r for r in rows if r["model_id"] in scope] for r in rows: cfg = r["config"] or {} ck = bench_ontology.config_key(cfg, r["metric"]) trust = r["trust_level"] or bench_ontology.trust_level(r["source_key"], cfg, extractor=r["extractor"] or "deterministic") variant = r["variant"] or bench_ontology.variant_from_config(cfg) rg = r["run_group"] or bench_ontology.run_group_from_config(cfg) current = r["is_current"] and r["valid_to"] is None if (ck, trust, variant, rg, current) == (r["config_key"], r["trust_level"], r["variant"], r["run_group"], r["is_current"]): continue rep.bump("results_backfilled") if apply: await execute(conn, "update benchmark_results set config_key = :ck, trust_level = :t, variant = :v, run_group = :rg, is_current = :cur where id = :id", ck=ck, t=trust, v=variant, rg=rg, cur=current, id=r["id"]) if scope is None: closed = await enforce_current_results(conn, dry_run=not apply) else: closed = sum([await enforce_current_results(conn, model_id=mid, dry_run=not apply) for mid in sorted(scope)]) if closed: rep.bump("older_run_rows_closed", closed) live = await fetch_one(conn, "select count(*) as n from benchmark_results where valid_to is null and is_current") rep.notes.append(f"current benchmark results after step: {int(live['n']) if live else 0}") # ---------------------------------------------------------------------------------------------- step: events async def step_events(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None: lag = timedelta(days=BACKFILL_LAG_DAYS) runs = await fetch_all(conn, """select connector_name, started_at, finished_at from connector_runs where status in ('success', 'unchanged', 'suspect', 'released') order by connector_name, started_at""") second_start: dict[str, datetime | None] = {} per: dict[str, list[dict[str, Any]]] = defaultdict(list) for r in runs: per[r["connector_name"]].append(r) for name, lst in per.items(): second_start[name] = lst[1]["started_at"] if len(lst) > 1 else None before = await fetch_one(conn, """select count(*) as n from change_events where is_backfill = false and event_type <> 'DOCUMENT_CHANGED' and observed_at > now() - interval '24 hours'""") 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, e.attributes->>'release_date' as release_date, e.attributes->>'published_at' as published_at from change_events ev left join entities e on e.id = ev.entity_id""") from aiatlas.sdk.extract.dates import parse_datetime if scope is not None: rows = [r for r in rows if r["entity_id"] in scope] for ev in rows: obs = ev["observed_at"] bf = False if ev["connector_name"] in STRUCTURAL_CONNECTORS: bf = True # merges, folds, derived corrections: bookkeeping about our own data, never news elif ev["connector_name"] and ev["connector_name"] in per: s2 = second_start.get(ev["connector_name"]) if s2 is None or obs < s2: bf = True elif ev["connector_name"] and ev["connector_name"] not in per: bf = True # connector without any successful run yet → initial load if not bf and ev["effective_at"] is not None and ev["effective_at"] < obs - lag: bf = True if not bf and ev["event_type"].startswith("NEW_"): hint = parse_datetime(ev["release_date"]) if ev["release_date"] else (parse_datetime(ev["published_at"]) if ev["published_at"] else None) if hint is not None: hint = hint if hint.tzinfo else hint.replace(tzinfo=UTC) if hint < obs - lag: bf = True gk = group_key_for(ev["event_type"], ev["entity_id"], ev["effective_at"], obs) if bf == ev["is_backfill"] and gk == ev["group_key"]: continue if bf != ev["is_backfill"]: rep.bump("backfill_flag_set" if bf else "backfill_flag_cleared") if gk != ev["group_key"]: rep.bump("group_key_set") if apply: await execute(conn, "update change_events set is_backfill = :bf, group_key = :gk where id = :id", bf=bf, gk=gk, id=ev["id"]) after = await fetch_one(conn, """select count(*) as n from change_events where is_backfill = false and event_type <> 'DOCUMENT_CHANGED' and observed_at > now() - interval '24 hours'""") 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}" + ("" if apply else " (dry-run: after = before)")) # ---------------------------------------------------------------------------------------------- step: anomalies async def _collect_anomalies(conn: AsyncConnection) -> list[Anomaly]: found: list[Anomaly] = [] 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"): found.extend(check_model(r["id"], r["canonical_name"], r["attributes"] or {})) for r in await fetch_all(conn, "select id, canonical_name, attributes from entities where entity_type = 'hardware' and merged_into is null"): found.extend(check_hardware(r["id"], r["canonical_name"], r["attributes"] or {})) for r in await fetch_all(conn, """select p.*, m.canonical_name as model_name, v.canonical_name as provider_name from prices p join entities m on m.id = p.model_id join entities v on v.id = p.provider_id where p.valid_to is null"""): found.extend(check_price(r)) 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, b.canonical_name as benchmark_name, m.attributes->>'release_date' as model_release_date from benchmark_results r join entities m on m.id = r.model_id join entities b on b.id = r.benchmark_id where r.valid_to is null and r.is_current"""): found.extend(check_result(r)) return found async def step_anomalies(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None: found = await _collect_anomalies(conn) if scope is not None: found = [a for a in found if a.entity_id in scope] keys: dict[str, Anomaly] = {} for a in found: keys.setdefault(a.dedupe_key[:400], a) existing = {r["dedupe_key"]: r for r in await fetch_all(conn, "select dedupe_key, status, severity from anomalies")} for key, a in keys.items(): rep.bump(f"_by_severity:{a.severity}") prev = existing.get(key) if prev is None: rep.bump("anomalies_opened") rep.example(f"[{a.severity}] {a.message}") elif prev["status"] == "resolved": rep.bump("anomalies_reopened") if apply: await record(conn, a) stale = [k for k, r in existing.items() if r["status"] == "open" and k not in keys] if scope is None else [] if stale: rep.bump("anomalies_resolved", len(stale)) if apply: await execute(conn, """update anomalies set status = 'resolved', resolved_at = now(), resolution = 'check no longer fires' where status = 'open' and dedupe_key = any(cast(:k as text[]))""", k=stale) _STEP_FUNCTIONS = { "duplicates": step_duplicates, "variants": step_variants, "artifacts": step_artifacts, "families": step_families, "licenses": step_licenses, "taxonomy": step_taxonomy, "results": step_results, "events": step_events, "anomalies": step_anomalies, } # ---------------------------------------------------------------------------------------------- quarantine release / discard async def release_quarantine(conn: AsyncConnection, quarantine_id: str, *, actor: str = "admin") -> dict[str, Any]: """Write the held facts of a quarantined run exactly as the connector would have (same source, snapshot, tier, run id).""" from aiatlas.connectors import get as get_connector q = await fetch_one(conn, "select * from quarantined_runs where id = :id", id=quarantine_id) if not q: raise LookupError(f"quarantined run {quarantine_id} not found") if q["status"] != "pending": raise ValueError(f"quarantined run {quarantine_id} is already {q['status']}") connector = get_connector(q["connector_name"]) 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"]) totals: dict[str, int] = defaultdict(int) for item in q["facts"] or []: facts = facts_from_json(item["facts"]) fetched_at = datetime.fromisoformat(item["fetched_at"]) writer = FactWriter(conn, source_id=state["source_id"] if state else None, snapshot_id=item.get("snapshot_id"), source_url=item.get("source_url"), tier=connector.tier, connector_name=q["connector_name"], extractor="deterministic", extractor_version=connector.parser_version, observed_at=fetched_at, run_id=q["run_id"], source_key=state["source_key"] if state else None) ws = await writer.write(facts) for k, v in ws.as_dict().items(): totals[k] += v main = facts.document_entity if main and main.id is None: await writer.resolver.resolve(main) if main and main.id and item.get("doc_id"): 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"]) if item.get("snapshot_id"): await execute(conn, "update snapshots set processing_status = 'extracted' where id = :id", id=item["snapshot_id"]) await execute(conn, "update quarantined_runs set status = 'released', resolved_at = now(), resolved_by = :who where id = :id", who=actor, id=quarantine_id) await execute(conn, "update connector_runs set status = 'released' where id = :r", r=q["run_id"]) await execute(conn, "update review_queue set status = 'approved', resolved_at = now() where dedupe_key = :d", d=f"quarantine:{quarantine_id}") observed = (q["stats"] or {}).get("observed") if observed and (q["stats"] or {}).get("full_extraction"): baseline = (await fetch_one(conn, "select baseline from connectors where name = :n", n=q["connector_name"]) or {}).get("baseline") nb = connector._next_baseline(baseline, observed) # noqa: SLF001 await execute(conn, "update connectors set baseline = cast(:b as jsonb) where name = :n", b=jsonb(nb), n=q["connector_name"]) await audit(conn, "quarantine.release", quarantine_id, {"connector": q["connector_name"], "run_id": q["run_id"], **totals}, actor=actor) return {"id": quarantine_id, "connector": q["connector_name"], **totals} async def discard_quarantine(conn: AsyncConnection, quarantine_id: str, *, actor: str = "admin", note: str | None = None) -> dict[str, Any]: q = await fetch_one(conn, "select id, run_id, connector_name, status, facts from quarantined_runs where id = :id", id=quarantine_id) if not q: raise LookupError(f"quarantined run {quarantine_id} not found") if q["status"] != "pending": raise ValueError(f"quarantined run {quarantine_id} is already {q['status']}") await execute(conn, "update quarantined_runs set status = 'discarded', resolved_at = now(), resolved_by = :who where id = :id", who=actor, id=quarantine_id) await execute(conn, "update connector_runs set status = 'discarded' where id = :r", r=q["run_id"]) await execute(conn, "update review_queue set status = 'rejected', resolved_at = now() where dedupe_key = :d", d=f"quarantine:{quarantine_id}") for item in q["facts"] or []: if item.get("snapshot_id"): await execute(conn, "update snapshots set processing_status = 'discarded' where id = :id", id=item["snapshot_id"]) await audit(conn, "quarantine.discard", quarantine_id, {"connector": q["connector_name"], "run_id": q["run_id"], "note": note}, actor=actor) return {"id": quarantine_id, "connector": q["connector_name"], "status": "discarded"} async def list_quarantine(conn: AsyncConnection, *, status: str = "pending", limit: int = 50) -> list[dict[str, Any]]: 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 documents from quarantined_runs where (cast(:s as text) = '' or status = :s) order by created_at desc limit :n""", s=status or "", n=limit) __all__ = ["CANON_VERSION", "STEPS", "Report", "StepReport", "canonicalize", "discard_quarantine", "list_quarantine", "release_quarantine"]