SPB Git
spboucher.ai

spb/ai-atlas

Public

SDK hardening: SSRF guard with per-hop redirect validation, run_id on every fact, same-source loophole closed, taxonomy at write time, result comparability, event semantics, hierarchy hints, quarantine

- fetch: validate_destination() on every request and redirect hop (max 5, manual loop): non-http(s), localhost/.local/.internal, RFC1918,
  loopback, link-local (metadata), CGNAT, IPv6 loopback/link-local/ULA, unspecified → FetchError("blocked destination")
- facts: facts_to_json / facts_from_json (tagged datetimes) for quarantined runs
- writer: run_id/source_key/is_first_run; a claim supersedes only when tier <= current or same source AND same extractor (LLM never
  overwrites deterministic from the same URL); licence/openness/modalities/status/kind/org_kind normalised via the ontology with value_raw
  + taxonomy_mappings, same assertion in another spelling re-encoded in place (no event); license_key claim; results get config_key,
  trust_level, variant, run_group, extractor, effort config for variant refs, one current row per (model, benchmark, metric, config_key),
  out-of-range scores → confidence low + anomaly; events get recorded_at, is_backfill (classify_backfill), group_key, deterministic
  importance; family/canonical hints materialised (family_id, canonical_id, artifact_kind, member_of_family, artifact_of); NEW_ARTIFACT 0,
  NEW_MODEL_FAMILY 1; conflicting claims recorded once; derived writers never open review items and mark their events backfill
- resolver: model/artifact compatible lookups; keep_separate decisions honoured; digit-collapsing aliases require the same variant_key;
  first_seen_hint → first_seen_at = min(now, hint); resolve_variant() folds evaluator-only effort variants onto the canonical model
- connector: run_id/source_key/first-run flag into the writer; --url reprocess rebuilds Targets from documents.meta._target (key, doc_type,
  meta, needs_llm, entity persisted at first fetch); quarantine for tier ≥ 2 connectors (rolling median baseline of the last 5 full runs:
  new entities > max(50, 1.3×), prices > 3×, entities < 0.8× / results < 0.5× on full extraction) → quarantined_runs + review item, nothing written
- services/events: classify_backfill, group_key_for, importance_for · services/anomalies: record / run_checks / list_anomalies
- merge: modes merge|alias|variant|family_member, keep_separate refusal, org-type and model/artifact pairs, dedupe/config_key recompute with
  collision handling, one current result per key, embeddings/sources/domains/review_queue/family_id/canonical_id re-pointed,
  resolution_decisions + admin_audit_log rows
- handlers: LLM extraction writer carries the snapshot's run_id

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Simon-Pierre Boucher committed (Sep 12, 2026) parent e74d6fc Browse files

13 changed files +1,757 −133

13 changed files
Load all 13 files Unified Split
modified src/aiatlas/sdk/connector.py +200 −17
@@ -11,25 +11,29 @@ from __future__ import annotations
11 import asyncio 11 import asyncio
12 import json 12 import json
13 import logging 13 import logging
14+import statistics
14 import time 15 import time
15 from dataclasses import dataclass, field 16 from dataclasses import dataclass, field
16 from datetime import UTC, datetime, timedelta 17 from datetime import UTC, datetime, timedelta
17 from typing import Any 18 from typing import Any
18 19
19−from aiatlas.db import execute, fetch_one, jsonb, transaction 20+from aiatlas.db import connection, execute, fetch_one, jsonb, transaction
20 from aiatlas.ids import new_id 21 from aiatlas.ids import new_id
21 from aiatlas.sdk import archive 22 from aiatlas.sdk import archive
22 from aiatlas.sdk.extract.feeds import FeedItem, parse_feed 23 from aiatlas.sdk.extract.feeds import FeedItem, parse_feed
23 from aiatlas.sdk.extract.html import HtmlDoc, parse_html 24 from aiatlas.sdk.extract.html import HtmlDoc, parse_html
24 from aiatlas.sdk.extract.markdown import MarkdownDoc, parse_markdown 25 from aiatlas.sdk.extract.markdown import MarkdownDoc, parse_markdown
25−from aiatlas.sdk.facts import Facts, Target 26+from aiatlas.sdk.facts import EntityRef, Facts, Target, facts_to_json
26 from aiatlas.sdk.fetch import BlockedError, Fetcher, FetchError, FetchResult, NotModified, canonicalize_url, file_result 27 from aiatlas.sdk.fetch import BlockedError, Fetcher, FetchError, FetchResult, NotModified, canonicalize_url, file_result
28+from aiatlas.sdk.resolution import Resolver
27 from aiatlas.sdk.writer import FactWriter 29 from aiatlas.sdk.writer import FactWriter
28 30
29 log = logging.getLogger(__name__) 31 log = logging.getLogger(__name__)
30 32
31 CIRCUIT_FAILURES = 3 33 CIRCUIT_FAILURES = 3
32 CIRCUIT_COOLDOWN_S = 1800 34 CIRCUIT_COOLDOWN_S = 1800
35+BASELINE_RUNS = 5 # rolling window (median) of full-extraction runs used by the quarantine detector
36+QUARANTINE_MIN_NEW_ENTITIES = 50 # a run may always create up to this many new entities without tripping the detector
33 37
34 38
35 class ConnectorError(Exception): 39 class ConnectorError(Exception):
@@ -77,6 +81,17 @@ class RunStats:
77 meta: dict[str, Any] = field(default_factory=dict) 81 meta: dict[str, Any] = field(default_factory=dict)
78 82
79 83
84+@dataclass
85+class Buffered:
86+ """Facts extracted from one document, held until the end of the run when the connector is quarantine-capable."""
87+ target: Target
88+ doc_id: str
89+ snapshot_id: str
90+ source_url: str
91+ fetched_at: datetime
92+ facts: Facts
93+
94+
80 @dataclass 95 @dataclass
81 class RunContext: 96 class RunContext:
82 run_id: str 97 run_id: str
@@ -89,6 +104,11 @@ class RunContext:
89 max_targets: int = 2000 104 max_targets: int = 2000
90 stats: RunStats = field(default_factory=RunStats) 105 stats: RunStats = field(default_factory=RunStats)
91 seen_urls: set[str] = field(default_factory=set) 106 seen_urls: set[str] = field(default_factory=set)
107+ is_first_run: bool = False # no earlier successful run → every event of this run is backfill
108+ source_key: str | None = None
109+ buffered: list[Buffered] = field(default_factory=list) # quarantine mode: facts held until the run-end anomaly check
110+ quarantined: bool = False
111+ quarantine_reason: str | None = None
92 log: logging.LoggerAdapter[logging.Logger] = field(init=False) 112 log: logging.LoggerAdapter[logging.Logger] = field(init=False)
93 113
94 def __post_init__(self) -> None: 114 def __post_init__(self) -> None:
@@ -114,11 +134,21 @@ class BaseConnector:
114 concurrency: int = 3 134 concurrency: int = 3
115 default_doc_type: str = "page" 135 default_doc_type: str = "page"
116 needs_llm: bool = False # queue documents for LLM extraction after deterministic extraction 136 needs_llm: bool = False # queue documents for LLM extraction after deterministic extraction
137+ # Hold every fact until the end of the run and compare the run's counts with the connector's rolling baseline before writing;
138+ # a collapse or an explosion parks the whole run in `quarantined_runs` for review. None = default (tier ≥ 2 hub/leaderboard/registry
139+ # connectors are quarantine-capable; tier-1 lab documentation is not — it is the primary source).
140+ quarantine: bool | None = None
117 141
118 def __init__(self, config: dict[str, Any] | None = None): 142 def __init__(self, config: dict[str, Any] | None = None):
119 self.config = config or {} 143 self.config = config or {}
120 self.source_id: str | None = None 144 self.source_id: str | None = None
121 145
146+ @property
147+ def quarantine_enabled(self) -> bool:
148+ if self.quarantine is not None:
149+ return self.quarantine
150+ return self.tier >= 2
151+
122 # ------------------------------------------------------------------------------------------ to implement 152 # ------------------------------------------------------------------------------------------ to implement
123 async def discover(self, ctx: RunContext) -> list[Target]: 153 async def discover(self, ctx: RunContext) -> list[Target]:
124 """Return the targets for this run (seed pages, feeds, sitemaps, org pages…).""" 154 """Return the targets for this run (seed pages, feeds, sitemaps, org pages…)."""
@@ -165,10 +195,14 @@ class BaseConnector:
165 ctx = RunContext(run_id=run_id, connector=self, fetcher=fetcher, started_at=started, force=force, reprocess=reprocess, 195 ctx = RunContext(run_id=run_id, connector=self, fetcher=fetcher, started_at=started, force=force, reprocess=reprocess,
166 file_overrides=file_overrides or {}, max_targets=max_targets or int(self.config.get("max_targets", 2000))) 196 file_overrides=file_overrides or {}, max_targets=max_targets or int(self.config.get("max_targets", 2000)))
167 async with transaction() as conn: 197 async with transaction() as conn:
168− state = await fetch_one(conn, "select source_id, enabled, circuit_open_until from connectors where name = :n", n=self.name) 198+ state = await fetch_one(conn, """select c.source_id, c.enabled, c.circuit_open_until, s.key as source_key,
199+ exists(select 1 from connector_runs r where r.connector_name = c.name and r.status in ('success','unchanged','suspect')) as has_success
200+ from connectors c left join sources s on s.id = c.source_id where c.name = :n""", n=self.name)
169 if state is None: 201 if state is None:
170 raise ConnectorError(f"connector {self.name!r} not registered (run `aia seed`)") 202 raise ConnectorError(f"connector {self.name!r} not registered (run `aia seed`)")
171 self.source_id = state["source_id"] 203 self.source_id = state["source_id"]
204+ ctx.source_key = state["source_key"] or self.source_key or None
205+ ctx.is_first_run = not state["has_success"]
172 if not state["enabled"] and not force: 206 if not state["enabled"] and not force:
173 ctx.log.info("connector disabled, skipping") 207 ctx.log.info("connector disabled, skipping")
174 await self._record(conn, ctx, "skipped", error="disabled") 208 await self._record(conn, ctx, "skipped", error="disabled")
@@ -184,21 +218,23 @@ class BaseConnector:
184 try: 218 try:
185 targets = await self.discover(ctx) 219 targets = await self.discover(ctx)
186 if only_urls: 220 if only_urls:
187− wanted = {canonicalize_url(u) for u in only_urls} 221+ targets = await self._select_targets(targets, only_urls)
188− targets = [t for t in targets if canonicalize_url(t.url) in wanted] or [Target(url=u) for u in only_urls]
189 ctx.stats.docs_discovered = len(targets) 222 ctx.stats.docs_discovered = len(targets)
190 await self._process_all(ctx, targets) 223 await self._process_all(ctx, targets)
191− status = self._final_status(ctx) 224+ if ctx.buffered:
225+ await self._flush_or_quarantine(ctx)
226+ status = "quarantined" if ctx.quarantined else self._final_status(ctx)
192 async with transaction() as conn: 227 async with transaction() as conn:
193 await self._record(conn, ctx, status) 228 await self._record(conn, ctx, status)
194 changed = ctx.stats.docs_changed > 0 229 changed = ctx.stats.docs_changed > 0
195− await execute(conn, """update connectors set consecutive_failures = 0, circuit_open_until = null, last_success_at = now(), 230+ await execute(conn, """update connectors set consecutive_failures = 0, circuit_open_until = null,
196− last_change_at = case when cast(:changed as boolean) then now() else last_change_at end, 231+ last_success_at = case when cast(:q as boolean) then last_success_at else now() end,
232+ last_change_at = case when cast(:changed as boolean) and not cast(:q as boolean) then now() else last_change_at end,
197 consecutive_unchanged = case when cast(:changed as boolean) then 0 else consecutive_unchanged + 1 end, 233 consecutive_unchanged = case when cast(:changed as boolean) then 0 else consecutive_unchanged + 1 end,
198 interval_seconds = cast(:interval as integer), next_run_at = now() + make_interval(secs => cast(:interval as double precision)), 234 interval_seconds = cast(:interval as integer), next_run_at = now() + make_interval(secs => cast(:interval as double precision)),
199 last_duration_ms = :d, health = :health, updated_at = now() where name = :n""", 235 last_duration_ms = :d, health = :health, updated_at = now() where name = :n""",
200− changed=changed, interval=await self._next_interval(conn, changed), d=int((time.perf_counter() - t0) * 1000), 236+ changed=changed, q=ctx.quarantined, interval=await self._next_interval(conn, changed), d=int((time.perf_counter() - t0) * 1000),
201− health="degraded" if status == "suspect" else "ok", n=self.name) 237+ health="degraded" if status in ("suspect", "quarantined") else "ok", n=self.name)
202 if status == "suspect": 238 if status == "suspect":
203 await execute(conn, """insert into review_queue (id, kind, entity_ids, payload, reason, dedupe_key) 239 await execute(conn, """insert into review_queue (id, kind, entity_ids, payload, reason, dedupe_key)
204 values (:id, 'parser_breakage', '{}', cast(:p as jsonb), :r, :d) on conflict (dedupe_key) do nothing""", 240 values (:id, 'parser_breakage', '{}', cast(:p as jsonb), :r, :d) on conflict (dedupe_key) do nothing""",
@@ -223,11 +259,58 @@ class BaseConnector:
223 raise 259 raise
224 return ctx 260 return ctx
225 261
262+ async def _select_targets(self, targets: list[Target], only_urls: list[str]) -> list[Target]:
263+ """`--url` / single-document reprocess: keep the discovered targets for those URLs; for URLs `discover()` no longer lists, rebuild
264+ the Target from what the first fetch persisted in `documents.meta` (key, doc_type, meta, needs_llm, entity) so extractors that
265+ branch on `target.key`/`target.meta` behave exactly as they did originally."""
266+ wanted = {canonicalize_url(u) for u in only_urls}
267+ selected = [t for t in targets if canonicalize_url(t.url) in wanted]
268+ missing = wanted - {canonicalize_url(t.url) for t in selected}
269+ if missing:
270+ async with transaction() as conn:
271+ for url in sorted(missing):
272+ doc = await fetch_one(conn, "select url, doc_type, meta, needs_llm, entity_id, priority from documents where url = :u", u=url)
273+ if not doc:
274+ selected.append(Target(url=url))
275+ continue
276+ meta = dict(doc["meta"] or {})
277+ stored = meta.pop("_target", {}) if isinstance(meta.get("_target"), dict) else {}
278+ entity = None
279+ if stored.get("entity"):
280+ try:
281+ from aiatlas.sdk.facts import _ref_from # noqa: PLC0415
282+
283+ entity = _ref_from(stored["entity"])
284+ except Exception: # noqa: BLE001
285+ entity = None
286+ elif doc["entity_id"]:
287+ row = await fetch_one(conn, "select entity_type, canonical_name from entities where id = :id", id=doc["entity_id"])
288+ if row:
289+ entity = EntityRef(entity_type=row["entity_type"], name=row["canonical_name"], id=doc["entity_id"])
290+ selected.append(Target(url=doc["url"], doc_type=stored.get("doc_type") or doc["doc_type"] or self.default_doc_type, entity=entity,
291+ meta=stored.get("meta") if isinstance(stored.get("meta"), dict) else meta, key=stored.get("key"),
292+ needs_llm=bool(stored.get("needs_llm", doc["needs_llm"])), priority=int(doc["priority"] or 2)))
293+ return selected
294+
295+ @staticmethod
296+ def _target_meta(target: Target, doc_type: str) -> dict[str, Any]:
297+ """What `documents.meta` remembers about the target that produced the document (rebuilt by `_select_targets`)."""
298+ stored: dict[str, Any] = {"key": target.key, "doc_type": doc_type, "meta": _jsonable(target.meta), "needs_llm": bool(target.needs_llm)}
299+ if target.entity is not None:
300+ from aiatlas.sdk.facts import _encode # noqa: PLC0415
301+
302+ stored["entity"] = _encode(target.entity)
303+ return {**_jsonable(target.meta), "_target": stored}
304+
305+ def _full_extraction(self, ctx: RunContext) -> bool:
306+ s = ctx.stats
307+ return s.docs_fetched > 0 and s.docs_changed >= s.docs_fetched - s.docs_failed and s.docs_unchanged == 0
308+
226 def _final_status(self, ctx: RunContext) -> str: 309 def _final_status(self, ctx: RunContext) -> str:
227 s = ctx.stats 310 s = ctx.stats
228 # Breakage detection only makes sense when every fetched document was (re)extracted: unchanged documents legitimately 311 # Breakage detection only makes sense when every fetched document was (re)extracted: unchanged documents legitimately
229 # produce zero records (304 / same hash), so an incremental run is never "suspect". 312 # produce zero records (304 / same hash), so an incremental run is never "suspect".
230− full_extraction = s.docs_fetched > 0 and s.docs_changed >= s.docs_fetched - s.docs_failed and s.docs_unchanged == 0 313+ full_extraction = self._full_extraction(ctx)
231 if self.expected_min_records and full_extraction and s.records < self.expected_min_records and not ctx.reprocess: 314 if self.expected_min_records and full_extraction and s.records < self.expected_min_records and not ctx.reprocess:
232 return "suspect" 315 return "suspect"
233 if s.docs_fetched and s.docs_failed == s.docs_fetched and s.docs_fetched > 0: 316 if s.docs_fetched and s.docs_failed == s.docs_fetched and s.docs_fetched > 0:
@@ -298,8 +381,11 @@ class BaseConnector:
298 if doc is None: 381 if doc is None:
299 doc_id = new_id("document") 382 doc_id = new_id("document")
300 await execute(conn, """insert into documents (id, source_id, connector_name, url, doc_type, priority, meta) values (:id, :s, :c, :u, :t, :p, cast(:m as jsonb)) 383 await execute(conn, """insert into documents (id, source_id, connector_name, url, doc_type, priority, meta) values (:id, :s, :c, :u, :t, :p, cast(:m as jsonb))
301− on conflict (url) do nothing""", id=doc_id, s=self.source_id, c=self.name, u=url, t=doc_type, p=target.priority, m=jsonb(target.meta)) 384+ on conflict (url) do nothing""", id=doc_id, s=self.source_id, c=self.name, u=url, t=doc_type, p=target.priority,
385+ m=jsonb(self._target_meta(target, doc_type)))
302 doc = await fetch_one(conn, "select * from documents where url = :u", u=url) 386 doc = await fetch_one(conn, "select * from documents where url = :u", u=url)
387+ elif (target.key or target.meta) and not ctx.reprocess and (doc["meta"] or {}).get("_target", {}).get("key") != target.key:
388+ await execute(conn, "update documents set meta = meta || cast(:m as jsonb) where id = :id", m=jsonb(self._target_meta(target, doc_type)), id=doc["id"])
303 assert doc 389 assert doc
304 # ---- reprocess mode: replay latest snapshot without network 390 # ---- reprocess mode: replay latest snapshot without network
305 if ctx.reprocess: 391 if ctx.reprocess:
@@ -407,15 +493,26 @@ class BaseConnector:
407 return [] 493 return []
408 if facts is None: 494 if facts is None:
409 return [] 495 return []
496+ item = Buffered(target=target, doc_id=doc["id"], snapshot_id=snapshot_id, source_url=res.final_url or res.url, fetched_at=res.fetched_at, facts=facts)
497+ ctx.stats.records += len(facts.entities) + len(facts.prices) + len(facts.results)
498+ if self.quarantine_enabled and not ctx.reprocess:
499+ ctx.buffered.append(item) # written (or quarantined) at the end of the run, see _flush_or_quarantine
500+ else:
501+ await self._write_buffered(ctx, item)
502+ return facts.targets
503+
504+ async def _write_buffered(self, ctx: RunContext, item: Buffered) -> None:
505+ facts, target, snapshot_id = item.facts, item.target, item.snapshot_id
410 async with transaction() as conn: 506 async with transaction() as conn:
411− writer = FactWriter(conn, source_id=self.source_id, snapshot_id=snapshot_id, source_url=res.final_url or res.url, tier=self.tier, 507+ writer = FactWriter(conn, source_id=self.source_id, snapshot_id=snapshot_id, source_url=item.source_url, tier=self.tier,
412− connector_name=self.name, extractor="deterministic", extractor_version=self.parser_version, observed_at=res.fetched_at) 508+ connector_name=self.name, extractor="deterministic", extractor_version=self.parser_version, observed_at=item.fetched_at,
509+ run_id=ctx.run_id, source_key=ctx.source_key, is_first_run=ctx.is_first_run)
413 ws = await writer.write(facts) 510 ws = await writer.write(facts)
414 main = facts.document_entity or target.entity 511 main = facts.document_entity or target.entity
415 if main and main.id is None: 512 if main and main.id is None:
416 await writer.resolver.resolve(main) 513 await writer.resolver.resolve(main)
417 if main and main.id: 514 if main and main.id:
418− await execute(conn, "update documents set entity_id = coalesce(entity_id, :e), title = coalesce(:t, title) where id = :id", e=main.id, t=facts.document_title, id=doc["id"]) 515+ await execute(conn, "update documents set entity_id = coalesce(entity_id, :e), title = coalesce(:t, title) where id = :id", e=main.id, t=facts.document_title, id=item.doc_id)
419 await execute(conn, "update snapshots set processing_status = :st where id = :id", 516 await execute(conn, "update snapshots set processing_status = :st where id = :id",
420 st="llm_pending" if (target.needs_llm or self.needs_llm or facts.llm_hint) else "extracted", id=snapshot_id) 517 st="llm_pending" if (target.needs_llm or self.needs_llm or facts.llm_hint) else "extracted", id=snapshot_id)
421 if target.needs_llm or self.needs_llm or facts.llm_hint: 518 if target.needs_llm or self.needs_llm or facts.llm_hint:
@@ -430,8 +527,94 @@ class BaseConnector:
430 s.claims_written += ws.claims 527 s.claims_written += ws.claims
431 s.relations_written += ws.relations 528 s.relations_written += ws.relations
432 s.events_emitted += ws.events 529 s.events_emitted += ws.events
433− s.records += len(facts.entities) + len(facts.prices) + len(facts.results) 530+
434− return facts.targets 531+ # ------------------------------------------------------------------------------------------ quarantine
532+ async def _observed_counts(self, ctx: RunContext) -> dict[str, int]:
533+ """Counts of what the buffered facts would write: entity references, prices, results and *new* entity names (refs that resolve to
534+ nothing today). Resolution runs in a transaction that is rolled back so the check leaves no trace."""
535+ refs: dict[str, EntityRef] = {}
536+ prices = results = 0
537+ for item in ctx.buffered:
538+ prices += len(item.facts.prices)
539+ results += len(item.facts.results)
540+ for ref in item.facts.entities:
541+ refs.setdefault(ref.key(), ref)
542+ new_entities = 0
543+ async with connection() as conn:
544+ trans = await conn.begin()
545+ try:
546+ resolver = Resolver(conn, source_tier=self.tier)
547+ for ref in refs.values():
548+ probe = EntityRef(entity_type=ref.entity_type, name=ref.name, identifiers=dict(ref.identifiers), aliases=list(ref.aliases), slug_hint=ref.slug_hint)
549+ if await resolver.resolve(probe, create=False) is None:
550+ new_entities += 1
551+ finally:
552+ await trans.rollback()
553+ return {"entities": len(refs), "prices": prices, "results": results, "new_entities": new_entities}
554+
555+ @staticmethod
556+ def _quarantine_reason(observed: dict[str, int], baseline: dict[str, Any] | None, *, full_extraction: bool) -> str | None:
557+ """Explosion rules always apply; collapse rules only when every document was re-extracted (an incremental run legitimately yields little)."""
558+ if not baseline or not baseline.get("history"):
559+ return None
560+ med = baseline.get("medians") or {}
561+ b_new, b_ent, b_pr, b_res = med.get("new_entities", 0), med.get("entities", 0), med.get("prices", 0), med.get("results", 0)
562+ if observed["new_entities"] > max(QUARANTINE_MIN_NEW_ENTITIES, 1.3 * b_new):
563+ return f"new entities {observed['new_entities']} > max({QUARANTINE_MIN_NEW_ENTITIES}, 1.3 × baseline {b_new:g})"
564+ if b_pr > 0 and observed["prices"] > 3 * b_pr:
565+ return f"price rows {observed['prices']} > 3 × baseline {b_pr:g}"
566+ if full_extraction:
567+ if b_ent > 0 and observed["entities"] < 0.8 * b_ent:
568+ return f"entity references {observed['entities']} < 0.8 × baseline {b_ent:g}"
569+ if b_res > 0 and observed["results"] < 0.5 * b_res:
570+ return f"benchmark results {observed['results']} < 0.5 × baseline {b_res:g}"
571+ return None
572+
573+ @staticmethod
574+ def _next_baseline(baseline: dict[str, Any] | None, observed: dict[str, int]) -> dict[str, Any]:
575+ history = list((baseline or {}).get("history") or [])
576+ history.append({k: int(observed[k]) for k in ("entities", "prices", "results", "new_entities")})
577+ history = history[-BASELINE_RUNS:]
578+ medians = {k: float(statistics.median(h[k] for h in history)) for k in ("entities", "prices", "results", "new_entities")}
579+ return {"history": history, "medians": medians, "runs": len(history), "updated_at": datetime.now(UTC).isoformat(timespec="seconds")}
580+
581+ async def _flush_or_quarantine(self, ctx: RunContext) -> None:
582+ observed = await self._observed_counts(ctx)
583+ full = self._full_extraction(ctx)
584+ async with transaction() as conn:
585+ row = await fetch_one(conn, "select baseline from connectors where name = :n", n=self.name)
586+ baseline = (row or {}).get("baseline")
587+ reason = None if ctx.force and ctx.file_overrides else self._quarantine_reason(observed, baseline, full_extraction=full)
588+ ctx.stats.meta["quarantine_check"] = {"observed": observed, "baseline": (baseline or {}).get("medians"), "full_extraction": full}
589+ if reason:
590+ ctx.quarantined = True
591+ ctx.quarantine_reason = reason
592+ ctx.log.warning("run quarantined", extra={"reason": reason, **observed})
593+ payload = [{"doc_id": b.doc_id, "snapshot_id": b.snapshot_id, "source_url": b.source_url, "fetched_at": b.fetched_at.isoformat(),
594+ "target": {"url": b.target.url, "doc_type": b.target.doc_type, "key": b.target.key, "needs_llm": b.target.needs_llm, "meta": _jsonable(b.target.meta)},
595+ "facts": facts_to_json(b.facts)} for b in ctx.buffered]
596+ qid = f"quar_{new_id('connector_run').split('_', 1)[1]}"
597+ async with transaction() as conn:
598+ await execute(conn, """insert into quarantined_runs (id, run_id, connector_name, reason, stats, facts) values (:id, :r, :c, :why, cast(:s as jsonb), cast(:f as jsonb))""",
599+ id=qid, r=ctx.run_id, c=self.name, why=reason, s=jsonb({"observed": observed, "baseline": baseline, "full_extraction": full, "documents": len(ctx.buffered)}),
600+ f=jsonb(payload))
601+ await execute(conn, "update connector_runs set quarantined = true, baseline = cast(:b as jsonb) where id = :id", b=jsonb(baseline), id=ctx.run_id)
602+ await execute(conn, "update snapshots set processing_status = 'quarantined' where id = any(cast(:ids as text[]))", ids=[b.snapshot_id for b in ctx.buffered])
603+ await execute(conn, """insert into review_queue (id, kind, entity_ids, payload, reason, dedupe_key) values (:id, 'quarantine', '{}', cast(:p as jsonb), :r, :d)
604+ on conflict (dedupe_key) do nothing""",
605+ id=new_id("review"), p=jsonb({"connector": self.name, "run_id": ctx.run_id, "quarantine_id": qid, "observed": observed,
606+ "baseline": (baseline or {}).get("medians")}),
607+ r=f"{self.name}: run held in quarantine — {reason}; nothing written, nothing deleted", d=f"quarantine:{qid}")
608+ ctx.buffered.clear()
609+ return
610+ for item in ctx.buffered:
611+ await self._write_buffered(ctx, item)
612+ ctx.buffered.clear()
613+ if full and not ctx.reprocess and not ctx.file_overrides:
614+ async with transaction() as conn:
615+ nb = self._next_baseline(baseline, observed)
616+ await execute(conn, "update connectors set baseline = cast(:b as jsonb) where name = :n", b=jsonb(nb), n=self.name)
617+ await execute(conn, "update connector_runs set baseline = cast(:b as jsonb) where id = :id", b=jsonb(nb), id=ctx.run_id)
435 618
436 async def _record(self, conn: Any, ctx: RunContext, status: str, *, error: str | None = None) -> None: 619 async def _record(self, conn: Any, ctx: RunContext, status: str, *, error: str | None = None) -> None:
437 s = ctx.stats 620 s = ctx.stats
modified src/aiatlas/sdk/facts.py +88 −0
@@ -227,6 +227,92 @@ EVENT_CATEGORY_BY_TYPE: dict[str, str] = {
227 "mcp_server": "tool", "researcher": "company", "license": "model", "product": "tool", "robot": "hardware", 227 "mcp_server": "tool", "researcher": "company", "license": "model", "product": "tool", "robot": "hardware",
228 } 228 }
229 229
230+
231+# ---------------------------------------------------------------------------------------------- JSON round-trip (quarantine)
232+# Quarantined runs hold their Facts in `quarantined_runs.facts` (jsonb) until an operator releases or discards them. Datetimes are
233+# tagged so they come back as real datetime objects (asyncpg needs them, not ISO strings). Entity references are inlined; the
234+# resolver caches by `EntityRef.key()`, so equal refs resolve to the same entity after a round-trip.
235+_DT_TAG = "__datetime__"
236+
237+
238+def _encode(value: Any) -> Any:
239+ if isinstance(value, datetime):
240+ return {_DT_TAG: value.isoformat()}
241+ if isinstance(value, EntityRef):
242+ return {"entity_type": value.entity_type, "name": value.name, "identifiers": dict(value.identifiers), "aliases": list(value.aliases),
243+ "slug_hint": value.slug_hint, "organization": _encode(value.organization), "description": value.description, "status": value.status,
244+ "attributes": _encode(value.attributes), "first_seen_hint": _encode(value.first_seen_hint), "family": _encode(value.family),
245+ "canonical": _encode(value.canonical), "artifact_kind": value.artifact_kind, "identity_confidence": value.identity_confidence, "id": value.id}
246+ if isinstance(value, dict):
247+ return {str(k): _encode(v) for k, v in value.items()}
248+ if isinstance(value, (list, tuple, set)):
249+ return [_encode(v) for v in value]
250+ if isinstance(value, (str, int, float, bool)) or value is None:
251+ return value
252+ return str(value)
253+
254+
255+def _decode(value: Any) -> Any:
256+ if isinstance(value, dict):
257+ if set(value) == {_DT_TAG}:
258+ return datetime.fromisoformat(value[_DT_TAG])
259+ return {k: _decode(v) for k, v in value.items()}
260+ if isinstance(value, list):
261+ return [_decode(v) for v in value]
262+ return value
263+
264+
265+def _ref_from(d: dict[str, Any] | None) -> EntityRef | None:
266+ if not d:
267+ return None
268+ return EntityRef(entity_type=d["entity_type"], name=d["name"], identifiers=dict(d.get("identifiers") or {}), aliases=list(d.get("aliases") or []),
269+ slug_hint=d.get("slug_hint"), organization=_ref_from(d.get("organization")), description=d.get("description"), status=d.get("status"),
270+ attributes=_decode(d.get("attributes") or {}), first_seen_hint=_decode(d.get("first_seen_hint")), family=_ref_from(d.get("family")),
271+ canonical=_ref_from(d.get("canonical")), artifact_kind=d.get("artifact_kind"), identity_confidence=d.get("identity_confidence"), id=d.get("id"))
272+
273+
274+def facts_to_json(facts: Facts) -> dict[str, Any]:
275+ """Serialise a Facts object to a JSON-compatible dict (see `facts_from_json`)."""
276+ def obj(o: Any) -> dict[str, Any]:
277+ return {k: _encode(v) for k, v in o.__dict__.items()}
278+
279+ return {
280+ "version": 1,
281+ "entities": [_encode(e) for e in facts.entities],
282+ "claims": [obj(c) for c in facts.claims],
283+ "relations": [obj(r) for r in facts.relations],
284+ "events": [obj(e) for e in facts.events],
285+ "prices": [obj(p) for p in facts.prices],
286+ "results": [obj(r) for r in facts.results],
287+ "targets": [obj(t) for t in facts.targets],
288+ "document_title": facts.document_title,
289+ "document_entity": _encode(facts.document_entity),
290+ "llm_hint": facts.llm_hint,
291+ }
292+
293+
294+def facts_from_json(data: dict[str, Any]) -> Facts:
295+ """Inverse of `facts_to_json`."""
296+ def ref_fields(d: dict[str, Any], *keys: str) -> dict[str, Any]:
297+ out = {k: _decode(v) for k, v in d.items() if k not in keys}
298+ for k in keys:
299+ out[k] = _ref_from(d.get(k))
300+ return out
301+
302+ facts = Facts()
303+ facts.entities = [r for r in (_ref_from(e) for e in data.get("entities") or []) if r]
304+ facts.claims = [Claim(**ref_fields(c, "entity")) for c in data.get("claims") or []]
305+ facts.relations = [Relation(**ref_fields(r, "subject", "object")) for r in data.get("relations") or []]
306+ facts.events = [Event(**ref_fields(e, "entity")) for e in data.get("events") or []]
307+ facts.prices = [PriceObs(**ref_fields(p, "model", "provider")) for p in data.get("prices") or []]
308+ facts.results = [ResultObs(**ref_fields(r, "model", "benchmark")) for r in data.get("results") or []]
309+ facts.targets = [Target(**ref_fields(t, "entity")) for t in data.get("targets") or []]
310+ facts.document_title = data.get("document_title")
311+ facts.document_entity = _ref_from(data.get("document_entity"))
312+ facts.llm_hint = data.get("llm_hint")
313+ return facts
314+
315+
230 __all__ = [ 316 __all__ = [
231 "CONFIDENCE", 317 "CONFIDENCE",
232 "EVENT_CATEGORY_BY_TYPE", 318 "EVENT_CATEGORY_BY_TYPE",
@@ -240,4 +326,6 @@ __all__ = [
240 "Relation", 326 "Relation",
241 "ResultObs", 327 "ResultObs",
242 "Target", 328 "Target",
329+ "facts_from_json",
330+ "facts_to_json",
243 ] 331 ]
modified src/aiatlas/sdk/fetch.py +187 −24
@@ -7,11 +7,13 @@ from __future__ import annotations
7 7
8 import asyncio 8 import asyncio
9 import hashlib 9 import hashlib
10+import ipaddress
10 import logging 11 import logging
12+import socket
11 import time 13 import time
12 from dataclasses import dataclass, field 14 from dataclasses import dataclass, field
13 from datetime import UTC, datetime 15 from datetime import UTC, datetime
14−from urllib.parse import urlparse, urlunparse 16+from urllib.parse import urljoin, urlparse, urlunparse
15 from urllib.robotparser import RobotFileParser 17 from urllib.robotparser import RobotFileParser
16 18
17 import httpx 19 import httpx
@@ -22,6 +24,100 @@ log = logging.getLogger(__name__)
22 24
23 TRANSIENT_STATUS = {408, 425, 429, 500, 502, 503, 504} 25 TRANSIENT_STATUS = {408, 425, 429, 500, 502, 503, 504}
24 BLOCK_STATUS = {401, 403, 451, 999} 26 BLOCK_STATUS = {401, 403, 451, 999}
27+REDIRECT_STATUS = {301, 302, 303, 307, 308}
28+MAX_REDIRECTS = 5
29+
30+# ---------------------------------------------------------------------------------------------- SSRF guard
31+# Follow-up targets come from crawled content (links in model cards, feeds, READMEs): every destination — and every redirect
32+# hop — is validated before a connection is opened. Nothing internal is ever fetched.
33+_BLOCKED_SUFFIXES = (".local", ".internal", ".localhost", ".localdomain", ".lan", ".home", ".corp", ".intranet")
34+_BLOCKED_HOSTS = {"localhost", "metadata.google.internal", "metadata", "instance-data", "kubernetes.default", "kubernetes.default.svc"}
35+_BLOCKED_NETWORKS = [ipaddress.ip_network(n) for n in (
36+ "0.0.0.0/8", # unspecified / "this network"
37+ "10.0.0.0/8", # RFC1918
38+ "100.64.0.0/10", # CGNAT
39+ "127.0.0.0/8", # loopback
40+ "169.254.0.0/16", # link-local (cloud metadata endpoints 169.254.169.254 …)
41+ "172.16.0.0/12", # RFC1918
42+ "192.0.0.0/24", # IETF protocol assignments
43+ "192.168.0.0/16", # RFC1918
44+ "198.18.0.0/15", # benchmarking
45+ "240.0.0.0/4", # reserved + broadcast
46+ "::/128", # unspecified
47+ "::1/128", # loopback
48+ "fc00::/7", # ULA
49+ "fe80::/10", # link-local
50+ "::ffff:0:0/96", # IPv4-mapped (checked again as IPv4 below)
51+ "64:ff9b::/96", # NAT64
52+)]
53+
54+
55+class BlockedDestination(ValueError):
56+ """The URL points at a private, local or otherwise non-public destination."""
57+
58+
59+def _ip_blocked(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
60+ if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
61+ ip = ip.ipv4_mapped
62+ if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_unspecified or ip.is_reserved or ip.is_multicast:
63+ return True
64+ return any(ip in net for net in _BLOCKED_NETWORKS)
65+
66+
67+def validate_destination(url: str, *, resolved_ips: list[str] | None = None) -> None:
68+ """Raise `BlockedDestination` for non-http(s) schemes, local/internal hostnames and private, loopback, link-local, CGNAT,
69+ IPv6 ULA/link-local or unspecified addresses. `resolved_ips` lets callers (and tests) inject the DNS answer."""
70+ p = urlparse(url.strip())
71+ if p.scheme.lower() not in ("http", "https"):
72+ raise BlockedDestination(f"blocked destination: scheme {p.scheme!r} for {url}")
73+ host = (p.hostname or "").strip().lower().rstrip(".")
74+ if not host:
75+ raise BlockedDestination(f"blocked destination: no host in {url}")
76+ if p.username or p.password:
77+ raise BlockedDestination(f"blocked destination: credentials in URL {url}")
78+ if host in _BLOCKED_HOSTS or host.endswith(_BLOCKED_SUFFIXES) or "." not in host and not _is_ip_literal(host):
79+ raise BlockedDestination(f"blocked destination: local host {host!r}")
80+ literal = _parse_ip(host)
81+ if literal is not None:
82+ if _ip_blocked(literal):
83+ raise BlockedDestination(f"blocked destination: non-public address {host}")
84+ return
85+ ips = resolved_ips if resolved_ips is not None else _resolve(host, p.port or (443 if p.scheme.lower() == "https" else 80))
86+ for raw in ips:
87+ ip = _parse_ip(raw)
88+ if ip is not None and _ip_blocked(ip):
89+ raise BlockedDestination(f"blocked destination: {host} resolves to non-public address {raw}")
90+
91+
92+def _is_ip_literal(host: str) -> bool:
93+ return _parse_ip(host) is not None
94+
95+
96+def _parse_ip(host: str) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
97+ try:
98+ return ipaddress.ip_address(host.strip("[]").split("%")[0])
99+ except ValueError:
100+ return None
101+
102+
103+def _resolve(host: str, port: int) -> list[str]:
104+ try:
105+ infos = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP)
106+ except socket.gaierror:
107+ return [] # unresolvable hosts fail later with a transport error, not as SSRF
108+ return sorted({info[4][0] for info in infos})
109+
110+
111+async def validate_destination_async(url: str) -> None:
112+ """Non-blocking wrapper (DNS resolution runs in a thread)."""
113+ p = urlparse(url.strip())
114+ host = (p.hostname or "").strip().lower()
115+ if not host or _parse_ip(host) is not None:
116+ validate_destination(url, resolved_ips=[])
117+ return
118+ validate_destination(url, resolved_ips=[]) # scheme / hostname rules first (no DNS)
119+ ips = await asyncio.to_thread(_resolve, host, p.port or (443 if p.scheme.lower() == "https" else 80))
120+ validate_destination(url, resolved_ips=ips)
25 121
26 122
27 class FetchError(Exception): 123 class FetchError(Exception):
@@ -138,7 +234,7 @@ class _Robots:
138 if cached is None or cached[0] < time.monotonic(): 234 if cached is None or cached[0] < time.monotonic():
139 rp: RobotFileParser | None = RobotFileParser() 235 rp: RobotFileParser | None = RobotFileParser()
140 try: 236 try:
141− r = await client.get(f"{key}/robots.txt", timeout=15, follow_redirects=True) 237+ r = await client.get(f"{key}/robots.txt", timeout=15, follow_redirects=False) # host already validated by the caller
142 if r.status_code == 200 and len(r.content) < 512 * 1024: 238 if r.status_code == 200 and len(r.content) < 512 * 1024:
143 rp.parse(r.text.splitlines()) # type: ignore[union-attr] 239 rp.parse(r.text.splitlines()) # type: ignore[union-attr]
144 else: 240 else:
@@ -170,7 +266,8 @@ class Fetcher:
170 self._http2 = http2 266 self._http2 = http2
171 267
172 async def __aenter__(self) -> Fetcher: 268 async def __aenter__(self) -> Fetcher:
173− self._client = httpx.AsyncClient(headers=self.headers, timeout=httpx.Timeout(self.timeout_s, connect=20), follow_redirects=True, 269+ # redirects are followed manually in `get()` so that every hop goes through the SSRF guard
270+ self._client = httpx.AsyncClient(headers=self.headers, timeout=httpx.Timeout(self.timeout_s, connect=20), follow_redirects=False,
174 http2=self._http2, limits=httpx.Limits(max_connections=16, max_keepalive_connections=8)) 271 http2=self._http2, limits=httpx.Limits(max_connections=16, max_keepalive_connections=8))
175 return self 272 return self
176 273
@@ -187,9 +284,13 @@ class Fetcher:
187 284
188 async def get(self, url: str, *, etag: str | None = None, last_modified: str | None = None, min_bytes: int = 1, 285 async def get(self, url: str, *, etag: str | None = None, last_modified: str | None = None, min_bytes: int = 1,
189 escalate: bool = False, retries: int = 2, accept: str | None = None, rate_per_min: int | None = None) -> FetchResult: 286 escalate: bool = False, retries: int = 2, accept: str | None = None, rate_per_min: int | None = None) -> FetchResult:
190− """Direct fetch with conditional headers. Raises NotModified (304), FetchError, BlockedError.""" 287+ """Direct fetch with conditional headers. Raises NotModified (304), FetchError, BlockedError.
288+ Every destination and every redirect hop (max 5) is validated against the SSRF guard before a connection is opened."""
191 client = self.client 289 client = self.client
192− domain = domain_of(url) 290+ try:
291+ await validate_destination_async(url)
292+ except BlockedDestination as exc:
293+ raise FetchError(f"blocked destination: {exc}", url=url) from exc
193 if self.robots and not await _robots.allowed(client, url): 294 if self.robots and not await _robots.allowed(client, url):
194 raise BlockedError(f"robots.txt disallows {url}", status=None, url=url) 295 raise BlockedError(f"robots.txt disallows {url}", status=None, url=url)
195 headers: dict[str, str] = {} 296 headers: dict[str, str] = {}
@@ -200,13 +301,34 @@ class Fetcher:
200 if accept: 301 if accept:
201 headers["Accept"] = accept 302 headers["Accept"] = accept
202 last_exc: Exception | None = None 303 last_exc: Exception | None = None
304+ current = url
305+ hops = 0
203 for attempt in range(retries + 1): 306 for attempt in range(retries + 1):
204− await _limiter.wait(domain, rate_per_min or self.rate_per_min) 307+ await _limiter.wait(domain_of(current), rate_per_min or self.rate_per_min)
205 t0 = time.perf_counter() 308 t0 = time.perf_counter()
206 try: 309 try:
207− async with client.stream("GET", url, headers=headers) as r: 310+ async with client.stream("GET", current, headers=headers) as r:
208 if r.status_code == 304: 311 if r.status_code == 304:
209 raise NotModified() 312 raise NotModified()
313+ if r.status_code in REDIRECT_STATUS:
314+ location = r.headers.get("location")
315+ if not location:
316+ raise FetchError(f"http {r.status_code} without Location for {current}", status=r.status_code, url=url)
317+ hops += 1
318+ if hops > MAX_REDIRECTS:
319+ raise FetchError(f"too many redirects (> {MAX_REDIRECTS}) for {url}", status=r.status_code, url=url)
320+ nxt = urljoin(current, location)
321+ try:
322+ await validate_destination_async(nxt)
323+ except BlockedDestination as exc:
324+ raise FetchError(f"blocked destination: redirect {current} → {nxt}: {exc}", status=r.status_code, url=url) from exc
325+ if domain_of(nxt) != domain_of(current):
326+ headers.pop("If-None-Match", None) # validators belong to the original resource only
327+ headers.pop("If-Modified-Since", None)
328+ current = nxt
329+ # a redirect is not a retry: loop again without consuming the attempt budget
330+ return await self._follow(client, current, headers, url=url, hops=hops, min_bytes=min_bytes, escalate=escalate,
331+ retries=retries, rate_per_min=rate_per_min)
210 if r.status_code in TRANSIENT_STATUS and attempt < retries: 332 if r.status_code in TRANSIENT_STATUS and attempt < retries:
211 retry_after = r.headers.get("retry-after") 333 retry_after = r.headers.get("retry-after")
212 delay = min(60.0, float(retry_after)) if retry_after and retry_after.isdigit() else 2.0 * (2 ** attempt) 334 delay = min(60.0, float(retry_after)) if retry_after and retry_after.isdigit() else 2.0 * (2 ** attempt)
@@ -219,22 +341,7 @@ class Fetcher:
219 raise BlockedError(f"http {r.status_code} for {url}", status=r.status_code, url=url) 341 raise BlockedError(f"http {r.status_code} for {url}", status=r.status_code, url=url)
220 if r.status_code >= 400: 342 if r.status_code >= 400:
221 raise FetchError(f"http {r.status_code} for {url}", status=r.status_code, url=url) 343 raise FetchError(f"http {r.status_code} for {url}", status=r.status_code, url=url)
222− chunks: list[bytes] = [] 344+ return await self._read(r, url=url, final_url=current, t0=t0, min_bytes=min_bytes, escalate=escalate)
223− size = 0
224− async for chunk in r.aiter_bytes():
225− size += len(chunk)
226− if size > settings.max_body_bytes:
227− raise FetchError(f"body exceeds {settings.max_body_bytes} bytes", status=r.status_code, url=url)
228− chunks.append(chunk)
229− content = b"".join(chunks)
230− if len(content) < min_bytes:
231− raise FetchError(f"suspiciously short body ({len(content)} bytes) for {url}", status=r.status_code, url=url)
232− res = FetchResult(url=url, final_url=str(r.url), status=r.status_code, headers={k.lower(): v for k, v in r.headers.items()},
233− content=content, content_type=r.headers.get("content-type", "").lower(), fetched_at=datetime.now(UTC),
234− duration_ms=int((time.perf_counter() - t0) * 1000), transport="direct")
235− if escalate and res.is_html and _looks_like_challenge(content):
236− return await self._escalate(url, reason="anti-bot challenge page")
237− return res
238 except NotModified: 345 except NotModified:
239 raise 346 raise
240 except (BlockedError, FetchError): 347 except (BlockedError, FetchError):
@@ -247,6 +354,61 @@ class Fetcher:
247 raise FetchError(f"{exc.__class__.__name__}: {exc}", url=url) from exc 354 raise FetchError(f"{exc.__class__.__name__}: {exc}", url=url) from exc
248 raise FetchError(f"fetch failed: {last_exc}", url=url) 355 raise FetchError(f"fetch failed: {last_exc}", url=url)
249 356
357+ async def _follow(self, client: httpx.AsyncClient, current: str, headers: dict[str, str], *, url: str, hops: int, min_bytes: int,
358+ escalate: bool, retries: int, rate_per_min: int | None) -> FetchResult:
359+ """Continue a redirect chain: each hop is validated, capped at MAX_REDIRECTS, rate-limited per domain."""
360+ while True:
361+ await _limiter.wait(domain_of(current), rate_per_min or self.rate_per_min)
362+ t0 = time.perf_counter()
363+ try:
364+ async with client.stream("GET", current, headers=headers) as r:
365+ if r.status_code in REDIRECT_STATUS:
366+ location = r.headers.get("location")
367+ if not location:
368+ raise FetchError(f"http {r.status_code} without Location for {current}", status=r.status_code, url=url)
369+ hops += 1
370+ if hops > MAX_REDIRECTS:
371+ raise FetchError(f"too many redirects (> {MAX_REDIRECTS}) for {url}", status=r.status_code, url=url)
372+ nxt = urljoin(current, location)
373+ try:
374+ await validate_destination_async(nxt)
375+ except BlockedDestination as exc:
376+ raise FetchError(f"blocked destination: redirect {current} → {nxt}: {exc}", status=r.status_code, url=url) from exc
377+ if domain_of(nxt) != domain_of(current):
378+ headers.pop("If-None-Match", None)
379+ headers.pop("If-Modified-Since", None)
380+ current = nxt
381+ continue
382+ if r.status_code == 304:
383+ raise NotModified()
384+ if r.status_code in BLOCK_STATUS:
385+ if escalate:
386+ return await self._escalate(url, reason=f"http {r.status_code}")
387+ raise BlockedError(f"http {r.status_code} for {current}", status=r.status_code, url=url)
388+ if r.status_code >= 400:
389+ raise FetchError(f"http {r.status_code} for {current}", status=r.status_code, url=url)
390+ return await self._read(r, url=url, final_url=current, t0=t0, min_bytes=min_bytes, escalate=escalate)
391+ except (httpx.TimeoutException, httpx.TransportError) as exc:
392+ raise FetchError(f"{exc.__class__.__name__}: {exc}", url=url) from exc
393+
394+ async def _read(self, r: httpx.Response, *, url: str, final_url: str, t0: float, min_bytes: int, escalate: bool) -> FetchResult:
395+ chunks: list[bytes] = []
396+ size = 0
397+ async for chunk in r.aiter_bytes():
398+ size += len(chunk)
399+ if size > settings.max_body_bytes:
400+ raise FetchError(f"body exceeds {settings.max_body_bytes} bytes", status=r.status_code, url=url)
401+ chunks.append(chunk)
402+ content = b"".join(chunks)
403+ if len(content) < min_bytes:
404+ raise FetchError(f"suspiciously short body ({len(content)} bytes) for {url}", status=r.status_code, url=url)
405+ res = FetchResult(url=url, final_url=final_url, status=r.status_code, headers={k.lower(): v for k, v in r.headers.items()},
406+ content=content, content_type=r.headers.get("content-type", "").lower(), fetched_at=datetime.now(UTC),
407+ duration_ms=int((time.perf_counter() - t0) * 1000), transport="direct")
408+ if escalate and res.is_html and _looks_like_challenge(content):
409+ return await self._escalate(url, reason="anti-bot challenge page")
410+ return res
411+
250 # ---------------------------------------------------------------------------------------------- escalation (optional) 412 # ---------------------------------------------------------------------------------------------- escalation (optional)
251 async def _escalate(self, url: str, *, reason: str) -> FetchResult: 413 async def _escalate(self, url: str, *, reason: str) -> FetchResult:
252 log.warning("escalating fetch", extra={"url": url, "reason": reason}) 414 log.warning("escalating fetch", extra={"url": url, "reason": reason})
@@ -326,4 +488,5 @@ def file_result(path: str, *, url: str, content_type: str = "application/octet-s
326 _limiter = _RateLimiter() 488 _limiter = _RateLimiter()
327 _robots = _Robots() 489 _robots = _Robots()
328 490
329−__all__ = ["BlockedError", "FetchError", "FetchResult", "Fetcher", "NotModified", "canonicalize_url", "domain_of", "file_result"] 491+__all__ = ["MAX_REDIRECTS", "BlockedDestination", "BlockedError", "FetchError", "FetchResult", "Fetcher", "NotModified", "canonicalize_url", "domain_of",
492+ "file_result", "validate_destination", "validate_destination_async"]
modified src/aiatlas/sdk/resolution.py +124 −11
@@ -3,11 +3,19 @@
3 1. identifiers (scheme, value) → exact entity 3 1. identifiers (scheme, value) → exact entity
4 2. normalized alias within the same entity type (disambiguated by organization when several match) 4 2. normalized alias within the same entity type (disambiguated by organization when several match)
5 3. slug collision within the same type 5 3. slug collision within the same type
6− 4. otherwise create, and park ambiguous cases in the review queue as merge candidates 6+ 4. evaluation-effort variants of an existing model ("gpt-5-4-mini-medium") fold onto the canonical model — they are a result
7+ configuration, never an entity of their own
8+ 5. otherwise create, and park ambiguous cases in the review queue as merge candidates
9+
10+Guards: persisted `resolution_decisions` (keep_separate) are honoured — two entities an operator kept apart are never re-fused by an
11+alias match; aliases whose normalisation collapses digits and dots ("Qwen3-8B" ≡ "Qwen 38B") additionally require the same
12+`variant_key`; `model` and `artifact` are compatible types for lookups (an artifact used to be typed model — the same hub repo must
13+keep resolving to the same row).
7 """ 14 """
8 from __future__ import annotations 15 from __future__ import annotations
9 16
10 import logging 17 import logging
18+import re
11 from datetime import UTC, datetime 19 from datetime import UTC, datetime
12 from typing import Any 20 from typing import Any
13 21
@@ -15,6 +23,7 @@ from sqlalchemy.ext.asyncio import AsyncConnection
15 23
16 from aiatlas.db import execute, fetch_all, fetch_one, jsonb 24 from aiatlas.db import execute, fetch_all, fetch_one, jsonb
17 from aiatlas.ids import ENTITY_TYPES, new_id, normalize_alias, slugify 25 from aiatlas.ids import ENTITY_TYPES, new_id, normalize_alias, slugify
26+from aiatlas.ontology.models import analyze_model_name, base_name, variant_key
18 from aiatlas.sdk.facts import EntityRef 27 from aiatlas.sdk.facts import EntityRef
19 28
20 log = logging.getLogger(__name__) 29 log = logging.getLogger(__name__)
@@ -22,16 +31,28 @@ log = logging.getLogger(__name__)
22 # Types whose slug should be prefixed by the organization slug to stay unique and readable (models: `qwen-qwen3-8b` is ugly → 31 # Types whose slug should be prefixed by the organization slug to stay unique and readable (models: `qwen-qwen3-8b` is ugly →
23 # we keep the model name and only prefix on collision). 32 # we keep the model name and only prefix on collision).
24 GENERIC_NAMES = {"model", "models", "api", "pricing", "docs", "blog", "news", "research", "overview"} 33 GENERIC_NAMES = {"model", "models", "api", "pricing", "docs", "blog", "news", "research", "overview"}
34+# lookup-compatible type groups: a ref of one type may resolve to a stored entity of another type in the same group
35+_COMPATIBLE: dict[str, tuple[str, ...]] = {"model": ("model", "artifact"), "artifact": ("artifact", "model")}
36+# names whose normalised alias is ambiguous: a digit, a separator, a digit ("Qwen3-8B" / "Qwen 38B" → "qwen38b")
37+_DIGIT_SEP_DIGIT = re.compile(r"\d[.\-\s_]\d")
38+# identifier schemes issued by evaluators: an entity known ONLY through these may be an evaluation configuration rather than a model
39+EVALUATOR_SCHEMES = frozenset({"artificial_analysis", "livebench_model_id", "aider_model", "lmarena"})
40+
41+
42+def compatible_types(entity_type: str) -> tuple[str, ...]:
43+ return _COMPATIBLE.get(entity_type, (entity_type,))
25 44
26 45
27 class Resolver: 46 class Resolver:
28− def __init__(self, conn: AsyncConnection, *, snapshot_id: str | None = None, source_tier: int = 2): 47+ def __init__(self, conn: AsyncConnection, *, snapshot_id: str | None = None, source_tier: int = 2, variant_index: dict[str, str] | None = None):
29 self.conn = conn 48 self.conn = conn
30 self.snapshot_id = snapshot_id 49 self.snapshot_id = snapshot_id
31 self.tier = source_tier 50 self.tier = source_tier
51+ self.variant_index = variant_index # optional precomputed variant_key → canonical model id (canonicalization)
32 self._cache: dict[str, str] = {} 52 self._cache: dict[str, str] = {}
33 self.created: list[str] = [] 53 self.created: list[str] = []
34 self.updated: set[str] = set() 54 self.updated: set[str] = set()
55+ self.folded: dict[str, tuple[str, dict[str, str]]] = {} # ref key → (canonical id, effort config) for folded variants
35 56
36 async def resolve(self, ref: EntityRef, *, create: bool = True) -> str | None: 57 async def resolve(self, ref: EntityRef, *, create: bool = True) -> str | None:
37 if ref.id: 58 if ref.id:
@@ -51,6 +72,12 @@ class Resolver:
51 found = await self._by_alias(ref, org_id) 72 found = await self._by_alias(ref, org_id)
52 if found is None: 73 if found is None:
53 found = await self._by_slug(ref) 74 found = await self._by_slug(ref)
75+ if found is None and ref.entity_type == "model" and self.tier >= 2 and set(ref.identifiers) <= EVALUATOR_SCHEMES and not ref.attributes.get("hf_repo"):
76+ # only evaluator-only references fold: a ref carrying official/hub identifiers (or a tier-1 source) names a real model
77+ variant = await self.resolve_variant(ref, org_id=org_id)
78+ if variant:
79+ found = variant[0]
80+ self.folded[key] = variant
54 if found is None and not create: 81 if found is None and not create:
55 return None 82 return None
56 if found is None: 83 if found is None:
@@ -64,12 +91,13 @@ class Resolver:
64 91
65 # ---------------------------------------------------------------------------------------------- lookups 92 # ---------------------------------------------------------------------------------------------- lookups
66 async def _by_identifiers(self, ref: EntityRef) -> str | None: 93 async def _by_identifiers(self, ref: EntityRef) -> str | None:
94+ types = compatible_types(ref.entity_type)
67 for scheme, value in ref.identifiers.items(): 95 for scheme, value in ref.identifiers.items():
68 row = await fetch_one(self.conn, """select ei.entity_id, e.entity_type, e.merged_into from entity_identifiers ei 96 row = await fetch_one(self.conn, """select ei.entity_id, e.entity_type, e.merged_into from entity_identifiers ei
69 join entities e on e.id = ei.entity_id where ei.scheme = :s and ei.value = :v""", 97 join entities e on e.id = ei.entity_id where ei.scheme = :s and ei.value = :v""",
70 s=scheme, v=str(value)) 98 s=scheme, v=str(value))
71 if row: 99 if row:
72− if row["entity_type"] != ref.entity_type: 100+ if row["entity_type"] not in types:
73 log.warning("identifier type mismatch", extra={"scheme": scheme, "value": value, "have": row["entity_type"], "want": ref.entity_type}) 101 log.warning("identifier type mismatch", extra={"scheme": scheme, "value": value, "have": row["entity_type"], "want": ref.entity_type})
74 continue 102 continue
75 return row["merged_into"] or row["entity_id"] 103 return row["merged_into"] or row["entity_id"]
@@ -89,12 +117,27 @@ class Resolver:
89 norms = {normalize_alias(n) for n in names if n and normalize_alias(n)} 117 norms = {normalize_alias(n) for n in names if n and normalize_alias(n)}
90 if not norms: 118 if not norms:
91 return None 119 return None
92− rows = await fetch_all(self.conn, """select distinct e.id, e.organization_id, e.canonical_name, e.merged_into from entity_aliases a 120+ rows = await fetch_all(self.conn, """select distinct e.id, e.organization_id, e.canonical_name, e.merged_into, a.alias from entity_aliases a
93 join entities e on e.id = a.entity_id 121 join entities e on e.id = a.entity_id
94− where a.alias_norm = any(cast(:norms as text[])) and e.entity_type = :t""", norms=list(norms), t=ref.entity_type) 122+ where a.alias_norm = any(cast(:norms as text[])) and e.entity_type = any(cast(:types as text[]))""",
123+ norms=list(norms), types=list(compatible_types(ref.entity_type)))
95 if not rows: 124 if not rows:
96 return None 125 return None
97 rows = [{**r, "id": r["merged_into"] or r["id"]} for r in rows] 126 rows = [{**r, "id": r["merged_into"] or r["id"]} for r in rows]
127+ # collision safety: when the normalised alias erases digit separators, the candidate must share the ref's variant key
128+ # (or match one of the ref's names textually) — "Qwen 38B" must not land on "Qwen3-8B"
129+ if ref.entity_type in ("model", "artifact") and any(_DIGIT_SEP_DIGIT.search(n) for n in names if n):
130+ lower_names = {n.strip().lower() for n in names if n}
131+ ref_vk = variant_key(ref.name)
132+ safe = []
133+ for r in rows:
134+ if (r["alias"] or "").strip().lower() in lower_names or variant_key(r["canonical_name"]) == ref_vk:
135+ safe.append(r)
136+ else:
137+ log.info("alias collision refused", extra={"name": ref.name, "candidate": r["canonical_name"]})
138+ rows = safe
139+ if not rows:
140+ return None
98 kept = [] 141 kept = []
99 for r in rows: 142 for r in rows:
100 if await self._conflicting_identifier(r["id"], ref): 143 if await self._conflicting_identifier(r["id"], ref):
@@ -106,6 +149,13 @@ class Resolver:
106 if not rows: 149 if not rows:
107 return None 150 return None
108 ids = {r["id"] for r in rows} 151 ids = {r["id"] for r in rows}
152+ # persisted decisions: a candidate an operator kept separate from another entity is only accepted with a matching organisation
153+ separate = await self._kept_separate(sorted(ids))
154+ if separate:
155+ rows = [r for r in rows if r["id"] not in separate or (org_id and r["organization_id"] == org_id)]
156+ ids = {r["id"] for r in rows}
157+ if not rows:
158+ return None
109 if len(ids) == 1: 159 if len(ids) == 1:
110 return rows[0]["id"] 160 return rows[0]["id"]
111 if org_id: 161 if org_id:
@@ -117,22 +167,82 @@ class Resolver:
117 {"name": ref.name, "identifiers": ref.identifiers}) 167 {"name": ref.name, "identifiers": ref.identifiers})
118 return None 168 return None
119 169
170+ async def _kept_separate(self, ids: list[str]) -> set[str]:
171+ """Ids among `ids` that carry a `keep_separate` decision with any entity."""
172+ if not ids:
173+ return set()
174+ rows = await fetch_all(self.conn, """select a_id, b_id from resolution_decisions where decision = 'keep_separate'
175+ and (a_id = any(cast(:ids as text[])) or b_id = any(cast(:ids as text[])))""", ids=ids)
176+ out: set[str] = set()
177+ for r in rows:
178+ out.update(x for x in (r["a_id"], r["b_id"]) if x in ids)
179+ return out
180+
181+ async def kept_separate(self, a: str, b: str) -> bool:
182+ row = await fetch_one(self.conn, """select 1 from resolution_decisions where decision = 'keep_separate'
183+ and ((a_id = :a and b_id = :b) or (a_id = :b and b_id = :a)) limit 1""", a=a, b=b)
184+ return row is not None
185+
120 async def _by_slug(self, ref: EntityRef) -> str | None: 186 async def _by_slug(self, ref: EntityRef) -> str | None:
121 slug = ref.slug_hint or slugify(ref.name) 187 slug = ref.slug_hint or slugify(ref.name)
122 row = await fetch_one(self.conn, "select id, entity_type, organization_id, merged_into from entities where slug = :s", s=slug) 188 row = await fetch_one(self.conn, "select id, entity_type, organization_id, merged_into from entities where slug = :s", s=slug)
123− if row and row["entity_type"] == ref.entity_type and not await self._conflicting_identifier(row["merged_into"] or row["id"], ref): 189+ if row and row["entity_type"] in compatible_types(ref.entity_type) and not await self._conflicting_identifier(row["merged_into"] or row["id"], ref):
124 return row["merged_into"] or row["id"] 190 return row["merged_into"] or row["id"]
125 return None 191 return None
126 192
193+ async def resolve_variant(self, ref: EntityRef, *, org_id: str | None = None) -> tuple[str, dict[str, str]] | None:
194+ """Evaluation-effort variant ("claude-opus-5-medium", "qwen3-6-27b-non-reasoning") → (canonical model id, effort config) when a
195+ canonical model exists: same `artificial_analysis` identifier or alias as the base name, or the same `variant_key`
196+ (through the optional precomputed index). Returns None when the base cannot be resolved or is ambiguous."""
197+ a = analyze_model_name(ref.name)
198+ if not a.is_effort_variant:
199+ return None
200+ base = base_name(ref.name)
201+ base_norms = {normalize_alias(base), normalize_alias(a.base_key)} - {""}
202+ rows = await fetch_all(self.conn, """select distinct e.id, e.canonical_name, e.organization_id, e.merged_into from entities e
203+ left join entity_identifiers ei on ei.entity_id = e.id
204+ left join entity_aliases al on al.entity_id = e.id
205+ where e.entity_type = 'model' and e.merged_into is null
206+ and ((ei.scheme = 'artificial_analysis' and ei.value = any(cast(:bases as text[])))
207+ or al.alias_norm = any(cast(:norms as text[])) or e.slug = any(cast(:bases as text[])))""",
208+ bases=sorted({base.lower(), a.base_key}), norms=sorted(base_norms))
209+ candidates = {r["id"]: r for r in rows if r["id"] != ref.id}
210+ if self.variant_index:
211+ vk = variant_key(base)
212+ cid = self.variant_index.get(vk)
213+ if cid and cid != ref.id and cid not in candidates:
214+ row = await fetch_one(self.conn, "select id, canonical_name, organization_id, merged_into from entities where id = :id and merged_into is null", id=cid)
215+ if row:
216+ candidates[cid] = row
217+ # never fold onto another effort variant
218+ candidates = {k: v for k, v in candidates.items() if not analyze_model_name(v["canonical_name"]).is_effort_variant}
219+ if not candidates:
220+ return None
221+ if len(candidates) > 1 and org_id:
222+ same = {k: v for k, v in candidates.items() if v["organization_id"] == org_id}
223+ if same:
224+ candidates = same
225+ if len(candidates) != 1:
226+ await self._review("merge_candidate", sorted(candidates), f"effort variant '{ref.name}' matches {len(candidates)} base models",
227+ {"name": ref.name, "base": base})
228+ return None
229+ cid = next(iter(candidates))
230+ return cid, dict(a.effort)
231+
127 # ---------------------------------------------------------------------------------------------- writes 232 # ---------------------------------------------------------------------------------------------- writes
128 async def _create(self, ref: EntityRef, org_id: str | None) -> str: 233 async def _create(self, ref: EntityRef, org_id: str | None) -> str:
129 eid = new_id(ref.entity_type) 234 eid = new_id(ref.entity_type)
130 slug = await self._unique_slug(ref, org_id) 235 slug = await self._unique_slug(ref, org_id)
131− first_seen = datetime.now(UTC) 236+ now = datetime.now(UTC)
132− await execute(self.conn, """insert into entities (id, entity_type, canonical_name, slug, description, status, organization_id, first_seen_at, last_seen_at) 237+ first_seen = now
133− values (:id, :t, :n, :slug, :d, :status, :org, :fs, now())""", 238+ if ref.first_seen_hint is not None:
239+ hint = ref.first_seen_hint if ref.first_seen_hint.tzinfo else ref.first_seen_hint.replace(tzinfo=UTC)
240+ first_seen = min(now, hint)
241+ await execute(self.conn, """insert into entities (id, entity_type, canonical_name, slug, description, status, organization_id, first_seen_at, last_seen_at,
242+ identity_confidence, artifact_kind)
243+ values (:id, :t, :n, :slug, :d, :status, :org, :fs, now(), :ic, :ak)""",
134 id=eid, t=ref.entity_type, n=ref.name.strip()[:300], slug=slug, d=(ref.description or None), status=ref.status or "active", 244 id=eid, t=ref.entity_type, n=ref.name.strip()[:300], slug=slug, d=(ref.description or None), status=ref.status or "active",
135− org=org_id, fs=first_seen) 245+ org=org_id, fs=first_seen, ic=ref.identity_confidence or "high", ak=ref.artifact_kind if ref.entity_type == "artifact" else None)
136 self.created.append(eid) 246 self.created.append(eid)
137 return eid 247 return eid
138 248
@@ -145,6 +255,9 @@ class Resolver:
145 if ref.description: 255 if ref.description:
146 sets.append("description = case when description is null or length(description) < 40 then :d else description end") 256 sets.append("description = case when description is null or length(description) < 40 then :d else description end")
147 params["d"] = ref.description 257 params["d"] = ref.description
258+ if ref.first_seen_hint is not None:
259+ sets.append("first_seen_at = least(first_seen_at, :fs)")
260+ params["fs"] = ref.first_seen_hint if ref.first_seen_hint.tzinfo else ref.first_seen_hint.replace(tzinfo=UTC)
148 await execute(self.conn, f"update entities set {', '.join(sets)} where id = :id", **params) 261 await execute(self.conn, f"update entities set {', '.join(sets)} where id = :id", **params)
149 self.updated.add(eid) 262 self.updated.add(eid)
150 263
@@ -185,4 +298,4 @@ class Resolver:
185 on conflict (dedupe_key) do nothing""", id=new_id("review"), k=kind, ids=entity_ids, p=jsonb(payload), r=reason, d=dedupe) 298 on conflict (dedupe_key) do nothing""", id=new_id("review"), k=kind, ids=entity_ids, p=jsonb(payload), r=reason, d=dedupe)
186 299
187 300
188−__all__ = ["Resolver"] 301+__all__ = ["EVALUATOR_SCHEMES", "Resolver", "compatible_types"]
modified src/aiatlas/sdk/writer.py +226 −69
@@ -3,9 +3,14 @@
3 Temporal rules (per entity × property): 3 Temporal rules (per entity × property):
4 * no current claim → insert current claim, set attribute, NEW_* event for new entities 4 * no current claim → insert current claim, set attribute, NEW_* event for new entities
5 * same value → confirm (observed_at bumped), nothing else 5 * same value → confirm (observed_at bumped), nothing else
6+ * same assertion, older encoding (taxonomy) → re-encode the current claim in place (canonical value + value_raw), never an event
6 * different, source ≥ tier → supersede (valid_to = observed), insert new current, update attribute, CHANGE event for material props 7 * different, source ≥ tier → supersede (valid_to = observed), insert new current, update attribute, CHANGE event for material props
8+ * different, same source AND same extractor → supersede (a source correcting itself); an LLM claim never supersedes a deterministic
9+ claim from the same URL — it is stored as conflicting
7 * different, worse source → store as `conflicting`, flag confidence, review-queue item — never overwrite 10 * different, worse source → store as `conflicting`, flag confidence, review-queue item — never overwrite
8−Prices and benchmark results have their own append-only tables with the same close/open semantics. 11+Prices and benchmark results have their own append-only tables with the same close/open semantics; results additionally keep ONE
12+current row per (model, benchmark, metric, config_key): a newer run group closes the older rows so leaderboards show the latest run.
13+Every fact carries `run_id` (batch rollback) and every event carries `recorded_at`, `is_backfill` and `group_key`.
9 """ 14 """
10 from __future__ import annotations 15 from __future__ import annotations
11 16
@@ -19,6 +24,10 @@ from sqlalchemy.ext.asyncio import AsyncConnection
19 24
20 from aiatlas.db import execute, fetch_one, jsonb 25 from aiatlas.db import execute, fetch_one, jsonb
21 from aiatlas.ids import new_id, normalize_alias 26 from aiatlas.ids import new_id, normalize_alias
27+from aiatlas.ontology import benchmarks as bench_ontology
28+from aiatlas.ontology.anomalies import check_price_movement, check_result
29+from aiatlas.ontology.models import effort_config
30+from aiatlas.ontology.taxonomy import TAXONOMY_PROPERTIES, normalize_property
22 from aiatlas.sdk.facts import ( 31 from aiatlas.sdk.facts import (
23 EVENT_CATEGORY_BY_TYPE, 32 EVENT_CATEGORY_BY_TYPE,
24 MATERIAL_PROPERTIES, 33 MATERIAL_PROPERTIES,
@@ -29,13 +38,18 @@ from aiatlas.sdk.facts import (
29 ResultObs, 38 ResultObs,
30 ) 39 )
31 from aiatlas.sdk.resolution import Resolver 40 from aiatlas.sdk.resolution import Resolver
41+from aiatlas.services.events import classify_backfill, group_key_for, importance_for
32 42
33 log = logging.getLogger(__name__) 43 log = logging.getLogger(__name__)
34 44
35 TIER_CONFIDENCE = {1: "high", 2: "medium", 3: "low", 4: "low"} 45 TIER_CONFIDENCE = {1: "high", 2: "medium", 3: "low", 4: "low"}
36 SOFT_PROPERTIES = {"description", "summary", "tagline", "availability_note", "abstract", "notes", "training_data_notes", "safety_notes", "hardware_requirements"} 46 SOFT_PROPERTIES = {"description", "summary", "tagline", "availability_note", "abstract", "notes", "training_data_notes", "safety_notes", "hardware_requirements"}
47+# derived / secondary encodings of another property: never their own change event
48+NO_EVENT_PROPERTIES = {"license_key"}
37 NEW_IMPORTANCE = {"model": 3, "company": 2, "provider": 2, "paper": 1, "dataset": 1, "benchmark": 2, "framework": 1, "hardware": 2, 49 NEW_IMPORTANCE = {"model": 3, "company": 2, "provider": 2, "paper": 1, "dataset": 1, "benchmark": 2, "framework": 1, "hardware": 2,
38− "tool": 1, "repository": 0, "release": 2, "regulation": 2, "incident": 2, "organization": 1, "researcher": 0} 50+ "tool": 1, "repository": 0, "release": 2, "regulation": 2, "incident": 2, "organization": 1, "researcher": 0,
51+ "artifact": 0, "model_family": 1, "license": 0}
52+SILENT_NEW_TYPES = {"researcher", "country", "license", "quantization"}
39 53
40 54
41 def _norm_value(v: Any) -> Any: 55 def _norm_value(v: Any) -> Any:
@@ -55,6 +69,14 @@ def _same(a: Any, b: Any) -> bool:
55 return json.dumps(_norm_value(a), sort_keys=True, default=str) == json.dumps(_norm_value(b), sort_keys=True, default=str) 69 return json.dumps(_norm_value(a), sort_keys=True, default=str) == json.dumps(_norm_value(b), sort_keys=True, default=str)
56 70
57 71
72+def _cfg_hash(config: dict[str, Any] | None) -> str:
73+ return hashlib.sha1(json.dumps(config or {}, sort_keys=True, default=str).encode()).hexdigest()[:12]
74+
75+
76+def result_dedupe_key(model_id: str, benchmark_id: str, config: dict[str, Any] | None, metric: str | None) -> str:
77+ return f"{model_id}:{benchmark_id}:{_cfg_hash(config)}:{metric or ''}"
78+
79+
58 class WriteStats: 80 class WriteStats:
59 def __init__(self) -> None: 81 def __init__(self) -> None:
60 self.entities_created = 0 82 self.entities_created = 0
@@ -65,6 +87,7 @@ class WriteStats:
65 self.prices = 0 87 self.prices = 0
66 self.results = 0 88 self.results = 0
67 self.conflicts = 0 89 self.conflicts = 0
90+ self.folded_variants = 0
68 91
69 def as_dict(self) -> dict[str, int]: 92 def as_dict(self) -> dict[str, int]:
70 return dict(self.__dict__) 93 return dict(self.__dict__)
@@ -73,7 +96,8 @@ class WriteStats:
73 class FactWriter: 96 class FactWriter:
74 def __init__(self, conn: AsyncConnection, *, source_id: str | None, snapshot_id: str | None, source_url: str | None, 97 def __init__(self, conn: AsyncConnection, *, source_id: str | None, snapshot_id: str | None, source_url: str | None,
75 tier: int = 2, connector_name: str | None = None, extractor: str = "deterministic", extractor_version: str = "1", 98 tier: int = 2, connector_name: str | None = None, extractor: str = "deterministic", extractor_version: str = "1",
76− observed_at: datetime | None = None): 99+ observed_at: datetime | None = None, run_id: str | None = None, source_key: str | None = None, is_first_run: bool = False,
100+ variant_index: dict[str, str] | None = None):
77 self.conn = conn 101 self.conn = conn
78 self.source_id = source_id 102 self.source_id = source_id
79 self.snapshot_id = snapshot_id 103 self.snapshot_id = snapshot_id
@@ -83,14 +107,29 @@ class FactWriter:
83 self.extractor = extractor 107 self.extractor = extractor
84 self.extractor_version = extractor_version 108 self.extractor_version = extractor_version
85 self.observed_at = observed_at or datetime.now(UTC) 109 self.observed_at = observed_at or datetime.now(UTC)
86− self.resolver = Resolver(conn, snapshot_id=snapshot_id, source_tier=tier) 110+ self.run_id = run_id
111+ self._source_key = source_key
112+ self.is_first_run = is_first_run # connector's first successful run → every event is backfill (initial corpus)
113+ self.resolver = Resolver(conn, snapshot_id=snapshot_id, source_tier=tier, variant_index=variant_index)
87 self.stats = WriteStats() 114 self.stats = WriteStats()
88 115
116+ @property
117+ def derived(self) -> bool:
118+ """Derived writers (canonicalization) store disagreements as conflicting claims but never open review items."""
119+ return self.extractor == "derived"
120+
121+ async def source_key(self) -> str | None:
122+ if self._source_key is None and self.source_id:
123+ row = await fetch_one(self.conn, "select key from sources where id = :id", id=self.source_id)
124+ self._source_key = row["key"] if row else ""
125+ return self._source_key or None
126+
89 # ---------------------------------------------------------------------------------------------- entry point 127 # ---------------------------------------------------------------------------------------------- entry point
90 async def write(self, facts: Facts) -> WriteStats: 128 async def write(self, facts: Facts) -> WriteStats:
91 new_before = len(self.resolver.created) 129 new_before = len(self.resolver.created)
92 for ref in facts.entities: 130 for ref in facts.entities:
93 await self.resolver.resolve(ref) 131 await self.resolver.resolve(ref)
132+ await self._apply_hierarchy(ref)
94 for prop, value in ref.attributes.items(): 133 for prop, value in ref.attributes.items():
95 await self.write_claim(ref, prop, value) 134 await self.write_claim(ref, prop, value)
96 for c in facts.claims: 135 for c in facts.claims:
@@ -111,8 +150,31 @@ class FactWriter:
111 await self._new_entity_event(eid) 150 await self._new_entity_event(eid)
112 self.stats.entities_created = len(self.resolver.created) 151 self.stats.entities_created = len(self.resolver.created)
113 self.stats.entities_updated = len(self.resolver.updated - set(self.resolver.created)) 152 self.stats.entities_updated = len(self.resolver.updated - set(self.resolver.created))
153+ self.stats.folded_variants = len(self.resolver.folded)
114 return self.stats 154 return self.stats
115 155
156+ # ---------------------------------------------------------------------------------------------- hierarchy hints
157+ async def _apply_hierarchy(self, ref: EntityRef) -> None:
158+ """Materialise `family` / `canonical` hints: entities.family_id + member_of_family, entities.canonical_id + artifact_of, identity_confidence."""
159+ if not ref.id:
160+ return
161+ if ref.family is not None:
162+ if ref.family.entity_type != "model_family":
163+ ref.family.entity_type = "model_family"
164+ fid = await self.resolver.resolve(ref.family)
165+ if fid and fid != ref.id:
166+ await execute(self.conn, "update entities set family_id = :f where id = :e and family_id is distinct from :f", f=fid, e=ref.id)
167+ await self.write_relation(ref, "member_of_family", ref.family)
168+ if ref.canonical is not None:
169+ cid = await self.resolver.resolve(ref.canonical)
170+ if cid and cid != ref.id:
171+ await execute(self.conn, """update entities set canonical_id = :c, artifact_kind = coalesce(cast(:k as text), artifact_kind)
172+ where id = :e and (canonical_id is distinct from :c or (cast(:k as text) is not null and artifact_kind is null))""",
173+ c=cid, k=ref.artifact_kind, e=ref.id)
174+ await self.write_relation(ref, "artifact_of", ref.canonical, {"artifact_kind": ref.artifact_kind} if ref.artifact_kind else None)
175+ if ref.identity_confidence:
176+ await execute(self.conn, "update entities set identity_confidence = :ic where id = :e and identity_confidence <> :ic", ic=ref.identity_confidence, e=ref.id)
177+
116 # ---------------------------------------------------------------------------------------------- claims 178 # ---------------------------------------------------------------------------------------------- claims
117 async def write_claim(self, ref: EntityRef, prop: str, value: Any, *, unit: str | None = None, confidence: str | None = None, 179 async def write_claim(self, ref: EntityRef, prop: str, value: Any, *, unit: str | None = None, confidence: str | None = None,
118 observed_at: datetime | None = None, effective_at: datetime | None = None, source_url: str | None = None) -> None: 180 observed_at: datetime | None = None, effective_at: datetime | None = None, source_url: str | None = None) -> None:
@@ -121,67 +183,114 @@ class FactWriter:
121 eid = await self.resolver.resolve(ref) 183 eid = await self.resolver.resolve(ref)
122 assert eid 184 assert eid
123 value = _norm_value(value) 185 value = _norm_value(value)
186+ value_raw: str | None = None
187+ if prop in TAXONOMY_PROPERTIES:
188+ value, value_raw, mappings = normalize_property(ref.entity_type, prop, value)
189+ for domain, raw, canon in mappings:
190+ await self._record_mapping(domain, raw, canon)
191+ value = _norm_value(value)
124 observed = observed_at or self.observed_at 192 observed = observed_at or self.observed_at
125 conf = confidence or TIER_CONFIDENCE.get(self.tier, "medium") 193 conf = confidence or TIER_CONFIDENCE.get(self.tier, "medium")
126 url = source_url or self.source_url 194 url = source_url or self.source_url
127 noisy = prop.startswith(NOISY_PREFIXES) 195 noisy = prop.startswith(NOISY_PREFIXES)
128− current = await fetch_one(self.conn, """select id, value, tier, source_url, observed_at from claims 196+ current = await fetch_one(self.conn, """select id, value, value_raw, tier, source_url, observed_at, extractor from claims
129 where entity_id = :e and property = :p and status = 'current' order by valid_from desc limit 1""", 197 where entity_id = :e and property = :p and status = 'current' order by valid_from desc limit 1""",
130 e=eid, p=prop) 198 e=eid, p=prop)
131 is_new_entity = eid in self.resolver.created 199 is_new_entity = eid in self.resolver.created
200+ await self._apply_claim(ref, eid, prop, value, value_raw, current, unit=unit, conf=conf, observed=observed, effective_at=effective_at, url=url,
201+ noisy=noisy, is_new_entity=is_new_entity)
202+ if prop == "license" and isinstance(value, str):
203+ from aiatlas.ontology.licenses import LICENSES
204+
205+ if value in LICENSES: # canonical licence key as its own property (filterable), never its own event
206+ await self.write_claim(ref, "license_key", value, confidence=confidence, observed_at=observed_at, effective_at=effective_at, source_url=source_url)
207+
208+ async def _apply_claim(self, ref: EntityRef, eid: str, prop: str, value: Any, value_raw: str | None, current: dict[str, Any] | None, *, unit: str | None,
209+ conf: str, observed: datetime, effective_at: datetime | None, url: str | None, noisy: bool, is_new_entity: bool) -> None:
132 if current is None: 210 if current is None:
133− await self._insert_claim(eid, prop, value, unit, conf, "current", observed, effective_at, url) 211+ await self._insert_claim(eid, prop, value, unit, conf, "current", observed, effective_at, url, value_raw)
134− await self._set_attribute(eid, prop, value, unit, conf, url, observed) 212+ await self._set_attribute(eid, prop, value, unit, conf, url, observed, value_raw)
135 return 213 return
136 if _same(current["value"], value): 214 if _same(current["value"], value):
137 await execute(self.conn, "update claims set observed_at = greatest(observed_at, :o) where id = :id", o=observed, id=current["id"]) 215 await execute(self.conn, "update claims set observed_at = greatest(observed_at, :o) where id = :id", o=observed, id=current["id"])
138 if noisy: 216 if noisy:
139− await self._set_attribute(eid, prop, value, unit, conf, url, observed) 217+ await self._set_attribute(eid, prop, value, unit, conf, url, observed, value_raw)
218+ return
219+ if prop in TAXONOMY_PROPERTIES and _same(normalize_property(ref.entity_type, prop, current["value"])[0], value):
220+ # same assertion in an older encoding ("apache-2.0" → "Apache-2.0"): re-encode in place, keep the source label, no event
221+ raw_keep = current["value_raw"] or (current["value"] if isinstance(current["value"], str) else json.dumps(current["value"], ensure_ascii=False))
222+ raw_keep = raw_keep if raw_keep != value else None
223+ await execute(self.conn, """update claims set value = cast(:v as jsonb), value_text = :vt, value_raw = :raw, observed_at = greatest(observed_at, :o) where id = :id""",
224+ v=jsonb(value), vt=value[:2000] if isinstance(value, str) else None, raw=raw_keep, o=observed, id=current["id"])
225+ await self._set_attribute(eid, prop, value, unit, conf, url, observed, raw_keep, keep_provenance=True)
140 return 226 return
141 same_source = bool(url and current["source_url"] == url) 227 same_source = bool(url and current["source_url"] == url)
228+ same_extractor = (current["extractor"] or "deterministic") == self.extractor
142 if prop in SOFT_PROPERTIES and not same_source: 229 if prop in SOFT_PROPERTIES and not same_source:
143 # soft text (descriptions, notes): first statement wins until *its own* source changes; never an event, never a conflict 230 # soft text (descriptions, notes): first statement wins until *its own* source changes; never an event, never a conflict
144 return 231 return
145− if self.tier <= (current["tier"] or 2) or same_source: 232+ if self.tier <= (current["tier"] or 2) or (same_source and same_extractor):
146 # supersede 233 # supersede
147 await execute(self.conn, "update claims set status = 'superseded', valid_to = :o where id = :id", o=observed, id=current["id"]) 234 await execute(self.conn, "update claims set status = 'superseded', valid_to = :o where id = :id", o=observed, id=current["id"])
148− await self._insert_claim(eid, prop, value, unit, conf, "current", observed, effective_at, url) 235+ await self._insert_claim(eid, prop, value, unit, conf, "current", observed, effective_at, url, value_raw)
149− await self._set_attribute(eid, prop, value, unit, conf, url, observed) 236+ await self._set_attribute(eid, prop, value, unit, conf, url, observed, value_raw)
150− if not noisy and not is_new_entity and prop not in SOFT_PROPERTIES: 237+ if not noisy and not is_new_entity and prop not in SOFT_PROPERTIES and prop not in NO_EVENT_PROPERTIES:
151 await self._property_change_event(eid, prop, current["value"], value, observed, effective_at, url) 238 await self._property_change_event(eid, prop, current["value"], value, observed, effective_at, url)
152− else: 239+ return
153− await self._insert_claim(eid, prop, value, unit, conf, "conflicting", observed, effective_at, url) 240+ # the same disagreement from the same source/extractor is recorded once (re-observed, not re-inserted)
154− await execute(self.conn, "update claims set confidence = 'conflicted' where id = :id", id=current["id"]) 241+ dup = await fetch_one(self.conn, """select id from claims where entity_id = :e and property = :p and status = 'conflicting' and value = cast(:v as jsonb)
155− await execute(self.conn, """update entities set quality = quality || jsonb_build_object('conflicts', coalesce((quality->>'conflicts')::int, 0) + 1) 242+ and extractor = :ex and coalesce(source_url, '') = :url limit 1""", e=eid, p=prop, v=jsonb(value), ex=self.extractor, url=url or "")
156− where id = :id""", id=eid) 243+ if dup:
157− self.stats.conflicts += 1 244+ await execute(self.conn, "update claims set observed_at = greatest(observed_at, :o) where id = :id", o=observed, id=dup["id"])
245+ return
246+ await self._insert_claim(eid, prop, value, unit, conf, "conflicting", observed, effective_at, url, value_raw)
247+ await execute(self.conn, "update claims set confidence = 'conflicted' where id = :id", id=current["id"])
248+ await execute(self.conn, """update entities set quality = quality || jsonb_build_object('conflicts', coalesce((quality->>'conflicts')::int, 0) + 1)
249+ where id = :id""", id=eid)
250+ self.stats.conflicts += 1
251+ if not self.derived:
158 dedupe = f"conflict:{eid}:{prop}:{hashlib.sha1(json.dumps(value, sort_keys=True, default=str).encode()).hexdigest()[:10]}" 252 dedupe = f"conflict:{eid}:{prop}:{hashlib.sha1(json.dumps(value, sort_keys=True, default=str).encode()).hexdigest()[:10]}"
159 await execute(self.conn, """insert into review_queue (id, kind, entity_ids, payload, reason, dedupe_key) 253 await execute(self.conn, """insert into review_queue (id, kind, entity_ids, payload, reason, dedupe_key)
160 values (:id, 'conflict', :ids, cast(:p as jsonb), :r, :d) on conflict (dedupe_key) do nothing""", 254 values (:id, 'conflict', :ids, cast(:p as jsonb), :r, :d) on conflict (dedupe_key) do nothing""",
161 id=new_id("review"), ids=[eid], p=jsonb({"property": prop, "current": current["value"], "current_source": current["source_url"], 255 id=new_id("review"), ids=[eid], p=jsonb({"property": prop, "current": current["value"], "current_source": current["source_url"],
162− "claimed": value, "claimed_source": url, "tier": self.tier}), 256+ "claimed": value, "claimed_source": url, "tier": self.tier, "extractor": self.extractor}),
163 r=f"source disagrees on {prop}", d=dedupe) 257 r=f"source disagrees on {prop}", d=dedupe)
164 258
259+ async def _record_mapping(self, domain: str, raw: str, canonical: str | None) -> None:
260+ if canonical is not None and raw == canonical:
261+ return # identity mappings carry no information
262+ await execute(self.conn, """insert into taxonomy_mappings (domain, raw, canonical) values (:d, :r, :c)
263+ on conflict (domain, raw) do update set count = taxonomy_mappings.count + 1, last_seen_at = now(),
264+ canonical = coalesce(excluded.canonical, taxonomy_mappings.canonical)""", d=domain, r=raw[:300], c=canonical)
265+
165 async def _insert_claim(self, eid: str, prop: str, value: Any, unit: str | None, conf: str, status: str, observed: datetime, 266 async def _insert_claim(self, eid: str, prop: str, value: Any, unit: str | None, conf: str, status: str, observed: datetime,
166− effective_at: datetime | None, url: str | None) -> None: 267+ effective_at: datetime | None, url: str | None, value_raw: str | None = None) -> None:
167 text_val = value if isinstance(value, str) else None 268 text_val = value if isinstance(value, str) else None
168 num_val = float(value) if isinstance(value, (int, float)) and not isinstance(value, bool) else None 269 num_val = float(value) if isinstance(value, (int, float)) and not isinstance(value, bool) else None
169 await execute(self.conn, """insert into claims (id, entity_id, property, value, value_text, value_num, unit, source_id, snapshot_id, source_url, tier, 270 await execute(self.conn, """insert into claims (id, entity_id, property, value, value_text, value_num, unit, source_id, snapshot_id, source_url, tier,
170− confidence, status, extractor, extractor_version, observed_at, effective_at, valid_from) 271+ confidence, status, extractor, extractor_version, observed_at, effective_at, valid_from, run_id, value_raw)
171− values (:id, :e, :p, cast(:v as jsonb), :vt, :vn, :u, :src, :snap, :url, :tier, :conf, :status, :ex, :exv, :o, :eff, :vf)""", 272+ values (:id, :e, :p, cast(:v as jsonb), :vt, :vn, :u, :src, :snap, :url, :tier, :conf, :status, :ex, :exv, :o, :eff, :vf, :run, :raw)""",
172 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, 273 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,
173 snap=self.snapshot_id, url=url, tier=self.tier, conf=conf, status=status, ex=self.extractor, exv=self.extractor_version, 274 snap=self.snapshot_id, url=url, tier=self.tier, conf=conf, status=status, ex=self.extractor, exv=self.extractor_version,
174− o=observed, eff=effective_at, vf=effective_at or observed) 275+ o=observed, eff=effective_at, vf=effective_at or observed, run=self.run_id, raw=value_raw[:500] if value_raw else None)
175 self.stats.claims += 1 276 self.stats.claims += 1
176 277
177− async def _set_attribute(self, eid: str, prop: str, value: Any, unit: str | None, conf: str, url: str | None, observed: datetime) -> None: 278+ async def _set_attribute(self, eid: str, prop: str, value: Any, unit: str | None, conf: str, url: str | None, observed: datetime,
279+ value_raw: str | None = None, *, keep_provenance: bool = False) -> None:
178 prov = {"source_id": self.source_id, "snapshot_id": self.snapshot_id, "url": url, "observed_at": observed.isoformat(timespec="seconds"), 280 prov = {"source_id": self.source_id, "snapshot_id": self.snapshot_id, "url": url, "observed_at": observed.isoformat(timespec="seconds"),
179 "tier": self.tier, "confidence": conf, "extractor": self.extractor} 281 "tier": self.tier, "confidence": conf, "extractor": self.extractor}
180 if unit: 282 if unit:
181 prov["unit"] = unit 283 prov["unit"] = unit
182− await execute(self.conn, """update entities set attributes = attributes || jsonb_build_object(cast(:p as text), cast(:v as jsonb)), 284+ attrs = {prop: value}
183− provenance = provenance || jsonb_build_object(cast(:p as text), cast(:prov as jsonb)), last_seen_at = greatest(last_seen_at, :o) 285+ if value_raw is not None:
184− where id = :id""", p=prop, v=jsonb(value), prov=jsonb(prov), o=observed, id=eid) 286+ attrs[f"{prop}_raw"] = value_raw
287+ if keep_provenance:
288+ await execute(self.conn, """update entities set attributes = (attributes - cast(:praw as text)) || cast(:a as jsonb), last_seen_at = greatest(last_seen_at, :o) where id = :id""",
289+ praw=f"{prop}_raw", a=jsonb(attrs), o=observed, id=eid)
290+ else:
291+ await execute(self.conn, """update entities set attributes = (attributes - cast(:praw as text)) || cast(:a as jsonb),
292+ provenance = provenance || jsonb_build_object(cast(:p as text), cast(:prov as jsonb)), last_seen_at = greatest(last_seen_at, :o)
293+ where id = :id""", praw=f"{prop}_raw", a=jsonb(attrs), p=prop, prov=jsonb(prov), o=observed, id=eid)
185 if prop == "status" and isinstance(value, str): 294 if prop == "status" and isinstance(value, str):
186 await execute(self.conn, "update entities set status = :s where id = :id", s=value[:40], id=eid) 295 await execute(self.conn, "update entities set status = :s where id = :id", s=value[:40], id=eid)
187 if prop == "description" and isinstance(value, str): 296 if prop == "description" and isinstance(value, str):
@@ -192,9 +301,9 @@ class FactWriter:
192 event_type, importance = MATERIAL_PROPERTIES.get(prop, ("PROPERTY_CHANGED", 0)) 301 event_type, importance = MATERIAL_PROPERTIES.get(prop, ("PROPERTY_CHANGED", 0))
193 row = await fetch_one(self.conn, "select canonical_name, entity_type from entities where id = :id", id=eid) 302 row = await fetch_one(self.conn, "select canonical_name, entity_type from entities where id = :id", id=eid)
194 name = row["canonical_name"] if row else eid 303 name = row["canonical_name"] if row else eid
195− category = EVENT_CATEGORY_BY_TYPE.get(row["entity_type"] if row else "", "update") 304+ etype = row["entity_type"] if row else ""
196− if prop == "status" and str(new).lower() in ("deprecated", "retired", "discontinued"): 305+ category = EVENT_CATEGORY_BY_TYPE.get(etype, "update")
197− importance = 3 306+ importance = importance_for(event_type, etype, old, new, tier=self.tier, default=importance)
198 summary = f"{name}: {prop.replace('_', ' ')} changed from {_short(old)} to {_short(new)}" 307 summary = f"{name}: {prop.replace('_', ' ')} changed from {_short(old)} to {_short(new)}"
199 dedupe = f"{event_type}:{eid}:{prop}:{hashlib.sha1(json.dumps([old, new], sort_keys=True, default=str).encode()).hexdigest()[:12]}" 308 dedupe = f"{event_type}:{eid}:{prop}:{hashlib.sha1(json.dumps([old, new], sort_keys=True, default=str).encode()).hexdigest()[:12]}"
200 await self.emit_event(event_type, category, summary, entity_id=eid, old_value=old, new_value=new, importance=importance, 309 await self.emit_event(event_type, category, summary, entity_id=eid, old_value=old, new_value=new, importance=importance,
@@ -217,10 +326,10 @@ class FactWriter:
217 else: 326 else:
218 await execute(self.conn, "update relations set observed_at = :o where id = :id", o=self.observed_at, id=existing["id"]) 327 await execute(self.conn, "update relations set observed_at = :o where id = :id", o=self.observed_at, id=existing["id"])
219 return 328 return
220− 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) 329+ 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, run_id)
221− values (:id, :s, :p, :o, cast(:a as jsonb), :src, :snap, :url, :tier, :conf, :obs, :obs)""", 330+ values (:id, :s, :p, :o, cast(:a as jsonb), :src, :snap, :url, :tier, :conf, :obs, :obs, :run)""",
222 id=new_id("relation"), s=sid, p=predicate, o=oid, a=jsonb(attributes or {}), src=self.source_id, snap=self.snapshot_id, 331 id=new_id("relation"), s=sid, p=predicate, o=oid, a=jsonb(attributes or {}), src=self.source_id, snap=self.snapshot_id,
223− url=source_url or self.source_url, tier=self.tier, conf=conf, obs=self.observed_at) 332+ url=source_url or self.source_url, tier=self.tier, conf=conf, obs=self.observed_at, run=self.run_id)
224 self.stats.relations += 1 333 self.stats.relations += 1
225 334
226 # ---------------------------------------------------------------------------------------------- prices 335 # ---------------------------------------------------------------------------------------------- prices
@@ -242,16 +351,17 @@ class FactWriter:
242 await execute(self.conn, "update prices set observed_at = :o where id = :id", o=self.observed_at, id=current["id"]) 351 await execute(self.conn, "update prices set observed_at = :o where id = :id", o=self.observed_at, id=current["id"])
243 return 352 return
244 await execute(self.conn, "update prices set valid_to = :o where id = :id", o=self.observed_at, id=current["id"]) 353 await execute(self.conn, "update prices set valid_to = :o where id = :id", o=self.observed_at, id=current["id"])
354+ price_id = new_id("price")
245 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, 355 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,
246 batch_input_per_mtok, batch_output_per_mtok, per_image, per_request, currency, context_length, max_output_tokens, features, 356 batch_input_per_mtok, batch_output_per_mtok, per_image, per_request, currency, context_length, max_output_tokens, features,
247− observed_at, valid_from, source_id, snapshot_id, source_url, tier, meta) 357+ observed_at, valid_from, source_id, snapshot_id, source_url, tier, meta, run_id)
248− 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))""", 358+ 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), :run)""",
249− 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, 359+ id=price_id, 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,
250 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, 360 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,
251 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, 361 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,
252− url=url, tier=self.tier, meta=jsonb(p.meta)) 362+ url=url, tier=self.tier, meta=jsonb(p.meta), run=self.run_id)
253 self.stats.prices += 1 363 self.stats.prices += 1
254− model = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=mid) 364+ model = await fetch_one(self.conn, "select canonical_name, entity_type from entities where id = :id", id=mid)
255 provider = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=pid) 365 provider = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=pid)
256 mname = model["canonical_name"] if model else mid 366 mname = model["canonical_name"] if model else mid
257 pname = provider["canonical_name"] if provider else pid 367 pname = provider["canonical_name"] if provider else pid
@@ -260,8 +370,14 @@ class FactWriter:
260 new = {"input_per_mtok": p.input_per_mtok, "output_per_mtok": p.output_per_mtok} 370 new = {"input_per_mtok": p.input_per_mtok, "output_per_mtok": p.output_per_mtok}
261 summary = f"{pname} changed pricing for {mname}: {_fmt_price(old)} → {_fmt_price(new)}" 371 summary = f"{pname} changed pricing for {mname}: {_fmt_price(old)} → {_fmt_price(new)}"
262 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]}" 372 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]}"
263− await self.emit_event("PRICE_CHANGED", "price", summary, entity_id=mid, old_value=old, new_value=new, importance=2, dedupe_key=dedupe, 373+ importance = importance_for("PRICE_CHANGED", model["entity_type"] if model else None, old, new, tier=self.tier)
374+ await self.emit_event("PRICE_CHANGED", "price", summary, entity_id=mid, old_value=old, new_value=new, importance=importance, dedupe_key=dedupe,
264 source_url=url, meta={"provider_id": pid, "provider": pname}) 375 source_url=url, meta={"provider_id": pid, "provider": pname})
376+ for anomaly in check_price_movement({**old, "model_id": mid, "provider_id": pid}, {**new, "model_id": mid, "provider_id": pid}):
377+ from aiatlas.services.anomalies import record
378+
379+ anomaly.detail["price_id"] = price_id
380+ await record(self.conn, anomaly)
265 else: 381 else:
266 new = {"input_per_mtok": p.input_per_mtok, "output_per_mtok": p.output_per_mtok} 382 new = {"input_per_mtok": p.input_per_mtok, "output_per_mtok": p.output_per_mtok}
267 summary = f"{pname} lists {mname} at {_fmt_price(new)}" 383 summary = f"{pname} lists {mname} at {_fmt_price(new)}"
@@ -277,51 +393,87 @@ class FactWriter:
277 if not mid or not bid: 393 if not mid or not bid:
278 return 394 return
279 url = r.source_url or self.source_url 395 url = r.source_url or self.source_url
280− cfg_hash = hashlib.sha1(json.dumps(r.config, sort_keys=True, default=str).encode()).hexdigest()[:12] 396+ # evaluation-effort variants ("gpt-5-4-mini-medium") are a *configuration* of the canonical model: the effort dict lands in the
281− dedupe = f"{mid}:{bid}:{cfg_hash}:{r.metric or ''}" 397+ # config so folded results stay distinguishable and comparable (reasoning_effort is a condition key, not a task key)
282− existing = await fetch_one(self.conn, "select id, score from benchmark_results where dedupe_key = :d", d=dedupe) 398+ config = effort_config(r.model.name, r.config)
399+ metric = r.metric
400+ dedupe = result_dedupe_key(mid, bid, config, metric)
401+ config_key = bench_ontology.config_key(config, metric)
402+ trust = r.trust_level or bench_ontology.trust_level(await self.source_key(), config, extractor=self.extractor)
403+ variant = r.variant or bench_ontology.variant_from_config(config)
404+ run_group = r.run_group or bench_ontology.run_group_from_config(config)
283 conf = r.confidence or TIER_CONFIDENCE.get(self.tier, "medium") 405 conf = r.confidence or TIER_CONFIDENCE.get(self.tier, "medium")
406+ lo, hi = bench_ontology.metric_bounds(metric, r.unit)
407+ out_of_range = (hi is not None and r.score > hi + 1e-9) or (lo is not None and r.score < lo - 1e-9)
408+ if out_of_range:
409+ conf = "low"
410+ existing = await fetch_one(self.conn, "select id, score from benchmark_results where dedupe_key = :d", d=dedupe)
411+ names: dict[str, Any] | None = None
284 if existing: 412 if existing:
285 if abs((existing["score"] or 0) - r.score) > 1e-9: 413 if abs((existing["score"] or 0) - r.score) > 1e-9:
286− await execute(self.conn, "update benchmark_results set valid_to = :o, dedupe_key = dedupe_key || ':' || :suffix where id = :id", 414+ await execute(self.conn, "update benchmark_results set valid_to = :o, is_current = false, dedupe_key = dedupe_key || ':' || :suffix where id = :id",
287 o=self.observed_at, suffix=new_id("result")[-10:], id=existing["id"]) 415 o=self.observed_at, suffix=new_id("result")[-10:], id=existing["id"])
288− model = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=mid) 416+ names = await self._names(mid, bid)
289− bench = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=bid) 417+ await self.emit_event("BENCHMARK_UPDATED", "benchmark", f"{names['model']} on {names['bench']}: {existing['score']:g} → {r.score:g}",
290− await self.emit_event("BENCHMARK_UPDATED", "benchmark",
291− f"{model['canonical_name'] if model else mid} on {bench['canonical_name'] if bench else bid}: {existing['score']:g} → {r.score:g}",
292 entity_id=mid, old_value=existing["score"], new_value=r.score, importance=1, 418 entity_id=mid, old_value=existing["score"], new_value=r.score, importance=1,
293 dedupe_key=f"BENCHMARK_UPDATED:{dedupe}:{r.score:g}", source_url=url, meta={"benchmark_id": bid}) 419 dedupe_key=f"BENCHMARK_UPDATED:{dedupe}:{r.score:g}", source_url=url, meta={"benchmark_id": bid})
294 else: 420 else:
295 await execute(self.conn, "update benchmark_results set observed_at = :o where id = :id", o=self.observed_at, id=existing["id"]) 421 await execute(self.conn, "update benchmark_results set observed_at = :o where id = :id", o=self.observed_at, id=existing["id"])
296 return 422 return
423+ rid = new_id("result")
297 await execute(self.conn, """insert into benchmark_results (id, model_id, benchmark_id, score, metric, unit, higher_is_better, config, evaluated_at, observed_at, 424 await execute(self.conn, """insert into benchmark_results (id, model_id, benchmark_id, score, metric, unit, higher_is_better, config, evaluated_at, observed_at,
298− source_id, snapshot_id, source_url, tier, confidence, dedupe_key) 425+ source_id, snapshot_id, source_url, tier, confidence, dedupe_key, config_key, trust_level, variant, run_group, is_current, extractor, run_id)
299− values (:id, :m, :b, :s, :metric, :unit, :hib, cast(:cfg as jsonb), :ev, :obs, :src, :snap, :url, :tier, :conf, :d)""", 426+ values (:id, :m, :b, :s, :metric, :unit, :hib, cast(:cfg as jsonb), :ev, :obs, :src, :snap, :url, :tier, :conf, :d, :ck, :trust, :variant, :rg, true, :ex, :run)""",
300− 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), 427+ id=rid, m=mid, b=bid, s=r.score, metric=metric, unit=r.unit, hib=r.higher_is_better, cfg=jsonb(config), ev=r.evaluated_at, obs=self.observed_at,
301− 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) 428+ src=self.source_id, snap=self.snapshot_id, url=url, tier=self.tier, conf=conf, d=dedupe, ck=config_key, trust=trust, variant=variant, rg=run_group,
429+ ex=self.extractor, run=self.run_id)
302 self.stats.results += 1 430 self.stats.results += 1
431+ # one current row per (model, benchmark, metric, config_key): a different (older) run group is closed so the leaderboard shows the latest run
432+ await execute(self.conn, """update benchmark_results set is_current = false, valid_to = coalesce(valid_to, :obs)
433+ where model_id = :m and benchmark_id = :b and coalesce(metric, '') = :metric and config_key = :ck and id <> :id and is_current
434+ and coalesce(run_group, '') <> :rg and observed_at <= :obs""",
435+ obs=self.observed_at, m=mid, b=bid, metric=metric or "", ck=config_key, id=rid, rg=run_group or "")
436+ if out_of_range:
437+ from aiatlas.services.anomalies import record
438+
439+ names = names or await self._names(mid, bid)
440+ for anomaly in check_result({"id": rid, "model_id": mid, "benchmark_id": bid, "score": r.score, "metric": metric, "unit": r.unit,
441+ "model_name": names["model"], "benchmark_name": names["bench"]}):
442+ await record(self.conn, anomaly)
303 if not existing: 443 if not existing:
304− model = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=mid) 444+ names = names or await self._names(mid, bid)
305− bench = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=bid) 445+ await self.emit_event("BENCHMARK_RESULT", "benchmark", f"{names['model']} scores {r.score:g}{r.unit or ''} on {names['bench']}",
306− await self.emit_event("BENCHMARK_RESULT", "benchmark",
307− f"{model['canonical_name'] if model else mid} scores {r.score:g}{r.unit or ''} on {bench['canonical_name'] if bench else bid}",
308 entity_id=mid, new_value=r.score, importance=1, dedupe_key=f"BENCHMARK_RESULT:{dedupe}", source_url=url, 446 entity_id=mid, new_value=r.score, importance=1, dedupe_key=f"BENCHMARK_RESULT:{dedupe}", source_url=url,
309− meta={"benchmark_id": bid, "metric": r.metric}) 447+ meta={"benchmark_id": bid, "metric": metric, "config_key": config_key})
310 await self.write_relation(r.model, "evaluated_on", r.benchmark, source_url=url) 448 await self.write_relation(r.model, "evaluated_on", r.benchmark, source_url=url)
311 449
450+ async def _names(self, mid: str, bid: str) -> dict[str, Any]:
451+ model = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=mid)
452+ bench = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=bid)
453+ return {"model": model["canonical_name"] if model else mid, "bench": bench["canonical_name"] if bench else bid}
454+
312 # ---------------------------------------------------------------------------------------------- events 455 # ---------------------------------------------------------------------------------------------- events
313 async def emit_event(self, event_type: str, category: str, summary: str, *, entity_id: str | None = None, old_value: Any = None, 456 async def emit_event(self, event_type: str, category: str, summary: str, *, entity_id: str | None = None, old_value: Any = None,
314 new_value: Any = None, importance: int = 2, effective_at: datetime | None = None, dedupe_key: str | None = None, 457 new_value: Any = None, importance: int = 2, effective_at: datetime | None = None, dedupe_key: str | None = None,
315− source_url: str | None = None, meta: dict[str, Any] | None = None, observed_at: datetime | None = None) -> None: 458+ source_url: str | None = None, meta: dict[str, Any] | None = None, observed_at: datetime | None = None,
459+ entity_first_seen: datetime | None = None, is_backfill: bool | None = None) -> None:
316 dedupe = dedupe_key or f"{event_type}:{entity_id or ''}:{normalize_alias(summary)[:120]}" 460 dedupe = dedupe_key or f"{event_type}:{entity_id or ''}:{normalize_alias(summary)[:120]}"
461+ observed = observed_at or self.observed_at
462+ if self.derived:
463+ # a derived writer re-encodes what we already know: its events are bookkeeping, never news
464+ is_backfill, importance = True, min(importance, 1)
465+ backfill = is_backfill if is_backfill is not None else classify_backfill(event_type, effective_at, observed, is_first_run=self.is_first_run,
466+ entity_first_seen=entity_first_seen)
467+ group_key = group_key_for(event_type, entity_id, effective_at, observed)
317 await execute(self.conn, """insert into change_events (id, entity_id, event_type, category, property, old_value, new_value, summary, importance, observed_at, 468 await execute(self.conn, """insert into change_events (id, entity_id, event_type, category, property, old_value, new_value, summary, importance, observed_at,
318− effective_at, source_id, snapshot_id, source_url, connector_name, dedupe_key, meta) 469+ effective_at, source_id, snapshot_id, source_url, connector_name, dedupe_key, meta, run_id, recorded_at, is_backfill, group_key)
319− 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)) 470+ 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),
471+ :run, now(), :bf, :gk)
320 on conflict (dedupe_key) do nothing""", 472 on conflict (dedupe_key) do nothing""",
321 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, 473 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,
322− 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, 474+ n=jsonb(new_value) if new_value is not None else None, s=summary[:500], imp=max(0, min(3, importance)), obs=observed,
323 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], 475 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],
324− meta=jsonb(meta or {})) 476+ meta=jsonb(meta or {}), run=self.run_id, bf=backfill, gk=group_key)
325 self.stats.events += 1 477 self.stats.events += 1
326 478
327 async def _new_entity_event(self, eid: str) -> None: 479 async def _new_entity_event(self, eid: str) -> None:
@@ -329,25 +481,30 @@ class FactWriter:
329 if not row: 481 if not row:
330 return 482 return
331 etype = row["entity_type"] 483 etype = row["entity_type"]
332− if etype in ("researcher", "country", "license", "quantization"): 484+ if etype in SILENT_NEW_TYPES:
333 return 485 return
334− importance = NEW_IMPORTANCE.get(etype, 1)
335− if self.tier > 2:
336− importance = max(0, importance - 1)
337 org = None 486 org = None
487+ org_models = 0
338 if row["organization_id"]: 488 if row["organization_id"]:
339− o = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=row["organization_id"]) 489+ o = await fetch_one(self.conn, """select canonical_name, (select count(*) from entities m where m.organization_id = e.id and m.entity_type = 'model'
490+ and m.merged_into is null) as models from entities e where e.id = :id""", id=row["organization_id"])
340 org = o["canonical_name"] if o else None 491 org = o["canonical_name"] if o else None
492+ org_models = int(o["models"] or 0) if o else 0
493+ attrs = row["attributes"] or {}
494+ if etype == "model":
495+ importance = importance_for("NEW_MODEL", etype, tier=self.tier, org_model_count=org_models, openness=attrs.get("openness"))
496+ else:
497+ importance = importance_for(f"NEW_{etype.upper()}", etype, tier=self.tier, default=NEW_IMPORTANCE.get(etype, 1))
341 label = etype.replace("_", " ") 498 label = etype.replace("_", " ")
342 summary = f"New {label}: {row['canonical_name']}" + (f" ({org})" if org else "") 499 summary = f"New {label}: {row['canonical_name']}" + (f" ({org})" if org else "")
343 effective = None 500 effective = None
344− rel = row["attributes"].get("release_date") or row["attributes"].get("published_at") 501+ rel = attrs.get("release_date") or attrs.get("published_at")
345 if isinstance(rel, str): 502 if isinstance(rel, str):
346 from aiatlas.sdk.extract.dates import parse_datetime 503 from aiatlas.sdk.extract.dates import parse_datetime
347 504
348 effective = parse_datetime(rel) 505 effective = parse_datetime(rel)
349 await self.emit_event(f"NEW_{etype.upper()}", EVENT_CATEGORY_BY_TYPE.get(etype, "update"), summary, entity_id=eid, importance=importance, 506 await self.emit_event(f"NEW_{etype.upper()}", EVENT_CATEGORY_BY_TYPE.get(etype, "update"), summary, entity_id=eid, importance=importance,
350− dedupe_key=f"NEW_{etype.upper()}:{eid}", effective_at=effective) 507+ dedupe_key=f"NEW_{etype.upper()}:{eid}", effective_at=effective, entity_first_seen=effective)
351 508
352 509
353 def _short(v: Any) -> str: 510 def _short(v: Any) -> str:
@@ -365,4 +522,4 @@ def _fmt_price(p: dict[str, Any]) -> str:
365 return " / ".join(parts) + " per 1M tokens" if parts else "n/a" 522 return " / ".join(parts) + " per 1M tokens" if parts else "n/a"
366 523
367 524
368−__all__ = ["FactWriter", "WriteStats"] 525+__all__ = ["FactWriter", "WriteStats", "result_dedupe_key"]
added src/aiatlas/services/anomalies.py +82 −0
@@ -0,0 +1,82 @@
1+"""Anomaly persistence: the ontology's deterministic checks (`aiatlas.ontology.anomalies`) produce flags; this module upserts them in the
2+`anomalies` table (one row per dedupe key, reopened when a check fires again, auto-resolved when it stops firing). Never deletes,
3+never edits the flagged value."""
4+from __future__ import annotations
5+
6+import logging
7+from typing import Any
8+
9+from sqlalchemy.ext.asyncio import AsyncConnection
10+from ulid import ULID
11+
12+from aiatlas.db import execute, fetch_all, fetch_one, jsonb
13+from aiatlas.ontology.anomalies import Anomaly, check_hardware, check_model, check_price, check_result
14+
15+log = logging.getLogger(__name__)
16+
17+
18+async 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 end
26+ 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 ""
30+
31+
32+async 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 whose
34+ 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 p
43+ 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_date
48+ from benchmark_results r join entities m on m.id = r.model_id join entities b on b.id = r.benchmark_id
49+ 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+ continue
58+ keys.add(key)
59+ await record(conn, a)
60+ by_sev[a.severity] = by_sev.get(a.severity, 0) + 1
61+ resolved = 0
62+ 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 0
67+ return {"flagged": len(keys), "by_severity": by_sev, "resolved": resolved}
68+
69+
70+async 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_id
72+ 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)
75+
76+
77+async 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)
80+
81+
82+__all__ = ["list_anomalies", "record", "run_checks", "set_status"]
added src/aiatlas/services/events.py +128 −0
@@ -0,0 +1,128 @@
1+"""Event semantics — deterministic, database-free rules shared by the writer and `aia canonicalize events`.
2+
3+Three clocks on every change event:
4+ occurred_at = coalesce(effective_at, observed_at) when the thing happened (release date, price effective date)
5+ observed_at when a connector saw it
6+ recorded_at when the row was written
7+
8+`is_backfill` separates *history being loaded* from *news*: the "+N in 24 h" counters, /changes and the pulse feed only count live
9+(non-backfill) events; timelines and /asof use every event.
10+"""
11+from __future__ import annotations
12+
13+from datetime import datetime, timedelta
14+from typing import Any
15+
16+BACKFILL_LAG_DAYS = 3
17+RELEASE_LIKE = {"RELEASE", "ANNOUNCEMENT", "NEW_MODEL", "VERSION_RELEASED"}
18+
19+
20+def classify_backfill(event_type: str, effective_at: datetime | None, observed_at: datetime, *, is_first_run: bool = False,
21+ entity_first_seen: datetime | None = None) -> bool:
22+ """True when the event describes something that happened well before we observed it.
23+
24+ * `effective_at` more than BACKFILL_LAG_DAYS before `observed_at` → backfill (an old release date being loaded);
25+ * the connector run is the connector's first successful run → backfill (initial corpus, not news);
26+ * NEW_* events for an entity whose first-seen hint (release/publication date) predates observation by more than the lag → backfill.
27+ """
28+ if is_first_run:
29+ return True
30+ lag = timedelta(days=BACKFILL_LAG_DAYS)
31+ if effective_at is not None and observed_at is not None and _aware(effective_at) < _aware(observed_at) - lag:
32+ return True
33+ if event_type.startswith("NEW_") and entity_first_seen is not None and observed_at is not None and _aware(entity_first_seen) < _aware(observed_at) - lag:
34+ return True
35+ return False
36+
37+
38+def group_key_for(event_type: str, entity_id: str | None, effective_at: datetime | None, observed_at: datetime | None) -> str | None:
39+ """One release seen in several documents (blog post, docs page, hub listing, provider listing) groups under one key per entity × month."""
40+ if event_type not in RELEASE_LIKE or not entity_id:
41+ return None
42+ when = effective_at or observed_at
43+ if when is None:
44+ return None
45+ return f"release:{entity_id}:{when.strftime('%Y-%m')}"
46+
47+
48+def _aware(dt: datetime) -> datetime:
49+ from datetime import UTC
50+
51+ return dt if dt.tzinfo is not None else dt.replace(tzinfo=UTC)
52+
53+
54+# ---------------------------------------------------------------------------------------------- importance (0 minor … 3 major)
55+def _num(v: Any) -> float | None:
56+ if isinstance(v, bool):
57+ return None
58+ if isinstance(v, (int, float)):
59+ return float(v)
60+ if isinstance(v, str):
61+ try:
62+ return float(v.replace(",", ""))
63+ except ValueError:
64+ return None
65+ return None
66+
67+
68+def _price_change_ratio(old: Any, new: Any) -> float:
69+ """Largest relative move among input/output per-million prices (0 when unknown)."""
70+ if not isinstance(old, dict) or not isinstance(new, dict):
71+ a, b = _num(old), _num(new)
72+ return abs(b - a) / a if a and b is not None and a > 0 else 0.0
73+ best = 0.0
74+ for k in ("input_per_mtok", "output_per_mtok", "cached_input_per_mtok"):
75+ a, b = _num(old.get(k)), _num(new.get(k))
76+ if a is not None and b is not None and a > 0:
77+ best = max(best, abs(b - a) / a)
78+ elif a in (None, 0) and b:
79+ best = max(best, 1.0) # newly priced / free → paid counts as a full move
80+ return best
81+
82+
83+def importance_for(event_type: str, entity_type: str | None, old: Any = None, new: Any = None, *, tier: int = 2, org_model_count: int = 0,
84+ openness: str | None = None, leader_change: bool = False, default: int = 2) -> int:
85+ """Deterministic importance:
86+ artifact events 0 · model_family events 1 · metadata corrections (PROPERTY_CHANGED) 0
87+ frontier-model release (organisation with ≥ 3 models, source tier ≤ 2) 3 · open-weight release 3 · other NEW_MODEL 2
88+ price change ≥ 50 % → 3, ≥ 20 % → 2, else 1 · context ≥ 5× → 3 (shrink → 1) · deprecation / retirement 3 · benchmark leader change 2
89+ Tier > 2 sources lose one point (never below 0)."""
90+ et = event_type.upper()
91+ if entity_type == "artifact":
92+ return 0
93+ if entity_type == "model_family":
94+ return min(1, default)
95+ if et == "PROPERTY_CHANGED":
96+ return 0
97+ if et == "PRICE_CHANGED":
98+ ratio = _price_change_ratio(old, new)
99+ imp = 3 if ratio >= 0.5 else 2 if ratio >= 0.2 else 1
100+ elif et == "CONTEXT_CHANGED":
101+ a, b = _num(old), _num(new)
102+ if a and b and a > 0:
103+ imp = 3 if b / a >= 5 else 1 if b < a else 2
104+ else:
105+ imp = 2
106+ elif et in ("DEPRECATION_ANNOUNCED", "RETIREMENT_ANNOUNCED"):
107+ imp = 3
108+ elif et == "STATUS_CHANGED":
109+ imp = 3 if str(new).lower() in ("deprecated", "retired", "discontinued") else 2
110+ elif et == "OPENNESS_CHANGED":
111+ imp = 3 if str(new).startswith("open") else 2
112+ elif et == "NEW_MODEL":
113+ if openness and str(openness).startswith("open"):
114+ imp = 3
115+ elif org_model_count >= 3 and tier <= 2:
116+ imp = 3
117+ else:
118+ imp = 2
119+ elif et in ("BENCHMARK_LEADER_CHANGED",) or (et.startswith("BENCHMARK") and leader_change):
120+ imp = 2
121+ else:
122+ imp = default
123+ if tier > 2:
124+ imp -= 1
125+ return max(0, min(3, imp))
126+
127+
128+__all__ = ["BACKFILL_LAG_DAYS", "RELEASE_LIKE", "classify_backfill", "group_key_for", "importance_for"]
modified src/aiatlas/services/handlers.py +2 −1
Large diff (9 lines) is not loaded by default to keep the page fast.Load diff
modified src/aiatlas/services/merge.py +182 −11
Large diff (231 lines) is not loaded by default to keep the page fast.Load diff
added tests/test_events.py +47 −0
Large diff (47 lines) is not loaded by default to keep the page fast.Load diff
added tests/test_fetch_ssrf.py +102 −0
Large diff (102 lines) is not loaded by default to keep the page fast.Load diff
added tests/test_resolver.py +123 −0
Large diff (123 lines) is not loaded by default to keep the page fast.Load diff
added tests/test_writer.py +266 −0
Large diff (267 lines) is not loaded by default to keep the page fast.Load diff
Menu

Theme
↑↓ navigate ↵ open esc close / or ⌘K anywhere