spb/ai-atlas

1"""Anomaly persistence: the ontology's deterministic checks (`aiatlas.ontology.anomalies`) produce flags; this module upserts them in the2`anomalies` table (one row per dedupe key, reopened when a check fires again, auto-resolved when it stops firing). Never deletes,3never edits the flagged value."""4from __future__ import annotations56import logging7from typing import Any89from sqlalchemy.ext.asyncio import AsyncConnection10from ulid import ULID1112from aiatlas.db import execute, fetch_all, fetch_one, jsonb13from aiatlas.ontology.anomalies import Anomaly, check_hardware, check_model, check_price, check_result1415log = logging.getLogger(__name__)161718async def record(conn: AsyncConnection, anomaly: Anomaly) -> str:19    """Upsert one anomaly. Rows an operator marked `ignored`/`fixed` keep their status; `resolved` rows reopen."""20    row = await fetch_one(conn, """insert into anomalies (id, entity_id, check_name, severity, message, value, detail, status, dedupe_key)21                                   values (:id, :e, :c, :sev, :m, cast(:v as jsonb), cast(:d as jsonb), 'open', :k)22                                   on conflict (dedupe_key) do update set last_seen_at = now(), message = excluded.message, value = excluded.value,23                                       detail = excluded.detail, severity = excluded.severity,24                                       status = case when anomalies.status = 'resolved' then 'open' else anomalies.status end,25                                       resolved_at = case when anomalies.status = 'resolved' then null else anomalies.resolved_at end26                                   returning id""",27                         id=f"anom_{ULID()}", e=anomaly.entity_id, c=anomaly.check, sev=anomaly.severity, m=anomaly.message[:1000],28                         v=jsonb(anomaly.value) if anomaly.value is not None else None, d=jsonb(anomaly.detail or {}), k=anomaly.dedupe_key[:400])29    return row["id"] if row else ""303132async def run_checks(conn: AsyncConnection, *, resolve_stale: bool = True) -> dict[str, Any]:33    """Run every ontology check over live models/artifacts, hardware, live prices and live results; upsert flags; auto-resolve open flags whose34    check no longer fires. Returns counts by severity and the number resolved."""35    found: list[Anomaly] = []36    rows = await fetch_all(conn, "select id, canonical_name, attributes from entities where entity_type in ('model','artifact') and merged_into is null")37    for r in rows:38        found.extend(check_model(r["id"], r["canonical_name"], r["attributes"] or {}))39    rows = await fetch_all(conn, "select id, canonical_name, attributes from entities where entity_type = 'hardware' and merged_into is null")40    for r in rows:41        found.extend(check_hardware(r["id"], r["canonical_name"], r["attributes"] or {}))42    rows = await fetch_all(conn, """select p.*, m.canonical_name as model_name, v.canonical_name as provider_name from prices p43                                    join entities m on m.id = p.model_id join entities v on v.id = p.provider_id where p.valid_to is null""")44    for r in rows:45        found.extend(check_price(r))46    rows = await fetch_all(conn, """select r.id, r.model_id, r.benchmark_id, r.score, r.metric, r.unit, r.evaluated_at, m.canonical_name as model_name,47                                           b.canonical_name as benchmark_name, m.attributes->>'release_date' as model_release_date48                                    from benchmark_results r join entities m on m.id = r.model_id join entities b on b.id = r.benchmark_id49                                    where r.valid_to is null and r.is_current""")50    for r in rows:51        found.extend(check_result(r))52    keys: set[str] = set()53    by_sev: dict[str, int] = {}54    for a in found:55        key = a.dedupe_key[:400]56        if key in keys:57            continue58        keys.add(key)59        await record(conn, a)60        by_sev[a.severity] = by_sev.get(a.severity, 0) + 161    resolved = 062    if resolve_stale:63        row = await fetch_one(conn, """with s as (update anomalies set status = 'resolved', resolved_at = now(), resolution = 'check no longer fires'64                                       where status = 'open' and not (dedupe_key = any(cast(:keys as text[]))) returning 1) select count(*) as n from s""",65                              keys=sorted(keys))66        resolved = int(row["n"]) if row else 067    return {"flagged": len(keys), "by_severity": by_sev, "resolved": resolved}686970async def list_anomalies(conn: AsyncConnection, *, severity: str | None = None, status: str = "open", limit: int = 200) -> list[dict[str, Any]]:71    return await fetch_all(conn, """select a.*, e.canonical_name, e.slug, e.entity_type from anomalies a left join entities e on e.id = a.entity_id72                                    where (cast(:st as text) = '' or a.status = :st) and (cast(:sev as text) = '' or a.severity = :sev)73                                    order by case a.severity when 'critical' then 0 when 'warning' then 1 else 2 end, a.last_seen_at desc limit :n""",74                           st=status or "", sev=severity or "", n=limit)757677async def set_status(conn: AsyncConnection, anomaly_id: str, status: str, *, resolution: str | None = None) -> None:78    await execute(conn, "update anomalies set status = :s, resolution = coalesce(:r, resolution), resolved_at = case when :s in ('resolved','fixed','ignored') then now() else null end where id = :id",79                  s=status, r=resolution, id=anomaly_id)808182__all__ = ["list_anomalies", "record", "run_checks", "set_status"]83