HTML 77.2%
TypeScript 10.5%
Python 9.6%
JavaScript 2.5%
1"""Canonicalization steps on synthetic rows, inside a rolled-back transaction on the dev database. Every step must be idempotent:2a second call right after `apply=True` reports zero changes."""3from __future__ import annotations45import uuid6from collections.abc import AsyncIterator7from datetime import UTC, datetime, timedelta89import pytest10from sqlalchemy.ext.asyncio import AsyncConnection1112from aiatlas import db13from aiatlas.db import execute, fetch_all, fetch_one14from aiatlas.sdk.facts import Facts15from aiatlas.sdk.writer import FactWriter16from aiatlas.services import canonical as canon17from aiatlas.services.merge import enforce_current_results, merge_entities181920@pytest.fixture21async def conn() -> AsyncIterator[AsyncConnection]:22 async with db.engine().connect() as c:23 trans = await c.begin()24 try:25 yield c26 finally:27 await trans.rollback()28 await db.dispose()293031def _tag() -> str:32 return uuid.uuid4().hex[:8]333435async def _write(conn: AsyncConnection, facts: Facts, *, tier: int = 2, source_key: str = "", observed_at: datetime | None = None) -> None:36 await FactWriter(conn, source_id=None, snapshot_id=None, source_url="https://example.com/x", tier=tier, connector_name="test", run_id="run_test",37 source_key=source_key, observed_at=observed_at).write(facts)383940async def _run(step, conn: AsyncConnection, apply: bool, *ids: str) -> canon.StepReport: # type: ignore[no-untyped-def]41 """Run one step scoped to the test's own entities (the dev database is shared with other work)."""42 rep = canon.StepReport(name=step.__name__)43 await step(conn, rep, apply, scope=set(ids) if ids else None)44 return rep454647# ---------------------------------------------------------------------------------------------- variants48async def test_variant_folding(conn: AsyncConnection) -> None:49 t = _tag()50 f = Facts()51 org = f.entity("company", f"Zorg {t}")52 base = f.entity("model", f"Zeta {t}", organization=org, identifiers={"artificial_analysis": f"zeta-{t}"})53 b = f.entity("benchmark", f"GPQA {t}", identifiers={"registry_benchmark": f"gpqa-{t}"})54 f.result(model=base, benchmark=b, score=70.0, metric="accuracy", unit="%", config={"evaluator": "AA", "index_version": "4.3", "aa_slug": f"zeta-{t}"})55 await _write(conn, f, source_key="artificialanalysis.ai")56 # the variant was created earlier as its own entity (pre-upgrade behaviour): insert it directly57 await execute(conn, """insert into entities (id, entity_type, canonical_name, slug, status) values (:id, 'model', :n, :slug, 'active')""",58 id=f"model_test{t}v", n=f"zeta-{t}-xhigh", slug=f"zeta-{t}-xhigh")59 await execute(conn, "insert into entity_identifiers (entity_id, scheme, value) values (:e, 'artificial_analysis', :v)", e=f"model_test{t}v", v=f"zeta-{t}-xhigh")60 await execute(conn, "insert into entity_aliases (entity_id, alias, alias_norm) values (:e, :a, :n)", e=f"model_test{t}v", a=f"zeta-{t}-xhigh", n=f"zeta{t}xhigh")61 await execute(conn, """insert into benchmark_results (id, model_id, benchmark_id, score, metric, unit, config, dedupe_key, tier)62 values (:id, :m, :b, 75.0, 'accuracy', '%', cast(:c as jsonb), :d, 2)""",63 id=f"res_test{t}", m=f"model_test{t}v", b=b.id, c='{"evaluator": "AA", "index_version": "4.3", "aa_slug": "zeta-%s-xhigh"}' % t, d=f"test{t}")64 rep = await _run(canon.step_variants, conn, False, base.id, f"model_test{t}v")65 assert rep.counts.get("folded", 0) >= 1 and not await fetch_one(conn, "select 1 from entities where id = :id and merged_into is not null", id=f"model_test{t}v")66 rep = await _run(canon.step_variants, conn, True, base.id, f"model_test{t}v")67 assert rep.counts["folded"] >= 168 v = await fetch_one(conn, "select merged_into, status from entities where id = :id", id=f"model_test{t}v")69 assert v["merged_into"] == base.id and v["status"] == "merged"70 rows = await fetch_all(conn, "select score, config, model_id, is_current from benchmark_results where model_id = :m order by score", m=base.id)71 assert [r["score"] for r in rows] == [70.0, 75.0]72 assert rows[1]["config"]["reasoning_effort"] == "xhigh" and rows[1]["config"]["aa_variant_slug"] == f"zeta-{t}-xhigh"73 assert all(r["is_current"] for r in rows) # same run group, different condition → both current74 assert await fetch_one(conn, "select 1 from resolution_decisions where a_id = :a and b_id = :b and decision = 'merge' and applied", a=f"model_test{t}v", b=base.id)75 again = await _run(canon.step_variants, conn, True, base.id, f"model_test{t}v")76 assert again.counts.get("folded", 0) == 0 and again.counts.get("unresolved_flagged", 0) == 0777879async def test_variant_with_official_identifier_is_not_folded(conn: AsyncConnection) -> None:80 t = _tag()81 f = Facts()82 org = f.entity("company", f"Zorg {t}")83 f.entity("model", f"Zeta {t}", organization=org, identifiers={"artificial_analysis": f"zeta-{t}"})84 real = f.entity("model", f"Zeta {t} Thinking", organization=org, identifiers={"hf_repo": f"zorg/Zeta-{t}-Thinking"})85 await _write(conn, f)86 rep = await _run(canon.step_variants, conn, True, real.id)87 assert rep.counts.get("_kept_real_model", 0) >= 188 assert (await fetch_one(conn, "select merged_into from entities where id = :id", id=real.id))["merged_into"] is None899091# ---------------------------------------------------------------------------------------------- artifacts92async def test_artifact_detection(conn: AsyncConnection) -> None:93 t = _tag()94 f = Facts()95 org = f.entity("company", f"Zorg {t}", attributes={"hf_org": f"zorg{t}"})96 base = f.entity("model", f"Zeta-{t}-9B", organization=org, identifiers={"hf_repo": f"zorg{t}/Zeta-{t}-9B"}, attributes={"hf_repo": f"zorg{t}/Zeta-{t}-9B"})97 third = f.entity("model", f"unsloth/Zeta-{t}-9B-GGUF", organization=f.entity("organization", "unsloth-test", attributes={"hf_org": "unsloth"}),98 identifiers={"hf_repo": f"unsloth/Zeta-{t}-9B-GGUF"}, attributes={"hf_repo": f"unsloth/Zeta-{t}-9B-GGUF", "quant_format": "gguf", "is_quantized": True})99 official_fp8 = f.entity("model", f"Zeta-{t}-9B-FP8", organization=org, identifiers={"hf_repo": f"zorg{t}/Zeta-{t}-9B-FP8"}, attributes={"hf_repo": f"zorg{t}/Zeta-{t}-9B-FP8"})100 native = f.entity("model", f"Zeta-{t}-Big", organization=org, identifiers={"hf_repo": f"zorg{t}/Zeta-{t}-Big"}, attributes={"hf_repo": f"zorg{t}/Zeta-{t}-Big", "quant_format": "fp8", "is_quantized": True})101 endpoint = f.entity("model", f"Zeta {t} Served", organization=org, identifiers={"openrouter": f"zorg/zeta-{t}-served"}, attributes={"quant_format": "fp8"})102 await _write(conn, f)103 ids = [base.id, third.id, official_fp8.id, native.id, endpoint.id]104 rep = await _run(canon.step_artifacts, conn, True, *ids)105 assert rep.counts.get("artifacts_marked", 0) >= 2106 rows = {r["id"]: r for r in await fetch_all(conn, "select id, entity_type, artifact_kind, canonical_id, identity_confidence from entities where id = any(cast(:ids as text[]))",107 ids=[base.id, third.id, official_fp8.id, native.id, endpoint.id])}108 assert rows[base.id]["entity_type"] == "model"109 assert rows[third.id]["entity_type"] == "artifact" and rows[third.id]["artifact_kind"] == "quantization" and rows[third.id]["canonical_id"] == base.id110 assert rows[official_fp8.id]["entity_type"] == "artifact" and rows[official_fp8.id]["canonical_id"] == base.id # same org has the plain model111 assert rows[native.id]["entity_type"] == "model" # developer repo in native FP8 with no plain sibling = the model112 assert rows[endpoint.id]["entity_type"] == "model" # provider endpoint precision says nothing about identity113 assert await fetch_one(conn, "select 1 from relations where subject_id = :a and predicate = 'artifact_of' and object_id = :m and valid_to is null", a=third.id, m=base.id)114 # slug, identifiers and claims untouched115 assert (await fetch_one(conn, "select slug from entities where id = :a", a=third.id))["slug"].startswith("unsloth")116 assert await fetch_one(conn, "select 1 from entity_identifiers where entity_id = :a", a=third.id)117 again = await _run(canon.step_artifacts, conn, True, *ids)118 assert again.counts.get("artifacts_marked", 0) == 0 and again.counts.get("artifacts_unresolved", 0) == 0119120121async def test_unresolved_artifact_gets_review(conn: AsyncConnection) -> None:122 t = _tag()123 f = Facts()124 lone = f.entity("model", f"bartowski/Omega-{t}-70B-GGUF", identifiers={"hf_repo": f"bartowski/Omega-{t}-70B-GGUF"}, attributes={"hf_repo": f"bartowski/Omega-{t}-70B-GGUF"})125 await _write(conn, f)126 rep = await _run(canon.step_artifacts, conn, True, lone.id)127 assert rep.counts.get("artifacts_unresolved", 0) >= 1128 row = await fetch_one(conn, "select entity_type, canonical_id, identity_confidence from entities where id = :id", id=lone.id)129 assert row["entity_type"] == "artifact" and row["canonical_id"] is None and row["identity_confidence"] == "low"130 assert await fetch_one(conn, "select 1 from review_queue where kind = 'unresolved_artifact' and :id = any(entity_ids)", id=lone.id)131132133# ---------------------------------------------------------------------------------------------- families134async def test_families(conn: AsyncConnection) -> None:135 t = _tag()136 f = Facts()137 org = f.entity("company", f"Meta test {t}")138 v = str(int(t, 16) % 900 + 100)139 a = f.entity("model", f"Llama {v}.1 9B Test", organization=org)140 b = f.entity("model", f"Llama {v}.1 70B Test", organization=org)141 await _write(conn, f)142 rep = await _run(canon.step_families, conn, True, a.id, b.id)143 assert rep.counts.get("families_created", 0) >= 1 and rep.counts.get("models_linked", 0) >= 2144 rows = await fetch_all(conn, "select e.family_id, fam.canonical_name, fam.entity_type, fam.attributes from entities e join entities fam on fam.id = e.family_id where e.id in (:a, :b)", a=a.id, b=b.id)145 assert len(rows) == 2 and len({r["family_id"] for r in rows}) == 1 and rows[0]["canonical_name"] == f"Llama {v}.1" and rows[0]["entity_type"] == "model_family"146 assert rows[0]["attributes"]["family_root"] == "Llama"147 assert (await fetch_one(conn, "select attributes->>'family' as f from entities where id = :a", a=a.id))["f"] == f"Llama {v}.1"148 again = await _run(canon.step_families, conn, True, a.id, b.id)149 assert again.changes == 0150151152# ---------------------------------------------------------------------------------------------- licences / openness153async def test_licenses_and_openness(conn: AsyncConnection) -> None:154 t = _tag()155 f = Facts()156 m = f.entity("model", f"Zeta {t}", identifiers={"hf_repo": f"zorg/zeta-{t}"}, attributes={"hf_repo": f"zorg/zeta-{t}", "license": "Apache 2.0"})157 llama = f.entity("model", f"Zeta {t} L", identifiers={"hf_repo": f"zorg/zeta-{t}-l"}, attributes={"hf_repo": f"zorg/zeta-{t}-l", "license": "llama3.1", "openness": "restricted"})158 closed = f.entity("model", f"Zeta {t} API", attributes={"openness": "proprietary"})159 await _write(conn, f)160 rep = await _run(canon.step_licenses, conn, True, m.id, llama.id, closed.id)161 assert rep.counts.get("uses_license_relations", 0) >= 2162 lic = await fetch_one(conn, "select id, attributes from entities where entity_type = 'license' and slug = 'apache-2.0'")163 assert lic and lic["attributes"]["commercial_use"] is True164 assert await fetch_one(conn, "select 1 from relations where subject_id = :m and predicate = 'uses_license' and object_id = :l", m=m.id, l=lic["id"])165 attrs = (await fetch_one(conn, "select attributes from entities where id = :m", m=m.id))["attributes"]166 assert attrs["openness"] == "open-weights" and attrs["weights_available"] is True and attrs["commercial_use_allowed"] is True and attrs["license_key"] == "Apache-2.0"167 attrs = (await fetch_one(conn, "select attributes from entities where id = :m", m=llama.id))["attributes"]168 assert attrs["openness"] == "restricted-weights" and attrs["openness_raw"] == "restricted" and attrs["license_key"] == "Llama-3.1-Community"169 attrs = (await fetch_one(conn, "select attributes from entities where id = :m", m=closed.id))["attributes"]170 assert attrs["weights_available"] is False and attrs["openness"] == "proprietary"171 claim = await fetch_one(conn, "select tier, extractor, source_id from claims where entity_id = :m and property = 'weights_available' and status = 'current'", m=m.id)172 assert claim["tier"] == 2 and claim["extractor"] == "derived"173 again = await _run(canon.step_licenses, conn, True, m.id, llama.id, closed.id)174 assert again.changes == 0175176177async def test_derived_openness_never_supersedes_tier1(conn: AsyncConnection) -> None:178 t = _tag()179 f = Facts()180 m = f.entity("model", f"Zeta {t}", identifiers={"hf_repo": f"zorg/zeta-{t}"}, attributes={"hf_repo": f"zorg/zeta-{t}", "license": "llama3.1", "openness": "open-weights"})181 await _write(conn, f, tier=1)182 await _run(canon.step_licenses, conn, True, m.id)183 cur = await fetch_one(conn, "select value, tier from claims where entity_id = :m and property = 'openness' and status = 'current'", m=m.id)184 assert cur["value"] == "open-weights" and cur["tier"] == 1185 assert await fetch_one(conn, "select 1 from claims where entity_id = :m and property = 'openness' and status = 'conflicting' and extractor = 'derived'", m=m.id)186187188# ---------------------------------------------------------------------------------------------- taxonomy in place189async def test_taxonomy_in_place(conn: AsyncConnection) -> None:190 t = _tag()191 f = Facts()192 hw = f.entity("hardware", f"Box {t}")193 await _write(conn, f)194 # legacy encodings inserted directly (pre-upgrade rows)195 await execute(conn, """insert into claims (id, entity_id, property, value, value_text, tier, status) values (:id, :e, 'kind', '"computer"', 'computer', 1, 'current')""",196 id=f"claim_test{t}", e=hw.id)197 await execute(conn, """update entities set attributes = attributes || '{"kind": "computer", "status": "limited availability"}' where id = :e""", e=hw.id)198 rep = await _run(canon.step_taxonomy, conn, True, hw.id)199 assert rep.counts.get("claims_normalized:kind", 0) >= 1 and rep.counts.get("attributes_normalized:status", 0) >= 1200 c = await fetch_one(conn, "select value, value_raw from claims where id = :id", id=f"claim_test{t}")201 assert c["value"] == "system" and c["value_raw"] == "computer"202 row = await fetch_one(conn, "select attributes, status from entities where id = :e", e=hw.id)203 assert row["attributes"]["kind"] == "system" and row["attributes"]["kind_raw"] == "computer" and row["attributes"]["status"] == "limited-availability" and row["status"] == "limited-availability"204 assert (await fetch_one(conn, "select canonical from taxonomy_mappings where domain = 'hardware_kind' and raw = 'computer'"))["canonical"] == "system"205 again = await _run(canon.step_taxonomy, conn, True, hw.id)206 assert again.changes == 0207208209# ---------------------------------------------------------------------------------------------- results210async def test_results_backfill_and_current(conn: AsyncConnection) -> None:211 t = _tag()212 f = Facts()213 m = f.entity("model", f"Zeta {t}")214 b = f.entity("benchmark", f"LiveBench {t}", identifiers={"registry_benchmark": f"lb-{t}"})215 await _write(conn, f)216 now = datetime.now(UTC)217 for i, rel in enumerate(["2026-04-01", "2026-05-01", "2026-06-25"]):218 await execute(conn, """insert into benchmark_results (id, model_id, benchmark_id, score, metric, unit, config, dedupe_key, tier, observed_at)219 values (:id, :m, :b, :s, 'global_average', '%', cast(:c as jsonb), :d, 2, :o)""",220 id=f"res_t{t}{i}", m=m.id, b=b.id, s=50 + i, c='{"release": "%s"}' % rel, d=f"t{t}{i}", o=now + timedelta(seconds=i))221 rep = await _run(canon.step_results, conn, True, m.id)222 assert rep.counts["results_backfilled"] >= 3 and rep.counts["older_run_rows_closed"] >= 2223 rows = await fetch_all(conn, "select score, is_current, valid_to, run_group, config_key, trust_level from benchmark_results where model_id = :m order by score", m=m.id)224 assert [(r["score"], r["is_current"], r["valid_to"] is None) for r in rows] == [(50, False, False), (51, False, False), (52, True, True)]225 assert len({r["config_key"] for r in rows}) == 1 and rows[0]["run_group"] == "2026-04-01" and rows[0]["trust_level"] == "unverified"226 assert await enforce_current_results(conn, model_id=m.id, dry_run=True) == 0227 again = await _run(canon.step_results, conn, True, m.id)228 assert again.changes == 0229230231# ---------------------------------------------------------------------------------------------- events232async def test_events_backfill_classification(conn: AsyncConnection) -> None:233 t = _tag()234 f = Facts()235 m = f.entity("model", f"Zeta {t}")236 await _write(conn, f, observed_at=datetime.now(UTC))237 old = datetime.now(UTC) - timedelta(days=30)238 await execute(conn, """insert into change_events (id, entity_id, event_type, category, summary, importance, observed_at, effective_at, connector_name, dedupe_key)239 values (:id, :e, 'RELEASE', 'model', 'old release', 2, now(), :eff, 'curation', :d)""", id=f"evt_t{t}", e=m.id, eff=old, d=f"t{t}")240 rep = await _run(canon.step_events, conn, True, m.id)241 assert rep.counts.get("backfill_flag_set", 0) >= 1242 ev = await fetch_one(conn, "select is_backfill, group_key from change_events where id = :id", id=f"evt_t{t}")243 assert ev["is_backfill"] is True and ev["group_key"] == f"release:{m.id}:{old:%Y-%m}"244 again = await _run(canon.step_events, conn, True, m.id)245 assert again.changes == 0246247248# ---------------------------------------------------------------------------------------------- merge modes & duplicates249async def test_merge_modes(conn: AsyncConnection) -> None:250 t = _tag()251 f = Facts()252 fam = f.entity("model_family", f"Zeta fam {t}")253 a = f.entity("model", f"Zeta {t}")254 b = f.entity("model", f"Zeta {t} dup")255 c = f.entity("model", f"Zeta {t} FP8")256 await _write(conn, f)257 res = await merge_entities(conn, c.id, a.id, mode="variant", payload={"artifact_kind": "conversion"})258 row = await fetch_one(conn, "select entity_type, canonical_id, merged_into from entities where id = :id", id=c.id)259 assert res["mode"] == "variant" and row["entity_type"] == "artifact" and row["canonical_id"] == a.id and row["merged_into"] is None260 await merge_entities(conn, a.id, fam.id, mode="family_member")261 assert (await fetch_one(conn, "select family_id from entities where id = :id", id=a.id))["family_id"] == fam.id262 res = await merge_entities(conn, b.id, a.id, mode="alias")263 row = await fetch_one(conn, "select merged_into, status from entities where id = :id", id=b.id)264 assert row["merged_into"] == a.id and row["status"] == "merged"265 assert await fetch_one(conn, "select 1 from resolution_decisions where a_id = :a and b_id = :b and decision = 'alias'", a=b.id, b=a.id)266 assert await fetch_one(conn, "select 1 from admin_audit_log where action = 'entity.alias' and target = :t", t=a.id)267 with pytest.raises(ValueError):268 await merge_entities(conn, b.id, a.id) # already merged269270271async def test_duplicates_step(conn: AsyncConnection) -> None:272 t = _tag()273 f = Facts()274 keep = f.entity("company", f"Kwai {t}", attributes={"hf_org": f"kwai{t}"})275 dup = f.entity("organization", f"kwai{t}", attributes={"hf_org": f"kwai{t}"})276 x = f.entity("model", f"Dup {t}", identifiers={"openai_model_id": f"dup-{t}-a"})277 y = f.entity("model", f"Dup {t} ", identifiers={"openai_model_id": f"dup-{t}-b"}) # same name, conflicting identifier scheme → review only278 await _write(conn, f)279 rep = await _run(canon.step_duplicates, conn, True, keep.id, dup.id, x.id, y.id)280 assert rep.counts.get("merged", 0) >= 1 and rep.counts.get("keep_separate_recorded", 0) >= 1281 assert (await fetch_one(conn, "select merged_into from entities where id = :id", id=dup.id))["merged_into"] == keep.id282 assert await fetch_one(conn, "select 1 from resolution_decisions where decision = 'keep_separate' and a_id in (:x, :y) and b_id in (:x, :y)", x=x.id, y=y.id)283 assert (await fetch_one(conn, "select attributes->>'org_kind' as k from entities where id = :id", id=keep.id))["k"] == "company"284 for i in (x.id, y.id):285 assert (await fetch_one(conn, "select merged_into from entities where id = :id", id=i))["merged_into"] is None286 again = await _run(canon.step_duplicates, conn, True, keep.id, dup.id, x.id, y.id)287 assert again.counts.get("merged", 0) == 0 and again.counts.get("keep_separate_recorded", 0) == 0288289290async def test_snapshot_identifier_is_not_a_conflict_and_routers_become_products(conn: AsyncConnection) -> None:291 t = _tag()292 f = Facts()293 org = f.entity("company", f"Google test {t}")294 plain = f.entity("model", f"Gemini {t} Flash", organization=org, identifiers={"gemini_model_id": f"gemini-{t}-flash", "openrouter": f"google/gemini-{t}-flash"})295 prefixed = f.entity("model", f"Gemini {t} Flash ", organization=org, identifiers={"gemini_model_id": f"gemini-{t}-flash-preview-06-17"}, slug_hint=f"google-gemini-{t}-flash")296 router = f.entity("model", f"Auto Router {t}", identifiers={"openrouter": f"openrouter/auto-{t}"}, attributes={"context_length": 2_000_000})297 rel_org = f.entity("provider", f"OpenRouter {t}")298 f.relate(rel_org, "develops", router)299 await _write(conn, f)300 rep = await _run(canon.step_duplicates, conn, True, plain.id, prefixed.id, router.id)301 assert rep.counts.get("merged", 0) == 1 and rep.counts.get("routers_retyped_as_product") == 1302 row = await fetch_one(conn, "select merged_into, slug from entities where id = :id", id=prefixed.id)303 assert row["merged_into"] == plain.id # the plain-slug entity survives, both identifiers are kept304 ids = {r["value"] for r in await fetch_all(conn, "select value from entity_identifiers where entity_id = :id and scheme = 'gemini_model_id'", id=plain.id)}305 assert ids == {f"gemini-{t}-flash", f"gemini-{t}-flash-preview-06-17"}306 r = await fetch_one(conn, "select entity_type, attributes->>'kind' as kind from entities where id = :id", id=router.id)307 assert r["entity_type"] == "product" and r["kind"] == "router"308 assert not await fetch_one(conn, "select 1 from relations where object_id = :id and predicate = 'develops' and valid_to is null", id=router.id)309 again = await _run(canon.step_duplicates, conn, True, plain.id, prefixed.id, router.id)310 assert again.changes == 0311312313# ---------------------------------------------------------------------------------------------- result re-homing & family slugs314async def test_results_rehomed_to_variant_benchmarks(conn: AsyncConnection) -> None:315 t = _tag()316 f = Facts()317 m = f.entity("model", f"Zeta {t}")318 head = f.entity("benchmark", f"GPQA {t}", identifiers={"registry_benchmark": f"gpqa-{t}"}, slug_hint=f"gpqa-{t}")319 diamond = f.entity("benchmark", f"GPQA {t} Diamond", identifiers={"registry_benchmark": f"gpqa-{t}-diamond"}, slug_hint=f"gpqa-{t}-diamond",320 aliases=[f"GPQA-{t}-Diamond"], attributes={"variant": "Diamond", "metric": "accuracy"})321 f.relate(diamond, "variant_of", head)322 lb = f.entity("benchmark", f"LiveBench {t}", slug_hint=f"livebench-{t}")323 lb_reason = f.entity("benchmark", f"LiveBench {t} Reasoning", slug_hint=f"livebench-reasoning-{t}", attributes={"variant": "Reasoning", "metric": "average score"})324 f.relate(lb_reason, "variant_of", lb)325 f.result(model=m, benchmark=head, score=71.0, metric="accuracy", unit="%", config={"variant": f"GPQA {t} Diamond", "evaluator": "AA"})326 f.result(model=m, benchmark=head, score=60.0, metric="accuracy", unit="%", config={"evaluator": "AA"}) # main set stays327 f.result(model=m, benchmark=lb, score=55.0, metric="global_average", unit="%", config={"release": "2026-06-25"})328 await _write(conn, f, source_key="artificialanalysis.ai")329 # a category metric sitting on the head (pre-upgrade LiveBench rows); the variant slug must be livebench-<slug(Name)>330 await execute(conn, "update entities set slug = :s where id = :id", s=f"livebench-reasoning-{t}", id=lb_reason.id)331 await execute(conn, """insert into benchmark_results (id, model_id, benchmark_id, score, metric, unit, config, dedupe_key, tier)332 values (:id, :m, :b, 48.0, :metric, '%', '{"release": "2026-06-25"}', :d, 2)""",333 id=f"res_cat{t}", m=m.id, b=lb.id, metric="category:Reasoning", d=f"cat{t}")334 # slug(Name) lookup uses exactly `livebench-reasoning`; emulate it by renaming the head-side entity for this test335 await execute(conn, "update entities set slug = 'livebench-reasoning' where id = :id and not exists (select 1 from entities where slug = 'livebench-reasoning')", id=lb_reason.id)336 rep = await _run(canon.step_results, conn, True, m.id)337 assert rep.counts.get(f"results_rehomed:gpqa-{t}-diamond") == 1338 rows = {r["score"]: r for r in await fetch_all(conn, "select r.score, b.slug, r.metric, r.config, r.dedupe_key, r.is_current from benchmark_results r join entities b on b.id = r.benchmark_id where r.model_id = :m", m=m.id)}339 assert rows[71.0]["slug"] == f"gpqa-{t}-diamond" and rows[71.0]["dedupe_key"].startswith(f"{m.id}:{diamond.id}:") and rows[71.0]["is_current"]340 assert rows[60.0]["slug"] == f"gpqa-{t}" and rows[60.0]["is_current"]341 lb_slug = (await fetch_one(conn, "select slug from entities where id = :id", id=lb_reason.id))["slug"]342 if lb_slug == "livebench-reasoning":343 assert rows[48.0]["slug"] == "livebench-reasoning" and rows[48.0]["metric"] == "average score" and rows[48.0]["config"]["variant"] == "Reasoning"344 rels = {(r["object_id"], r["valid_to"] is None) for r in await fetch_all(conn, "select object_id, valid_to from relations where subject_id = :m and predicate = 'evaluated_on'", m=m.id)}345 assert (diamond.id, True) in rels and (head.id, True) in rels # head keeps a live edge: the main-set row is still there346 again = await _run(canon.step_results, conn, True, m.id)347 assert again.changes == 0348349350async def test_family_slug_collision_uses_family_suffix(conn: AsyncConnection) -> None:351 t = _tag()352 v = str(int(t, 16) % 900 + 100)353 f = Facts()354 org = f.entity("company", f"Mistral test {t}")355 taken = f.entity("model", f"Gemma {v}", organization=org) # occupies the slug `gemma-<v>`356 a = f.entity("model", f"Gemma {v} 9B Test", organization=org)357 await _write(conn, f)358 rep = await _run(canon.step_families, conn, True, a.id, taken.id)359 assert rep.counts.get("families_created", 0) >= 1360 fam = await fetch_one(conn, "select slug from entities where entity_type = 'model_family' and canonical_name = :n", n=f"Gemma {v}")361 assert fam["slug"] == f"gemma-{v}-family"362 # a family created under the old rule (org prefix) is re-slugged and keeps the old slug as an alias363 await execute(conn, "update entities set slug = :s where id = (select id from entities where entity_type = 'model_family' and canonical_name = :n)", s=f"mistral-gemma-{v}", n=f"Gemma {v}")364 rep = await _run(canon.step_families, conn, True, a.id)365 assert rep.counts.get("families_reslugged") == 1366 fam = await fetch_one(conn, "select id, slug from entities where entity_type = 'model_family' and canonical_name = :n", n=f"Gemma {v}")367 assert fam["slug"] == f"gemma-{v}-family"368 assert await fetch_one(conn, "select 1 from entity_aliases where entity_id = :id and alias = :a and kind = 'former_name'", id=fam["id"], a=f"mistral-gemma-{v}")369 again = await _run(canon.step_families, conn, True, a.id)370 assert again.changes == 0371