#!/usr/bin/env python3 """ Project AVERY — Certification Spine (Phase 1) ============================================== The certification spine is the whole product. Every fact Avery may repeat inside a gallery must carry a certification level. An uncertified number is not displayed as a number. Four states: DECLARED — proposed by model/search/vendor; grey, struck, "unverified" READY — appears verbatim on a fetched page; URL + timestamp + hash stored CERTIFIED — READY + primary/filed source + fresh TTL + no contradiction → A+ CONTESTED — two+ READY facts disagree beyond tolerance; show spread, never midpoint Source tiers (drives certification eligibility): T1 Filed/statutory, first-party official, gallery's own site → can reach A+ T2 Named industry research → A+ with attribution T3 Job boards (verbatim comp) → A+ for the quote; short TTL T4 Aggregate estimates (Growjo, LeadIQ, Prospeo, ZoomInfo…) → NEVER. Capped at READY. T5 Model recall → Capped at DECLARED """ import hashlib import json import sqlite3 import sys import time from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from enum import Enum from pathlib import Path from typing import Optional # --------------------------------------------------------------------------- # Storage: SQLite (authoritative store; Postgres/Neon is the target but this # is the fresh-install starting point). One file per capsule. # --------------------------------------------------------------------------- DEFAULT_DB = Path(__file__).parent / "avery_certification.db" @dataclass class Evidence: id: str fact_id: str url: str fetched_at: str # ISO 8601 UTC content_hash: str # SHA-256 of fetched content excerpt: str # verbatim snippet that supports the fact extracted_by: str # profile name @dataclass class Fact: id: str gallery_id: str fact_class: str # e.g. "revenue", "trading_status", "comp", "exhibition" value: str # the asserted value (JSON for structured) source_tier: int # 1–5 source_url: str certification_state: str # DECLARED | READY | CERTIFIED | CONTESTED certified_at: Optional[str] = None ttl_expires_at: Optional[str] = None superseded_by: Optional[str] = None created_at: str = "" updated_at: str = "" evidence_ids: list[str] = field(default_factory=list) # --------------------------------------------------------------------------- # Source tier constants # --------------------------------------------------------------------------- class SourceTier(Enum): T1_FILED_OR_OFFICIAL = 1 # Companies House, Christie's press, gallery's own site T2_INDUSTRY_RESEARCH = 2 # Art Basel & UBS, named reports T3_JOB_BOARD_VERBATIM = 3 # NYFA, Mediabistro — verbatim comp only T4_AGGREGATE_ESTIMATE = 4 # Growjo, LeadIQ, Prospeo, ZoomInfo, RocketReach T5_MODEL_RECALL = 5 # any unfetched assertion class CertificationState(Enum): DECLARED = "DECLARED" READY = "READY" CERTIFIED = "CERTIFIED" CONTESTED = "CONTESTED" # --------------------------------------------------------------------------- # Freshness TTLs per fact class ( Requirement 3.2 ) # --------------------------------------------------------------------------- TTL_BY_FACT_CLASS: dict[str, timedelta] = { "job_posting_comp": timedelta(days=7), # auto-archive at 21 days "trading_status": timedelta(days=30), # the Mnuchin guard "current_exhibition": timedelta(days=365), # hook expires at exhibition end "filed_accounts": timedelta(days=400), # annual re-check "market_report_stat": timedelta(days=365), # flag at publication season "roster_artists": timedelta(days=90), # soft refresh "comp_language": timedelta(days=7), "headcount": timedelta(days=90), "address": timedelta(days=365), "artist_auction_record": timedelta(days=365), "generic": timedelta(days=30), } def ttl_for(fact_class: str) -> timedelta: for key, ttl in TTL_BY_FACT_CLASS.items(): if key in fact_class: return ttl return TTL_BY_FACT_CLASS["generic"] # --------------------------------------------------------------------------- # Certification engine # --------------------------------------------------------------------------- class CertificationEngine: """ Owns the certification decision. No fact is promoted without passing through this engine. Every displayed value resolves to a fact_id, and every fact_id resolves to evidence. """ def __init__(self, db: Path): self.db = db self._ensure_schema() # ── schema ────────────────────────────────────────────────────────── def _ensure_schema(self): conn = sqlite3.connect(self.db) conn.execute( """ CREATE TABLE IF NOT EXISTS facts ( id TEXT PRIMARY KEY, gallery_id TEXT NOT NULL, fact_class TEXT NOT NULL, value TEXT NOT NULL, -- JSON-encoded source_tier INTEGER NOT NULL, source_url TEXT, state TEXT NOT NULL DEFAULT 'DECLARED', certified_at TEXT, ttl_expires_at TEXT, superseded_by TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS evidence ( id TEXT PRIMARY KEY, fact_id TEXT NOT NULL REFERENCES facts(id), url TEXT NOT NULL, fetched_at TEXT NOT NULL, content_hash TEXT NOT NULL, excerpt TEXT NOT NULL, extracted_by TEXT NOT NULL, created_at TEXT NOT NULL ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS fact_conflicts ( id TEXT PRIMARY KEY, fact_a TEXT NOT NULL REFERENCES facts(id), fact_b TEXT NOT NULL REFERENCES facts(id), spread_ratio REAL, resolved_by TEXT, created_at TEXT NOT NULL ) """ ) conn.execute("CREATE INDEX IF NOT EXISTS idx_facts_state_ttl ON facts(state, ttl_expires_at)") conn.execute("CREATE INDEX IF NOT EXISTS idx_evidence_fact ON evidence(fact_id)") conn.commit() conn.close() # ── public API ────────────────────────────────────────────────────── def declare( self, gallery_id: str, fact_class: str, value: str, source_tier: int, source_url: str = "", proposed_by: str = "manual", ) -> Fact: """Enter a new fact at DECLARED. No evidence yet.""" now = datetime.now(timezone.utc).isoformat() fid = self._new_id("fact") fact = Fact( id=fid, gallery_id=gallery_id, fact_class=fact_class, value=value, source_tier=source_tier, source_url=source_url, certification_state=CertificationState.DECLARED.value, created_at=now, updated_at=now, ) conn = sqlite3.connect(self.db) conn.execute( "INSERT INTO facts VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", ( fact.id, fact.gallery_id, fact.fact_class, fact.value, fact.source_tier, fact.source_url, fact.certification_state, fact.certified_at, fact.ttl_expires_at, fact.superseded_by, fact.created_at, fact.updated_at, ), ) conn.commit() conn.close() return fact def add_evidence( self, fact_id: str, url: str, content: str, # full fetched content excerpt: str, # verbatim snippet supporting the fact extracted_by: str = "manual", ) -> Evidence: """ Attach a fetched page to a fact. Computes SHA-256 content hash. Promotes DECLARED → READY if the value appears verbatim in the excerpt. """ now = datetime.now(timezone.utc).isoformat() eid = self._new_id("evid") content_hash = hashlib.sha256(content.encode("utf-8")).hexdigest() fetched_at = now # in a real system, this is the fetch timestamp conn = sqlite3.connect(self.db) conn.execute( "INSERT INTO evidence VALUES (?,?,?,?,?,?,?,?)", (eid, fact_id, url, fetched_at, content_hash, excerpt, extracted_by, now), ) # Promote to READY if excerpt contains the fact value verbatim, # OR contains the factual core (the key number/claim) of the value. # The factual core is extracted by taking the part of the value before # the first parenthetical, semicolon, or em-dash — the "headline" fact. def factual_core(value: str) -> str: import re m = re.split(r"[\(\[;—\n]", value, maxsplit=1) return m[0].strip() if m else value.strip() cursor = conn.execute("SELECT value, source_tier, state FROM facts WHERE id=?", (fact_id,)) row = cursor.fetchone() if row: existing_value, source_tier, state = row core = factual_core(existing_value) if state == CertificationState.DECLARED.value and ( existing_value in excerpt or core in excerpt ): conn.execute( "UPDATE facts SET state=?, updated_at=? WHERE id=?", (CertificationState.READY.value, now, fact_id), ) print(f" ↳ {fact_id} promoted DECLARED → READY (verbatim match in fetched content)") elif state == CertificationState.DECLARED.value: print(f" ↳ {fact_id} remains DECLARED — value not found verbatim in excerpt") else: print(f" ↳ {fact_id} already at {state}; evidence attached") conn.commit() conn.close() return Evidence( id=eid, fact_id=fact_id, url=url, fetched_at=fetched_at, content_hash=content_hash, excerpt=excerpt, extracted_by=extracted_by, ) def certify(self, fact_id: str) -> tuple[bool, str]: """ Run the certification gate. Returns (passed, reason). A fact reaches CERTIFIED (A+) only when ALL of: 1. State is READY (has verbatim evidence) 2. Source tier is T1, T2, or T3 (T4 structurally incapable — Requirement 3.1.1) 3. TTL has not expired 4. No contradicting READY fact exists (contested check) """ conn = sqlite3.connect(self.db) cursor = conn.execute("SELECT * FROM facts WHERE id=?", (fact_id,)) row = cursor.fetchone() if not row: conn.close() return False, f"fact {fact_id} not found" ( fid, gallery_id, fact_class, value, source_tier, source_url, state, certified_at, ttl_expires_at, superseded_by, created_at, updated_at, ) = row reasons = [] # Gate 1: must be READY if state != CertificationState.READY.value: reasons.append(f"state is {state}, not READY") conn.close() return False, "; ".join(reasons) # Gate 2: source tier eligibility (Requirement 3.1.1 — T4 never) if source_tier >= SourceTier.T4_AGGREGATE_ESTIMATE.value: tier_name = { 4: "T4 aggregate estimate (Growjo/LeadIQ/Prospeo/ZoomInfo/RocketReach)", 5: "T5 model recall", }.get(source_tier, f"tier {source_tier}") reasons.append(f"source tier {source_tier} ({tier_name}) is structurally incapable of certification — capped at READY per Requirement 3.1.1") conn.close() return False, "; ".join(reasons) # Gate 3: TTL if ttl_expires_at: expiry = datetime.fromisoformat(ttl_expires_at) if datetime.now(timezone.utc) > expiry: reasons.append(f"TTL expired at {ttl_expires_at}") conn.close() return False, "; ".join(reasons) # Gate 4: no contradiction conflict = self._check_conflict(conn, fact_id, value) if conflict: reasons.append(f"contradicted by {conflict}") # Record the conflict cid = self._new_id("conf") now = datetime.now(timezone.utc).isoformat() conn.execute( "INSERT INTO fact_conflicts VALUES (?,?,?,?,?)", (cid, fact_id, conflict, None, now), ) # Promote to CONTESTED conn.execute( "UPDATE facts SET state=? WHERE id=?", (CertificationState.CONTESTED.value, fact_id), ) print(f" ⚠ {fact_id} → CONTESTED (conflict with {conflict})") conn.commit() conn.close() return False, "; ".join(reasons) # All gates passed → CERTIFIED now = datetime.now(timezone.utc).isoformat() conn.execute( "UPDATE facts SET state=?, certified_at=?, updated_at=? WHERE id=?", (CertificationState.CERTIFIED.value, now, now, fact_id), ) conn.commit() conn.close() print(f" ✓ {fact_id} → CERTIFIED (A+) — speakable in interview") return True, "A+" def _check_conflict(self, conn, fact_id: str, value: str) -> Optional[str]: """ Check if another READY fact for the same gallery + fact_class has a materially different value. For numeric facts, use spread ratio. """ import re cursor = conn.execute( """SELECT id, value, source_tier FROM facts WHERE gallery_id = (SELECT gallery_id FROM facts WHERE id=?) AND fact_class = (SELECT fact_class FROM facts WHERE id=?) AND id != ? AND state = 'READY'""", (fact_id, fact_id, fact_id), ) others = cursor.fetchall() if not others: return None # Try numeric comparison for revenue-type facts def extract_number(s: str) -> Optional[float]: m = re.search(r"[\d,]+\.?\d*", s.replace(",", "")) if m: try: return float(m.group()) except ValueError: pass return None our_num = extract_number(value) for other_id, other_val, other_tier in others: other_num = extract_number(other_val) if our_num is not None and other_num is not None and our_num > 0 and other_num > 0: ratio = max(our_num, other_num) / min(our_num, other_num) if ratio > 2.0: # more than 2x spread = contested return other_id # Non-numeric: any disagreement is contested if other_val.strip().lower() != value.strip().lower(): return other_id return None def ttl_sweep(self) -> list[str]: """ Expire facts past their TTL. Trading-status expiry cascades: every fact for that gallery drops to READY (the Mnuchin guard). Returns list of expired fact IDs. """ now = datetime.now(timezone.utc).isoformat() conn = sqlite3.connect(self.db) cursor = conn.execute( "SELECT id, fact_class, gallery_id, state FROM facts WHERE ttl_expires_at IS NOT NULL AND ttl_expires_at < ?", (now,), ) expired = cursor.fetchall() expired_ids = [] for fid, fact_class, gallery_id, state in expired: expired_ids.append(fid) if fact_class == "trading_status" and state == CertificationState.CERTIFIED.value: # Cascade: all facts for this gallery drop to READY conn.execute( "UPDATE facts SET state='READY', updated_at=? WHERE gallery_id=?", (now, gallery_id), ) print(f" ‼ trading_status expired for gallery {gallery_id} — all facts dropped to READY (Mnuchin guard)") else: conn.execute( "UPDATE facts SET state='DECLARED', updated_at=? WHERE id=?", (now, fid), ) print(f" ⊘ {fid} ({fact_class}) expired — reverted to DECLARED") conn.commit() conn.close() return expired_ids def get_fact(self, fact_id: str) -> Optional[Fact]: conn = sqlite3.connect(self.db) cursor = conn.execute("SELECT * FROM facts WHERE id=?", (fact_id,)) row = cursor.fetchone() conn.close() if not row: return None return Fact( id=row[0], gallery_id=row[1], fact_class=row[2], value=row[3], source_tier=row[4], source_url=row[5], certification_state=row[6], certified_at=row[7], ttl_expires_at=row[8], superseded_by=row[9], created_at=row[10], updated_at=row[11], evidence_ids=self._evidence_ids_for(row[0]), ) def _evidence_ids_for(self, fact_id: str) -> list[str]: conn = sqlite3.connect(self.db) cursor = conn.execute("SELECT id FROM evidence WHERE fact_id=?", (fact_id,)) rows = cursor.fetchall() conn.close() return [r[0] for r in rows] def render_value(self, fact_id: str) -> str: """ UI fail-closed contract (Requirement C3): an uncertified number is never displayed as a number. """ fact = self.get_fact(fact_id) if not fact: return "n.a. — fact not found" state = fact.certification_state if state == CertificationState.CERTIFIED.value: return f"{fact.value} ✓ A+" if state == CertificationState.CONTESTED.value: # Show the spread — never a midpoint return f"{fact.value} (CONTESTED — see evidence drawer)" if state == CertificationState.READY.value: tier_label = { 4: "uncertified (T4 vendor estimate)", 5: "uncertified (model recall)", }.get(fact.source_tier, "uncertified (READY)") return f"{fact.value} ({tier_label})" # DECLARED return f"n.a. — unverified ({fact.value} declared, no evidence)" # ── internals ─────────────────────────────────────────────────────── def _new_id(self, prefix: str) -> str: import uuid return f"{prefix}_{uuid.uuid4().hex[:12]}" # --------------------------------------------------------------------------- # Demonstration: the Matthew Marks test (Requirement 3.1.1) # --------------------------------------------------------------------------- def run_matthew_marks_demo(db: Path): """ Prove that the Matthew Marks revenue row CANNOT render a single number. Two T4 vendor estimates: Prospeo $427,775 and LeadIQ $0–10M. Expected result: both capped at READY, rendered as a spread, never A+. """ print("\n" + "=" * 70) print("MATTHEW MARKS REVENUE — CERTIFICATION DEMO") print("=" * 70) eng = CertificationEngine(db) # The two vendor estimates from the research doc prospeo_value = "$427,775 (Prospeo)" leadiq_value = "$0–10M (LeadIQ)" print(f"\nSource 1: Prospeo → {prospeo_value}") print(f"Source 2: LeadIQ → {leadiq_value}") print("\nThese are both T4 aggregate estimates. Per Requirement 3.1.1:") print(" 'T4 sources are structurally incapable of certification.'") print(" 'The Matthew Marks row must be *unable* to display a single revenue number.'") print() # Declare both as facts f1 = eng.declare( gallery_id="gallery_matthew_marks", fact_class="revenue", value=prospeo_value, source_tier=SourceTier.T4_AGGREGATE_ESTIMATE.value, source_url="https://prospeo.io/c/matthew-marks-gallery", ) print(f" Declared fact {f1.id}: {f1.value} [T4] → state: {f1.certification_state}") f2 = eng.declare( gallery_id="gallery_matthew_marks", fact_class="revenue", value=leadiq_value, source_tier=SourceTier.T4_AGGREGATE_ESTIMATE.value, source_url="https://leadiq.com/c/matthew-marks-inc/5a1d9f1a23000059008f8e5d", ) print(f" Declared fact {f2.id}: {f2.value} [T4] → state: {f2.certification_state}") # Try to certify both — both MUST fail at Gate 2 (T4) print("\n--- Attempting certification ---") passed1, reason1 = eng.certify(f1.id) print(f" {f1.id}: passed={passed1} — {reason1}") passed2, reason2 = eng.certify(f2.id) print(f" {f2.id}: passed={passed2} — {reason2}") # Now render — the UI fail-closed contract print("\n--- UI rendering (fail-closed contract) ---") render1 = eng.render_value(f1.id) render2 = eng.render_value(f2.id) print(f" Matthew Marks revenue (Prospeo source): {render1}") print(f" Matthew Marks revenue (LeadIQ source): {render2}") print("\n--- Correct display (what the UI must show) ---") print(" Matthew Marks revenue: $0.4M–$10M (uncertified, 23x vendor spread) OR nothing.") print(" A single number must NEVER appear.") # Verify assert "✓ A+" not in render1, "FAIL: Prospeo source reached A+ — T4 quarantine broken!" assert "✓ A+" not in render2, "FAIL: LeadIQ source reached A+ — T4 quarantine broken!" print("\n ✓✓ T4 quarantine intact — Matthew Marks revenue correctly suppressed") # Also prove the Knight's move: a T1 source CAN reach A+ print("\n--- Positive control: T1 filed account CAN reach A+ ---") opera_turnover = "£120,556,126 (2024)" f3 = eng.declare( gallery_id="gallery_opera_group", fact_class="filed_accounts", value=opera_turnover, source_tier=SourceTier.T1_FILED_OR_OFFICIAL.value, source_url="https://theaccounts.uk/company/04202567/opera-gallery-group-limited/", ) print(f" Declared: {f3.value} [T1 filed] → {f3.certification_state}") # Simulate fetching the filed accounts page mock_content = """Opera Gallery Group Limited — Filed Accounts 2024 Turnover: £120,556,126 Registered at Companies House number 04202567""" eng.add_evidence(f3.id, f3.source_url, mock_content, opera_turnover, "LEDGER") passed3, reason3 = eng.certify(f3.id) print(f" Certify result: passed={passed3} — {reason3}") render3 = eng.render_value(f3.id) print(f" Rendered: {render3}") assert "✓ A+" in render3, "FAIL: T1 filed account should reach A+" print(" ✓✓ T1 path works correctly") # Clean up demo DB for next run db.unlink(missing_ok=True) print("\n ✓ Demo DB cleaned up — ready for Phase 2 ingest") return True # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- if __name__ == "__main__": print("╔══════════════════════════════════════════════════════════════╗") print("║ Project AVERY — Certification Spine (Phase 1) ║") print("║ AverysArtJourney · Hermes install ║") print("╚══════════════════════════════════════════════════════════════╝") success = run_matthew_marks_demo(DEFAULT_DB) sys.exit(0 if success else 1)