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 logging17from collections import defaultdict18from dataclasses import dataclass, field19from datetime import UTC, datetime, timedelta20from typing import Any2122from sqlalchemy.ext.asyncio import AsyncConnection2324from aiatlas.db import execute, fetch_all, fetch_one, jsonb, transaction25from aiatlas.ids import new_id, normalize_alias, slugify26from aiatlas.ontology import benchmarks as bench_ontology27from aiatlas.ontology.anomalies import Anomaly, check_hardware, check_model, check_price, check_result28from aiatlas.ontology.licenses import LICENSES, normalize_license29from aiatlas.ontology.models import (30 CONVERTER_ORGS,31 analyze_model_name,32 base_name,33 effort_config,34 family_hint,35 family_release_hint,36 official_orgs,37 variant_key,38)39from aiatlas.ontology.openness import derive_openness, normalize_openness, openness_dimensions40from aiatlas.ontology.taxonomy import ORG_TYPE_DEFAULT_KIND, TAXONOMY_PROPERTIES, normalize_property41from aiatlas.sdk.facts import EntityRef, Facts, facts_from_json42from aiatlas.sdk.resolution import EVALUATOR_SCHEMES, Resolver43from aiatlas.sdk.writer import FactWriter, _same44from aiatlas.services.anomalies import record45from aiatlas.services.events import BACKFILL_LAG_DAYS, group_key_for46from aiatlas.services.merge import (47 ORG_TYPES,48 audit,49 enforce_current_results,50 kept_separate,51 merge_entities,52 record_decision,53 registry_source_id,54 upsert_relation,55)5657log = logging.getLogger(__name__)5859CANON_VERSION = "2026.09"60STRUCTURAL_CONNECTORS = {"curation", "canonicalize"}61STEPS = ("duplicates", "variants", "artifacts", "families", "licenses", "taxonomy", "results", "events", "anomalies")62DEDUPE_TYPES = ("model", "provider", "benchmark", "dataset", "framework", "library", "hardware")63LICENSED_TYPES = ("model", "artifact", "dataset", "framework", "library", "repository")64QUANT_ATTR_FORMATS = {"gguf", "awq", "gptq", "exl2", "exl3", "int4", "int8", "fp8", "nvfp4", "mxfp4", "fp4", "bnb", "quantized"}656667@dataclass68class StepReport:69 name: str70 counts: dict[str, int] = field(default_factory=dict)71 examples: list[str] = field(default_factory=list)72 notes: list[str] = field(default_factory=list)7374 def bump(self, key: str, n: int = 1) -> None:75 self.counts[key] = self.counts.get(key, 0) + n7677 def example(self, text: str, *, limit: int = 12) -> None:78 if len(self.examples) < limit:79 self.examples.append(text)8081 @property82 def changes(self) -> int:83 return sum(v for k, v in self.counts.items() if not k.startswith("_"))848586@dataclass87class Report:88 apply: bool89 steps: list[StepReport] = field(default_factory=list)90 started_at: datetime = field(default_factory=lambda: datetime.now(UTC))9192 @property93 def changes(self) -> int:94 return sum(s.changes for s in self.steps)9596 def render(self) -> str:97 mode = "APPLY" if self.apply else "DRY-RUN"98 lines = [f"aia canonicalize — {mode} — {self.started_at:%Y-%m-%d %H:%M:%S} UTC — {self.changes} change(s)"]99 for s in self.steps:100 lines.append(f"\n[{s.name}] {s.changes} change(s)")101 for k, v in sorted(s.counts.items()):102 lines.append(f" {k:<40} {v}")103 for n in s.notes:104 lines.append(f" · {n}")105 for e in s.examples:106 lines.append(f" - {e}")107 return "\n".join(lines)108109 def as_dict(self) -> dict[str, Any]:110 return {"apply": self.apply, "started_at": self.started_at.isoformat(), "changes": self.changes,111 "steps": [{"name": s.name, "counts": s.counts, "examples": s.examples, "notes": s.notes} for s in self.steps]}112113114async def canonicalize(*, apply: bool = False, steps: list[str] | None = None, scope: set[str] | None = None) -> Report:115 """Run the steps in canonical order. `scope` (entity ids) restricts the rows a step *acts on* — lookups (bases, families, licences)116 still see the whole corpus; used by tests and targeted re-runs."""117 wanted = [s for s in STEPS if not steps or s in steps]118 unknown = set(steps or []) - set(STEPS)119 if unknown:120 raise ValueError(f"unknown step(s): {sorted(unknown)}; known: {STEPS}")121 report = Report(apply=apply)122 for name in wanted:123 rep = StepReport(name=name)124 fn = _STEP_FUNCTIONS[name]125 async with transaction() as conn:126 await fn(conn, rep, apply, scope=scope)127 if apply and rep.changes:128 await audit(conn, f"canonicalize.{name}", None, {"counts": rep.counts, "version": CANON_VERSION}, actor="canonicalize")129 report.steps.append(rep)130 log.info("canonicalize step done", extra={"step": name, "apply": apply, **{k: v for k, v in rep.counts.items()}})131 return report132133134# ---------------------------------------------------------------------------------------------- shared helpers135def _derived_writer(conn: AsyncConnection, source_id: str | None) -> FactWriter:136 return FactWriter(conn, source_id=source_id, snapshot_id=None, source_url=None, tier=2, connector_name="canonicalize", extractor="derived",137 extractor_version=CANON_VERSION, run_id=f"canon_{datetime.now(UTC):%Y%m%d}")138139140async def _review(conn: AsyncConnection, kind: str, entity_ids: list[str], reason: str, payload: dict[str, Any], *, apply: bool = True) -> bool:141 """Queue a review item once (dedupe key). Returns True when the item is new (or would be, in dry-run) so reports stay idempotent."""142 dedupe = f"{kind}:{':'.join(entity_ids)}:{normalize_alias(reason)[:80]}"143 if await fetch_one(conn, "select 1 from review_queue where dedupe_key = :d", d=dedupe):144 return False145 if not apply:146 return True147 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)148 on conflict (dedupe_key) do nothing returning id""", id=new_id("review"), k=kind, ids=entity_ids, p=jsonb(payload), r=reason, d=dedupe)149 return row is not None150151152async def _set_attributes(conn: AsyncConnection, entity_id: str, attrs: dict[str, Any], *, source_id: str | None) -> None:153 """Direct attribute write for structural/derived facts (family label, license_key, org_kind default) with `derived` provenance."""154 prov = {k: {"source_id": source_id, "tier": 2, "confidence": "high", "extractor": "derived", "observed_at": datetime.now(UTC).isoformat(timespec="seconds")} for k in attrs}155 await execute(conn, "update entities set attributes = attributes || cast(:a as jsonb), provenance = provenance || cast(:p as jsonb), updated_at = now() where id = :id",156 a=jsonb(attrs), p=jsonb(prov), id=entity_id)157158159async def _claim_counts(conn: AsyncConnection, ids: list[str]) -> dict[str, int]:160 if not ids:161 return {}162 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)163 return {r["entity_id"]: int(r["n"]) for r in rows}164165166async def _identifiers(conn: AsyncConnection, ids: list[str]) -> dict[str, dict[str, set[str]]]:167 out: dict[str, dict[str, set[str]]] = defaultdict(lambda: defaultdict(set))168 if not ids:169 return out170 rows = await fetch_all(conn, "select entity_id, scheme, value from entity_identifiers where entity_id = any(cast(:ids as text[]))", ids=ids)171 for r in rows:172 out[r["entity_id"]][r["scheme"]].add(r["value"])173 return out174175176def _identifiers_conflict(a: dict[str, set[str]], b: dict[str, set[str]]) -> bool:177 return any(a[s] and b[s] and a[s] != b[s] for s in set(a) & set(b))178179180def _org_lookup_keys(row: dict[str, Any]) -> set[str]:181 attrs = row.get("attributes") or {}182 keys = {row["slug"], normalize_alias(row["canonical_name"])}183 for k in ("hf_org", "github_org"):184 v = attrs.get(k)185 if isinstance(v, str) and v.strip():186 keys.add(v.strip().lower())187 return {k for k in keys if k}188189190# ---------------------------------------------------------------------------------------------- step: duplicates191def _in_scope(scope: set[str] | None, entity_id: str) -> bool:192 return scope is None or entity_id in scope193194195async def step_duplicates(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:196 source_id = await registry_source_id(conn)197 rows = await fetch_all(conn, """select id, entity_type, canonical_name, slug, organization_id, attributes, first_seen_at from entities198 where merged_into is null and entity_type = any(cast(:types as text[]))""", types=list(DEDUPE_TYPES) + sorted(ORG_TYPES))199 if scope is not None:200 rows = [r for r in rows if r["id"] in scope]201 # group: exact normalised name within a type; organisations across the org group (+ shared hf_org/github_org)202 groups: dict[str, list[dict[str, Any]]] = defaultdict(list)203 for r in rows:204 norm = normalize_alias(r["canonical_name"])205 if r["entity_type"] in ORG_TYPES:206 groups[f"org:{norm}"].append(r)207 for k in ("hf_org", "github_org"):208 v = (r["attributes"] or {}).get(k)209 if isinstance(v, str) and v.strip() and normalize_alias(v) != norm:210 groups[f"org:{normalize_alias(v)}"].append(r)211 else:212 groups[f"{r['entity_type']}:{norm}"].append(r)213 # union overlapping org groups214 parent: dict[str, str] = {}215216 def find(x: str) -> str:217 while parent.setdefault(x, x) != x:218 x = parent[x]219 return x220221 for members in groups.values():222 ids = [m["id"] for m in members]223 for other in ids[1:]:224 parent[find(other)] = find(ids[0])225 clusters: dict[str, list[dict[str, Any]]] = defaultdict(list)226 seen: set[str] = set()227 for members in groups.values():228 for m in members:229 if m["id"] not in seen:230 seen.add(m["id"])231 clusters[find(m["id"])].append(m)232 dup_clusters = [c for c in clusters.values() if len(c) > 1]233 all_ids = [m["id"] for c in dup_clusters for m in c]234 claims = await _claim_counts(conn, all_ids)235 idents = await _identifiers(conn, all_ids)236 type_rank = {"company": 0, "lab": 0, "university": 0, "organization": 1} # curated org types beat the generic hub "organization"237 for cluster in dup_clusters:238 cluster.sort(key=lambda m: (type_rank.get(m["entity_type"], 0), -claims.get(m["id"], 0), m["first_seen_at"]))239 survivor = cluster[0]240 for other in cluster[1:]:241 pair = sorted([survivor["id"], other["id"]])242 if await kept_separate(conn, survivor["id"], other["id"]):243 rep.bump("_kept_separate")244 continue245 if _identifiers_conflict(idents[survivor["id"]], idents[other["id"]]):246 if await _review(conn, "merge_candidate", pair, f"'{other['canonical_name']}' duplicates '{survivor['canonical_name']}' but identifiers conflict",247 {"slugs": [other["slug"], survivor["slug"]], "step": "duplicates"}, apply=apply):248 rep.bump("review_conflicting_identifiers")249 rep.example(f"review: {other['canonical_name']} ({other['slug']}) vs {survivor['canonical_name']} ({survivor['slug']}) — identifiers differ")250 continue251 if survivor["organization_id"] and other["organization_id"] and survivor["organization_id"] != other["organization_id"] and other["entity_type"] not in ORG_TYPES:252 if await _review(conn, "merge_candidate", pair, f"'{other['canonical_name']}' and '{survivor['canonical_name']}' share a name but have different organisations",253 {"slugs": [other["slug"], survivor["slug"]], "step": "duplicates"}, apply=apply):254 rep.bump("review_different_organizations")255 rep.example(f"review: {other['slug']} and {survivor['slug']} share a name but belong to different organisations")256 continue257 rep.bump("merged")258 rep.example(f"merge {other['entity_type']} {other['slug']} → {survivor['slug']}")259 if apply:260 await merge_entities(conn, other["id"], survivor["id"], mode="merge", actor="canonicalize", note="exact normalised-name duplicate",261 payload={"step": "duplicates"})262 # junk organisations (single-letter names) → review only263 for r in rows:264 if r["entity_type"] in ORG_TYPES and len(r["canonical_name"].strip()) <= 1:265 if await _review(conn, "junk_entity", [r["id"]], f"organisation '{r['canonical_name']}' ({r['slug']}) looks like extraction noise", {"slug": r["slug"]}, apply=apply):266 rep.bump("review_junk_organization")267 rep.example(f"review junk organisation {r['slug']!r}")268 # org_kind defaults / normalisation269 for r in rows:270 if r["entity_type"] not in ORG_TYPES:271 continue272 attrs = r["attributes"] or {}273 current = attrs.get("org_kind")274 default = ORG_TYPE_DEFAULT_KIND.get(r["entity_type"])275 canon, _raw, _m = normalize_property(r["entity_type"], "org_kind", current) if current else (None, None, [])276 target = canon if isinstance(canon, str) and canon in ("company", "lab", "university", "nonprofit", "government", "community", "consortium", "individual") else default277 if target and current != target:278 rep.bump("org_kind_set")279 if apply:280 await _set_attributes(conn, r["id"], {"org_kind": target}, source_id=source_id)281282283# ---------------------------------------------------------------------------------------------- step: variants284async def _models(conn: AsyncConnection, *, types: tuple[str, ...] = ("model",)) -> list[dict[str, Any]]:285 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,286 e.identity_confidence, o.slug as org_slug, o.canonical_name as org_name, o.attributes->>'hf_org' as org_hf287 from entities e left join entities o on o.id = e.organization_id288 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))289290291def _variant_index(models: list[dict[str, Any]]) -> tuple[dict[str, str], dict[str, list[dict[str, Any]]]]:292 """variant_key → canonical model id (unique keys only) and the full multi-map."""293 multi: dict[str, list[dict[str, Any]]] = defaultdict(list)294 for m in models:295 a = analyze_model_name(m["canonical_name"])296 if a.is_effort_variant or a.is_artifact:297 continue298 multi[variant_key(m["canonical_name"])].append(m)299 return {k: v[0]["id"] for k, v in multi.items() if len(v) == 1}, multi300301302async def step_variants(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:303 models = await _models(conn)304 index, _ = _variant_index(models)305 resolver = Resolver(conn, source_tier=2, variant_index=index)306 candidates = [m for m in models if _in_scope(scope, m["id"]) and analyze_model_name(m["canonical_name"]).is_effort_variant]307 idents = await _identifiers(conn, [m["id"] for m in candidates])308 official = {r["entity_id"] for r in await fetch_all(conn, """select distinct entity_id from claims where tier = 1 and status = 'current'309 and entity_id = any(cast(:ids as text[]))""", ids=[m["id"] for m in candidates])} if candidates else set()310 for m in candidates:311 a = analyze_model_name(m["canonical_name"])312 rep.bump("_effort_variants_seen")313 schemes = set(idents.get(m["id"], {}))314 if not schemes <= EVALUATOR_SCHEMES or (m["attributes"] or {}).get("hf_repo") or m["id"] in official:315 # known to an official source, a hub or a provider → a real model that happens to end in "thinking"/"high"; never folded316 rep.bump("_kept_real_model")317 continue318 ref = EntityRef(entity_type="model", name=m["canonical_name"], id=m["id"])319 folded = await resolver.resolve_variant(ref, org_id=m["organization_id"])320 if folded is None:321 hint = base_name(m["canonical_name"])322 attrs = m["attributes"] or {}323 if attrs.get("evaluation_variant_of_hint") == hint and m["identity_confidence"] == "medium":324 continue325 rep.bump("unresolved_flagged")326 rep.example(f"unresolved variant {m['slug']} (base '{hint}' not found) → identity medium + review")327 if apply:328 await _set_attributes(conn, m["id"], {"evaluation_variant_of_hint": hint}, source_id=await registry_source_id(conn))329 await execute(conn, "update entities set identity_confidence = 'medium' where id = :id", id=m["id"])330 await _review(conn, "variant_candidate", [m["id"]], f"'{m['canonical_name']}' looks like an evaluation-effort variant of '{hint}' but no such model exists",331 {"slug": m["slug"], "base": hint, "effort": a.effort}, apply=apply)332 continue333 cid, effort = folded334 rep.bump("folded")335 target = next((x for x in models if x["id"] == cid), None)336 rep.example(f"fold {m['slug']} → {target['slug'] if target else cid} {effort}")337 if apply:338 rows = await fetch_all(conn, "select id, config from benchmark_results where model_id = :m", m=m["id"])339 for r in rows:340 cfg = effort_config(m["canonical_name"], r["config"])341 if not _same(cfg, r["config"]):342 await execute(conn, "update benchmark_results set config = cast(:c as jsonb) where id = :id", c=jsonb(cfg), id=r["id"])343 rep.bump("results_reconfigured", len(rows))344 await merge_entities(conn, m["id"], cid, mode="merge", actor="canonicalize", note="evaluation-effort variant folded into its canonical model",345 payload={"step": "variants", "effort": effort, "variant_slug": m["slug"]})346347348# ---------------------------------------------------------------------------------------------- step: artifacts349async def step_artifacts(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:350 source_id = await registry_source_id(conn)351 models = await _models(conn)352 _, multi = _variant_index(models)353 by_id = {m["id"]: m for m in models}354 # only `quantized_from` names the packaged model; `derived_from`/`fine_tuned_from` point at a *base* model (a fine-tune is a new model)355 rel_rows = await fetch_all(conn, """select r.subject_id, r.object_id from relations r join entities o on o.id = r.object_id356 where r.predicate = 'quantized_from' and r.valid_to is null and o.merged_into is null and o.entity_type = 'model'""")357 base_of: dict[str, str] = {}358 for r in rel_rows:359 base_of.setdefault(r["subject_id"], r["object_id"])360 official_idents = await _identifiers(conn, [m["id"] for m in models if (m["attributes"] or {}).get("hf_repo")])361 for m in models:362 if not _in_scope(scope, m["id"]):363 continue364 attrs = m["attributes"] or {}365 hf_repo = attrs.get("hf_repo") if isinstance(attrs.get("hf_repo"), str) else None366 probe = hf_repo or m["canonical_name"]367 a = analyze_model_name(probe)368 a_name = analyze_model_name(m["canonical_name"]) if hf_repo else a369 repo_org = (a.repo_org or "").lower()370 # repo attributes (is_quantized / quant_format) describe a hub repository: without hf_repo they come from a provider endpoint371 # (OpenRouter lists the serving precision) and say nothing about the model's identity372 quant_attr = str(attrs.get("quant_format") or "").lower() if hf_repo else ""373 attr_quantized = attrs.get("is_quantized") is True and bool(hf_repo)374 has_token = bool(a.quant_formats or a.precision or a_name.quant_formats or a_name.precision)375 flagged = has_token or a.is_artifact or a_name.is_artifact or attr_quantized or bool(quant_attr) or repo_org in CONVERTER_ORGS376 if not flagged:377 continue378 # "official" = the repo belongs to the model's developer: the family's known publisher, or the entity's own organisation when that379 # organisation is not a redistributor (hub-derived entities are attached to the redistributor org, e.g. `bartowski`)380 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}381 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)382 vk_candidates = [c for c in multi.get(variant_key(probe), []) if c["id"] != m["id"]]383 cid: str | None = None384 if official:385 # the developer's own repo is the model (NVIDIA-Nemotron-3-Ultra-…-BF16, DeepSeek-R1 in native FP8) — it only becomes an386 # artifact when the same organisation also has the plain model entity (tencent/Hy-MT2-7B-GGUF next to Hy-MT2-7B) AND nobody387 # else (evaluators, providers) knows it under its own identity; otherwise the pair is a merge candidate for review388 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]389 if len(same_org) != 1 or not has_token:390 rep.bump("_official_checkpoint_kept_as_model")391 continue392 schemes = set(official_idents.get(m["id"], {})) - {"hf_repo"}393 if schemes:394 if await _review(conn, "merge_candidate", sorted([m["id"], same_org[0]["id"]]),395 f"'{m['canonical_name']}' and '{same_org[0]['canonical_name']}' look like one model published as two official checkpoints",396 {"slugs": [m["slug"], same_org[0]["slug"]], "step": "artifacts"}, apply=apply):397 rep.bump("review_official_checkpoint_pair")398 rep.example(f"review: {m['slug']} (official repo with dtype tag, known to {sorted(schemes)}) vs {same_org[0]['slug']}")399 continue400 cid = same_org[0]["id"]401 if attr_quantized or a.is_quantized or a_name.is_quantized or quant_attr in QUANT_ATTR_FORMATS:402 kind = "quantization"403 elif a.is_conversion or a_name.is_conversion or a.precision or quant_attr:404 kind = "conversion"405 else:406 kind = "packaging"407 if cid is None:408 cid = base_of.get(m["id"])409 if cid and (cid not in by_id or cid == m["id"]):410 cid = None411 if cid is None:412 candidates = vk_candidates413 if len(candidates) > 1:414 same_org = [c for c in candidates if c["organization_id"] == m["organization_id"]]415 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))]416 candidates = same_org or official_c or candidates417 if len(candidates) == 1:418 cid = candidates[0]["id"]419 if m["entity_type"] == "artifact" and m["canonical_id"] == cid:420 continue421 rep.bump("artifacts_marked" if cid else "artifacts_unresolved")422 rep.example(f"artifact[{kind}] {m['slug']} → {by_id[cid]['slug'] if cid else 'UNRESOLVED'}")423 if apply:424 await execute(conn, """update entities set entity_type = 'artifact', artifact_kind = coalesce(artifact_kind, :k), canonical_id = :c,425 identity_confidence = :ic, updated_at = now() where id = :id""",426 k=kind, c=cid, ic="high" if cid else "low", id=m["id"])427 if cid:428 await upsert_relation(conn, m["id"], "artifact_of", cid, {"artifact_kind": kind}, source_id=source_id)429 await record_decision(conn, m["id"], cid, "variant_of", actor="canonicalize", payload={"artifact_kind": kind, "step": "artifacts"})430 if not cid:431 await _review(conn, "unresolved_artifact", [m["id"]], f"'{m['canonical_name']}' is a {kind} artifact but its base model is unknown",432 {"slug": m["slug"], "hf_repo": hf_repo, "variant_key": variant_key(probe)}, apply=apply)433434435# ---------------------------------------------------------------------------------------------- step: families436async def step_families(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:437 source_id = await registry_source_id(conn)438 models = [m for m in await _models(conn) if _in_scope(scope, m["id"])]439 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))440 org_by_key: dict[str, str] = {}441 for o in orgs:442 for k in _org_lookup_keys(o):443 org_by_key.setdefault(k, o["id"])444 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")445 families: dict[str, dict[str, Any]] = {f["canonical_name"].lower(): f for f in fam_rows} # one family per label446 slugs_taken = {r["slug"] for r in await fetch_all(conn, "select slug from entities")}447 for m in models:448 label = family_release_hint(m["canonical_name"])449 if not label:450 rep.bump("_no_family_hint")451 continue452 root = family_hint(m["canonical_name"]) or label453 official = official_orgs(m["canonical_name"])454 org_id = next((org_by_key[k] for k in official if k in org_by_key), None) or m["organization_id"]455 fam = families.get(label.lower())456 if fam is None:457 # slug = slugify(label); on collision with any other entity (the model "gpt-5.5" itself) prefix with the organisation slug, then "family-"458 base_slug = slugify(label)459 org_slug = next((o["slug"] for o in orgs if o["id"] == org_id), None)460 candidates = [base_slug] + ([f"{org_slug}-{base_slug}"] if org_slug else []) + [f"family-{base_slug}"] + [f"family-{base_slug}-{n}" for n in range(2, 20)]461 slug = next(c for c in candidates if c not in slugs_taken)462 rep.bump("families_created")463 rep.example(f"family '{label}' ({slug}) root={root}")464 fam = {"id": new_id("model_family"), "slug": slug, "canonical_name": label, "organization_id": org_id, "_new": True}465 families[label.lower()] = fam466 slugs_taken.add(slug)467 if apply:468 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)469 values (:id, 'model_family', :n, :slug, 'active', :org, cast(:a as jsonb), cast(:p as jsonb), :fs, now(), 'high')""",470 id=fam["id"], n=label, slug=slug, org=org_id, a=jsonb({"family_root": root, "label": label}),471 p=jsonb({"family_root": {"source_id": source_id, "tier": 2, "extractor": "derived"}}), fs=m["first_seen_at"])472 await execute(conn, "insert into entity_aliases (entity_id, alias, alias_norm, kind) values (:e, :a, :n, 'alias') on conflict do nothing",473 e=fam["id"], a=label, n=normalize_alias(label))474 if m["family_id"] != fam["id"]:475 rep.bump("models_linked")476 if apply:477 await execute(conn, "update entities set family_id = :f, first_seen_at = first_seen_at where id = :id", f=fam["id"], id=m["id"])478 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"])479 await upsert_relation(conn, m["id"], "member_of_family", fam["id"], source_id=source_id)480 if not (m["attributes"] or {}).get("family"):481 rep.bump("family_attribute_set")482 if apply:483 await _set_attributes(conn, m["id"], {"family": label}, source_id=source_id)484485486# ---------------------------------------------------------------------------------------------- step: licenses487async def step_licenses(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:488 source_id = await registry_source_id(conn)489 rows = await fetch_all(conn, """select id, entity_type, canonical_name, slug, attributes from entities where merged_into is null490 and entity_type = any(cast(:t as text[])) and (attributes ? 'license' or attributes ? 'openness' or attributes ? 'hf_repo')491 order by first_seen_at, id""", t=list(LICENSED_TYPES))492 if scope is not None:493 rows = [r for r in rows if r["id"] in scope]494 lic_rows = await fetch_all(conn, "select id, slug from entities where entity_type = 'license'")495 license_entities: dict[str, str] = {r["slug"]: r["id"] for r in lic_rows}496 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")}497 # derived claims that a higher-tier source already contradicts: stored once as `conflicting`, never re-proposed498 contested: dict[tuple[str, str], Any] = {(r["entity_id"], r["property"]): r["value"] for r in await fetch_all(499 conn, "select entity_id, property, value from claims where extractor = 'derived' and status = 'conflicting'")}500 writer = _derived_writer(conn, source_id) if apply else None501 for r in rows:502 attrs = r["attributes"] or {}503 raw = attrs.get("license")504 key = normalize_license(raw) if isinstance(raw, str) else None505 if key is None and isinstance(attrs.get("license_key"), str) and attrs["license_key"] in LICENSES:506 key = attrs["license_key"]507 if isinstance(raw, str) and key is None:508 rep.bump("_license_unclassified")509 rep.example(f"unclassified licence {raw!r} on {r['slug']}") if len(rep.examples) < 4 else None510 if key:511 lslug = key.lower()512 lid = license_entities.get(lslug)513 if lid is None:514 info = LICENSES[key]515 lid = new_id("license")516 license_entities[lslug] = lid517 rep.bump("license_entities_created")518 if apply:519 await execute(conn, """insert into entities (id, entity_type, canonical_name, slug, status, description, attributes, provenance, identity_confidence)520 values (:id, 'license', :n, :slug, 'active', :d, cast(:a as jsonb), cast(:p as jsonb), 'high')""",521 id=lid, n=info.label, slug=lslug, d=f"{info.label} — {info.category} licence" + (f" (SPDX {info.spdx})" if info.spdx else ""),522 a=jsonb({**info.as_dict(), "license_key": key}), p=jsonb({"license_key": {"source_id": source_id, "tier": 2, "extractor": "derived"}}))523 await execute(conn, "insert into entity_aliases (entity_id, alias, alias_norm, kind) values (:e, :a, :n, 'alias') on conflict do nothing",524 e=lid, a=info.label, n=normalize_alias(info.label))525 for alias in (key, info.spdx or key):526 await execute(conn, "insert into entity_aliases (entity_id, alias, alias_norm, kind) values (:e, :a, :n, 'alias') on conflict do nothing",527 e=lid, a=alias, n=normalize_alias(alias))528 if (r["id"], lid) not in existing_rel:529 rep.bump("uses_license_relations")530 existing_rel.add((r["id"], lid))531 if apply:532 await upsert_relation(conn, r["id"], "uses_license", lid, source_id=source_id)533 if attrs.get("license_key") != key:534 rep.bump("license_key_attribute_set")535 if apply:536 await _set_attributes(conn, r["id"], {"license_key": key}, source_id=source_id)537 if r["entity_type"] not in ("model", "artifact"):538 continue539 # openness dimensions (derived claims, tier 2, never supersede a tier-1 statement)540 openness_raw = attrs.get("openness")541 openness_now = normalize_openness(openness_raw) if isinstance(openness_raw, str) else None542 weights: bool | None = None543 if attrs.get("hf_repo") or attrs.get("model_card_url") or attrs.get("weights_url") or (openness_now or "").startswith(("open", "restricted")):544 weights = True545 elif openness_now == "proprietary":546 weights = False547 if weights is None:548 rep.bump("_openness_unknown_skipped")549 continue550 dims = openness_dimensions(weights_available=weights, license_key=key)551 derived = derive_openness(dims, license_key=key)552 wanted: dict[str, Any] = {"weights_available": dims["weights_available"]}553 for k in ("commercial_use_allowed", "redistribution_allowed", "derivatives_allowed"):554 if dims[k] is not None:555 wanted[k] = dims[k]556 if derived != "unknown":557 wanted["openness"] = derived558 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)}559 if not changed:560 continue561 rep.bump("openness_claims_written", len(changed))562 if "openness" in changed:563 rep.bump("openness_category_changed")564 rep.example(f"{r['slug']}: openness {openness_raw!r} → {derived} (licence {key})")565 if writer is not None:566 facts = Facts()567 ref = EntityRef(entity_type=r["entity_type"], name=r["canonical_name"], id=r["id"])568 for k, v in changed.items():569 facts.claim(ref, k, v, confidence="high")570 await writer.write(facts)571 if writer is not None:572 rep.bump("_conflicting_derived_claims", writer.stats.conflicts)573574575# ---------------------------------------------------------------------------------------------- step: taxonomy576async def step_taxonomy(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:577 props = sorted(TAXONOMY_PROPERTIES)578 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_id579 where c.status = 'current' and c.property = any(cast(:p as text[]))""", p=props)580 if scope is not None:581 rows = [r for r in rows if r["entity_id"] in scope]582 mappings: dict[tuple[str, str], str | None] = {}583584 def note(maps: list[tuple[str, str, str | None]]) -> None:585 for d, rw, cn in maps:586 if cn is not None and rw == cn:587 continue # identity mapping: nothing to learn588 mappings[(d, rw)] = cn if cn is not None else mappings.get((d, rw))589590 for r in rows:591 canon, raw, maps = normalize_property(r["entity_type"], r["property"], r["value"])592 note(maps)593 if _same(canon, r["value"]):594 continue595 rep.bump(f"claims_normalized:{r['property']}")596 rep.example(f"{r['entity_type']} {r['property']}: {json.dumps(r['value'], ensure_ascii=False)[:40]} → {json.dumps(canon, ensure_ascii=False)[:40]}")597 if apply:598 keep_raw = r["value_raw"] or (r["value"] if isinstance(r["value"], str) else json.dumps(r["value"], ensure_ascii=False))599 keep_raw = keep_raw if keep_raw != canon else None600 await execute(conn, "update claims set value = cast(:v as jsonb), value_text = :vt, value_raw = :raw where id = :id",601 v=jsonb(canon), vt=canon[:2000] if isinstance(canon, str) else None, raw=keep_raw, id=r["id"])602 attrs = {r["property"]: canon}603 if keep_raw:604 attrs[f"{r['property']}_raw"] = keep_raw605 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"])606 if r["property"] == "status" and isinstance(canon, str):607 await execute(conn, "update entities set status = :s where id = :id", s=canon[:40], id=r["entity_id"])608 # attributes without a current claim (seeded/imported values)609 for prop in props:610 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 ? :p611 and not exists (select 1 from claims c where c.entity_id = e.id and c.property = :p and c.status = 'current')""", p=prop)612 for e in ents:613 if not _in_scope(scope, e["id"]):614 continue615 canon, raw, maps = normalize_property(e["entity_type"], prop, e["value"])616 note(maps)617 if _same(canon, e["value"]):618 continue619 rep.bump(f"attributes_normalized:{prop}")620 if apply:621 keep_raw = e["value"] if isinstance(e["value"], str) else json.dumps(e["value"], ensure_ascii=False)622 attrs = {prop: canon, f"{prop}_raw": keep_raw}623 await execute(conn, "update entities set attributes = attributes || cast(:a as jsonb), updated_at = now() where id = :id", a=jsonb(attrs), id=e["id"])624 if prop == "status" and isinstance(canon, str):625 await execute(conn, "update entities set status = :s where id = :id", s=canon[:40], id=e["id"])626 existing = {(m["domain"], m["raw"]): m["canonical"] for m in await fetch_all(conn, "select domain, raw, canonical from taxonomy_mappings")}627 for (domain, raw_full), canon in mappings.items():628 raw = raw_full[:300]629 if (domain, raw) in existing and (existing[(domain, raw)] == canon or canon is None):630 continue631 rep.bump("taxonomy_mappings_upserted")632 if apply:633 await execute(conn, """insert into taxonomy_mappings (domain, raw, canonical) values (:d, :r, :c)634 on conflict (domain, raw) do update set canonical = coalesce(excluded.canonical, taxonomy_mappings.canonical), last_seen_at = now()""",635 d=domain, r=raw, c=canon)636 unknown = sorted({f"{d}:{r}" for (d, r), c in mappings.items() if c is None})637 if unknown:638 rep.notes.append(f"{len(unknown)} raw value(s) without canonical mapping kept as-is: {', '.join(unknown[:15])}{'…' if len(unknown) > 15 else ''}")639640641# ---------------------------------------------------------------------------------------------- step: results642async def step_results(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:643 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,644 r.is_current, s.key as source_key from benchmark_results r left join sources s on s.id = r.source_id""")645 if scope is not None:646 rows = [r for r in rows if r["model_id"] in scope]647 for r in rows:648 cfg = r["config"] or {}649 ck = bench_ontology.config_key(cfg, r["metric"])650 trust = r["trust_level"] or bench_ontology.trust_level(r["source_key"], cfg, extractor=r["extractor"] or "deterministic")651 variant = r["variant"] or bench_ontology.variant_from_config(cfg)652 rg = r["run_group"] or bench_ontology.run_group_from_config(cfg)653 current = r["is_current"] and r["valid_to"] is None654 if (ck, trust, variant, rg, current) == (r["config_key"], r["trust_level"], r["variant"], r["run_group"], r["is_current"]):655 continue656 rep.bump("results_backfilled")657 if apply:658 await execute(conn, "update benchmark_results set config_key = :ck, trust_level = :t, variant = :v, run_group = :rg, is_current = :cur where id = :id",659 ck=ck, t=trust, v=variant, rg=rg, cur=current, id=r["id"])660 if scope is None:661 closed = await enforce_current_results(conn, dry_run=not apply)662 else:663 closed = sum([await enforce_current_results(conn, model_id=mid, dry_run=not apply) for mid in sorted(scope)])664 if closed:665 rep.bump("older_run_rows_closed", closed)666 live = await fetch_one(conn, "select count(*) as n from benchmark_results where valid_to is null and is_current")667 rep.notes.append(f"current benchmark results after step: {int(live['n']) if live else 0}")668669670# ---------------------------------------------------------------------------------------------- step: events671async def step_events(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:672 lag = timedelta(days=BACKFILL_LAG_DAYS)673 runs = await fetch_all(conn, """select connector_name, started_at, finished_at from connector_runs where status in ('success', 'unchanged', 'suspect', 'released')674 order by connector_name, started_at""")675 second_start: dict[str, datetime | None] = {}676 per: dict[str, list[dict[str, Any]]] = defaultdict(list)677 for r in runs:678 per[r["connector_name"]].append(r)679 for name, lst in per.items():680 second_start[name] = lst[1]["started_at"] if len(lst) > 1 else None681 before = await fetch_one(conn, """select count(*) as n from change_events where is_backfill = false and event_type <> 'DOCUMENT_CHANGED'682 and observed_at > now() - interval '24 hours'""")683 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,684 e.attributes->>'release_date' as release_date, e.attributes->>'published_at' as published_at685 from change_events ev left join entities e on e.id = ev.entity_id""")686 from aiatlas.sdk.extract.dates import parse_datetime687688 if scope is not None:689 rows = [r for r in rows if r["entity_id"] in scope]690 for ev in rows:691 obs = ev["observed_at"]692 bf = False693 if ev["connector_name"] in STRUCTURAL_CONNECTORS:694 bf = True # merges, folds, derived corrections: bookkeeping about our own data, never news695 elif ev["connector_name"] and ev["connector_name"] in per:696 s2 = second_start.get(ev["connector_name"])697 if s2 is None or obs < s2:698 bf = True699 elif ev["connector_name"] and ev["connector_name"] not in per:700 bf = True # connector without any successful run yet → initial load701 if not bf and ev["effective_at"] is not None and ev["effective_at"] < obs - lag:702 bf = True703 if not bf and ev["event_type"].startswith("NEW_"):704 hint = parse_datetime(ev["release_date"]) if ev["release_date"] else (parse_datetime(ev["published_at"]) if ev["published_at"] else None)705 if hint is not None:706 hint = hint if hint.tzinfo else hint.replace(tzinfo=UTC)707 if hint < obs - lag:708 bf = True709 gk = group_key_for(ev["event_type"], ev["entity_id"], ev["effective_at"], obs)710 if bf == ev["is_backfill"] and gk == ev["group_key"]:711 continue712 if bf != ev["is_backfill"]:713 rep.bump("backfill_flag_set" if bf else "backfill_flag_cleared")714 if gk != ev["group_key"]:715 rep.bump("group_key_set")716 if apply:717 await execute(conn, "update change_events set is_backfill = :bf, group_key = :gk where id = :id", bf=bf, gk=gk, id=ev["id"])718 after = await fetch_one(conn, """select count(*) as n from change_events where is_backfill = false and event_type <> 'DOCUMENT_CHANGED'719 and observed_at > now() - interval '24 hours'""")720 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}"721 + ("" if apply else " (dry-run: after = before)"))722723724# ---------------------------------------------------------------------------------------------- step: anomalies725async def _collect_anomalies(conn: AsyncConnection) -> list[Anomaly]:726 found: list[Anomaly] = []727 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"):728 found.extend(check_model(r["id"], r["canonical_name"], r["attributes"] or {}))729 for r in await fetch_all(conn, "select id, canonical_name, attributes from entities where entity_type = 'hardware' and merged_into is null"):730 found.extend(check_hardware(r["id"], r["canonical_name"], r["attributes"] or {}))731 for r in await fetch_all(conn, """select p.*, m.canonical_name as model_name, v.canonical_name as provider_name from prices p732 join entities m on m.id = p.model_id join entities v on v.id = p.provider_id where p.valid_to is null"""):733 found.extend(check_price(r))734 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,735 b.canonical_name as benchmark_name, m.attributes->>'release_date' as model_release_date736 from benchmark_results r join entities m on m.id = r.model_id join entities b on b.id = r.benchmark_id737 where r.valid_to is null and r.is_current"""):738 found.extend(check_result(r))739 return found740741742async def step_anomalies(conn: AsyncConnection, rep: StepReport, apply: bool, *, scope: set[str] | None = None) -> None:743 found = await _collect_anomalies(conn)744 if scope is not None:745 found = [a for a in found if a.entity_id in scope]746 keys: dict[str, Anomaly] = {}747 for a in found:748 keys.setdefault(a.dedupe_key[:400], a)749 existing = {r["dedupe_key"]: r for r in await fetch_all(conn, "select dedupe_key, status, severity from anomalies")}750 for key, a in keys.items():751 rep.bump(f"_by_severity:{a.severity}")752 prev = existing.get(key)753 if prev is None:754 rep.bump("anomalies_opened")755 rep.example(f"[{a.severity}] {a.message}")756 elif prev["status"] == "resolved":757 rep.bump("anomalies_reopened")758 if apply:759 await record(conn, a)760 stale = [k for k, r in existing.items() if r["status"] == "open" and k not in keys] if scope is None else []761 if stale:762 rep.bump("anomalies_resolved", len(stale))763 if apply:764 await execute(conn, """update anomalies set status = 'resolved', resolved_at = now(), resolution = 'check no longer fires'765 where status = 'open' and dedupe_key = any(cast(:k as text[]))""", k=stale)766767768_STEP_FUNCTIONS = {769 "duplicates": step_duplicates, "variants": step_variants, "artifacts": step_artifacts, "families": step_families, "licenses": step_licenses,770 "taxonomy": step_taxonomy, "results": step_results, "events": step_events, "anomalies": step_anomalies,771}772773774# ---------------------------------------------------------------------------------------------- quarantine release / discard775async def release_quarantine(conn: AsyncConnection, quarantine_id: str, *, actor: str = "admin") -> dict[str, Any]:776 """Write the held facts of a quarantined run exactly as the connector would have (same source, snapshot, tier, run id)."""777 from aiatlas.connectors import get as get_connector778779 q = await fetch_one(conn, "select * from quarantined_runs where id = :id", id=quarantine_id)780 if not q:781 raise LookupError(f"quarantined run {quarantine_id} not found")782 if q["status"] != "pending":783 raise ValueError(f"quarantined run {quarantine_id} is already {q['status']}")784 connector = get_connector(q["connector_name"])785 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"])786 totals: dict[str, int] = defaultdict(int)787 for item in q["facts"] or []:788 facts = facts_from_json(item["facts"])789 fetched_at = datetime.fromisoformat(item["fetched_at"])790 writer = FactWriter(conn, source_id=state["source_id"] if state else None, snapshot_id=item.get("snapshot_id"), source_url=item.get("source_url"),791 tier=connector.tier, connector_name=q["connector_name"], extractor="deterministic", extractor_version=connector.parser_version,792 observed_at=fetched_at, run_id=q["run_id"], source_key=state["source_key"] if state else None)793 ws = await writer.write(facts)794 for k, v in ws.as_dict().items():795 totals[k] += v796 main = facts.document_entity797 if main and main.id is None:798 await writer.resolver.resolve(main)799 if main and main.id and item.get("doc_id"):800 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"])801 if item.get("snapshot_id"):802 await execute(conn, "update snapshots set processing_status = 'extracted' where id = :id", id=item["snapshot_id"])803 await execute(conn, "update quarantined_runs set status = 'released', resolved_at = now(), resolved_by = :who where id = :id", who=actor, id=quarantine_id)804 await execute(conn, "update connector_runs set status = 'released' where id = :r", r=q["run_id"])805 await execute(conn, "update review_queue set status = 'approved', resolved_at = now() where dedupe_key = :d", d=f"quarantine:{quarantine_id}")806 observed = (q["stats"] or {}).get("observed")807 if observed and (q["stats"] or {}).get("full_extraction"):808 baseline = (await fetch_one(conn, "select baseline from connectors where name = :n", n=q["connector_name"]) or {}).get("baseline")809 nb = connector._next_baseline(baseline, observed) # noqa: SLF001810 await execute(conn, "update connectors set baseline = cast(:b as jsonb) where name = :n", b=jsonb(nb), n=q["connector_name"])811 await audit(conn, "quarantine.release", quarantine_id, {"connector": q["connector_name"], "run_id": q["run_id"], **totals}, actor=actor)812 return {"id": quarantine_id, "connector": q["connector_name"], **totals}813814815async def discard_quarantine(conn: AsyncConnection, quarantine_id: str, *, actor: str = "admin", note: str | None = None) -> dict[str, Any]:816 q = await fetch_one(conn, "select id, run_id, connector_name, status, facts from quarantined_runs where id = :id", id=quarantine_id)817 if not q:818 raise LookupError(f"quarantined run {quarantine_id} not found")819 if q["status"] != "pending":820 raise ValueError(f"quarantined run {quarantine_id} is already {q['status']}")821 await execute(conn, "update quarantined_runs set status = 'discarded', resolved_at = now(), resolved_by = :who where id = :id", who=actor, id=quarantine_id)822 await execute(conn, "update connector_runs set status = 'discarded' where id = :r", r=q["run_id"])823 await execute(conn, "update review_queue set status = 'rejected', resolved_at = now() where dedupe_key = :d", d=f"quarantine:{quarantine_id}")824 for item in q["facts"] or []:825 if item.get("snapshot_id"):826 await execute(conn, "update snapshots set processing_status = 'discarded' where id = :id", id=item["snapshot_id"])827 await audit(conn, "quarantine.discard", quarantine_id, {"connector": q["connector_name"], "run_id": q["run_id"], "note": note}, actor=actor)828 return {"id": quarantine_id, "connector": q["connector_name"], "status": "discarded"}829830831async def list_quarantine(conn: AsyncConnection, *, status: str = "pending", limit: int = 50) -> list[dict[str, Any]]:832 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 documents833 from quarantined_runs where (cast(:s as text) = '' or status = :s) order by created_at desc limit :n""", s=status or "", n=limit)834835836__all__ = ["CANON_VERSION", "STEPS", "Report", "StepReport", "canonicalize", "discard_quarantine", "list_quarantine", "release_quarantine"]837