"""Admin workbenches (API 1.1, `x-aia-admin-token`): data-health dashboard, entity resolution, anomalies, extraction debugger, quarantine, audit log, run rollback. Every call is audited (router dependency shared with `admin.py`). Nothing here deletes data.""" from __future__ import annotations import inspect import re from collections import defaultdict from typing import Any from fastapi import APIRouter, Depends, Query, Request from pydantic import BaseModel, Field from aiatlas.api.common import ADMIN_DEPENDENCIES, MAIN_ENTITY_TYPES, STATUS_VOCAB, ApiError, audit, client_ip from aiatlas.api.routers.admin import SNAPSHOT_TEXT_LIMIT, audit_request from aiatlas.db import connection, execute, fetch_all, fetch_one, fetch_val, jsonb, transaction from aiatlas.ids import new_id from aiatlas.ontology.licenses import normalize_license from aiatlas.ontology.models import analyze_model_name, variant_key from aiatlas.ontology.openness import OPENNESS_CATEGORIES from aiatlas.sdk import archive from aiatlas.services import cache from aiatlas.services import merge as merge_service from aiatlas.services.frontier import all_primary_groups, frontier_model_ids, rank_rows router = APIRouter(prefix="/api/v1/admin", tags=["admin-quality"], dependencies=[*ADMIN_DEPENDENCIES, Depends(audit_request)]) SAMPLE = 8 DECISIONS = ("merge", "alias", "variant_of", "family_member", "keep_separate", "defer") async def _count_sample(conn: Any, sql_from_where: str, select: str, order: str = "1", **params: Any) -> dict[str, Any]: count = await fetch_val(conn, f"select count(*) {sql_from_where}", **params) sample = await fetch_all(conn, f"select {select} {sql_from_where} order by {order} limit {SAMPLE}", **params) return {"count": int(count or 0), "sample": sample} # ------------------------------------------------------------------------------------------------------------------ /admin/quality @router.get("/quality") async def quality_dashboard() -> dict[str, Any]: out: dict[str, Any] = {} ent = "e.id, e.slug, e.canonical_name as name, e.entity_type" async with connection() as conn: out["duplicate_candidates"] = { "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"), "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"), } 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") 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[]))", f"{ent}, e.attributes->>'openness' as value", "e.updated_at desc", cats=list(OPENNESS_CATEGORIES)) 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", st=[*STATUS_VOCAB, "merged", "archived"]) 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") bad_lic = [{"raw": r["raw"], "models": int(r["n"])} for r in lic_rows if r["raw"] and normalize_license(r["raw"]) is None] out["taxonomy_violations"] = {"unmapped_taxonomy_rows": tax_unmapped, "openness_unknown_vocab": bad_open, "status_unknown_vocab": bad_status, "license_unclassified": {"count": sum(x["models"] for x in bad_lic), "sample": sorted(bad_lic, key=lambda x: -x["models"])[:SAMPLE]}} 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") 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], "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")} 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", f"c.id as claim_id, c.property, c.value, c.source_url, {ent}", "c.observed_at desc") base_models = "from entities e where e.entity_type = 'model' and e.merged_into is null" out["models_without_organization"] = await _count_sample(conn, f"{base_models} and e.organization_id is null", ent, "e.updated_at desc") 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") out["models_without_parameters"] = await _count_sample(conn, f"{base_models} and not e.attributes ? 'parameter_count'", ent, "e.updated_at desc") 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')", "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") 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)", ent, "e.canonical_name") 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 null and not exists (select 1 from entity_identifiers i where i.entity_id = p.model_id and i.value = p.provider_model_id)""", "p.id as price_id, p.provider_model_id, m.slug as model, pv.slug as provider", "p.observed_at desc") 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") quants = [] for r in names: a = analyze_model_name(r["hf_repo"] or r["canonical_name"]) if a.is_artifact or r["is_q"] == "true": 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 []), "kind": "quantization" if (a.is_quantized or r["is_q"] == "true") else "conversion"}) out["quantisations_typed_as_models"] = {"count": len(quants), "sample": quants[:SAMPLE]} 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)", "c.name, c.health, c.last_success_at, c.interval_seconds, c.consecutive_failures", "c.last_success_at") 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")} 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)]} 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") out["review_queue_priority"] = await _review_priority(conn) out["note"] = "Counts are live. Nothing is deleted by these checks; they point at review actions (/admin/entity-resolution, /admin/anomalies, /admin/quarantine)." return out async def _review_priority(conn: Any) -> list[dict[str, Any]]: """Homepage-visible items first: frontier models, largest params/context claims, price anomalies, benchmark leaders, major orgs, duplicate canonical models.""" frontier, _ = await frontier_model_ids(conn) groups = await all_primary_groups(conn) leaders = {rank_rows(g["rows"], g["higher_is_better"])[0]["model_id"] for g in groups.values() if g["rows"]} 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.]+$' order by (attributes->>'parameter_count')::double precision desc limit 20""")} 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]+$' order by (attributes->>'context_length')::bigint desc limit 20""")} 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")} 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_id where a.status = 'open' order by a.last_seen_at desc limit 2000""") 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") items = [] for a in anomalies: reasons = [] eid = a["entity_id"] if eid in frontier: reasons.append("frontier model") if eid in leaders: reasons.append("benchmark leader") if eid in big: reasons.append("among the largest params/context claims") if a["check_name"] in ("negative_price", "price_too_high", "zero_output_price", "price_jump_100x", "cached_gt_input_price"): reasons.append("price anomaly") if a["organization_id"] in major_orgs: reasons.append("major organization") if reasons or a["severity"] == "critical": 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"], "_p": (0 if reasons else 1, {"critical": 0, "warning": 1}.get(a["severity"], 2))}) for m in merges: ids = list(m["entity_ids"] or []) hits = [i for i in ids if i in frontier or i in leaders] if hits: 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)}) items.sort(key=lambda x: x["_p"]) for x in items: x.pop("_p") return items[:50] # ------------------------------------------------------------------------------------------------------------------ /admin/entity-resolution async def _side(conn: Any, eid: str) -> dict[str, Any] | None: 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, o.slug as org_slug, o.canonical_name as org_name, f.canonical_name as family_name, (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, (select count(distinct c.source_id) from claims c where c.entity_id = e.id and c.status = 'current') as sources, (select count(*) from claims c where c.entity_id = e.id and c.status = 'current') as claims, (select count(*) from prices p where p.model_id = e.id and p.valid_to is null) as prices, (select count(*) from benchmark_results r where r.model_id = e.id and r.valid_to is null) as results, (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, (select jsonb_agg(a.alias order by a.alias) from entity_aliases a where a.entity_id = e.id) as aliases 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) if not row: return None a = row["attributes"] or {} 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"], "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"), "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"), "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), "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"], "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"])} def _analysis(name: str) -> dict[str, Any]: a = analyze_model_name(name) 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, "parameter_count": a.parameter_count, "snapshot_date": a.snapshot_date} @router.get("/entity-resolution") async 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]: """Candidate pairs: review merge_candidates + trigram similarity + same `variant_key` (ontology.models) — side by side.""" pairs: dict[tuple[str, str], dict[str, Any]] = {} async with connection() as conn: 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")} if status in ("pending", "all"): 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): ids = list(r["entity_ids"] or []) if len(ids) >= 2: pairs.setdefault(tuple(sorted(ids[:2])), {"sources": [], "similarity": None})["sources"].append({"kind": "review_merge_candidate", "review_id": r["id"], "reason": r["reason"]}) where = "a.entity_type = :t" if type else "a.entity_type in ('model','company','organization','lab','provider','benchmark','researcher')" 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 b on b.entity_type = a.entity_type and b.id > a.id and a.canonical_name % b.canonical_name 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""", t=type, th=threshold, lim=limit * 2): p = pairs.setdefault(tuple(sorted((r["a_id"], r["b_id"]))), {"sources": [], "similarity": None}) p["similarity"] = round(float(r["sim"]), 3) p["sources"].append({"kind": "trigram_similarity", "similarity": round(float(r["sim"]), 3)}) if type in (None, "model"): 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") by_key: dict[str, list[str]] = defaultdict(list) for n in names: k = variant_key(n["hf"] or n["canonical_name"]) if k: by_key[k].append(n["id"]) for k, ids in by_key.items(): if 2 <= len(ids) <= 6: for i in range(len(ids)): for j in range(i + 1, len(ids)): pairs.setdefault(tuple(sorted((ids[i], ids[j]))), {"sources": [], "similarity": None})["sources"].append({"kind": "same_variant_key", "variant_key": k}) items = [] for (a, b), p in pairs.items(): d = decided.get((a, b)) or decided.get((b, a)) if status == "pending" and d and d["decision"] != "defer": continue if status == "decided" and not d: continue sa, sb = await _side(conn, a), await _side(conn, b) if not sa or not sb: continue 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"]), "same_organization": sa["organization_slug"] == sb["organization_slug"] and sa["organization_slug"] is not None, "decision": d, "hint": ("artifact of the other" if sa["name_analysis"]["is_artifact"] != sb["name_analysis"]["is_artifact"] else "effort variant" if (sa["name_analysis"]["effort"] or sb["name_analysis"]["effort"]) else "possible duplicate")}) if len(items) >= limit: break items.sort(key=lambda x: (-len(x["signals"]), -(x["similarity"] or 0))) return {"items": items, "total": len(items), "threshold": threshold, "status": status, "decisions": list(DECISIONS), "note": "Signals are independent (review queue, trigram similarity, shared variant_key); nothing is merged until POST /admin/entity-resolution/{a}/{b}."} class ResolutionBody(BaseModel): decision: str = Field(..., pattern="^(merge|alias|variant_of|family_member|keep_separate|defer)$") note: str | None = Field(None, max_length=2000) @router.post("/entity-resolution/{a}/{b}") async def resolve_pair(a: str, b: str, body: ResolutionBody, request: Request) -> dict[str, Any]: """Persist the decision, then apply it (merge_entities with `mode=` when the service supports it). `a` is folded INTO `b`.""" applied = False effect: dict[str, Any] | None = None async with transaction() as conn: 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) 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) if not ea or not eb: raise ApiError(404, "entity a or b not found") if ea["id"] == eb["id"]: raise ApiError(400, "a and b are the same entity") did = new_id("review") sig = inspect.signature(merge_service.merge_entities) supports_mode = "mode" in sig.parameters try: if body.decision == "merge": 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"])) applied = True elif body.decision == "variant_of": if supports_mode: effect = await merge_service.merge_entities(conn, ea["id"], eb["id"], mode="variant_of") else: await execute(conn, "update entities set canonical_id = :b where id = :a", a=ea["id"], b=eb["id"]) effect = await merge_service.merge_entities(conn, ea["id"], eb["id"]) effect["fallback"] = "canonical_id set then folded with merge_entities (service has no mode= yet)" applied = True elif body.decision == "alias": if supports_mode: effect = await merge_service.merge_entities(conn, ea["id"], eb["id"], mode="alias") else: from aiatlas.ids import normalize_alias 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", t=eb["id"], al=ea["canonical_name"], n=normalize_alias(ea["canonical_name"])) effect = {"alias_added": ea["canonical_name"], "to": eb["id"], "fallback": "alias row only (service has no mode= yet); entities kept separate"} applied = True elif body.decision == "family_member": if supports_mode: effect = await merge_service.merge_entities(conn, ea["id"], eb["id"], mode="family_member") applied = True elif eb["entity_type"] == "model_family": await execute(conn, "update entities set family_id = :b, updated_at = now() where id = :a", a=ea["id"], b=eb["id"]) effect = {"family_id": eb["id"]} applied = True else: raise ApiError(400, "family_member requires b to be a model_family entity") except (ValueError, LookupError) as exc: raise ApiError(400, str(exc)) from exc 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) on conflict (a_id, b_id, decision) do update set note = excluded.note, payload = excluded.payload, applied = excluded.applied, created_at = now()""", id=did, a=ea["id"], b=eb["id"], d=body.decision, n=body.note, p=jsonb({"effect": effect, "mode_supported": supports_mode}), ap=applied) if applied: 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[])", r=jsonb({"via": "entity-resolution", "decision": body.decision}), ids=[ea["id"], eb["id"]]) if applied: await cache.cache_invalidate() await audit("entity-resolution", f"{ea['id']}→{eb['id']}", {"decision": body.decision, "applied": applied, "effect": effect}, client_ip(request)) return {"ok": True, "a": ea["id"], "b": eb["id"], "decision": body.decision, "applied": applied, "effect": effect} # ------------------------------------------------------------------------------------------------------------------ /admin/anomalies @router.get("/anomalies") async 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]: where = ["true"] if status == "all" else ["a.status = :st"] params: dict[str, Any] = {"st": status, "lim": limit, "off": offset} if severity: where.append("a.severity = :sev") params["sev"] = severity if check: where.append("a.check_name = :chk") params["chk"] = check w = " and ".join(where) async with connection() as conn: 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} 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) 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")}) by = await fetch_all(conn, "select check_name, severity, status, count(*) as n from anomalies group by 1, 2, 3 order by 4 desc") return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset, "by_check": [{**r, "n": int(r["n"])} for r in by]} class AnomalyBody(BaseModel): status: str = Field(..., pattern="^(resolved|ignored|open)$") note: str | None = Field(None, max_length=2000) @router.post("/anomalies/{anomaly_id}") async def anomaly_action(anomaly_id: str, body: AnomalyBody) -> dict[str, Any]: async with transaction() as conn: 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", st=body.status, n=body.note, id=anomaly_id) if not row: raise ApiError(404, "anomaly not found") return {"ok": True, **row} # ------------------------------------------------------------------------------------------------------------------ /admin/extractions/{snapshot_id} _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}", 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), 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), lambda v: f"{v / 1e3:g}k" if abs(v) >= 1e3 else str(v), lambda v: f"${v:g}", lambda v: f"${v:.2f}") def _locate(text: str, value: Any) -> dict[str, Any]: cands: list[str] = [] if isinstance(value, bool): cands = [str(value).lower()] elif isinstance(value, (int, float)): for f in _NUM_FORMATS: try: cands.append(f(float(value))) except Exception: # noqa: BLE001 pass elif isinstance(value, str) and value.strip(): cands = [value.strip()] for c in dict.fromkeys(cands): m = re.search(re.escape(c), text, flags=re.IGNORECASE) if m: s, e = max(0, m.start() - 80), min(len(text), m.end() + 80) return {"found": True, "offset": m.start(), "match": text[m.start():m.end()], "context": text[s:e]} return {"found": False, "tried": cands[:6]} @router.get("/extractions/{snap_id}") async def extraction_debugger(snap_id: str, text_limit: int = Query(SNAPSHOT_TEXT_LIMIT, ge=0, le=SNAPSHOT_TEXT_LIMIT)) -> dict[str, Any]: async with connection() as conn: 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, 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, d.doc_type, d.connector_name, d.entity_id, d.title, e.slug as entity_slug, e.canonical_name as entity_name 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) if not snap: raise ApiError(404, "snapshot not found") 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"]) 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) 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) 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) 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) 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) 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) 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} touched.discard(None) 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 [] text, text_error, text_len = None, None, 0 if snap.get("text_path"): try: full = archive.load_text(snap["text_path"]) text_len = len(full) text = full[:text_limit] except OSError as exc: text_error = f"cleaned text unavailable ({exc.__class__.__name__})" full = "" else: full = "" spans = [] for c in claims: v = c["value"] if isinstance(v, (int, float, str, bool)) and not (isinstance(v, str) and len(v) > 200): spans.append({"claim_id": c["id"], "property": c["property"], "value": v, **(_locate(full, v) if full else {"found": False, "reason": "no cleaned text"})}) out = {k: v for k, v in snap.items() if k != "text_path"} 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, "previous_snapshot": prev, "diff": snap.get("diff"), "claims": claims, "relations": relations, "results": results, "prices": prices, "events": events, "llm_jobs": llm, "entity_candidates": cands, "spans": spans, "spans_found": sum(1 for s in spans if s.get("found")), "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."}) return out # ------------------------------------------------------------------------------------------------------------------ /admin/quarantine @router.get("/quarantine") async def quarantine(status: str = Query("pending"), limit: int = Query(50, ge=1, le=500), offset: int = Query(0, ge=0)) -> dict[str, Any]: where = "true" if status == "all" else "q.status = :st" async with connection() as conn: 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", st=status, lim=limit, off=offset) total = await fetch_val(conn, f"select count(*) from quarantined_runs q where {where}", st=status) return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset} class QuarantineBody(BaseModel): action: str = Field(..., pattern="^(release|discard)$") note: str | None = Field(None, max_length=2000) @router.post("/quarantine/{quarantine_id}") async def quarantine_action(quarantine_id: str, body: QuarantineBody) -> dict[str, Any]: try: from aiatlas.services import canonical except ImportError: canonical = None # type: ignore[assignment] fn = getattr(canonical, f"{body.action}_quarantine", None) if canonical else None if fn is None: raise ApiError(501, f"quarantine {body.action} is not available yet: services.canonical.{body.action}_quarantine is missing (Stream A)") async with transaction() as conn: row = await fetch_one(conn, "select * from quarantined_runs where id = :id for update", id=quarantine_id) if not row: raise ApiError(404, "quarantined run not found") if row["status"] != "pending": raise ApiError(409, f"quarantined run already {row['status']}") result = await fn(conn, quarantine_id) if inspect.iscoroutinefunction(fn) else fn(conn, quarantine_id) await execute(conn, "update quarantined_runs set status = :st, resolved_at = now(), resolved_by = 'admin' where id = :id and status = 'pending'", st="released" if body.action == "release" else "discarded", id=quarantine_id) await cache.cache_invalidate() return {"ok": True, "id": quarantine_id, "action": body.action, "result": result} # ------------------------------------------------------------------------------------------------------------------ /admin/audit · rollback @router.get("/audit") async def audit_log(limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0), action: str | None = None) -> dict[str, Any]: where = "action ilike :a" if action else "true" async with connection() as conn: 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) total = await fetch_val(conn, f"select count(*) from admin_audit_log where {where}", a=f"%{action}%" if action else None) return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset} @router.post("/runs/{run_id}/rollback") async def rollback_run(run_id: str, request: Request) -> dict[str, Any]: """Undo one connector run WITHOUT deleting: retract its claims, close its relations/prices/results, flag its events as back-fill (meta.rolled_back).""" async with transaction() as conn: run = await fetch_one(conn, "select id, connector_name, status from connector_runs where id = :id", id=run_id) n: dict[str, int] = {} 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) 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) 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) 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) 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 = :r and not coalesce((meta->>'rolled_back')::boolean, false) returning 1) select count(*) from u""", r=run_id) or 0) # re-materialise attributes whose current claim was retracted: fall back to the best remaining claim affected = await fetch_all(conn, "select distinct entity_id, property from claims where run_id = :r and status = 'retracted'", r=run_id) restored = 0 for a in affected: 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"]) if prev: await execute(conn, "update claims set status = 'current', valid_to = null where id = :id", id=prev["id"]) 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"]) restored += 1 else: 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"]) n["attributes_restored"] = restored if run: 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) if not run and not any(n.values()): raise ApiError(404, f"no run or facts found for run_id {run_id!r}") await cache.cache_invalidate() await audit("rollback", run_id, {"counts": n, "connector": run["connector_name"] if run else None}, client_ip(request)) 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"} __all__ = ["DECISIONS", "router"]