1"""FactWriter — turns `Facts` into versioned database state.23Temporal rules (per entity × property):4 * no current claim → insert current claim, set attribute, NEW_* event for new entities5 * same value → confirm (observed_at bumped), nothing else6 * different, source ≥ tier → supersede (valid_to = observed), insert new current, update attribute, CHANGE event for material props7 * different, worse source → store as `conflicting`, flag confidence, review-queue item — never overwrite8Prices and benchmark results have their own append-only tables with the same close/open semantics.9"""10from __future__ import annotations1112import hashlib13import json14import logging15from datetime import UTC, datetime16from typing import Any1718from sqlalchemy.ext.asyncio import AsyncConnection1920from aiatlas.db import execute, fetch_one, jsonb21from aiatlas.ids import new_id, normalize_alias22from aiatlas.sdk.facts import EVENT_CATEGORY_BY_TYPE, MATERIAL_PROPERTIES, NOISY_PREFIXES, EntityRef, Facts, PriceObs, ResultObs23from aiatlas.sdk.resolution import Resolver2425log = logging.getLogger(__name__)2627TIER_CONFIDENCE = {1: "high", 2: "medium", 3: "low", 4: "low"}28SOFT_PROPERTIES = {"description", "summary", "tagline", "availability_note", "abstract", "notes", "training_data_notes", "safety_notes", "hardware_requirements"}29NEW_IMPORTANCE = {"model": 3, "company": 2, "provider": 2, "paper": 1, "dataset": 1, "benchmark": 2, "framework": 1, "hardware": 2,30 "tool": 1, "repository": 0, "release": 2, "regulation": 2, "incident": 2, "organization": 1, "researcher": 0}313233def _norm_value(v: Any) -> Any:34 if isinstance(v, datetime):35 return v.astimezone(UTC).isoformat(timespec="seconds")36 if isinstance(v, (set, tuple)):37 return sorted(v) if all(isinstance(x, str) for x in v) else list(v)38 if isinstance(v, float) and v.is_integer() and abs(v) < 1e15:39 return int(v)40 return v414243def _same(a: Any, b: Any) -> bool:44 return json.dumps(_norm_value(a), sort_keys=True, default=str) == json.dumps(_norm_value(b), sort_keys=True, default=str)454647class WriteStats:48 def __init__(self) -> None:49 self.entities_created = 050 self.entities_updated = 051 self.claims = 052 self.relations = 053 self.events = 054 self.prices = 055 self.results = 056 self.conflicts = 05758 def as_dict(self) -> dict[str, int]:59 return dict(self.__dict__)606162class FactWriter:63 def __init__(self, conn: AsyncConnection, *, source_id: str | None, snapshot_id: str | None, source_url: str | None,64 tier: int = 2, connector_name: str | None = None, extractor: str = "deterministic", extractor_version: str = "1",65 observed_at: datetime | None = None):66 self.conn = conn67 self.source_id = source_id68 self.snapshot_id = snapshot_id69 self.source_url = source_url70 self.tier = tier71 self.connector_name = connector_name72 self.extractor = extractor73 self.extractor_version = extractor_version74 self.observed_at = observed_at or datetime.now(UTC)75 self.resolver = Resolver(conn, snapshot_id=snapshot_id, source_tier=tier)76 self.stats = WriteStats()7778 # ---------------------------------------------------------------------------------------------- entry point79 async def write(self, facts: Facts) -> WriteStats:80 new_before = len(self.resolver.created)81 for ref in facts.entities:82 await self.resolver.resolve(ref)83 for prop, value in ref.attributes.items():84 await self.write_claim(ref, prop, value)85 for c in facts.claims:86 await self.write_claim(c.entity, c.property, c.value, unit=c.unit, confidence=c.confidence, observed_at=c.observed_at,87 effective_at=c.effective_at, source_url=c.source_url)88 for r in facts.relations:89 await self.write_relation(r.subject, r.predicate, r.object, r.attributes, confidence=r.confidence, source_url=r.source_url)90 for p in facts.prices:91 await self.write_price(p)92 for res in facts.results:93 await self.write_result(res)94 for e in facts.events:95 eid = await self.resolver.resolve(e.entity) if e.entity else None96 await self.emit_event(e.event_type, e.category, e.summary, entity_id=eid, old_value=e.old_value, new_value=e.new_value,97 importance=e.importance, effective_at=e.effective_at, dedupe_key=e.dedupe_key, source_url=e.source_url, meta=e.meta)98 # NEW_* events for every entity created in this write99 for eid in self.resolver.created[new_before:]:100 await self._new_entity_event(eid)101 self.stats.entities_created = len(self.resolver.created)102 self.stats.entities_updated = len(self.resolver.updated - set(self.resolver.created))103 return self.stats104105 # ---------------------------------------------------------------------------------------------- claims106 async def write_claim(self, ref: EntityRef, prop: str, value: Any, *, unit: str | None = None, confidence: str | None = None,107 observed_at: datetime | None = None, effective_at: datetime | None = None, source_url: str | None = None) -> None:108 if value is None or value == "" or value == [] or value == {}:109 return110 eid = await self.resolver.resolve(ref)111 assert eid112 value = _norm_value(value)113 observed = observed_at or self.observed_at114 conf = confidence or TIER_CONFIDENCE.get(self.tier, "medium")115 url = source_url or self.source_url116 noisy = prop.startswith(NOISY_PREFIXES)117 current = await fetch_one(self.conn, """select id, value, tier, source_url, observed_at from claims118 where entity_id = :e and property = :p and status = 'current' order by valid_from desc limit 1""",119 e=eid, p=prop)120 is_new_entity = eid in self.resolver.created121 if current is None:122 await self._insert_claim(eid, prop, value, unit, conf, "current", observed, effective_at, url)123 await self._set_attribute(eid, prop, value, unit, conf, url, observed)124 return125 if _same(current["value"], value):126 await execute(self.conn, "update claims set observed_at = greatest(observed_at, :o) where id = :id", o=observed, id=current["id"])127 if noisy:128 await self._set_attribute(eid, prop, value, unit, conf, url, observed)129 return130 same_source = bool(url and current["source_url"] == url)131 if prop in SOFT_PROPERTIES and not same_source:132 # soft text (descriptions, notes): first statement wins until *its own* source changes; never an event, never a conflict133 return134 if self.tier <= (current["tier"] or 2) or same_source:135 # supersede136 await execute(self.conn, "update claims set status = 'superseded', valid_to = :o where id = :id", o=observed, id=current["id"])137 await self._insert_claim(eid, prop, value, unit, conf, "current", observed, effective_at, url)138 await self._set_attribute(eid, prop, value, unit, conf, url, observed)139 if not noisy and not is_new_entity and prop not in SOFT_PROPERTIES:140 await self._property_change_event(eid, prop, current["value"], value, observed, effective_at, url)141 else:142 await self._insert_claim(eid, prop, value, unit, conf, "conflicting", observed, effective_at, url)143 await execute(self.conn, "update claims set confidence = 'conflicted' where id = :id", id=current["id"])144 await execute(self.conn, """update entities set quality = quality || jsonb_build_object('conflicts', coalesce((quality->>'conflicts')::int, 0) + 1)145 where id = :id""", id=eid)146 self.stats.conflicts += 1147 dedupe = f"conflict:{eid}:{prop}:{hashlib.sha1(json.dumps(value, sort_keys=True, default=str).encode()).hexdigest()[:10]}"148 await execute(self.conn, """insert into review_queue (id, kind, entity_ids, payload, reason, dedupe_key)149 values (:id, 'conflict', :ids, cast(:p as jsonb), :r, :d) on conflict (dedupe_key) do nothing""",150 id=new_id("review"), ids=[eid], p=jsonb({"property": prop, "current": current["value"], "current_source": current["source_url"],151 "claimed": value, "claimed_source": url, "tier": self.tier}),152 r=f"source disagrees on {prop}", d=dedupe)153154 async def _insert_claim(self, eid: str, prop: str, value: Any, unit: str | None, conf: str, status: str, observed: datetime,155 effective_at: datetime | None, url: str | None) -> None:156 text_val = value if isinstance(value, str) else None157 num_val = float(value) if isinstance(value, (int, float)) and not isinstance(value, bool) else None158 await execute(self.conn, """insert into claims (id, entity_id, property, value, value_text, value_num, unit, source_id, snapshot_id, source_url, tier,159 confidence, status, extractor, extractor_version, observed_at, effective_at, valid_from)160 values (:id, :e, :p, cast(:v as jsonb), :vt, :vn, :u, :src, :snap, :url, :tier, :conf, :status, :ex, :exv, :o, :eff, :vf)""",161 id=new_id("claim"), e=eid, p=prop, v=jsonb(value), vt=text_val[:2000] if text_val else None, vn=num_val, u=unit, src=self.source_id,162 snap=self.snapshot_id, url=url, tier=self.tier, conf=conf, status=status, ex=self.extractor, exv=self.extractor_version,163 o=observed, eff=effective_at, vf=effective_at or observed)164 self.stats.claims += 1165166 async def _set_attribute(self, eid: str, prop: str, value: Any, unit: str | None, conf: str, url: str | None, observed: datetime) -> None:167 prov = {"source_id": self.source_id, "snapshot_id": self.snapshot_id, "url": url, "observed_at": observed.isoformat(timespec="seconds"),168 "tier": self.tier, "confidence": conf, "extractor": self.extractor}169 if unit:170 prov["unit"] = unit171 await execute(self.conn, """update entities set attributes = attributes || jsonb_build_object(cast(:p as text), cast(:v as jsonb)),172 provenance = provenance || jsonb_build_object(cast(:p as text), cast(:prov as jsonb)), last_seen_at = greatest(last_seen_at, :o)173 where id = :id""", p=prop, v=jsonb(value), prov=jsonb(prov), o=observed, id=eid)174 if prop == "status" and isinstance(value, str):175 await execute(self.conn, "update entities set status = :s where id = :id", s=value[:40], id=eid)176 if prop == "description" and isinstance(value, str):177 await execute(self.conn, "update entities set description = :d where id = :id", d=value[:4000], id=eid)178179 async def _property_change_event(self, eid: str, prop: str, old: Any, new: Any, observed: datetime, effective_at: datetime | None,180 url: str | None) -> None:181 event_type, importance = MATERIAL_PROPERTIES.get(prop, ("PROPERTY_CHANGED", 0))182 row = await fetch_one(self.conn, "select canonical_name, entity_type from entities where id = :id", id=eid)183 name = row["canonical_name"] if row else eid184 category = EVENT_CATEGORY_BY_TYPE.get(row["entity_type"] if row else "", "update")185 if prop == "status" and str(new).lower() in ("deprecated", "retired", "discontinued"):186 importance = 3187 summary = f"{name}: {prop.replace('_', ' ')} changed from {_short(old)} to {_short(new)}"188 dedupe = f"{event_type}:{eid}:{prop}:{hashlib.sha1(json.dumps([old, new], sort_keys=True, default=str).encode()).hexdigest()[:12]}"189 await self.emit_event(event_type, category, summary, entity_id=eid, old_value=old, new_value=new, importance=importance,190 effective_at=effective_at, dedupe_key=dedupe, source_url=url, meta={"property": prop}, observed_at=observed)191192 # ---------------------------------------------------------------------------------------------- relations193 async def write_relation(self, subject: EntityRef, predicate: str, obj: EntityRef, attributes: dict[str, Any] | None = None, *,194 confidence: str | None = None, source_url: str | None = None) -> None:195 sid = await self.resolver.resolve(subject)196 oid = await self.resolver.resolve(obj)197 if not sid or not oid or sid == oid:198 return199 conf = confidence or TIER_CONFIDENCE.get(self.tier, "medium")200 existing = await fetch_one(self.conn, "select id, attributes from relations where subject_id = :s and predicate = :p and object_id = :o and valid_to is null",201 s=sid, p=predicate, o=oid)202 if existing:203 if attributes and not _same(existing["attributes"], {**existing["attributes"], **attributes}):204 await execute(self.conn, "update relations set attributes = attributes || cast(:a as jsonb), observed_at = :o where id = :id",205 a=jsonb(attributes), o=self.observed_at, id=existing["id"])206 else:207 await execute(self.conn, "update relations set observed_at = :o where id = :id", o=self.observed_at, id=existing["id"])208 return209 await execute(self.conn, """insert into relations (id, subject_id, predicate, object_id, attributes, source_id, snapshot_id, source_url, tier, confidence, observed_at, valid_from)210 values (:id, :s, :p, :o, cast(:a as jsonb), :src, :snap, :url, :tier, :conf, :obs, :obs)""",211 id=new_id("relation"), s=sid, p=predicate, o=oid, a=jsonb(attributes or {}), src=self.source_id, snap=self.snapshot_id,212 url=source_url or self.source_url, tier=self.tier, conf=conf, obs=self.observed_at)213 self.stats.relations += 1214215 # ---------------------------------------------------------------------------------------------- prices216 async def write_price(self, p: PriceObs) -> None:217 mid = await self.resolver.resolve(p.model)218 pid = await self.resolver.resolve(p.provider)219 if not mid or not pid:220 return221 if all(v is None for v in p.price_tuple()[:-1]):222 return223 url = p.source_url or self.source_url224 current = await fetch_one(self.conn, """select * from prices where model_id = :m and provider_id = :p and coalesce(provider_model_id,'') = :pm and valid_to is null""",225 m=mid, p=pid, pm=p.provider_model_id or "")226 new_tuple = p.price_tuple()227 if current:228 cur_tuple = (current["input_per_mtok"], current["output_per_mtok"], current["cached_input_per_mtok"], current["cache_write_per_mtok"],229 current["batch_input_per_mtok"], current["batch_output_per_mtok"], current["per_image"], current["per_request"], current["currency"])230 if _same(cur_tuple, new_tuple) and (p.context_length in (None, current["context_length"])):231 await execute(self.conn, "update prices set observed_at = :o where id = :id", o=self.observed_at, id=current["id"])232 return233 await execute(self.conn, "update prices set valid_to = :o where id = :id", o=self.observed_at, id=current["id"])234 await execute(self.conn, """insert into prices (id, model_id, provider_id, provider_model_id, input_per_mtok, output_per_mtok, cached_input_per_mtok, cache_write_per_mtok,235 batch_input_per_mtok, batch_output_per_mtok, per_image, per_request, currency, context_length, max_output_tokens, features,236 observed_at, valid_from, source_id, snapshot_id, source_url, tier, meta)237 values (:id, :m, :p, :pm, :i, :o, :ci, :cw, :bi, :bo, :img, :req, :cur, :ctx, :mo, cast(:f as jsonb), :obs, :obs, :src, :snap, :url, :tier, cast(:meta as jsonb))""",238 id=new_id("price"), m=mid, p=pid, pm=p.provider_model_id, i=p.input_per_mtok, o=p.output_per_mtok, ci=p.cached_input_per_mtok,239 cw=p.cache_write_per_mtok, bi=p.batch_input_per_mtok, bo=p.batch_output_per_mtok, img=p.per_image, req=p.per_request, cur=p.currency,240 ctx=p.context_length, mo=p.max_output_tokens, f=jsonb(p.features), obs=self.observed_at, src=self.source_id, snap=self.snapshot_id,241 url=url, tier=self.tier, meta=jsonb(p.meta))242 self.stats.prices += 1243 model = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=mid)244 provider = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=pid)245 mname = model["canonical_name"] if model else mid246 pname = provider["canonical_name"] if provider else pid247 if current:248 old = {"input_per_mtok": current["input_per_mtok"], "output_per_mtok": current["output_per_mtok"]}249 new = {"input_per_mtok": p.input_per_mtok, "output_per_mtok": p.output_per_mtok}250 summary = f"{pname} changed pricing for {mname}: {_fmt_price(old)} → {_fmt_price(new)}"251 dedupe = f"PRICE_CHANGED:{mid}:{pid}:{p.provider_model_id or ''}:{hashlib.sha1(json.dumps([old, new], sort_keys=True, default=str).encode()).hexdigest()[:12]}"252 await self.emit_event("PRICE_CHANGED", "price", summary, entity_id=mid, old_value=old, new_value=new, importance=2, dedupe_key=dedupe,253 source_url=url, meta={"provider_id": pid, "provider": pname})254 else:255 new = {"input_per_mtok": p.input_per_mtok, "output_per_mtok": p.output_per_mtok}256 summary = f"{pname} lists {mname} at {_fmt_price(new)}"257 dedupe = f"PROVIDER_LISTED:{mid}:{pid}:{p.provider_model_id or ''}"258 await self.emit_event("PROVIDER_LISTED", "provider", summary, entity_id=mid, new_value=new, importance=1, dedupe_key=dedupe, source_url=url,259 meta={"provider_id": pid, "provider": pname})260 await self.write_relation(p.model, "available_through", p.provider, {"provider_model_id": p.provider_model_id}, source_url=url)261262 # ---------------------------------------------------------------------------------------------- benchmark results263 async def write_result(self, r: ResultObs) -> None:264 mid = await self.resolver.resolve(r.model)265 bid = await self.resolver.resolve(r.benchmark)266 if not mid or not bid:267 return268 url = r.source_url or self.source_url269 cfg_hash = hashlib.sha1(json.dumps(r.config, sort_keys=True, default=str).encode()).hexdigest()[:12]270 dedupe = f"{mid}:{bid}:{cfg_hash}:{r.metric or ''}"271 existing = await fetch_one(self.conn, "select id, score from benchmark_results where dedupe_key = :d", d=dedupe)272 conf = r.confidence or TIER_CONFIDENCE.get(self.tier, "medium")273 if existing:274 if abs((existing["score"] or 0) - r.score) > 1e-9:275 await execute(self.conn, "update benchmark_results set valid_to = :o, dedupe_key = dedupe_key || ':' || :suffix where id = :id",276 o=self.observed_at, suffix=self.observed_at.strftime("%Y%m%d%H%M%S"), id=existing["id"])277 model = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=mid)278 bench = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=bid)279 await self.emit_event("BENCHMARK_UPDATED", "benchmark",280 f"{model['canonical_name'] if model else mid} on {bench['canonical_name'] if bench else bid}: {existing['score']:g} → {r.score:g}",281 entity_id=mid, old_value=existing["score"], new_value=r.score, importance=1,282 dedupe_key=f"BENCHMARK_UPDATED:{dedupe}:{r.score:g}", source_url=url, meta={"benchmark_id": bid})283 else:284 await execute(self.conn, "update benchmark_results set observed_at = :o where id = :id", o=self.observed_at, id=existing["id"])285 return286 await execute(self.conn, """insert into benchmark_results (id, model_id, benchmark_id, score, metric, unit, higher_is_better, config, evaluated_at, observed_at,287 source_id, snapshot_id, source_url, tier, confidence, dedupe_key)288 values (:id, :m, :b, :s, :metric, :unit, :hib, cast(:cfg as jsonb), :ev, :obs, :src, :snap, :url, :tier, :conf, :d)""",289 id=new_id("result"), m=mid, b=bid, s=r.score, metric=r.metric, unit=r.unit, hib=r.higher_is_better, cfg=jsonb(r.config),290 ev=r.evaluated_at, obs=self.observed_at, src=self.source_id, snap=self.snapshot_id, url=url, tier=self.tier, conf=conf, d=dedupe)291 self.stats.results += 1292 if not existing:293 model = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=mid)294 bench = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=bid)295 await self.emit_event("BENCHMARK_RESULT", "benchmark",296 f"{model['canonical_name'] if model else mid} scores {r.score:g}{r.unit or ''} on {bench['canonical_name'] if bench else bid}",297 entity_id=mid, new_value=r.score, importance=1, dedupe_key=f"BENCHMARK_RESULT:{dedupe}", source_url=url,298 meta={"benchmark_id": bid, "metric": r.metric})299 await self.write_relation(r.model, "evaluated_on", r.benchmark, source_url=url)300301 # ---------------------------------------------------------------------------------------------- events302 async def emit_event(self, event_type: str, category: str, summary: str, *, entity_id: str | None = None, old_value: Any = None,303 new_value: Any = None, importance: int = 2, effective_at: datetime | None = None, dedupe_key: str | None = None,304 source_url: str | None = None, meta: dict[str, Any] | None = None, observed_at: datetime | None = None) -> None:305 dedupe = dedupe_key or f"{event_type}:{entity_id or ''}:{normalize_alias(summary)[:120]}"306 await execute(self.conn, """insert into change_events (id, entity_id, event_type, category, property, old_value, new_value, summary, importance, observed_at,307 effective_at, source_id, snapshot_id, source_url, connector_name, dedupe_key, meta)308 values (:id, :e, :t, :c, :p, cast(:o as jsonb), cast(:n as jsonb), :s, :imp, :obs, :eff, :src, :snap, :url, :conn, :d, cast(:meta as jsonb))309 on conflict (dedupe_key) do nothing""",310 id=new_id("change_event"), e=entity_id, t=event_type, c=category, p=(meta or {}).get("property"), o=jsonb(old_value) if old_value is not None else None,311 n=jsonb(new_value) if new_value is not None else None, s=summary[:500], imp=max(0, min(3, importance)), obs=observed_at or self.observed_at,312 eff=effective_at, src=self.source_id, snap=self.snapshot_id, url=source_url or self.source_url, conn=self.connector_name, d=dedupe[:400],313 meta=jsonb(meta or {}))314 self.stats.events += 1315316 async def _new_entity_event(self, eid: str) -> None:317 row = await fetch_one(self.conn, "select canonical_name, entity_type, attributes, organization_id from entities where id = :id", id=eid)318 if not row:319 return320 etype = row["entity_type"]321 if etype in ("researcher", "country", "license", "quantization"):322 return323 importance = NEW_IMPORTANCE.get(etype, 1)324 if self.tier > 2:325 importance = max(0, importance - 1)326 org = None327 if row["organization_id"]:328 o = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=row["organization_id"])329 org = o["canonical_name"] if o else None330 label = etype.replace("_", " ")331 summary = f"New {label}: {row['canonical_name']}" + (f" ({org})" if org else "")332 effective = None333 rel = row["attributes"].get("release_date") or row["attributes"].get("published_at")334 if isinstance(rel, str):335 from aiatlas.sdk.extract.dates import parse_datetime336337 effective = parse_datetime(rel)338 await self.emit_event(f"NEW_{etype.upper()}", EVENT_CATEGORY_BY_TYPE.get(etype, "update"), summary, entity_id=eid, importance=importance,339 dedupe_key=f"NEW_{etype.upper()}:{eid}", effective_at=effective)340341342def _short(v: Any) -> str:343 s = json.dumps(v, default=str, ensure_ascii=False) if not isinstance(v, str) else v344 return s if len(s) <= 60 else s[:57] + "…"345346347def _fmt_price(p: dict[str, Any]) -> str:348 i, o = p.get("input_per_mtok"), p.get("output_per_mtok")349 parts = []350 if i is not None:351 parts.append(f"${i:g} in")352 if o is not None:353 parts.append(f"${o:g} out")354 return " / ".join(parts) + " per 1M tokens" if parts else "n/a"355356357__all__ = ["FactWriter", "WriteStats"]358