HTML 77.2%
TypeScript 10.5%
Python 9.6%
JavaScript 2.5%
1"""Admin workbenches (API 1.1, `x-aia-admin-token`): data-health dashboard, entity resolution, anomalies, extraction debugger, quarantine,2audit log, run rollback. Every call is audited (router dependency shared with `admin.py`). Nothing here deletes data."""3from __future__ import annotations45import inspect6import re7from collections import defaultdict8from typing import Any910from fastapi import APIRouter, Depends, Query, Request11from pydantic import BaseModel, Field1213from aiatlas.api.common import ADMIN_DEPENDENCIES, MAIN_ENTITY_TYPES, STATUS_VOCAB, ApiError, audit, client_ip14from aiatlas.api.routers.admin import SNAPSHOT_TEXT_LIMIT, audit_request15from aiatlas.db import connection, execute, fetch_all, fetch_one, fetch_val, jsonb, transaction16from aiatlas.ids import new_id17from aiatlas.ontology.licenses import normalize_license18from aiatlas.ontology.models import analyze_model_name, variant_key19from aiatlas.ontology.openness import OPENNESS_CATEGORIES20from aiatlas.sdk import archive21from aiatlas.services import cache22from aiatlas.services import merge as merge_service23from aiatlas.services.frontier import all_primary_groups, frontier_model_ids, rank_rows2425router = APIRouter(prefix="/api/v1/admin", tags=["admin-quality"], dependencies=[*ADMIN_DEPENDENCIES, Depends(audit_request)])26SAMPLE = 827DECISIONS = ("merge", "alias", "variant_of", "family_member", "keep_separate", "defer")282930async def _count_sample(conn: Any, sql_from_where: str, select: str, order: str = "1", **params: Any) -> dict[str, Any]:31 count = await fetch_val(conn, f"select count(*) {sql_from_where}", **params)32 sample = await fetch_all(conn, f"select {select} {sql_from_where} order by {order} limit {SAMPLE}", **params)33 return {"count": int(count or 0), "sample": sample}343536# ------------------------------------------------------------------------------------------------------------------ /admin/quality373839@router.get("/quality")40async def quality_dashboard() -> dict[str, Any]:41 out: dict[str, Any] = {}42 ent = "e.id, e.slug, e.canonical_name as name, e.entity_type"43 async with connection() as conn:44 out["duplicate_candidates"] = {45 "pending_decisions": await _count_sample(conn, "from resolution_decisions d where d.decision = 'defer' and not d.applied", "d.id, d.a_id, d.b_id, d.note, d.created_at", "d.created_at desc"),46 "review_merge_candidates": await _count_sample(conn, "from review_queue r where r.kind = 'merge_candidate' and r.status = 'pending'", "r.id, r.entity_ids, r.reason, r.created_at", "r.created_at desc"),47 }48 tax_unmapped = await _count_sample(conn, "from taxonomy_mappings t where t.canonical is null", "t.domain, t.raw, t.count, t.last_seen_at", "t.count desc")49 bad_open = await _count_sample(conn, "from entities e where e.entity_type = 'model' and e.merged_into is null and coalesce(e.attributes->>'openness', '') <> '' and e.attributes->>'openness' <> all(cast(:cats as text[]))",50 f"{ent}, e.attributes->>'openness' as value", "e.updated_at desc", cats=list(OPENNESS_CATEGORIES))51 bad_status = await _count_sample(conn, "from entities e where e.entity_type = 'model' and e.merged_into is null and e.status <> all(cast(:st as text[]))", f"{ent}, e.status as value", "e.updated_at desc",52 st=[*STATUS_VOCAB, "merged", "archived"])53 lic_rows = await fetch_all(conn, "select e.attributes->>'license' as raw, count(*) as n from entities e where e.entity_type = 'model' and e.merged_into is null and e.attributes ? 'license' and not e.attributes ? 'license_key' group by 1")54 bad_lic = [{"raw": r["raw"], "models": int(r["n"])} for r in lic_rows if r["raw"] and normalize_license(r["raw"]) is None]55 out["taxonomy_violations"] = {"unmapped_taxonomy_rows": tax_unmapped, "openness_unknown_vocab": bad_open, "status_unknown_vocab": bad_status,56 "license_unclassified": {"count": sum(x["models"] for x in bad_lic), "sample": sorted(bad_lic, key=lambda x: -x["models"])[:SAMPLE]}}57 by_check = await fetch_all(conn, "select check_name, severity, count(*) as n from anomalies where status = 'open' group by 1, 2 order by 3 desc")58 out["impossible_values"] = {"count": sum(int(r["n"]) for r in by_check), "by_check": [{**r, "n": int(r["n"])} for r in by_check],59 "sample": await fetch_all(conn, "select a.id, a.check_name, a.severity, a.message, a.entity_id, e.slug, a.value, a.last_seen_at from anomalies a left join entities e on e.id = a.entity_id where a.status = 'open' order by case a.severity when 'critical' then 0 when 'warning' then 1 else 2 end, a.last_seen_at desc limit 12")}60 out["conflicting_t1_claims"] = await _count_sample(conn, "from claims c join entities e on e.id = c.entity_id where c.status = 'conflicting' and c.tier = 1",61 f"c.id as claim_id, c.property, c.value, c.source_url, {ent}", "c.observed_at desc")62 base_models = "from entities e where e.entity_type = 'model' and e.merged_into is null"63 out["models_without_organization"] = await _count_sample(conn, f"{base_models} and e.organization_id is null", ent, "e.updated_at desc")64 out["models_without_release_source"] = await _count_sample(conn, f"{base_models} and not exists (select 1 from claims c where c.entity_id = e.id and c.property = 'release_date' and c.status = 'current')", ent, "e.updated_at desc")65 out["models_without_parameters"] = await _count_sample(conn, f"{base_models} and not e.attributes ? 'parameter_count'", ent, "e.updated_at desc")66 out["orphan_benchmark_results"] = await _count_sample(conn, "from benchmark_results r join entities m on m.id = r.model_id where r.valid_to is null and (m.merged_into is not null or m.entity_type <> 'model')",67 "r.id as result_id, r.benchmark_id, m.id, m.slug, m.canonical_name as name, m.entity_type, m.merged_into", "r.observed_at desc")68 out["benchmarks_without_results"] = await _count_sample(conn, "from entities e where e.entity_type = 'benchmark' and e.merged_into is null and not exists (select 1 from benchmark_results r where r.benchmark_id = e.id and r.valid_to is null)",69 ent, "e.canonical_name")70 out["unresolved_provider_deployments"] = await _count_sample(conn, """from prices p join entities m on m.id = p.model_id join entities pv on pv.id = p.provider_id where p.valid_to is null and p.provider_model_id is not null71 and not exists (select 1 from entity_identifiers i where i.entity_id = p.model_id and i.value = p.provider_model_id)""",72 "p.id as price_id, p.provider_model_id, m.slug as model, pv.slug as provider", "p.observed_at desc")73 names = await fetch_all(conn, "select e.id, e.slug, e.canonical_name, e.attributes->>'hf_repo' as hf_repo, e.attributes->>'is_quantized' as is_q, e.attributes->>'quant_format' as fmt from entities e where e.entity_type = 'model' and e.merged_into is null")74 quants = []75 for r in names:76 a = analyze_model_name(r["hf_repo"] or r["canonical_name"])77 if a.is_artifact or r["is_q"] == "true":78 quants.append({"id": r["id"], "slug": r["slug"], "name": r["canonical_name"], "hf_repo": r["hf_repo"], "quant_formats": a.quant_formats or ([r["fmt"]] if r["fmt"] else []),79 "kind": "quantization" if (a.is_quantized or r["is_q"] == "true") else "conversion"})80 out["quantisations_typed_as_models"] = {"count": len(quants), "sample": quants[:SAMPLE]}81 out["stale_sources"] = await _count_sample(conn, "from connectors c where c.enabled and c.last_success_at is not null and c.last_success_at < now() - make_interval(secs => 3 * c.interval_seconds)",82 "c.name, c.health, c.last_success_at, c.interval_seconds, c.consecutive_failures", "c.last_success_at")83 counts = {r["entity_type"]: int(r["n"]) for r in await fetch_all(conn, "select entity_type, count(*) as n from entities where merged_into is null group by 1")}84 out["empty_public_categories"] = {"count": sum(1 for t in MAIN_ENTITY_TYPES if not counts.get(t)), "sample": [t for t in MAIN_ENTITY_TYPES if not counts.get(t)]}85 out["quarantined_runs_pending"] = await _count_sample(conn, "from quarantined_runs q where q.status = 'pending'", "q.id, q.run_id, q.connector_name, q.reason, q.stats, q.created_at", "q.created_at desc")86 out["review_queue_priority"] = await _review_priority(conn)87 out["note"] = "Counts are live. Nothing is deleted by these checks; they point at review actions (/admin/entity-resolution, /admin/anomalies, /admin/quarantine)."88 return out899091async def _review_priority(conn: Any) -> list[dict[str, Any]]:92 """Homepage-visible items first: frontier models, largest params/context claims, price anomalies, benchmark leaders, major orgs, duplicate canonical models."""93 frontier, _ = await frontier_model_ids(conn)94 groups = await all_primary_groups(conn)95 leaders = {rank_rows(g["rows"], g["higher_is_better"])[0]["model_id"] for g in groups.values() if g["rows"]}96 big = {r["id"] for r in await fetch_all(conn, """select id from entities where entity_type = 'model' and merged_into is null and attributes->>'parameter_count' ~ '^[0-9.]+$'97 order by (attributes->>'parameter_count')::double precision desc limit 20""")}98 big |= {r["id"] for r in await fetch_all(conn, """select id from entities where entity_type = 'model' and merged_into is null and attributes->>'context_length' ~ '^[0-9]+$'99 order by (attributes->>'context_length')::bigint desc limit 20""")}100 major_orgs = {r["organization_id"] for r in await fetch_all(conn, "select organization_id from entities where entity_type = 'model' and merged_into is null and organization_id is not null group by 1 having count(*) >= 10")}101 anomalies = await fetch_all(conn, """select a.id, a.check_name, a.severity, a.message, a.entity_id, e.slug, e.organization_id, e.entity_type from anomalies a left join entities e on e.id = a.entity_id102 where a.status = 'open' order by a.last_seen_at desc limit 2000""")103 merges = await fetch_all(conn, "select r.id, r.entity_ids, r.reason from review_queue r where r.kind = 'merge_candidate' and r.status = 'pending' limit 500")104 items = []105 for a in anomalies:106 reasons = []107 eid = a["entity_id"]108 if eid in frontier:109 reasons.append("frontier model")110 if eid in leaders:111 reasons.append("benchmark leader")112 if eid in big:113 reasons.append("among the largest params/context claims")114 if a["check_name"] in ("negative_price", "price_too_high", "zero_output_price", "price_jump_100x", "cached_gt_input_price"):115 reasons.append("price anomaly")116 if a["organization_id"] in major_orgs:117 reasons.append("major organization")118 if reasons or a["severity"] == "critical":119 items.append({"kind": "anomaly", "id": a["id"], "check": a["check_name"], "severity": a["severity"], "message": a["message"], "entity_id": eid, "slug": a["slug"], "reasons": reasons or ["critical severity"],120 "_p": (0 if reasons else 1, {"critical": 0, "warning": 1}.get(a["severity"], 2))})121 for m in merges:122 ids = list(m["entity_ids"] or [])123 hits = [i for i in ids if i in frontier or i in leaders]124 if hits:125 items.append({"kind": "merge_candidate", "id": m["id"], "entity_ids": ids, "reason": m["reason"], "reasons": ["duplicate involving a frontier model / benchmark leader"], "_p": (0, 0)})126 items.sort(key=lambda x: x["_p"])127 for x in items:128 x.pop("_p")129 return items[:50]130131132# ------------------------------------------------------------------------------------------------------------------ /admin/entity-resolution133134135async def _side(conn: Any, eid: str) -> dict[str, Any] | None:136 row = await fetch_one(conn, """select e.id, e.slug, e.canonical_name, e.entity_type, e.attributes, e.first_seen_at, e.merged_into, e.family_id, e.canonical_id, e.identity_confidence,137 o.slug as org_slug, o.canonical_name as org_name, f.canonical_name as family_name,138 (select count(*) from relations r where (r.subject_id = e.id or r.object_id = e.id) and r.valid_to is null) as relations,139 (select count(distinct c.source_id) from claims c where c.entity_id = e.id and c.status = 'current') as sources,140 (select count(*) from claims c where c.entity_id = e.id and c.status = 'current') as claims,141 (select count(*) from prices p where p.model_id = e.id and p.valid_to is null) as prices,142 (select count(*) from benchmark_results r where r.model_id = e.id and r.valid_to is null) as results,143 (select jsonb_agg(jsonb_build_object('scheme', i.scheme, 'value', i.value) order by i.scheme) from entity_identifiers i where i.entity_id = e.id) as identifiers,144 (select jsonb_agg(a.alias order by a.alias) from entity_aliases a where a.entity_id = e.id) as aliases145 from entities e left join entities o on o.id = e.organization_id left join entities f on f.id = e.family_id where e.id = :id or e.slug = :id limit 1""", id=eid)146 if not row:147 return None148 a = row["attributes"] or {}149 return {"id": row["id"], "slug": row["slug"], "name": row["canonical_name"], "entity_type": row["entity_type"], "organization": row["org_name"], "organization_slug": row["org_slug"],150 "family": row["family_name"] or a.get("family"), "parameter_count": a.get("parameter_count"), "active_parameter_count": a.get("active_parameter_count"), "release_date": a.get("release_date"),151 "architecture": a.get("architecture"), "model_type": a.get("model_type"), "hf_repo": a.get("hf_repo"), "openness": a.get("openness"), "context_length": a.get("context_length"),152 "identifiers": row["identifiers"] or [], "aliases": row["aliases"] or [], "relations": int(row["relations"] or 0), "sources": int(row["sources"] or 0), "claims": int(row["claims"] or 0),153 "prices": int(row["prices"] or 0), "results": int(row["results"] or 0), "first_seen_at": row["first_seen_at"], "merged_into": row["merged_into"], "canonical_id": row["canonical_id"],154 "identity_confidence": row["identity_confidence"], "variant_key": variant_key(a.get("hf_repo") or row["canonical_name"]), "name_analysis": _analysis(a.get("hf_repo") or row["canonical_name"])}155156157def _analysis(name: str) -> dict[str, Any]:158 a = analyze_model_name(name)159 return {"base_key": a.base_key, "quant_formats": a.quant_formats, "precision": a.precision, "is_artifact": a.is_artifact, "effort": a.effort, "family_hint": a.family_hint,160 "parameter_count": a.parameter_count, "snapshot_date": a.snapshot_date}161162163@router.get("/entity-resolution")164async def entity_resolution(type: str | None = Query("model", alias="type"), status: str = Query("pending"), limit: int = Query(50, ge=1, le=200), threshold: float = Query(0.8, ge=0.3, le=1.0)) -> dict[str, Any]:165 """Candidate pairs: review merge_candidates + trigram similarity + same `variant_key` (ontology.models) — side by side."""166 pairs: dict[tuple[str, str], dict[str, Any]] = {}167 async with connection() as conn:168 decided = {(d["a_id"], d["b_id"]): d for d in await fetch_all(conn, "select a_id, b_id, decision, applied, created_at, note from resolution_decisions")}169 if status in ("pending", "all"):170 for r in await fetch_all(conn, "select r.id, r.entity_ids, r.reason, r.payload from review_queue r where r.kind = 'merge_candidate' and r.status = 'pending' order by r.created_at desc limit :lim", lim=limit * 2):171 ids = list(r["entity_ids"] or [])172 if len(ids) >= 2:173 pairs.setdefault(tuple(sorted(ids[:2])), {"sources": [], "similarity": None})["sources"].append({"kind": "review_merge_candidate", "review_id": r["id"], "reason": r["reason"]})174 where = "a.entity_type = :t" if type else "a.entity_type in ('model','company','organization','lab','provider','benchmark','researcher')"175 for r in await fetch_all(conn, f"""select a.id as a_id, b.id as b_id, similarity(a.canonical_name, b.canonical_name) as sim from entities a join entities b176 on b.entity_type = a.entity_type and b.id > a.id and a.canonical_name % b.canonical_name177 where {where} and a.merged_into is null and b.merged_into is null and similarity(a.canonical_name, b.canonical_name) > :th order by sim desc limit :lim""",178 t=type, th=threshold, lim=limit * 2):179 p = pairs.setdefault(tuple(sorted((r["a_id"], r["b_id"]))), {"sources": [], "similarity": None})180 p["similarity"] = round(float(r["sim"]), 3)181 p["sources"].append({"kind": "trigram_similarity", "similarity": round(float(r["sim"]), 3)})182 if type in (None, "model"):183 names = await fetch_all(conn, "select id, canonical_name, attributes->>'hf_repo' as hf from entities where entity_type = 'model' and merged_into is null")184 by_key: dict[str, list[str]] = defaultdict(list)185 for n in names:186 k = variant_key(n["hf"] or n["canonical_name"])187 if k:188 by_key[k].append(n["id"])189 for k, ids in by_key.items():190 if 2 <= len(ids) <= 6:191 for i in range(len(ids)):192 for j in range(i + 1, len(ids)):193 pairs.setdefault(tuple(sorted((ids[i], ids[j]))), {"sources": [], "similarity": None})["sources"].append({"kind": "same_variant_key", "variant_key": k})194 items = []195 for (a, b), p in pairs.items():196 d = decided.get((a, b)) or decided.get((b, a))197 if status == "pending" and d and d["decision"] != "defer":198 continue199 if status == "decided" and not d:200 continue201 sa, sb = await _side(conn, a), await _side(conn, b)202 if not sa or not sb:203 continue204 items.append({"a": sa, "b": sb, "signals": p["sources"], "similarity": p["similarity"], "same_variant_key": sa["variant_key"] == sb["variant_key"] and bool(sa["variant_key"]),205 "same_organization": sa["organization_slug"] == sb["organization_slug"] and sa["organization_slug"] is not None, "decision": d,206 "hint": ("artifact of the other" if sa["name_analysis"]["is_artifact"] != sb["name_analysis"]["is_artifact"] else207 "effort variant" if (sa["name_analysis"]["effort"] or sb["name_analysis"]["effort"]) else "possible duplicate")})208 if len(items) >= limit:209 break210 items.sort(key=lambda x: (-len(x["signals"]), -(x["similarity"] or 0)))211 return {"items": items, "total": len(items), "threshold": threshold, "status": status, "decisions": list(DECISIONS),212 "note": "Signals are independent (review queue, trigram similarity, shared variant_key); nothing is merged until POST /admin/entity-resolution/{a}/{b}."}213214215class ResolutionBody(BaseModel):216 decision: str = Field(..., pattern="^(merge|alias|variant_of|family_member|keep_separate|defer)$")217 note: str | None = Field(None, max_length=2000)218219220@router.post("/entity-resolution/{a}/{b}")221async def resolve_pair(a: str, b: str, body: ResolutionBody, request: Request) -> dict[str, Any]:222 """Persist the decision, then apply it (merge_entities with `mode=` when the service supports it). `a` is folded INTO `b`."""223 applied = False224 effect: dict[str, Any] | None = None225 async with transaction() as conn:226 ea = await fetch_one(conn, "select id, entity_type, canonical_name, merged_into from entities where id = :k or slug = :k limit 1", k=a)227 eb = await fetch_one(conn, "select id, entity_type, canonical_name, merged_into from entities where id = :k or slug = :k limit 1", k=b)228 if not ea or not eb:229 raise ApiError(404, "entity a or b not found")230 if ea["id"] == eb["id"]:231 raise ApiError(400, "a and b are the same entity")232 did = new_id("review")233 sig = inspect.signature(merge_service.merge_entities)234 supports_mode = "mode" in sig.parameters235 try:236 if body.decision == "merge":237 effect = await (merge_service.merge_entities(conn, ea["id"], eb["id"], mode="merge") if supports_mode else merge_service.merge_entities(conn, ea["id"], eb["id"]))238 applied = True239 elif body.decision == "variant_of":240 if supports_mode:241 effect = await merge_service.merge_entities(conn, ea["id"], eb["id"], mode="variant_of")242 else:243 await execute(conn, "update entities set canonical_id = :b where id = :a", a=ea["id"], b=eb["id"])244 effect = await merge_service.merge_entities(conn, ea["id"], eb["id"])245 effect["fallback"] = "canonical_id set then folded with merge_entities (service has no mode= yet)"246 applied = True247 elif body.decision == "alias":248 if supports_mode:249 effect = await merge_service.merge_entities(conn, ea["id"], eb["id"], mode="alias")250 else:251 from aiatlas.ids import normalize_alias252253 await execute(conn, "insert into entity_aliases (entity_id, alias, alias_norm, kind) values (:t, :al, :n, 'alias') on conflict (entity_id, alias_norm) do nothing",254 t=eb["id"], al=ea["canonical_name"], n=normalize_alias(ea["canonical_name"]))255 effect = {"alias_added": ea["canonical_name"], "to": eb["id"], "fallback": "alias row only (service has no mode= yet); entities kept separate"}256 applied = True257 elif body.decision == "family_member":258 if supports_mode:259 effect = await merge_service.merge_entities(conn, ea["id"], eb["id"], mode="family_member")260 applied = True261 elif eb["entity_type"] == "model_family":262 await execute(conn, "update entities set family_id = :b, updated_at = now() where id = :a", a=ea["id"], b=eb["id"])263 effect = {"family_id": eb["id"]}264 applied = True265 else:266 raise ApiError(400, "family_member requires b to be a model_family entity")267 except (ValueError, LookupError) as exc:268 raise ApiError(400, str(exc)) from exc269 await execute(conn, """insert into resolution_decisions (id, a_id, b_id, decision, actor, note, payload, applied) values (:id, :a, :b, :d, 'admin', :n, cast(:p as jsonb), :ap)270 on conflict (a_id, b_id, decision) do update set note = excluded.note, payload = excluded.payload, applied = excluded.applied, created_at = now()""",271 id=did, a=ea["id"], b=eb["id"], d=body.decision, n=body.note, p=jsonb({"effect": effect, "mode_supported": supports_mode}), ap=applied)272 if applied:273 await execute(conn, "update review_queue set status = 'approved', resolved_at = now(), resolution = cast(:r as jsonb) where kind = 'merge_candidate' and status = 'pending' and entity_ids @> cast(:ids as text[])",274 r=jsonb({"via": "entity-resolution", "decision": body.decision}), ids=[ea["id"], eb["id"]])275 if applied:276 await cache.cache_invalidate()277 await audit("entity-resolution", f"{ea['id']}→{eb['id']}", {"decision": body.decision, "applied": applied, "effect": effect}, client_ip(request))278 return {"ok": True, "a": ea["id"], "b": eb["id"], "decision": body.decision, "applied": applied, "effect": effect}279280281# ------------------------------------------------------------------------------------------------------------------ /admin/anomalies282283284@router.get("/anomalies")285async def anomalies(status: str = Query("open"), severity: str | None = None, check: str | None = None, limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0)) -> dict[str, Any]:286 where = ["true"] if status == "all" else ["a.status = :st"]287 params: dict[str, Any] = {"st": status, "lim": limit, "off": offset}288 if severity:289 where.append("a.severity = :sev")290 params["sev"] = severity291 if check:292 where.append("a.check_name = :chk")293 params["chk"] = check294 w = " and ".join(where)295 async with connection() as conn:296 rows = await fetch_all(conn, f"""select a.*, e.slug, e.canonical_name as entity_name, e.entity_type from anomalies a left join entities e on e.id = a.entity_id where {w}297 order by case a.severity when 'critical' then 0 when 'warning' then 1 else 2 end, a.last_seen_at desc limit :lim offset :off""", **params)298 total = await fetch_val(conn, f"select count(*) from anomalies a where {w}", **{k: v for k, v in params.items() if k not in ("lim", "off")})299 by = await fetch_all(conn, "select check_name, severity, status, count(*) as n from anomalies group by 1, 2, 3 order by 4 desc")300 return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset, "by_check": [{**r, "n": int(r["n"])} for r in by]}301302303class AnomalyBody(BaseModel):304 status: str = Field(..., pattern="^(resolved|ignored|open)$")305 note: str | None = Field(None, max_length=2000)306307308@router.post("/anomalies/{anomaly_id}")309async def anomaly_action(anomaly_id: str, body: AnomalyBody) -> dict[str, Any]:310 async with transaction() as conn:311 row = await fetch_one(conn, "update anomalies set status = :st, resolution = :n, resolved_at = case when :st = 'open' then null else now() end where id = :id returning id, check_name, status",312 st=body.status, n=body.note, id=anomaly_id)313 if not row:314 raise ApiError(404, "anomaly not found")315 return {"ok": True, **row}316317318# ------------------------------------------------------------------------------------------------------------------ /admin/extractions/{snapshot_id}319320_NUM_FORMATS = (lambda v: str(v), lambda v: f"{int(v):,}" if float(v).is_integer() else f"{v:,}", lambda v: f"{int(v)}" if float(v).is_integer() else f"{v}",321 lambda v: f"{v / 1e9:g}B" if abs(v) >= 1e9 else f"{v / 1e6:g}M" if abs(v) >= 1e6 else f"{v / 1e3:g}K" if abs(v) >= 1e3 else str(v),322 lambda v: f"{v / 1e9:.1f}B" if abs(v) >= 1e9 else f"{v / 1e6:.1f}M" if abs(v) >= 1e6 else f"{v / 1e3:.0f}K" if abs(v) >= 1e3 else str(v),323 lambda v: f"{v / 1e3:g}k" if abs(v) >= 1e3 else str(v), lambda v: f"${v:g}", lambda v: f"${v:.2f}")324325326def _locate(text: str, value: Any) -> dict[str, Any]:327 cands: list[str] = []328 if isinstance(value, bool):329 cands = [str(value).lower()]330 elif isinstance(value, (int, float)):331 for f in _NUM_FORMATS:332 try:333 cands.append(f(float(value)))334 except Exception: # noqa: BLE001335 pass336 elif isinstance(value, str) and value.strip():337 cands = [value.strip()]338 for c in dict.fromkeys(cands):339 m = re.search(re.escape(c), text, flags=re.IGNORECASE)340 if m:341 s, e = max(0, m.start() - 80), min(len(text), m.end() + 80)342 return {"found": True, "offset": m.start(), "match": text[m.start():m.end()], "context": text[s:e]}343 return {"found": False, "tried": cands[:6]}344345346@router.get("/extractions/{snap_id}")347async def extraction_debugger(snap_id: str, text_limit: int = Query(SNAPSHOT_TEXT_LIMIT, ge=0, le=SNAPSHOT_TEXT_LIMIT)) -> dict[str, Any]:348 async with connection() as conn:349 snap = await fetch_one(conn, """select s.id, s.document_id, s.run_id, s.url, s.final_url, s.observed_at, s.http_status, s.content_type, s.content_hash, s.byte_size, s.text_hash, s.structured, s.diff,350 s.parser_version, s.connector_version, s.transport, s.changed, s.processing_status, s.raw_path is not null as has_raw, s.text_path, d.url as document_url,351 d.doc_type, d.connector_name, d.entity_id, d.title, e.slug as entity_slug, e.canonical_name as entity_name352 from snapshots s join documents d on d.id = s.document_id left join entities e on e.id = d.entity_id where s.id = :id""", id=snap_id)353 if not snap:354 raise ApiError(404, "snapshot not found")355 prev = await fetch_one(conn, "select id, observed_at, content_hash from snapshots where document_id = :d and observed_at < :t and changed order by observed_at desc limit 1", d=snap["document_id"], t=snap["observed_at"])356 claims = await fetch_all(conn, "select c.id, c.entity_id, e.slug as entity_slug, c.property, c.value, c.value_raw, c.unit, c.status, c.confidence, c.extractor, c.tier, c.observed_at from claims c left join entities e on e.id = c.entity_id where c.snapshot_id = :id order by e.slug, c.property limit 1000", id=snap_id)357 relations = await fetch_all(conn, "select r.id, r.subject_id, r.predicate, r.object_id, r.attributes, r.valid_to from relations r where r.snapshot_id = :id limit 500", id=snap_id)358 results = await fetch_all(conn, "select r.id, r.model_id, r.benchmark_id, r.score, r.metric, r.config, r.config_key, r.trust_level, r.valid_to from benchmark_results r where r.snapshot_id = :id limit 500", id=snap_id)359 prices = await fetch_all(conn, "select p.id, p.model_id, p.provider_id, p.provider_model_id, p.input_per_mtok, p.output_per_mtok, p.valid_from, p.valid_to from prices p where p.snapshot_id = :id limit 500", id=snap_id)360 events = await fetch_all(conn, "select id, entity_id, event_type, category, property, summary, importance, occurred_at, is_backfill, group_key from change_events where snapshot_id = :id order by occurred_at desc limit 300", id=snap_id)361 llm = await fetch_all(conn, "select id, task_type, stage, model, status, input_tokens, output_tokens, duration_ms, error, created_at from llm_jobs where snapshot_id = :id order by created_at desc limit 50", id=snap_id)362 touched = {c["entity_id"] for c in claims} | {r["subject_id"] for r in relations} | {r["object_id"] for r in relations} | {r["model_id"] for r in results} | {p["model_id"] for p in prices}363 touched.discard(None)364 cands = await fetch_all(conn, "select id, slug, canonical_name, entity_type, merged_into, identity_confidence from entities where id = any(cast(:ids as text[])) order by canonical_name", ids=list(touched)) if touched else []365 text, text_error, text_len = None, None, 0366 if snap.get("text_path"):367 try:368 full = archive.load_text(snap["text_path"])369 text_len = len(full)370 text = full[:text_limit]371 except OSError as exc:372 text_error = f"cleaned text unavailable ({exc.__class__.__name__})"373 full = ""374 else:375 full = ""376 spans = []377 for c in claims:378 v = c["value"]379 if isinstance(v, (int, float, str, bool)) and not (isinstance(v, str) and len(v) > 200):380 spans.append({"claim_id": c["id"], "property": c["property"], "value": v, **(_locate(full, v) if full else {"found": False, "reason": "no cleaned text"})})381 out = {k: v for k, v in snap.items() if k != "text_path"}382 out.update({"has_text": bool(snap.get("text_path")), "text": text, "text_chars": text_len, "text_truncated": text_len > text_limit, "text_error": text_error,383 "previous_snapshot": prev, "diff": snap.get("diff"), "claims": claims, "relations": relations, "results": results, "prices": prices, "events": events, "llm_jobs": llm,384 "entity_candidates": cands, "spans": spans, "spans_found": sum(1 for s in spans if s.get("found")),385 "note": "spans are a best-effort textual search of each claim value in the cleaned text (several number formats tried); not-found is reported honestly, never inferred."})386 return out387388389# ------------------------------------------------------------------------------------------------------------------ /admin/quarantine390391392@router.get("/quarantine")393async def quarantine(status: str = Query("pending"), limit: int = Query(50, ge=1, le=500), offset: int = Query(0, ge=0)) -> dict[str, Any]:394 where = "true" if status == "all" else "q.status = :st"395 async with connection() as conn:396 rows = await fetch_all(conn, f"select q.id, q.run_id, q.connector_name, q.reason, q.stats, q.status, q.created_at, q.resolved_at, q.resolved_by, jsonb_array_length(q.facts) as facts from quarantined_runs q where {where} order by q.created_at desc limit :lim offset :off",397 st=status, lim=limit, off=offset)398 total = await fetch_val(conn, f"select count(*) from quarantined_runs q where {where}", st=status)399 return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset}400401402class QuarantineBody(BaseModel):403 action: str = Field(..., pattern="^(release|discard)$")404 note: str | None = Field(None, max_length=2000)405406407@router.post("/quarantine/{quarantine_id}")408async def quarantine_action(quarantine_id: str, body: QuarantineBody) -> dict[str, Any]:409 try:410 from aiatlas.services import canonical411 except ImportError:412 canonical = None # type: ignore[assignment]413 fn = getattr(canonical, f"{body.action}_quarantine", None) if canonical else None414 if fn is None:415 raise ApiError(501, f"quarantine {body.action} is not available yet: services.canonical.{body.action}_quarantine is missing (Stream A)")416 async with transaction() as conn:417 row = await fetch_one(conn, "select * from quarantined_runs where id = :id for update", id=quarantine_id)418 if not row:419 raise ApiError(404, "quarantined run not found")420 if row["status"] != "pending":421 raise ApiError(409, f"quarantined run already {row['status']}")422 result = await fn(conn, quarantine_id) if inspect.iscoroutinefunction(fn) else fn(conn, quarantine_id)423 await execute(conn, "update quarantined_runs set status = :st, resolved_at = now(), resolved_by = 'admin' where id = :id and status = 'pending'",424 st="released" if body.action == "release" else "discarded", id=quarantine_id)425 await cache.cache_invalidate()426 return {"ok": True, "id": quarantine_id, "action": body.action, "result": result}427428429# ------------------------------------------------------------------------------------------------------------------ /admin/audit · rollback430431432@router.get("/audit")433async def audit_log(limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0), action: str | None = None) -> dict[str, Any]:434 where = "action ilike :a" if action else "true"435 async with connection() as conn:436 rows = await fetch_all(conn, f"select id, actor, action, target, payload, ip, created_at from admin_audit_log where {where} order by id desc limit :lim offset :off", a=f"%{action}%" if action else None, lim=limit, off=offset)437 total = await fetch_val(conn, f"select count(*) from admin_audit_log where {where}", a=f"%{action}%" if action else None)438 return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset}439440441@router.post("/runs/{run_id}/rollback")442async def rollback_run(run_id: str, request: Request) -> dict[str, Any]:443 """Undo one connector run WITHOUT deleting: retract its claims, close its relations/prices/results, flag its events as back-fill (meta.rolled_back)."""444 async with transaction() as conn:445 run = await fetch_one(conn, "select id, connector_name, status from connector_runs where id = :id", id=run_id)446 n: dict[str, int] = {}447 n["claims_retracted"] = int(await fetch_val(conn, "with u as (update claims set status = 'retracted', valid_to = coalesce(valid_to, now()) where run_id = :r and status <> 'retracted' returning 1) select count(*) from u", r=run_id) or 0)448 n["relations_closed"] = int(await fetch_val(conn, "with u as (update relations set valid_to = now() where run_id = :r and valid_to is null returning 1) select count(*) from u", r=run_id) or 0)449 n["prices_closed"] = int(await fetch_val(conn, "with u as (update prices set valid_to = now() where run_id = :r and valid_to is null returning 1) select count(*) from u", r=run_id) or 0)450 n["results_closed"] = int(await fetch_val(conn, "with u as (update benchmark_results set valid_to = now(), is_current = false where run_id = :r and valid_to is null returning 1) select count(*) from u", r=run_id) or 0)451 n["events_flagged"] = int(await fetch_val(conn, """with u as (update change_events set is_backfill = true, meta = meta || '{"rolled_back": true}'::jsonb where run_id = :r452 and not coalesce((meta->>'rolled_back')::boolean, false) returning 1) select count(*) from u""", r=run_id) or 0)453 # re-materialise attributes whose current claim was retracted: fall back to the best remaining claim454 affected = await fetch_all(conn, "select distinct entity_id, property from claims where run_id = :r and status = 'retracted'", r=run_id)455 restored = 0456 for a in affected:457 prev = await fetch_one(conn, "select id, value from claims where entity_id = :e and property = :p and status in ('superseded','current') order by tier, valid_from desc limit 1", e=a["entity_id"], p=a["property"])458 if prev:459 await execute(conn, "update claims set status = 'current', valid_to = null where id = :id", id=prev["id"])460 await execute(conn, "update entities set attributes = attributes || jsonb_build_object(:p, cast(:v as jsonb)), updated_at = now() where id = :e", p=a["property"], v=jsonb(prev["value"]), e=a["entity_id"])461 restored += 1462 else:463 await execute(conn, "update entities set attributes = attributes - :p, provenance = provenance - :p, updated_at = now() where id = :e", p=a["property"], e=a["entity_id"])464 n["attributes_restored"] = restored465 if run:466 await execute(conn, "update connector_runs set meta = meta || cast(:m as jsonb) where id = :id", m=jsonb({"rolled_back": True, "rollback_counts": n}), id=run_id)467 if not run and not any(n.values()):468 raise ApiError(404, f"no run or facts found for run_id {run_id!r}")469 await cache.cache_invalidate()470 await audit("rollback", run_id, {"counts": n, "connector": run["connector_name"] if run else None}, client_ip(request))471 return {"ok": True, "run_id": run_id, "connector": run["connector_name"] if run else None, "counts": n, "note": "nothing deleted: claims retracted, live rows closed with valid_to, events flagged is_backfill + meta.rolled_back"}472473474__all__ = ["DECISIONS", "router"]475