""" HF Space entry point — FastAPI app serving the Différance Engine. Copy of site/app.py but with paths adjusted for the Space root layout. The Space root is under-erasure/ with: - pipeline/ Python pipeline code - data/ SQLite database + benchmarks - site/dist/ Astro-built static files - site/src/ Astro source (not needed at runtime) """ from __future__ import annotations import json import os import sys from pathlib import Path # Ensure pipeline is importable from Space root _space_root = Path(__file__).resolve().parent if str(_space_root) not in sys.path: sys.path.insert(0, str(_space_root)) from fastapi import FastAPI, HTTPException, Query from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse from fastapi.staticfiles import StaticFiles app = FastAPI(title="Différance Engine", version="0.2.0") DIST_DIR = _space_root / "site" / "dist" # Serve Astro static assets (CSS, JS, fonts) if (DIST_DIR / "_astro").exists(): app.mount("/_astro", StaticFiles(directory=str(DIST_DIR / "_astro")), name="astro_assets") @app.get("/") async def index(): """Dynamic index — queries live DB for stats and papers at render time.""" try: from pipeline.db import get_db db = get_db() db.connect() stats = db.stats() papers = db.get_displayable_papers(limit=30) db.close() return HTMLResponse(_render_index_html(stats, papers)) except Exception: return HTMLResponse(_fallback_html()) @app.get("/discourse") async def discourse_page(): """Translation trace: papers sorted by reduction rate, descending and ascending. NOT a quality leaderboard. Translation efficiency against a declared taxonomy is descriptive, not evaluative. High rate = vocabulary overlap. Low rate = where to look for genuine novelty OR taxonomy gaps OR bullshit. Reader decides. """ try: from pipeline.db import get_db db = get_db(); db.connect() papers = db.get_displayable_papers(limit=50) # Enrich with per-paper reduction rate enriched = [] for p in papers: reds = db.get_reductions_for_paper(p["arxiv_id"]) total = len(reds) mapped = sum(1 for r in reds if r["result_type"] in ("identity", "compositional")) rate = mapped / total if total > 0 else 0 unknown = total - mapped p["_reduction_rate"] = rate p["_mapped"] = mapped p["_unknown"] = unknown p["_total"] = total enriched.append(p) db.close() # Sort descending (most translated) and ascending (least translated) desc = sorted(enriched, key=lambda p: p["_reduction_rate"], reverse=True) asc = sorted(enriched, key=lambda p: p["_reduction_rate"]) return HTMLResponse(_render_discourse_html(desc, asc)) except Exception as e: return HTMLResponse(_fallback_html(str(e))) @app.get("/lookup") async def lookup_page(): lookup_path = DIST_DIR / "lookup" / "index.html" if lookup_path.exists(): return HTMLResponse(lookup_path.read_text()) return HTMLResponse(_fallback_html()) @app.get("/paper/{arxiv_id}") async def paper_page(arxiv_id: str): paper_path = DIST_DIR / "paper" / arxiv_id / "index.html" if paper_path.exists(): return HTMLResponse(paper_path.read_text()) try: from pipeline.db import get_db db = get_db() db.connect() paper = db.find_paper(arxiv_id) if paper: canonical_id = paper["arxiv_id"] reductions = db.get_reductions_for_paper(canonical_id) db.close() return HTMLResponse(_render_paper_html(paper, reductions)) db.close() except Exception: pass return HTMLResponse(_fallback_html(f"Paper {arxiv_id} not found."), status_code=404) # --- API --- @app.get("/api/stats") async def api_stats(): try: from pipeline.db import get_db db = get_db() db.connect() stats = db.stats() db.close() return stats except Exception as e: return {"error": str(e)} @app.get("/api/paper/{arxiv_id}") async def api_paper(arxiv_id: str): try: from pipeline.db import get_db db = get_db() db.connect() paper = db.find_paper(arxiv_id) if not paper: db.close() raise HTTPException(status_code=404) canonical_id = paper["arxiv_id"] reductions = db.get_reductions_for_paper(canonical_id) db.close() return {"paper": paper, "reductions": reductions} except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/lookup") async def api_lookup(q: str = Query(...)): try: from pipeline.db import get_db db = get_db() db.connect() paper = db.find_paper(q.strip()) db.close() if paper: return RedirectResponse(f"/paper/{paper['arxiv_id']}") raise HTTPException(status_code=404, detail=f"Not found: {q}") except HTTPException: raise @app.get("/api/index/analogs") async def api_index_analogs(): """All canonical analogs cited, with paper counts (cross-reference index).""" try: from pipeline.db import get_db db = get_db(); db.connect() analogs = db.get_canonical_analogs() db.close() return analogs except Exception as e: return {"error": str(e)} @app.get("/api/index/moves") async def api_index_moves(): """Deconstructive move counts across all concepts.""" try: from pipeline.db import get_db db = get_db(); db.connect() moves = db.get_move_counts() db.close() return moves except Exception as e: return {"error": str(e)} @app.get("/api/crossref/{arxiv_id}") async def api_crossref(arxiv_id: str): """Papers that share canonical analogs with this paper.""" try: from pipeline.db import get_db db = get_db(); db.connect() paper = db.find_paper(arxiv_id) if not paper: db.close() raise HTTPException(status_code=404) reductions = db.get_reductions_for_paper(paper["arxiv_id"]) # Collect unique canonical analogs for this paper analogs = list({r["canonical_analog"] for r in reductions if r.get("canonical_analog")}) # Find related papers for each analog related: dict[str, list] = {} for analog in analogs: papers = db.get_papers_by_analog(analog, limit=10) # Exclude self papers = [p for p in papers if p["arxiv_id"] != paper["arxiv_id"]] if papers: related[analog] = papers db.close() return { "paper_id": paper["arxiv_id"], "paper_title": paper["title"], "canonical_analogs": analogs, "related_papers": related, } except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/canonical/{formalism_id}") async def canonical_page(formalism_id: str): """Page for a canonical formalism, with its orienting paper deconstructed.""" try: from pipeline.match import load_engine from pipeline.db import get_db engine = load_engine() fm = engine._by_id.get(formalism_id) if not fm: raise HTTPException(status_code=404, detail=f"Unknown formalism: {formalism_id}") db = get_db(); db.connect() # Papers that cite this formalism as a canonical analog citing_papers = db.get_papers_by_analog(fm.name, limit=30) # Check if canonical paper is already deconstructed canonical_arxiv = getattr(fm, 'canonical_arxiv_id', None) # Also load from raw YAML for canonical_arxiv_id import yaml from pathlib import Path kb_path = Path(__file__).resolve().parent / "pipeline" / "kb" / "formalisms.yaml" with open(kb_path) as f: raw = yaml.safe_load(f) for entry in raw.get("formalisms", []): if entry.get("id") == formalism_id: canonical_arxiv = entry.get("canonical_arxiv_id") break canonical_paper = None canonical_reductions = [] if canonical_arxiv: paper = db.find_paper(canonical_arxiv) if paper and paper.get("status") == "matched": canonical_paper = paper canonical_reductions = db.get_reductions_for_paper(paper["arxiv_id"]) db.close() return HTMLResponse(_render_canonical_html( fm, citing_papers, canonical_arxiv, canonical_paper, canonical_reductions )) except HTTPException: raise except Exception as e: return HTMLResponse(_fallback_html(str(e))) @app.get("/api/canonical/{formalism_id}") async def api_canonical(formalism_id: str): """JSON API for a canonical formalism.""" try: from pipeline.match import load_engine from pipeline.db import get_db engine = load_engine() fm = engine._by_id.get(formalism_id) if not fm: raise HTTPException(status_code=404) # Get canonical_arxiv_id import yaml from pathlib import Path kb_path = Path(__file__).resolve().parent / "pipeline" / "kb" / "formalisms.yaml" with open(kb_path) as f: raw = yaml.safe_load(f) canonical_arxiv = None for entry in raw.get("formalisms", []): if entry.get("id") == formalism_id: canonical_arxiv = entry.get("canonical_arxiv_id") break db = get_db(); db.connect() citing = db.get_papers_by_analog(fm.name, limit=30) db.close() return { "id": fm.id, "name": fm.name, "year": fm.year, "origin": fm.origin, "signature": { "operation": fm.signature.operation, "domain": fm.signature.domain, "codomain": fm.signature.codomain, "objective_family": fm.signature.objective_family, }, "meso_type": fm.meso_type, "macro_type": fm.macro_type, "canonical_reference": fm.canonical_reference, "canonical_arxiv_id": canonical_arxiv, "citing_papers_count": len(citing), "citing_papers": citing[:10], } except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/cite/{arxiv_id}") async def api_cite(arxiv_id: str): """Look up citation count from Semantic Scholar.""" import urllib.request, json as _json url = f"https://api.semanticscholar.org/graph/v1/paper/ArXiv:{arxiv_id}?fields=citationCount,influentialCitationCount,title,year" try: req = urllib.request.Request(url, headers={"User-Agent": "DifferanceEngine/1.0"}) with urllib.request.urlopen(req, timeout=10) as resp: data = _json.loads(resp.read()) return { "arxiv_id": arxiv_id, "title": data.get("title", ""), "year": data.get("year"), "citation_count": data.get("citationCount", 0), "influential_citation_count": data.get("influentialCitationCount", 0), } except Exception as e: return {"arxiv_id": arxiv_id, "error": str(e), "citation_count": 0} @app.get("/api/models") async def api_models(): try: from pipeline.extract import ( select_extraction_model, _benchmark_cache, EXTRACTION_MODEL_CANDIDATES, ) api_key = os.environ.get("HF_API_KEY", "") selection = select_extraction_model(api_key) benchmarks = {} for mid, bench in _benchmark_cache.items(): benchmarks[mid] = { "passed": bench.passed, "correctness_score": bench.correctness_score, "completeness_score": bench.completeness_score, "latency_sec": bench.latency_sec, "cost_est": bench.cost_est, "error": bench.error, } return { "selected_model": selection.model_id, "selected_reason": selection.reason, "selected_at": selection.selected_at, "candidates": [ {"id": c["id"], "cost_per_1k_tokens": c["cost_per_1k_tokens"]} for c in EXTRACTION_MODEL_CANDIDATES ], "benchmarks": benchmarks, } except Exception as e: return {"error": str(e)} @app.post("/api/trigger") async def api_trigger(arxiv_id: str | None = None, retroactive: bool = False): api_key = os.environ.get("HF_API_KEY", "") if not api_key: raise HTTPException(status_code=401, detail="HF_API_KEY not configured") try: from pipeline.db import get_db from pipeline.ingest import ingest_single from pipeline.match import load_engine from pipeline.extract import extract_paper as _extract db = get_db() engine = load_engine() if arxiv_id: paper = ingest_single(arxiv_id.strip(), db=db) if not paper: raise HTTPException(status_code=404, detail=f"Could not fetch {arxiv_id}") # Use the canonical arXiv ID from the DB/API (may include version suffix) canonical_id = paper["arxiv_id"] import traceback try: extraction = _extract(title=paper["title"], abstract=paper["abstract"], api_key=api_key) except Exception as exc: return {"status": "extraction_error", "error": str(exc), "traceback": traceback.format_exc()} if extraction: db.delete_concepts_for_paper(canonical_id) db.delete_reductions_for_paper(canonical_id) for concept in extraction.get("concepts", []): cid = db.insert_concept(canonical_id, concept) match = engine.match_concept(concept) db.insert_reduction(cid, canonical_id, engine._result_to_dict(match)) db.update_status(canonical_id, "matched") return {"status": "matched", "concepts": len(extraction.get("concepts", []))} return {"status": "extraction_failed", "debug": "extract_paper returned None — check Space logs for details"} if retroactive: papers = db.get_papers_with_unknown(limit=30) for p in papers: pid = p["arxiv_id"] db.delete_reductions_for_paper(pid) return {"status": "retroactive_queued", "papers": len(papers)} return {"status": "no_action", "message": "Specify arxiv_id or retroactive=true"} except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # --- Helpers --- def _render_discourse_html(desc: list[dict], asc: list[dict]) -> str: """Render the discourse page: translation trace, not quality leaderboard.""" def _paper_row(p: dict) -> str: rate = p["_reduction_rate"] pct = int(rate * 100) bar_color = ( "#4ecdc4" if pct >= 67 else "#ffe66d" if pct >= 34 else "#ff6b35" ) title = (p.get("title") or "Untitled")[:80] aid = p.get("arxiv_id", "?") return f"""
These papers' vocabulary aligned heavily with our formal taxonomy. An invitation to understand the mathematical primitives they rest on — and to branch out from them.
These papers resisted translation. The gaps might be genuine novelty, taxonomy inadequacy, or underspecified language. The unknowns are the interesting part — not an error condition, but an invitation to look closer.
Translation trace against the Différance Engine taxonomy v1.0.0 (canonical analogs · deconstructive moves)
""" def _render_index_html(stats: dict, papers: list[dict]) -> str: """Render the main feed page with live DB data. Uses the same CSS as the Astro build so styling stays consistent. """ total = stats.get("total_papers", 0) matched = stats.get("by_status", {}).get("matched", 0) reduction_rate = stats.get("reduction_rate", 0) total_reductions = stats.get("total_reductions", 0) reduction_counts = stats.get("reductions_by_type", {}) paper_cards = "" if papers: for paper in papers: aid = paper.get("arxiv_id", "?") title = paper.get("title", "Untitled") updated = (paper.get("updated") or "")[:10] categories = paper.get("categories", []) if isinstance(categories, str): try: import json; categories = json.loads(categories) except Exception: categories = [] cat_tags = "".join( f'{c}' for c in categories[:2] ) # Fetch reductions for this paper reductions_html = "" try: from pipeline.db import get_db db = get_db() db.connect() reds = db.get_reductions_for_paper(aid) db.close() if reds: red_items = "" for r in reds: rtype = r.get("result_type", "unknown") display = r.get("display", "") delta = r.get("genuine_delta", "") # Parse display: ~~Term~~ ≡/≈/→ Reduction red_items += ( f'Trigger a pipeline run to populate the feed:
POST /api/trigger?arxiv_id=<id>
Set HF_API_KEY as a Space secret for LLM extraction.
Productive deconstruction — what creative destruction looks like under erasure.
HF_API_KEY as a Space secret and trigger a pipeline run."
)
return f"""
Productive deconstruction — what creative destruction looks like under erasure.
{message}
""" def _render_paper_html(paper: dict, reductions: list[dict]) -> str: title = paper.get("title", "Untitled") arxiv_id = paper.get("arxiv_id", "?") abstract = paper.get("abstract", "") # --- Reductions --- reds = "" for r in reductions: disp = r.get("display", f"~~{r.get('concept_name','')}~~ → {r.get('reduction','')}") delta = r.get("genuine_delta", "") rtype = r.get("result_type", "unknown") analog = r.get("canonical_analog", "") reds += f"""{_format_reduction_html(disp, delta)}
{rtype} micro: {r.get('micro','?')} | meso: {r.get('meso','?')} | macro: {r.get('macro','?')}
Also reduces to {short_analog}:
{arxiv_id} · {paper.get('updated','?')[:10]}
Canonical Paper (Orienting Document)
{cp_id}{' — ' + str(canonical_paper.get('citation_count','')) + ' citations' if canonical_paper.get('citation_count') else ''}
Canonical Paper: {canonical_arxiv}
Not yet ingested. Trigger deconstruction to see the raw formalism beneath the vocabulary.
{fm.year or '?'} · {_html.escape(fm.origin or '')} · {fm.status}
| Operation | {sig.operation or '?'} |
| Domain | {sig.domain or '?'} |
| Codomain | {sig.codomain or '?'} |
| Objective | {sig.objective_family or '?'} |
| Meso-type | {fm.meso_type or 'none'} |
| Macro-type | {fm.macro_type or 'none'} |
| Reference | {_html.escape(fm.canonical_reference or '')} |
Canonical formalism in the Différance Engine taxonomy. · JSON
"""