server.py

"""
FALSIFY live web server — FastAPI backend for the animated belief-revision UI.

This is the additive "web mode" for FALSIFY. The CLI (`python main.py`) is untouched;
this server exposes the same three core operations — build / revise / scoreboard — over
HTTP, plus a Server-Sent Events stream so a browser can *watch* belief revision happen:
the refute flash, the cascade sweep, the forget-dissolve.

Why SSE and not WebSockets: our event stream is one-directional (server -> browser), and
SSE is plain streaming HTTP. That makes it work behind the Hugging Face Spaces single-port
proxy AND across a Vercel(frontend)+Render(backend) split, neither of which hosts
long-lived WebSockets on their free tiers. See IMPLEMENTATION_PLAN.md §6.1.

Run locally:   uvicorn server:app --reload --port 8000   ->  http://localhost:8000
On HF Spaces:  uvicorn server:app --host 0.0.0.0 --port 7860
"""

from __future__ import annotations

import asyncio
import json
import logging
import os
from typing import Any, Dict, List, Optional

from fastapi import FastAPI, File, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

# Importing `falsify` first sets Cognee's env defaults (access-control off, etc.)
# before any Cognee module is imported — same ordering contract as main.py.
import falsify  # noqa: F401
from falsify import events, graph_ops
from falsify.falsify import (
    Scoreboard,
    build_diamond_graph,
    build_graph,
    revise,
    scoreboard,
    use_backend,
)
from falsify.models import TruthState
from falsify.seed import NEW_FACT, NEW_FACT_2, QUESTION_TEXT
from falsify.utils import _STATE_STYLE, _infer_type, get_belief_summary

logger = logging.getLogger("falsify.server")

app = FastAPI(title="FALSIFY Live", version="1.0")

# Wide-open CORS: the demo may be served same-origin (HF monolith) or cross-origin
# (Vercel frontend -> Render backend). FRONTEND_ORIGIN narrows it when set.
app.add_middleware(
    CORSMiddleware,
    allow_origins=[os.environ.get("FRONTEND_ORIGIN", "*")],
    allow_methods=["*"],
    allow_headers=["*"],
)

# ------------------------------------------------------------------ server state
_seeded = None                    # the active SeededGraph (set on startup / reset)
_scenario: str = "simple"         # "simple" | "diamond"
_backend_mode: str = "opensource" # "opensource" | "cloud"
_revise_lock = asyncio.Lock()     # serialize revise() — single-user demo safety


# ------------------------------------------------------------------ request models
class ChatRequest(BaseModel):
    message: str
    demo: bool = True


class ScenarioRequest(BaseModel):
    kind: str = "simple"          # "simple" | "diamond"


class ModeRequest(BaseModel):
    mode: str = "opensource"      # "opensource" | "cloud"
    url: Optional[str] = None
    api_key: Optional[str] = None


# ------------------------------------------------------------------ JSON shaping
async def _graph_json() -> Dict[str, Any]:
    """Current graph as UI-ready JSON: nodes (with truth-state color) + edges.

    Colors come straight from utils._STATE_STYLE so the web UI, the CLI console,
    and the static HTML export all share one palette.
    """
    nodes, edges = await graph_ops.load_graph()
    ids = [nid for nid, _p in nodes]
    truth = await graph_ops.get_truth(ids)

    out_nodes: List[Dict[str, Any]] = []
    for nid, props in nodes:
        align = truth.get(str(nid), [TruthState.ALIVE.value])
        state = align[0] if align else TruthState.ALIVE.value
        _tag, color = _STATE_STYLE.get(state, ("ALIVE", "#22c55e"))
        out_nodes.append({
            "id": str(nid),
            "label": graph_ops.node_label(props),
            "type": _infer_type(props),
            "state": state,
            "color": color,
        })

    out_edges = [
        {"source": str(s), "target": str(d), "relation": r}
        for (s, d, r, _p) in edges
    ]
    return {"nodes": out_nodes, "edges": out_edges,
            "scenario": _scenario, "backend": _backend_mode}


def _scoreboard_json(b: Scoreboard) -> Dict[str, Any]:
    return {
        "question": b.question,
        "falsify_answer": b.falsify_answer,
        "falsify_support": b.falsify_support,
        "rag_answer": b.rag_answer,
        "rag_citations": b.rag_citations,
        "stale": b.stale,
    }


def _report_json(r) -> Dict[str, Any]:
    return {
        "new_fact": r.new_fact,
        "refuted": r.refuted,
        "invalidated": r.invalidated,
        "forgotten": r.forgotten,
        "forgotten_labels": r.forgotten_labels,
        "hypothesis_actions": r.hypothesis_actions,
        "retained_provenance": r.retained_provenance,
        "new_evidence_id": r.new_evidence_id,
        "epoch": r.epoch,
        "revised": r.revised,
    }


# ------------------------------------------------------------------ SSE stream
@app.get("/api/events")
async def api_events():
    """Server-Sent Events: one frame per belief mutation, fanned from the event bus."""
    q = events.subscribe()

    async def gen():
        # Prime the stream so the client's onopen fires immediately behind proxies.
        yield ": connected\n\n"
        try:
            while True:
                try:
                    ev = await asyncio.wait_for(q.get(), timeout=15)
                    yield f"data: {json.dumps(ev)}\n\n"
                except asyncio.TimeoutError:
                    yield ": keepalive\n\n"
        finally:
            events.unsubscribe(q)

    return StreamingResponse(
        gen(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",   # stop HF/Render proxies buffering the stream
        },
    )


# ------------------------------------------------------------------ REST endpoints
@app.get("/api/graph")
async def api_graph():
    return await _graph_json()


@app.get("/api/scoreboard")
async def api_scoreboard(q: str = QUESTION_TEXT):
    board = await scoreboard(q, _seeded)
    return _scoreboard_json(board)


@app.post("/api/chat")
async def api_chat(req: ChatRequest):
    """A question (ends with '?') -> scoreboard; anything else -> a new fact to revise.

    Graph mutations emit SSE events *during* revise(), so the browser animates the
    cascade in real time while this call is still in flight.
    """
    msg = req.message.strip()
    if not msg:
        return {"type": "noop", "data": {}}

    if msg.endswith("?"):
        board = await scoreboard(msg, _seeded)
        return {"type": "answer", "data": _scoreboard_json(board)}

    async with _revise_lock:
        pinned = _seeded.refuted_target_id if (req.demo and _seeded) else None
        await events.emit_step("ingest", "New contradicting fact received")
        report = await revise(msg, pinned_target_id=pinned)
    return {"type": "revision", "data": _report_json(report)}


@app.post("/api/scenario")
async def api_scenario(req: ScenarioRequest):
    """(Re)build the graph as the simple or diamond investigation."""
    global _seeded, _scenario
    if req.kind == "diamond":
        _seeded, _scenario = await build_diamond_graph(), "diamond"
    else:
        _seeded, _scenario = await build_graph(), "simple"
    await events.emit_graph_reset()
    return await _graph_json()


@app.post("/api/reset")
async def api_reset():
    """Rebuild the current scenario from scratch (fresh belief state)."""
    return await api_scenario(ScenarioRequest(kind=_scenario))


@app.post("/api/demo")
async def api_demo():
    """One-click headline demo on the simple graph: drop the back-dated-QA fact.

    Rebuilds the simple investigation, then refutes E_qa (pinned, so it runs
    keyless). Graph mutations stream over SSE while this call is in flight, so the
    browser animates the cascade live; the returned report + scoreboard let the UI
    finish with the promotion 'rise' and the FALSIFY-vs-RAG panel.
    """
    global _seeded, _scenario
    async with _revise_lock:
        _seeded, _scenario = await build_graph(), "simple"
        await events.emit_graph_reset()
        await asyncio.sleep(0.4)
        await events.emit_step("ingest", "New fact: the March 2021 QA report was back-dated")
        report = await revise(NEW_FACT, pinned_target_id=_seeded.refuted_target_id)
    board = await scoreboard(
        QUESTION_TEXT, _seeded,
        rag_snapshot=report.rag_snapshot if report.revised else None,
    )
    return {"report": _report_json(report), "scoreboard": _scoreboard_json(board)}


@app.post("/api/diamond")
async def api_diamond():
    """Two-phase diamond story: K2 survives phase 1, then collapses in phase 2.

    Phase 1 refutes E_qa — the single-dependency conclusion K dies but the diamond
    K2 survives on E_email. A pause lets the UI show K2 standing, then phase 2
    refutes E_email and K2 finally collapses. Both targets are pinned so it runs
    keyless and deterministically.
    """
    global _seeded, _scenario
    async with _revise_lock:
        _seeded, _scenario = await build_diamond_graph(), "diamond"
        await events.emit_graph_reset()
        await asyncio.sleep(0.5)
        await events.emit_step("phase1", "Phase 1 — the March 2021 QA report was back-dated")
        r1 = await revise(NEW_FACT, pinned_target_id=_seeded.ids["E_qa"])
        await asyncio.sleep(1.6)  # hold on K2 surviving before the second blow
        await events.emit_step("phase2", "Phase 2 — the January 2021 supplier email was fabricated")
        r2 = await revise(NEW_FACT_2, pinned_target_id=_seeded.ids["E_email"])
    board = await scoreboard(
        QUESTION_TEXT, _seeded,
        rag_snapshot=r2.rag_snapshot if r2.revised else None,
    )
    return {"phase1": _report_json(r1), "phase2": _report_json(r2),
            "scoreboard": _scoreboard_json(board)}


@app.post("/api/upload")
async def api_upload(file: UploadFile = File(...)):
    """Ingest an uploaded document into memory via cognee.add + cognify.

    Turns the fixed demo into a real product: a judge can drop their own file and
    watch it become graph. Requires an LLM key for cognify; degrades with a clear
    error otherwise (the demo keeps working).
    """
    import cognee

    raw = await file.read()
    text = raw.decode("utf-8", "ignore").strip()
    if not text:
        return {"ok": False, "error": "empty or unreadable file"}
    try:
        await events.emit_step("upload", f"Ingesting {file.filename}…")
        await cognee.add(text)
        await cognee.cognify()
        await events.emit_graph_reset()
        return {"ok": True, "graph": await _graph_json()}
    except Exception as exc:  # most likely: no LLM key for cognify
        logger.warning("upload cognify failed: %s", exc)
        return {"ok": False, "error": str(exc),
                "hint": "Set LLM_API_KEY to enable document ingestion."}


@app.get("/api/verify")
async def api_verify():
    """Re-read the persisted belief state from the on-disk graph store.

    Truth-state (refuted / invalidated / superseded) is written ON the graph nodes
    in Cognee's file-backed Kuzu store — not held in process memory — so re-reading
    it here returns the *revised* beliefs, which is exactly what survives a full
    process or container restart.

    We deliberately do NOT evict the engine mid-request: in Ladybug subprocess mode
    a worker process holds the DB file lock, so an in-process cold-reopen would
    collide with that lock. The durable-proof claim rests on the truth_alignment
    living in storage, and on the graph coming back revised after a real restart.
    """
    try:
        summary = await get_belief_summary()
        return {"persisted": True, "summary": summary}
    except Exception as exc:
        logger.error("verify read failed: %s", exc)
        return {"persisted": False, "error": str(exc)}


@app.post("/api/mode")
async def api_mode(req: ModeRequest):
    """Toggle the Cognee backend between self-hosted (open source) and Cognee Cloud."""
    global _backend_mode
    # Cloud creds can come from the request or the environment.
    url = req.url or os.environ.get("COGNEE_CLOUD_URL")
    api_key = req.api_key or os.environ.get("COGNEE_CLOUD_API_KEY")
    _backend_mode = await use_backend(req.mode, url=url, api_key=api_key)
    return {"backend": _backend_mode}


@app.get("/api/health")
async def api_health():
    return {"ok": True, "scenario": _scenario, "backend": _backend_mode,
            "subscribers": events.has_subscribers()}


# ------------------------------------------------------------------ lifecycle
@app.on_event("startup")
async def _startup():
    """Seed the simple investigation so the first page load shows a live graph."""
    global _seeded
    try:
        _seeded = await build_graph()
        logger.info("startup: seeded simple investigation")
    except Exception as exc:  # don't let a seed hiccup take the server down
        logger.error("startup seed failed: %s", exc)


# Serve the built frontend LAST so /api/* routes take precedence. The directory is
# created by the Docker build (frontend/dist -> ./static); guard so local dev without
# a build still boots (API-only).
if os.path.isdir("static"):
    app.mount("/", StaticFiles(directory="static", html=True), name="static")