HTML 77.2%
TypeScript 10.5%
Python 9.6%
JavaScript 2.5%
1"""Admin API (`x-aia-admin-token`): overview, connectors, runs/errors, documents & snapshots (cleaned text only), queues, LLM2accounting, review queue with entity merging, duplicates, curation, infrastructure, cache. Never returns raw archive paths."""3from __future__ import annotations45import platform6import socket7import time8from datetime import UTC, datetime9from typing import Any1011from fastapi import APIRouter, Depends, Query, Request12from pydantic import BaseModel, Field1314from aiatlas.api.common import ADMIN_DEPENDENCIES, PAGINATION, ApiError, Pagination, audit, client_ip, page15from aiatlas.config import settings16from aiatlas.connectors import registry as connector_registry17from aiatlas.db import connection, execute, fetch_all, fetch_one, fetch_val, jsonb, transaction18from aiatlas.ids import new_id19from aiatlas.sdk import archive20from aiatlas.services import cache21from aiatlas.services.jobs import enqueue, queue_depth22from aiatlas.services.llm import gateway23from aiatlas.services.merge import merge_entities24from aiatlas.services.stats import live_counts2526SKIP_AUDIT_GET = ("/api/v1/admin/overview", "/api/v1/admin/infrastructure", "/api/v1/admin/llm/health", "/api/v1/admin/audit")272829async def audit_request(request: Request) -> None:30 """Every admin call leaves a row in `admin_audit_log` (except the dashboard polling GETs). Runs AFTER the rate limit and the token check."""31 if request.method == "GET" and request.url.path in SKIP_AUDIT_GET:32 return33 payload: dict[str, Any] = {"method": request.method, "path": request.url.path}34 if request.query_params:35 payload["query"] = dict(request.query_params)36 if request.method in ("POST", "PATCH", "PUT", "DELETE"):37 try:38 body = await request.body()39 if body:40 import orjson4142 payload["body"] = orjson.loads(body) if len(body) < 64 * 1024 else {"truncated": True, "bytes": len(body)}43 except Exception: # noqa: BLE00144 payload["body"] = {"unparsed": True}45 target = request.path_params.get("name") or request.path_params.get("review_id") or request.path_params.get("job_id") or request.path_params.get("snap_id") \46 or request.path_params.get("doc_id") or request.path_params.get("entity_id") or request.path_params.get("run_id") or request.path_params.get("anomaly_id") \47 or request.path_params.get("quarantine_id") or request.path_params.get("a")48 await audit(f"{request.method} {request.url.path}", str(target) if target else None, payload, client_ip(request))495051router = APIRouter(prefix="/api/v1/admin", tags=["admin"], dependencies=[*ADMIN_DEPENDENCIES, Depends(audit_request)])52_STARTED = time.time()53SNAPSHOT_TEXT_LIMIT = 20 * 102454HIDDEN = ("raw_path", "text_path")555657def _public(row: dict[str, Any]) -> dict[str, Any]:58 return {k: v for k, v in row.items() if k not in HIDDEN}596061# ------------------------------------------------------------------------------------------------------------------ overview626364@router.get("/overview")65async def overview() -> dict[str, Any]:66 async with connection() as conn:67 stats = await live_counts(conn)68 queue = await queue_depth(conn)69 health = await fetch_all(conn, "select health, count(*) as n from connectors group by 1")70 errors = await fetch_val(conn, "select count(*) from connector_errors where created_at > now() - interval '24 hours'")71 llm = await fetch_one(conn, """select count(*) as jobs_24h, coalesce(sum(input_tokens), 0) + coalesce(sum(output_tokens), 0) as tokens_24h,72 count(*) filter (where status <> 'ok') as failed_24h from llm_jobs where created_at > now() - interval '24 hours'""")73 by_stage = await fetch_all(conn, "select stage, status, count(*) as n from llm_jobs where created_at > now() - interval '24 hours' group by 1, 2 order by 1, 2")74 recent_runs = await fetch_all(conn, "select * from connector_runs order by started_at desc limit 12")75 review_kinds = await fetch_all(conn, "select kind, count(*) as n from review_queue where status = 'pending' group by 1")76 counts = {h["health"]: int(h["n"]) for h in health}77 return {"stats": stats, "queue": queue, "heartbeats": await cache.heartbeats(),78 "connectors": {k: counts.get(k, 0) for k in ("ok", "degraded", "failing", "disabled", "unknown")}, "review_pending": stats.get("review_pending", 0),79 "review_by_kind": {r["kind"]: int(r["n"]) for r in review_kinds}, "recent_errors": int(errors or 0),80 "llm": {"available": gateway.available, "jobs_24h": int(llm["jobs_24h"] or 0), "tokens_24h": int(llm["tokens_24h"] or 0), "failed_24h": int(llm["failed_24h"] or 0),81 "by_stage": [{**r, "n": int(r["n"])} for r in by_stage]},82 "recent_runs": recent_runs, "archive": archive.archive_size(), "computed_at": datetime.now(UTC)}838485# ------------------------------------------------------------------------------------------------------------------ connectors868788@router.get("/connectors")89async def connectors() -> dict[str, Any]:90 async with connection() as conn:91 rows = await fetch_all(conn, """92 select c.*, s.key as source_key, s.name as source_name, s.tier as source_tier, s.domain as source_domain,93 (select row_to_json(r) from connector_runs r where r.connector_name = c.name order by r.started_at desc limit 1) as last_run,94 (select count(*) from documents d where d.connector_name = c.name) as documents,95 (select count(*) from snapshots x join documents d on d.id = x.document_id where d.connector_name = c.name) as snapshots,96 (select count(*) from connector_errors e where e.connector_name = c.name and e.created_at > now() - interval '7 days') as errors_7d97 from connectors c left join sources s on s.id = c.source_id order by c.priority, c.name""")98 known = connector_registry()99 return {"items": [{**r, "documents": int(r["documents"]), "snapshots": int(r["snapshots"]), "errors_7d": int(r["errors_7d"]), "in_code": r["name"] in known,100 "run_now_pending": await _run_now_pending(r["name"])} for r in rows],101 "unregistered_in_db": sorted(set(known) - {r["name"] for r in rows})}102103104async def _run_now_pending(name: str) -> bool:105 try:106 return bool(await cache.redis().exists(f"aia:run-now:{name}"))107 except Exception: # noqa: BLE001108 return False109110111class RunBody(BaseModel):112 force: bool = False113114115@router.post("/connectors/{name}/run")116async def run_connector(name: str, body: RunBody | None = None) -> dict[str, Any]:117 async with connection() as conn:118 exists = await fetch_one(conn, "select name from connectors where name = :n", n=name)119 if not exists and name not in connector_registry():120 raise ApiError(404, f"unknown connector {name!r}")121 try:122 await cache.redis().set(f"aia:run-now:{name}", b"force" if (body and body.force) else b"1", ex=6 * 3600)123 except Exception as exc:124 raise ApiError(503, f"redis unavailable: {exc.__class__.__name__}") from exc125 return {"queued": True, "connector": name, "force": bool(body and body.force), "note": "consumed by the scheduler tick (aia:run-now:<name>)"}126127128class ConnectorPatch(BaseModel):129 enabled: bool | None = None130 interval_seconds: int | None = Field(None, ge=60, le=30 * 86400)131 priority: int | None = Field(None, ge=0, le=9)132133134@router.patch("/connectors/{name}")135async def patch_connector(name: str, body: ConnectorPatch) -> dict[str, Any]:136 sets, params = [], {"n": name}137 if body.enabled is not None:138 sets.append("enabled = :enabled")139 sets.append("health = case when :enabled then (case when health = 'disabled' then 'unknown' else health end) else 'disabled' end")140 sets.append("circuit_open_until = case when :enabled then null else circuit_open_until end")141 params["enabled"] = body.enabled142 if body.interval_seconds is not None:143 sets.append("interval_seconds = :iv")144 sets.append("next_run_at = least(coalesce(next_run_at, now()), coalesce(last_success_at, now()) + make_interval(secs => :iv))")145 params["iv"] = body.interval_seconds146 if body.priority is not None:147 sets.append("priority = :pr")148 params["pr"] = body.priority149 if not sets:150 raise ApiError(400, "nothing to update")151 async with transaction() as conn:152 row = await fetch_one(conn, f"update connectors set {', '.join(sets)}, updated_at = now() where name = :n returning *", **params)153 if not row:154 raise ApiError(404, f"unknown connector {name!r}")155 return row156157158@router.get("/runs")159async def runs(connector: str | None = None, status: str | None = None, limit: int = Query(50, ge=1, le=500), offset: int = Query(0, ge=0)) -> dict[str, Any]:160 where = ["true"]161 params: dict[str, Any] = {"lim": limit, "off": offset}162 if connector:163 where.append("connector_name = :c")164 params["c"] = connector165 if status:166 where.append("status = :s")167 params["s"] = status168 async with connection() as conn:169 rows = await fetch_all(conn, f"select * from connector_runs where {' and '.join(where)} order by started_at desc limit :lim offset :off", **params)170 total = await fetch_val(conn, f"select count(*) from connector_runs where {' and '.join(where)}", **{k: v for k, v in params.items() if k not in ('lim', 'off')})171 return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset}172173174@router.get("/errors")175async def errors(connector: str | None = None, limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0)) -> dict[str, Any]:176 where = "connector_name = :c" if connector else "true"177 async with connection() as conn:178 rows = await fetch_all(conn, f"select * from connector_errors where {where} order by created_at desc limit :lim offset :off", c=connector, lim=limit, off=offset)179 total = await fetch_val(conn, f"select count(*) from connector_errors where {where}", c=connector)180 return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset}181182183# ------------------------------------------------------------------------------------------------------------------ documents & snapshots184185186@router.get("/documents")187async def documents(connector: str | None = None, status: str | None = None, q: str | None = Query(None, max_length=300), entity: str | None = None,188 needs_llm: int | None = Query(None, ge=0, le=1), p: Pagination = PAGINATION) -> dict[str, Any]:189 where = ["true"]190 params: dict[str, Any] = {}191 if connector:192 where.append("d.connector_name = :c")193 params["c"] = connector194 if status:195 where.append("d.status = :s")196 params["s"] = status197 if q:198 where.append("(d.url ilike :q or d.title ilike :q)")199 params["q"] = f"%{q}%"200 if entity:201 where.append("(d.entity_id = :e or e.slug = :e)")202 params["e"] = entity203 if needs_llm is not None:204 where.append("d.needs_llm = :nl")205 params["nl"] = bool(needs_llm)206 where_sql = " and ".join(where)207 async with connection() as conn:208 rows = await fetch_all(conn, f"""select d.*, e.slug as entity_slug, e.canonical_name as entity_name, e.entity_type,209 (select count(*) from snapshots x where x.document_id = d.id) as snapshots,210 (select x.processing_status from snapshots x where x.document_id = d.id order by x.observed_at desc limit 1) as last_processing_status211 from documents d left join entities e on e.id = d.entity_id where {where_sql}212 order by d.last_fetched_at desc nulls last, d.first_seen_at desc limit :lim offset :off""", lim=p.limit, off=p.offset, **params)213 total = await fetch_val(conn, f"select count(*) from documents d left join entities e on e.id = d.entity_id where {where_sql}", **params)214 return page([{**r, "snapshots": int(r["snapshots"])} for r in rows], int(total or 0), p)215216217@router.get("/documents/{doc_id}")218async def document(doc_id: str) -> dict[str, Any]:219 async with connection() as conn:220 doc = await fetch_one(conn, """select d.*, e.slug as entity_slug, e.canonical_name as entity_name, e.entity_type, s.name as source_name, s.tier as source_tier221 from documents d left join entities e on e.id = d.entity_id left join sources s on s.id = d.source_id where d.id = :id or d.url = :id limit 1""", id=doc_id)222 if not doc:223 raise ApiError(404, "document not found")224 snaps = await fetch_all(conn, """select id, run_id, url, final_url, observed_at, http_status, content_type, content_hash, byte_size, text_hash, parser_version, connector_version,225 transport, changed, processing_status, created_at, structured is not null as has_structured, diff is not null as has_diff, text_path is not null as has_text226 from snapshots where document_id = :id order by observed_at desc limit 200""", id=doc["id"])227 claims = await fetch_val(conn, "select count(*) from claims c join snapshots s on s.id = c.snapshot_id where s.document_id = :id", id=doc["id"])228 return {**doc, "snapshots": snaps, "claims_from_document": int(claims or 0)}229230231@router.get("/snapshots/{snap_id}")232async def snapshot(snap_id: str, text_limit: int = Query(SNAPSHOT_TEXT_LIMIT, ge=0, le=SNAPSHOT_TEXT_LIMIT)) -> dict[str, Any]:233 async with connection() as conn:234 snap = await fetch_one(conn, """select s.*, d.url as document_url, d.doc_type, d.connector_name, d.entity_id, e.slug as entity_slug, e.canonical_name as entity_name235 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)236 if not snap:237 raise ApiError(404, "snapshot not found")238 claims = await fetch_all(conn, """select c.id, c.entity_id, e.slug as entity_slug, c.property, c.value, c.unit, c.status, c.confidence, c.extractor, c.observed_at239 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 500""", id=snap_id)240 events = await fetch_all(conn, "select id, entity_id, event_type, category, property, summary, importance, observed_at from change_events where snapshot_id = :id order by observed_at desc limit 200", id=snap_id)241 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)242 text, text_error, text_len = None, None, 0243 if snap.get("text_path") and text_limit:244 try:245 full = archive.load_text(snap["text_path"])246 text_len = len(full)247 text = full[:text_limit]248 except OSError as exc:249 text_error = f"cleaned text unavailable ({exc.__class__.__name__})"250 out = _public(snap)251 out.update({"text": text, "text_chars": text_len, "text_truncated": text_len > text_limit, "text_error": text_error, "has_raw": bool(snap.get("raw_path")),252 "claims": claims, "events": events, "llm_jobs": llm})253 return out254255256# ------------------------------------------------------------------------------------------------------------------ queues257258259@router.get("/jobs")260async def jobs(status: str | None = None, kind: str | None = None, limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0)) -> dict[str, Any]:261 where = ["true"]262 params: dict[str, Any] = {}263 if status:264 where.append("status = any(cast(:s as text[]))")265 params["s"] = [x.strip() for x in status.split(",") if x.strip()]266 if kind:267 where.append("kind = :k")268 params["k"] = kind269 where_sql = " and ".join(where)270 async with connection() as conn:271 rows = await fetch_all(conn, f"select * from jobs where {where_sql} order by case status when 'running' then 0 when 'queued' then 1 when 'failed' then 2 when 'dead' then 3 else 4 end, "272 f"priority, coalesce(finished_at, run_after) desc limit :lim offset :off", lim=limit, off=offset, **params)273 total = await fetch_val(conn, f"select count(*) from jobs where {where_sql}", **params)274 depth = await queue_depth(conn)275 return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset, "depth": depth}276277278@router.post("/jobs/{job_id}/retry")279async def retry_job(job_id: str) -> dict[str, Any]:280 async with transaction() as conn:281 row = await fetch_one(conn, """update jobs set status = 'queued', attempts = 0, error = null, locked_by = null, locked_at = null, finished_at = null, run_after = now()282 where id = :id and status in ('failed', 'dead', 'done') returning id, kind, status""", id=job_id)283 if not row:284 raise ApiError(404, "job not found or not retryable (must be failed, dead or done)")285 return {"ok": True, **row}286287288@router.post("/jobs/requeue-dead")289async def requeue_dead(kind: str | None = None) -> dict[str, Any]:290 where = "j.status = 'dead'" + (" and j.kind = :k" if kind else "")291 async with transaction() as conn:292 # respect the partial unique index on dedupe_key (queued|running): skip dead jobs already superseded by a live one,293 # and requeue only one dead job per dedupe_key (the most recent); the other duplicates are closed as superseded.294 n = await fetch_val(conn, f"""with pick as (295 select distinct on (coalesce(j.dedupe_key, j.id)) j.id from jobs j296 where {where} and (j.dedupe_key is null or not exists (select 1 from jobs q where q.dedupe_key = j.dedupe_key and q.status in ('queued','running')))297 order by coalesce(j.dedupe_key, j.id), j.created_at desc),298 u as (update jobs set status = 'queued', attempts = 0, error = null, locked_by = null, locked_at = null, finished_at = null, run_after = now()299 where id in (select id from pick) returning 1)300 select count(*) from u""", k=kind)301 await execute(conn, f"update jobs j set status = 'done', error = coalesce(error, '') || ' [superseded]' where {where}", k=kind)302 return {"requeued": int(n or 0)}303304305@router.get("/llm-jobs")306async def llm_jobs(limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0), status: str | None = None, task: str | None = None) -> dict[str, Any]:307 where = ["true"]308 params: dict[str, Any] = {}309 if status:310 where.append("status = :s")311 params["s"] = status312 if task:313 where.append("task_type = :t")314 params["t"] = task315 where_sql = " and ".join(where)316 async with connection() as conn:317 rows = await fetch_all(conn, f"select id, job_id, task_type, stage, engine, model, node, schema_name, snapshot_id, entity_id, input_tokens, output_tokens, duration_ms, status, error, created_at "318 f"from llm_jobs where {where_sql} order by created_at desc limit :lim offset :off", lim=limit, off=offset, **params)319 total = await fetch_val(conn, f"select count(*) from llm_jobs where {where_sql}", **params)320 totals = await fetch_all(conn, """select stage, model, status, count(*) as n, coalesce(sum(input_tokens), 0) as input_tokens, coalesce(sum(output_tokens), 0) as output_tokens,321 coalesce(avg(duration_ms), 0)::int as avg_ms from llm_jobs group by 1, 2, 3 order by 1, 2, 3""")322 return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset, "totals": [{**t, "n": int(t["n"]), "input_tokens": int(t["input_tokens"]), "output_tokens": int(t["output_tokens"])} for t in totals]}323324325@router.get("/llm/health")326async def llm_health() -> dict[str, Any]:327 h = await gateway.health()328 h["base_url_configured"] = bool(settings.llm_base_url)329 return h330331332class LLMEnqueue(BaseModel):333 limit: int = Field(200, ge=1, le=5000)334 task: str | None = None335 connector: str | None = None336337338@router.post("/llm/enqueue")339async def llm_enqueue(body: LLMEnqueue | None = None) -> dict[str, Any]:340 body = body or LLMEnqueue()341 where = "s.processing_status in ('stored', 'extracted', 'llm_pending') and s.text_path is not null and (d.needs_llm or s.processing_status = 'llm_pending')"342 params: dict[str, Any] = {"lim": body.limit}343 if body.connector:344 where += " and d.connector_name = :c"345 params["c"] = body.connector346 queued = 0347 async with transaction() as conn:348 snaps = await fetch_all(conn, f"select s.id from snapshots s join documents d on d.id = s.document_id where {where} order by d.priority, s.observed_at desc limit :lim", **params)349 for s in snaps:350 payload = {"snapshot_id": s["id"], **({"task": body.task} if body.task else {})}351 if await enqueue(conn, "llm_extract", payload, priority=6, dedupe_key=f"llm:{s['id']}"):352 queued += 1353 if snaps:354 await execute(conn, "update snapshots set processing_status = 'llm_pending' where id = any(cast(:ids as text[])) and processing_status <> 'llm_pending'", ids=[s["id"] for s in snaps])355 return {"queued": queued, "candidates": len(snaps), "llm_available": gateway.available}356357358class ReprocessBody(BaseModel):359 connector: str360 url: str | None = None361362363@router.post("/reprocess")364async def reprocess(body: ReprocessBody) -> dict[str, Any]:365 if body.connector not in connector_registry():366 raise ApiError(404, f"unknown connector {body.connector!r}")367 async with transaction() as conn:368 jid = await enqueue(conn, "reprocess_snapshot", {"connector": body.connector, **({"url": body.url} if body.url else {})}, priority=3,369 dedupe_key=f"reprocess:{body.connector}:{body.url or '*'}")370 return {"queued": jid is not None, "job_id": jid}371372373# ------------------------------------------------------------------------------------------------------------------ review queue & curation374375376@router.get("/review")377async def review(status: str = "pending", kind: str | None = None, limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0)) -> dict[str, Any]:378 where = ["r.status = :s"] if status != "all" else ["true"]379 params: dict[str, Any] = {"s": status}380 if kind:381 where.append("r.kind = :k")382 params["k"] = kind383 where_sql = " and ".join(where)384 async with connection() as conn:385 rows = await fetch_all(conn, f"""select r.*, (select jsonb_agg(jsonb_build_object('id', e.id, 'slug', e.slug, 'name', e.canonical_name, 'entity_type', e.entity_type, 'status', e.status) order by e.slug)386 from entities e where e.id = any(r.entity_ids)) as entities387 from review_queue r where {where_sql} order by r.created_at desc limit :lim offset :off""", lim=limit, off=offset, **params)388 total = await fetch_val(conn, f"select count(*) from review_queue r where {where_sql}", **params)389 kinds = await fetch_all(conn, "select kind, status, count(*) as n from review_queue group by 1, 2 order by 1, 2")390 return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset, "by_kind": [{**k, "n": int(k["n"])} for k in kinds]}391392393class ReviewAction(BaseModel):394 action: str = Field(..., pattern="^(approve|reject|edit)$")395 resolution: dict[str, Any] | None = None396397398@router.post("/review/{review_id}")399async def review_action(review_id: str, body: ReviewAction) -> dict[str, Any]:400 async with transaction() as conn:401 item = await fetch_one(conn, "select * from review_queue where id = :id for update", id=review_id)402 if not item:403 raise ApiError(404, "review item not found")404 if item["status"] != "pending":405 raise ApiError(409, f"review item already {item['status']}")406 resolution: dict[str, Any] = dict(body.resolution or {})407 effect: dict[str, Any] | None = None408 if body.action == "approve" and item["kind"] == "merge_candidate":409 payload = item["payload"] or {}410 ids = list(item["entity_ids"] or [])411 source = resolution.get("source_id") or payload.get("source_id") or (ids[0] if len(ids) > 1 else None)412 target = resolution.get("target_id") or payload.get("target_id") or (ids[1] if len(ids) > 1 else None)413 if not source or not target:414 raise ApiError(400, "merge_candidate needs source_id and target_id (payload or resolution)")415 try:416 effect = await merge_entities(conn, source, target)417 except (ValueError, LookupError) as exc:418 raise ApiError(400, str(exc)) from exc419 resolution.update({"merged": effect})420 elif body.action == "approve" and item["kind"] == "conflict":421 keep = resolution.get("keep_claim_id")422 if keep:423 effect = await _accept_claim(conn, keep)424 resolution["accepted"] = effect425 status = {"approve": "approved", "reject": "rejected", "edit": "edited"}[body.action]426 await execute(conn, "update review_queue set status = :st, resolution = cast(:r as jsonb), resolved_at = now() where id = :id", st=status, r=jsonb(resolution), id=review_id)427 await cache.cache_invalidate()428 return {"ok": True, "id": review_id, "status": status, "effect": effect}429430431async def _accept_claim(conn: Any, claim_id: str) -> dict[str, Any]:432 """Conflict resolution: promote one claim to current, supersede the others for the same property."""433 c = await fetch_one(conn, "select * from claims where id = :id", id=claim_id)434 if not c:435 raise ApiError(404, "claim not found")436 await execute(conn, "update claims set status = 'superseded', valid_to = now() where entity_id = :e and property = :p and id <> :id and status in ('current', 'conflicting')",437 e=c["entity_id"], p=c["property"], id=claim_id)438 await execute(conn, "update claims set status = 'current', confidence = case when confidence = 'conflicted' then 'high' else confidence end, valid_to = null where id = :id", id=claim_id)439 prov = {"source_id": c["source_id"], "snapshot_id": c["snapshot_id"], "url": c["source_url"], "tier": c["tier"], "confidence": "high", "extractor": c["extractor"],440 "observed_at": c["observed_at"].isoformat() if c["observed_at"] else None, "resolved_by": "review"}441 await execute(conn, "update entities set attributes = attributes || jsonb_build_object(:p, cast(:v as jsonb)), provenance = provenance || jsonb_build_object(:p, cast(:pv as jsonb)), updated_at = now() where id = :e",442 p=c["property"], v=jsonb(c["value"]), pv=jsonb(prov), e=c["entity_id"])443 return {"claim_id": claim_id, "entity_id": c["entity_id"], "property": c["property"]}444445446@router.get("/entities/duplicates")447async def duplicates(type: str | None = Query(None, alias="type"), limit: int = Query(100, ge=1, le=500), threshold: float = Query(0.8, ge=0.3, le=1.0)) -> dict[str, Any]:448 where = "a.entity_type = :t" if type else "true"449 async with connection() as conn:450 # `%` uses the pg_trgm GIN index at the session default threshold (0.3, a superset of any threshold ≥ 0.3); the exact bound is applied451 # with similarity() > :th — no set_limit(), which used to leak a GUC change into the pooled connection.452 rows = await fetch_all(conn, f"""453 select a.id as a_id, a.slug as a_slug, a.canonical_name as a_name, a.entity_type, a.first_seen_at as a_first_seen_at, oa.canonical_name as a_org,454 b.id as b_id, b.slug as b_slug, b.canonical_name as b_name, b.first_seen_at as b_first_seen_at, ob.canonical_name as b_org,455 similarity(a.canonical_name, b.canonical_name) as similarity,456 (select count(*) from claims c where c.entity_id = a.id and c.status = 'current') as a_claims,457 (select count(*) from claims c where c.entity_id = b.id and c.status = 'current') as b_claims458 from entities a join entities b on b.entity_type = a.entity_type and b.id > a.id and a.canonical_name % b.canonical_name459 left join entities oa on oa.id = a.organization_id left join entities ob on ob.id = b.organization_id460 where {where} and a.merged_into is null and b.merged_into is null and similarity(a.canonical_name, b.canonical_name) > :th461 order by similarity desc, a.canonical_name limit :lim""", t=type, th=threshold, lim=limit)462 return {"threshold": threshold, "items": [{"entity_type": r["entity_type"], "similarity": round(float(r["similarity"]), 3),463 "a": {"id": r["a_id"], "slug": r["a_slug"], "name": r["a_name"], "organization": r["a_org"], "first_seen_at": r["a_first_seen_at"], "claims": int(r["a_claims"])},464 "b": {"id": r["b_id"], "slug": r["b_slug"], "name": r["b_name"], "organization": r["b_org"], "first_seen_at": r["b_first_seen_at"], "claims": int(r["b_claims"])}}465 for r in rows]}466467468class MergeBody(BaseModel):469 source_id: str470 target_id: str471472473@router.post("/entities/merge")474async def merge(body: MergeBody) -> dict[str, Any]:475 async with transaction() as conn:476 src = await fetch_one(conn, "select id from entities where id = :k or slug = :k limit 1", k=body.source_id)477 dst = await fetch_one(conn, "select id from entities where id = :k or slug = :k limit 1", k=body.target_id)478 if not src or not dst:479 raise ApiError(404, "source or target entity not found")480 try:481 result = await merge_entities(conn, src["id"], dst["id"])482 except (ValueError, LookupError) as exc:483 raise ApiError(400, str(exc)) from exc484 await cache.cache_invalidate()485 return result486487488@router.post("/entities/{entity_id}/claims/{claim_id}/retract")489async def retract_claim(entity_id: str, claim_id: str) -> dict[str, Any]:490 async with transaction() as conn:491 ent = await fetch_one(conn, "select id from entities where id = :k or slug = :k limit 1", k=entity_id)492 if not ent:493 raise ApiError(404, "entity not found")494 c = await fetch_one(conn, "select * from claims where id = :c and entity_id = :e", c=claim_id, e=ent["id"])495 if not c:496 raise ApiError(404, "claim not found for this entity")497 was_current = c["status"] == "current"498 await execute(conn, "update claims set status = 'retracted', valid_to = coalesce(valid_to, now()) where id = :c", c=claim_id)499 restored = None500 if was_current:501 prev = await fetch_one(conn, """select * from claims where entity_id = :e and property = :p and id <> :c and status in ('superseded', 'conflicting')502 order by tier, valid_from desc limit 1""", e=ent["id"], p=c["property"], c=claim_id)503 if prev:504 await execute(conn, "update claims set status = 'current', valid_to = null where id = :id", id=prev["id"])505 prov = {"source_id": prev["source_id"], "snapshot_id": prev["snapshot_id"], "url": prev["source_url"], "tier": prev["tier"], "confidence": prev["confidence"],506 "extractor": prev["extractor"], "observed_at": prev["observed_at"].isoformat() if prev["observed_at"] else None, "restored_by": "retraction"}507 await execute(conn, "update entities set attributes = attributes || jsonb_build_object(:p, cast(:v as jsonb)), provenance = provenance || jsonb_build_object(:p, cast(:pv as jsonb)), updated_at = now() where id = :e",508 p=c["property"], v=jsonb(prev["value"]), pv=jsonb(prov), e=ent["id"])509 restored = prev["id"]510 else:511 await execute(conn, "update entities set attributes = attributes - :p, provenance = provenance - :p, updated_at = now() where id = :e", p=c["property"], e=ent["id"])512 await execute(conn, """insert into change_events (id, entity_id, event_type, category, property, old_value, summary, importance, connector_name, dedupe_key)513 values (:id, :e, 'CLAIM_RETRACTED', 'source', :p, cast(:v as jsonb), :s, 1, 'curation', :dk) on conflict (dedupe_key) do nothing""",514 id=new_id("change_event"), e=ent["id"], p=c["property"], v=jsonb(c["value"]),515 s=f"Retracted claim {c['property']}", dk=f"retract:{claim_id}")516 await cache.cache_invalidate()517 return {"ok": True, "claim_id": claim_id, "property": c["property"], "was_current": was_current, "restored_claim_id": restored}518519520# ------------------------------------------------------------------------------------------------------------------ infrastructure & maintenance521522523@router.get("/infrastructure")524async def infrastructure() -> dict[str, Any]:525 async with connection() as conn:526 db = await fetch_one(conn, """select pg_database_size(current_database()) as db_bytes, current_database() as database, version() as pg_version,527 (select count(*) from pg_stat_activity where datname = current_database()) as connections,528 (select setting from pg_settings where name = 'max_connections') as max_connections""")529 tables = await fetch_all(conn, """select c.relname as table, pg_total_relation_size(c.oid) as total_bytes, pg_relation_size(c.oid) as data_bytes,530 coalesce(s.n_live_tup, 0) as rows_estimate from pg_class c join pg_namespace n on n.oid = c.relnamespace531 left join pg_stat_user_tables s on s.relid = c.oid where n.nspname = 'public' and c.relkind = 'r'532 order by pg_total_relation_size(c.oid) desc limit 20""")533 extensions = await fetch_all(conn, "select extname, extversion from pg_extension order by 1")534 redis_info: dict[str, Any] = {"ok": False}535 try:536 info = await cache.redis().info("memory")537 redis_info = {"ok": True, "used_memory": info.get("used_memory"), "used_memory_human": info.get("used_memory_human"),538 "api_cache_keys": sum([1 async for _ in cache.redis().scan_iter(match="aia:api:*", count=1000)])}539 except Exception as exc: # noqa: BLE001540 redis_info = {"ok": False, "error": exc.__class__.__name__}541 return {"hostname": socket.gethostname(), "python": platform.python_version(), "platform": platform.platform(), "api_uptime_s": int(time.time() - _STARTED),542 "env": settings.app_env, "heartbeats": await cache.heartbeats(), "archive": archive.archive_size(), "data_dir_exists": settings.data_dir.exists(),543 "database": {**(db or {}), "db_bytes": int((db or {}).get("db_bytes") or 0), "tables": tables, "extensions": extensions}, "redis": redis_info,544 "llm": {"available": gateway.available, "engine": gateway.engine.name}, "scheduler_tick_s": settings.scheduler_tick_s}545546547CACHE_PREFIXES_AFTER_RUN = ("/api/v1/stats", "/api/v1/changes", "/api/v1/benchmarks", "/api/v1/prices", "/api/v1/models", "/api/v1/deployments", "/api/v1/frontier",548 "/api/v1/pulse", "/api/v1/open", "/api/v1/families", "/api/v1/timeline", "/api/v1/diff", "facets:", "frontier:")549550551@router.post("/cache/flush")552async def cache_flush(prefix: str = "", after_run: int = Query(0, ge=0, le=1)) -> dict[str, Any]:553 """Flush `aia:api:<prefix>*`. `after_run=1` flushes the read paths a connector run invalidates (the scheduler calls `services.cache.cache_invalidate`554 with these prefixes after every run — see docs/API.md)."""555 if after_run:556 flushed = {p: await cache.cache_invalidate(p) for p in CACHE_PREFIXES_AFTER_RUN}557 return {"flushed": sum(flushed.values()), "by_prefix": flushed}558 return {"flushed": await cache.cache_invalidate(prefix), "prefix": prefix or "*"}559560561@router.post("/stats/recompute")562async def stats_recompute() -> dict[str, Any]:563 from aiatlas.services.stats import compute_stats564565 counts = await compute_stats()566 await cache.cache_invalidate("/api/v1/stats")567 return {"ok": True, "entities_total": counts.get("entities_total")}568569570class QualityBody(BaseModel):571 entity_ids: list[str] | None = None572 limit: int = Field(20000, ge=1, le=200000)573574575@router.post("/quality/recompute")576async def quality_recompute(body: QualityBody | None = None) -> dict[str, Any]:577 from aiatlas.services.quality import recompute578579 body = body or QualityBody()580 res = await recompute(entity_ids=body.entity_ids, limit=body.limit)581 await cache.cache_invalidate()582 return {"ok": True, **res}583