cfpe_verify.py

#!/usr/bin/env python3
"""
CFPE Automated Verification Pipeline v1.0
===========================================
Checks falsification register predictions against arXiv API.
Designed for quarterly scheduled execution by LLM agents.
Produces structured verification report with calibration deltas.

Usage:
    python cfpe_verify.py                          # Full scan of all candidates
    python cfpe_verify.py --id CFPE-E1-C1           # Single candidate check
    python cfpe_verify.py --era E1                  # All Era 1 candidates
    python cfpe_verify.py --output report.json      # Custom output path
    python cfpe_verify.py --schedule                # Quarterly schedule info
"""

import json
import sys
import os
import urllib.request
import urllib.parse
import xml.etree.ElementTree as ET
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any, Tuple
import time


# ─── Configuration ───────────────────────────────────────────────

ARXIV_API = "http://export.arxiv.org/api/query"
SEMANTIC_SCHOLAR_API = "https://api.semanticscholar.org/graph/v1/paper/search"
REQUEST_DELAY = 3.0  # seconds between API calls (arXiv rate limit: 1 req/3s)
MAX_RESULTS = 10
REGISTER_PATH = os.path.join(os.path.dirname(__file__), "phase-0-adversarial-testing", "falsification_register.json")
OUTPUT_PATH = os.path.join(os.path.dirname(__file__), "phase-0-adversarial-testing", "verification_report.json")


# ─── Candidate Query Definitions ────────────────────────────────

CANDIDATE_QUERIES = {
    "CFPE-E1-C1": {
        "name": "FCI braiding >1K coherence",
        "query": '"fractional Chern insulator" AND (coherence OR dephasing OR T2)',
        "max_results": 20,
        "falsification_criterion": "no paper reports T2* > 1ms at T > 1K",
        "check_deadline": "2028-12-31",
        "parser": "fci_coherence"
    },
    "CFPE-E1-C2": {
        "name": "Gate-based QC thermal wall at 10k qubits",
        "query": '(quantum processor OR quantum computer) AND (qubit count OR scaling) AND (roadmap OR milestone)',
        "max_results": 10,
        "falsification_criterion": "no processor exceeds 10k qubits with >99.9% fidelity by 2030",
        "check_deadline": "2030-12-31",
        "parser": "gate_scaling"
    },
    "CFPE-E2-C1": {
        "name": "Bio-spintronic √SWAP gate at 310K",
        "query": '("radical pair" OR "spin coherence") AND ("room temperature" OR 310K) AND (gate OR √SWAP OR controllable)',
        "max_results": 20,
        "falsification_criterion": "no room-temperature gate operation in bio-mimetic channel",
        "check_deadline": "2030-12-31",
        "parser": "bio_spintronic"
    },
    "CFPE-E3-C1": {
        "name": "Superdeterminism CMB anomalies",
        "query": 'superdeterminism AND CMB AND (anomaly OR anisotropy OR "power asymmetry")',
        "max_results": 15,
        "falsification_criterion": "no p < 0.01 CMB anomaly attribution to superdeterminism",
        "check_deadline": "2032-12-31",
        "parser": "superdeterminism_cmb"
    },
    "CFPE-E4-C1": {
        "name": "SM constants from π/φ geometry",
        "query": '("fine structure constant" OR "coupling constant") AND ("golden ratio" OR phi OR "geometric derivation")',
        "max_results": 10,
        "falsification_criterion": "no geometric derivation achieves <1% precision",
        "check_deadline": "2035-12-31",
        "parser": "geometric_constants"
    }
}

# ─── Paper Classification Rules ─────────────────────────────────

def classify_paper(entry: Dict, candidate_id: str) -> Tuple[str, float]:
    """
    Classify a paper as SUPPORTING, WEAKENING, or NEUTRAL relative to a CFPE candidate.
    Returns (classification, confidence 0-1).
    """
    title = entry.get("title", "").lower()
    abstract = entry.get("abstract", "").lower()
    published = entry.get("published", "")
    
    if candidate_id == "CFPE-E1-C1":
        # FCI coherence at elevated temperatures
        if any(kw in title + abstract for kw in ["temperature", "coherence", "dephasing"]):
            if any(kw in title for kw in ["superconductivity", "superconductor"]):
                return ("NEUTRAL", 0.8)  # superconductivity in Chern bands, not qubit coherence
            if "kelvin" in abstract or "temperature" in abstract:
                # Paper mentions temperature and coherence together
                return ("WEAKENING", 0.4)  # potentially relevant but need manual review
            return ("NEUTRAL", 0.6)
        return ("NEUTRAL", 0.9)  # irrelevant to FCI coherence
    
    elif candidate_id == "CFPE-E1-C2":
        # Gate-based scaling
        if "qubit" in title and any(kw in title for kw in ["thousand", "k qubit", "scale", "roadmap"]):
            return ("WEAKENING", 0.5)  # suggests gate-based QC is still scaling
        return ("NEUTRAL", 0.7)
    
    elif candidate_id == "CFPE-E2-C1":
        # Bio-spintronic gates
        if "radical pair" in title or "radical pair" in abstract:
            if "gate" in title or "gate" in abstract or "controll" in abstract:
                return ("WEAKENING", 0.6)  # bio-spintronic gate demonstrated
            return ("NEUTRAL", 0.7)  # radical pair studied, no gate
        return ("NEUTRAL", 0.8)
    
    elif candidate_id == "CFPE-E3-C1":
        # Superdeterminism CMB
        if "superdeterminism" in title:
            if "cmb" in abstract or "cosmic" in abstract:
                return ("WEAKENING", 0.7)  # superdeterminism CMB analysis exists
            return ("NEUTRAL", 0.6)
        return ("NEUTRAL", 0.9)
    
    elif candidate_id == "CFPE-E4-C1":
        # Geometric constants
        if "fine structure" in title and any(kw in title for kw in ["geometric", "golden", "phi", "cross"]):
            if "derive" in abstract or "derivation" in abstract or "=" in abstract:
                return ("WEAKENING", 0.5)  # claimed geometric derivation
            return ("NEUTRAL", 0.6)
        return ("NEUTRAL", 0.9)
    
    return ("NEUTRAL", 0.5)


# ─── arXiv API Client ────────────────────────────────────────────

def search_arxiv(query: str, max_results: int = 10) -> List[Dict]:
    """
    Query arXiv API and return parsed paper entries.
    Returns list of dicts with: title, abstract, authors, published, arxiv_id, url.
    """
    params = urllib.parse.urlencode({
        "search_query": query,
        "start": 0,
        "max_results": max_results,
        "sortBy": "submittedDate",
        "sortOrder": "descending"
    })
    
    url = f"{ARXIV_API}?{params}"
    
    try:
        req = urllib.request.Request(url, headers={"User-Agent": "CFPE-Verification-Pipeline/1.0"})
        with urllib.request.urlopen(req, timeout=30) as response:
            data = response.read().decode("utf-8")
    except Exception as e:
        print(f"  [ERROR] arXiv API unreachable: {e}", file=sys.stderr)
        return []
    
    # Parse Atom XML
    ns = {
        "atom": "http://www.w3.org/2005/Atom",
        "arxiv": "http://arxiv.org/schemas/atom"
    }
    
    try:
        root = ET.fromstring(data)
    except ET.ParseError as e:
        print(f"  [ERROR] XML parse failed: {e}", file=sys.stderr)
        return []
    
    papers = []
    for entry_elem in root.findall("atom:entry", ns):
        title_elem = entry_elem.find("atom:title", ns)
        abstract_elem = entry_elem.find("atom:summary", ns)
        
        paper = {
            "title": title_elem.text.strip().replace("\n", " ") if title_elem is not None else "Unknown",
            "abstract": abstract_elem.text.strip().replace("\n", " ") if abstract_elem is not None else "",
            "published": entry_elem.find("atom:published", ns).text if entry_elem.find("atom:published", ns) is not None else "",
            "arxiv_id": extract_id(entry_elem.find("atom:id", ns)),
            "url": extract_url(entry_elem, ns),
            "authors": [author.find("atom:name", ns).text for author in entry_elem.findall("atom:author", ns)]
        }
        papers.append(paper)
    
    return papers


def extract_id(id_elem) -> str:
    """Extract arXiv ID from the Atom ID element."""
    if id_elem is None:
        return "unknown"
    text = id_elem.text
    if "abs/" in text:
        return text.split("abs/")[-1]
    return text.split("/")[-1]


def extract_url(entry_elem, ns) -> str:
    """Extract PDF URL from entry."""
    for link in entry_elem.findall("atom:link", ns):
        href = link.attrib.get("href", "")
        if link.attrib.get("title") == "pdf" or "pdf" in href:
            return href
    # Fallback: construct from ID
    id_url = entry_elem.find("atom:id", ns)
    if id_url is not None:
        return id_url.text.replace("abs", "pdf")
    return ""


# ─── Verification Report Generation ─────────────────────────────

def verify_candidate(candidate_id: str) -> Dict:
    """
    Run full verification for a single candidate.
    Returns structured verification result.
    """
    if candidate_id not in CANDIDATE_QUERIES:
        return {"error": f"Unknown candidate ID: {candidate_id}"}
    
    config = CANDIDATE_QUERIES[candidate_id]
    
    print(f"\n{'='*70}")
    print(f"  Checking: {candidate_id} — {config['name']}")
    print(f"  Query: {config['query']}")
    print(f"  Deadline: {config['check_deadline']}")
    print(f"{'='*70}")
    
    deadline_passed = datetime.strptime(config["check_deadline"], "%Y-%m-%d") < datetime.now()
    
    # Query arXiv
    print(f"  Querying arXiv ({config['max_results']} results)...")
    papers = search_arxiv(config["query"], config["max_results"])
    print(f"  Retrieved {len(papers)} papers")
    
    # Classify papers
    results = []
    supporting_count = 0
    weakening_count = 0
    neutral_count = 0
    
    for paper in papers:
        classification, confidence = classify_paper(paper, candidate_id)
        results.append({
            "title": paper["title"],
            "arxiv_id": paper["arxiv_id"],
            "published": paper["published"],
            "classification": classification,
            "confidence": confidence,
            "url": paper["url"]
        })
        
        if classification == "SUPPORTING":
            supporting_count += 1
        elif classification == "WEAKENING":
            weakening_count += 1
        else:
            neutral_count += 1
    
    # Determine verification status
    if deadline_passed and weakening_count == 0 and supporting_count == 0:
        status = "FALSIFIED"
        status_detail = "Deadline passed with no supporting evidence — prediction is refuted"
    elif weakening_count > 0:
        status = "WEAKENED"
        status_detail = f"{weakening_count} paper(s) potentially weaken this prediction"
    elif supporting_count > 0:
        status = "CONFIRMED"
        status_detail = f"{supporting_count} paper(s) support this prediction"
    elif deadline_passed:
        status = "INCONCLUSIVE"
        status_detail = "Deadline passed but insufficient evidence to classify"
    else:
        status = "HOLDS"
        status_detail = "No evidence against prediction and deadline not yet reached"
    
    verification_result = {
        "candidate_id": candidate_id,
        "candidate_name": config["name"],
        "check_date": datetime.now().strftime("%Y-%m-%d"),
        "check_deadline": config["check_deadline"],
        "deadline_passed": deadline_passed,
        "papers_retrieved": len(papers),
        "papers_supporting": supporting_count,
        "papers_weakening": weakening_count,
        "papers_neutral": neutral_count,
        "status": status,
        "status_detail": status_detail,
        "falsification_criterion": config["falsification_criterion"],
        "top_papers": sorted(
            [r for r in results if r["classification"] != "NEUTRAL"],
            key=lambda x: x["confidence"],
            reverse=True
        )[:5],
        "all_classifications": results
    }
    
    print(f"\n  Status: {status}")
    print(f"  {status_detail}")
    print(f"  Supporting: {supporting_count} | Weakening: {weakening_count} | Neutral: {neutral_count}")
    
    return verification_result


def run_full_verification(candidate_ids: Optional[List[str]] = None) -> Dict:
    """Run verification for specified candidates or all registered."""
    if candidate_ids is None:
        candidate_ids = list(CANDIDATE_QUERIES.keys())
    
    results = {}
    
    for candidate_id in candidate_ids:
        results[candidate_id] = verify_candidate(candidate_id)
        if len(candidate_ids) > 1:
            print(f"\n  Sleeping {REQUEST_DELAY}s for rate limit...")
            time.sleep(REQUEST_DELAY)
    
    # Summary statistics
    status_counts = {"HOLDS": 0, "WEAKENED": 0, "FALSIFIED": 0, "CONFIRMED": 0, "INCONCLUSIVE": 0}
    for r in results.values():
        status_counts[r["status"]] = status_counts.get(r["status"], 0) + 1
    
    report = {
        "pipeline_version": "1.0",
        "run_date": datetime.now().strftime("%Y-%m-%dT%H:%M:%S"),
        "summary": {
            "total_candidates": len(results),
            "status_counts": status_counts,
            "next_scheduled_check": (datetime.now() + timedelta(days=90)).strftime("%Y-%m-%d")
        },
        "results": results
    }
    
    return report


# ─── Output & Scheduling ────────────────────────────────────────

def save_report(report: Dict, output_path: str):
    """Save verification report to JSON file."""
    os.makedirs(os.path.dirname(output_path), exist_ok=True)
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(report, f, indent=2, ensure_ascii=False)
    print(f"\n  Report saved to: {output_path}")


def update_falsification_register(report: Dict, register_path: str):
    """Update the falsification register JSON with latest check results."""
    if not os.path.exists(register_path):
        print(f"  [WARN] Register not found at {register_path} — skipping update", file=sys.stderr)
        return
    
    with open(register_path, "r", encoding="utf-8") as f:
        register = json.load(f)
    
    for candidate in register.get("candidates", []):
        cid = candidate["id"]
        if cid in report["results"]:
            r = report["results"][cid]
            candidate["current_status"] = "checked"
            candidate["last_check_date"] = r["check_date"]
            candidate["check_result"] = r["status_detail"]
            candidate["auto_status"] = r["status"]
    
    register["summary_statistics"]["last_full_verification"] = report["run_date"]
    
    with open(register_path, "w", encoding="utf-8") as f:
        json.dump(register, f, indent=2, ensure_ascii=False)
    print(f"  Falsification register updated at: {register_path}")


def print_schedule():
    """Print quarterly schedule information."""
    now = datetime.now()
    next_checks = []
    for i in range(4):
        check_date = now + timedelta(days=90 * (i + 1))
        next_checks.append(check_date.strftime("%Y-%m-%d"))
    
    print("""
    CFPE Verification Pipeline — Quarterly Schedule
    ================================================
    
    Candidate deadlines:
      CFPE-E1-C1  — 2028-12-31 (FCI coherence)
      CFPE-E1-C2  — 2030-12-31 (Gate-based thermal wall)
      CFPE-E2-C1  — 2030-12-31 (Bio-spintronic gate)
      CFPE-E3-C1  — 2032-12-31 (Superdeterminism CMB)
      CFPE-E4-C1  — 2035-12-31 (Geometric constants)
    
    Next scheduled checks:
      Q1: {q1}
      Q2: {q2}
      Q3: {q3}
      Q4: {q4}
    
    Run command: python cfpe_verify.py
    Auto-update register: python cfpe_verify.py --update-register
    
    All deadlines are >1 year away — current checks serve as
    early-warning scans for preliminary evidence.
    """.format(q1=next_checks[0], q2=next_checks[1], q3=next_checks[2], q4=next_checks[3]))


# ─── CLI Entry Point ────────────────────────────────────────────

def main():
    import argparse
    
    parser = argparse.ArgumentParser(
        description="CFPE Automated Verification Pipeline — arXiv API checking"
    )
    parser.add_argument("--id", help="Check single candidate by ID (e.g., CFPE-E1-C1)")
    parser.add_argument("--era", help="Check all candidates in era (E1, E2, E3, E4)")
    parser.add_argument("--output", help="Custom output path for report")
    parser.add_argument("--update-register", action="store_true", help="Update falsification register with results")
    parser.add_argument("--schedule", action="store_true", help="Show quarterly schedule")
    parser.add_argument("--register-path", default=REGISTER_PATH, help="Path to falsification register JSON")
    
    args = parser.parse_args()
    
    if args.schedule:
        print_schedule()
        return
    
    # Determine which candidates to check
    candidates_to_check = []
    if args.id:
        candidates_to_check = [args.id]
    elif args.era:
        era_prefix = f"CFPE-{args.era.upper()}-"
        candidates_to_check = [cid for cid in CANDIDATE_QUERIES if cid.startswith(era_prefix)]
        if not candidates_to_check:
            print(f"No candidates found for era {args.era}", file=sys.stderr)
            sys.exit(1)
    else:
        candidates_to_check = list(CANDIDATE_QUERIES.keys())
    
    # Run verification
    print(f"\n  CFPE Verification Pipeline v1.0")
    print(f"  Checking {len(candidates_to_check)} candidate(s)...")
    print(f"  Delay: {REQUEST_DELAY}s between queries (arXiv rate limit)")
    
    report = run_full_verification(candidates_to_check)
    
    output_path = args.output or OUTPUT_PATH
    save_report(report, output_path)
    
    if args.update_register:
        update_falsification_register(report, args.register_path)
    
    # Print summary
    print(f"\n{'='*70}")
    print(f"  VERIFICATION SUMMARY")
    print(f"{'='*70}")
    for cid, result in report["results"].items():
        print(f"  {cid}: {result['status']:12s} | {result['status_detail']}")
    print(f"\n  Next check: {report['summary']['next_scheduled_check']}")
    
    return report


if __name__ == "__main__":
    main()