""" 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""" {pct}%
{title} {p['_mapped']}/{p['_total']} mapped """ desc_rows = "\n".join(_paper_row(p) for p in desc if p["_total"] > 0) asc_rows = "\n".join(_paper_row(p) for p in asc if p["_total"] > 0) return f""" Translation Trace — Différance Engine

← The Différance Engine

Translation Trace

This is not a quality leaderboard. It describes translation efficiency — what fraction of a paper's extracted concepts successfully mapped to a declared formal vocabulary (taxonomy v1.0). A high rate means the paper's vocabulary overlapped heavily with ours. A low rate means it didn't. The low-rate papers are where you should look — for genuine novelty, for taxonomy gaps, or for bullshit. The tool doesn't adjudicate. It shows the trace and lets you decide.

↓ Descending Most Translated

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.

{desc_rows}

↑ Ascending Least Translated

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.

{asc_rows}

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'
' f'{rtype} ' f'{_format_reduction_html(display, delta)}' f'
' ) reductions_html = f'
{red_items}
' except Exception: pass paper_cards += f"""
{updated} {aid} {cat_tags}

{title}

{reductions_html}
""" paper_list_html = ( f'
{paper_cards}
' if papers else f"""

No papers deconstructed yet

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.

""" ) return f""" The Différance Engine — Under Erasure

The Différance Engine

Productive deconstruction — what creative destruction looks like under erasure.

{total} papers ingested {matched} deconstructed {int(reduction_rate * 100)}% reduction rate {total_reductions} reductions
{paper_list_html}
""" def _format_reduction_html(display: str, delta: str) -> str: """Parse a sous-rature display string like '~~GLIGEN~~ ≡ Diffusion Process ∘ diffuse' into formatted HTML spans. Handles expandable canonical decomposition chains.""" import re import html as _html # Check for embedded expansion data (NUL-delimited sentinels from match.py) expand_html = "" if "\x00EXPAND\x00" in display: parts = display.split("\x00EXPAND\x00", 1) display = parts[0] expanded = parts[1].split("\x00/EXPAND\x00", 1)[0] if "\x00/EXPAND\x00" in parts[1] else "" if expanded: expanded = _html.escape(expanded) uid = abs(hash(expanded)) % 100000 expand_html = ( f' [+]' f'' ) # Match ~~struck term~~ then connector (=/~/->) then math term m = re.match(r"~~(.+?)~~\s*(≡|≈|→)\s*(.+?)(?:\s*\(Δ:\s*(.+?)\))?$", display) if m: struck = m.group(1) connector = m.group(2) math_term = m.group(3) inner_delta = m.group(4) or delta delta_html = f' (Δ: {inner_delta})' if inner_delta else "" # Link recognized formalism names to canonical pages math_term = _link_formalism_names(math_term) return ( f'{struck}' f'{connector} ' f'{math_term}' f'{delta_html}' f'{expand_html}' ) # Fallback: just escape and display (still check for expansion) return _html.escape(display) + expand_html # Cache for formalism name -> ID mapping (populated lazily) _formalism_name_to_id: dict[str, str] | None = None def _get_formalism_name_map() -> dict[str, str]: """Return a mapping from lowercased formalism names to their KB IDs. Used for linking formalism names in reductions to canonical pages.""" global _formalism_name_to_id if _formalism_name_to_id is not None: return _formalism_name_to_id try: import yaml from pathlib import Path kb_path = Path(__file__).resolve().parent / "pipeline" / "kb" / "formalisms.yaml" with open(kb_path) as f: data = yaml.safe_load(f) _formalism_name_to_id = {} for entry in data.get("formalisms", []): fid = entry.get("id", "") name = (entry.get("name", "") or "").lower() if name and fid: _formalism_name_to_id[name] = fid return _formalism_name_to_id except Exception: return {} def _link_formalism_names(text: str) -> str: """Wrap known formalism names in links to their canonical pages. Returns HTML with tags for recognized formalisms.""" import re as _re fm_map = _get_formalism_name_map() if not fm_map: return text # Sort by length descending to match longest names first for name in sorted(fm_map.keys(), key=len, reverse=True): fid = fm_map[name] # Case-insensitive replacement, but only whole-word-ish pattern = _re.compile(_re.escape(name), _re.IGNORECASE) text = pattern.sub( f'{name}', text ) return text def _fallback_html(msg: str = "") -> str: message = msg or ( "The Différance Engine is live but no papers have been deconstructed yet. " "Set HF_API_KEY as a Space secret and trigger a pipeline run." ) return f""" The Différance Engine

The Différance Engine

Productive deconstruction — what creative destruction looks like under erasure.

{message}

Stats · Models · Lookup

""" 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','?')}

""" # --- Cross-references: canonical analogs shared with other papers --- from pipeline.db import get_db xref_html = "" try: db = get_db(); db.connect() analogs_seen = set() for r in reductions: analog = r.get("canonical_analog", "") if analog and analog not in analogs_seen: analogs_seen.add(analog) related = db.get_papers_by_analog(analog, limit=8) related = [p for p in related if p["arxiv_id"] != arxiv_id] if related: short_analog = analog[:80] + ("…" if len(analog) > 80 else "") items = "".join( f'
  • {p["title"]} ({p["arxiv_id"]})
  • ' for p in related[:5] ) xref_html += f"""

    Also reduces to {short_analog}:

    """ db.close() except Exception: pass return f""" {title} — Différance Engine

    ← The Différance Engine

    {title}

    {arxiv_id} · {paper.get('updated','?')[:10]}

    Abstract{abstract}

    Reductions

    {reds} {f'

    Cross-References

    {xref_html}' if xref_html else ''} """ def _render_canonical_html(fm, citing_papers, canonical_arxiv, canonical_paper, canonical_reductions) -> str: """Render a page for a canonical formalism, optionally with its orienting paper.""" import html as _html sig = fm.signature sig_str = f"{sig.operation or '?'}({sig.domain or '?'} -> {sig.codomain or '?'})" if sig.objective_family: sig_str += f" objective={sig.objective_family}" # Canonical paper section canonical_html = "" if canonical_arxiv: if canonical_paper: cp_title = _html.escape(canonical_paper.get("title", "Untitled")) cp_id = canonical_paper["arxiv_id"] reds_html = "" for r in canonical_reductions: disp = r.get("display", "") delta = r.get("genuine_delta", "") rtype = r.get("result_type", "unknown") reds_html += ( f'
    ' f'{_format_reduction_html(disp, delta)}' f'{rtype}
    ' ) canonical_html = f"""

    Canonical Paper (Orienting Document)

    {cp_title}

    {cp_id}{' — ' + str(canonical_paper.get('citation_count','')) + ' citations' if canonical_paper.get('citation_count') else ''}

    Deconstruction ({len(canonical_reductions)} concepts) {reds_html}
    """ else: canonical_html = f"""

    Canonical Paper: {canonical_arxiv}

    Not yet ingested. Trigger deconstruction to see the raw formalism beneath the vocabulary.

    """ # Citing papers table citing_rows = "" if citing_papers: for p in citing_papers[:20]: aid = p.get("arxiv_id", "?") title = (p.get("title") or "Untitled")[:80] citing_rows += ( f'{_html.escape(title)}' f'{aid}' ) citing_section = ( f'

    ' f'Papers Reducing to This Formalism ({len(citing_papers)})

    ' f'{citing_rows}
    ' ) if citing_papers else "" return f""" {fm.name} — Canonical Formalism — Différance Engine

    ← The Différance Engine

    {_html.escape(fm.name)}

    {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_html} {citing_section}

    Canonical formalism in the Différance Engine taxonomy. · JSON

    """