falsify/falsify.py

"""
FALSIFY orchestration — the belief-revision copilot's public API.

Three verbs tie the engine together:

    build_graph()          -> seed the Session-1 investigation graph (clean slate).
    revise(new_fact)       -> run the full revision pipeline for an incoming fact:
                              detect -> propagate -> promote -> record supersede -> forget.
    scoreboard(question)   -> the money shot: FALSIFY's revised answer (reads truth
                              state, skips dead branches) vs a plain-RAG baseline
                              (raw vector search, no truth filter) that still cites the
                              refuted fact.

Everything is persisted on the graph (truth-state on nodes), so a fresh process /
second session sees the revised beliefs — the cross-session guarantee.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

from falsify import graph_ops
from falsify.edges import SUPERSEDES, SUPPORTS
from falsify.models import Evidence, TruthState
from falsify.seed import SeededGraph, build_diamond_investigation, build_investigation
from falsify.tasks import (
    Contradiction,
    cascade_forget,
    detect_contradictions,
    promote_competing_hypothesis,
    propagate_refutation,
)

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

_ALIVE = TruthState.ALIVE.value
_EVIDENCE_COLLECTION = "Evidence_claim"


@dataclass
class RevisionReport:
    """Full record of one :func:`revise` run, for demo output and tests."""

    new_fact: str
    contradictions: List[Contradiction] = field(default_factory=list)
    refuted: List[str] = field(default_factory=list)
    invalidated: List[str] = field(default_factory=list)
    hypothesis_actions: Dict[str, str] = field(default_factory=dict)
    forgotten: List[str] = field(default_factory=list)
    forgotten_labels: Dict[str, str] = field(default_factory=dict)
    retained_provenance: List[str] = field(default_factory=list)
    new_evidence_id: Optional[str] = None
    epoch: int = 0
    rag_snapshot: Optional[List[Dict]] = None

    @property
    def revised(self) -> bool:
        """True if the fact actually triggered a belief change."""
        return bool(self.refuted or self.invalidated)


async def build_graph() -> SeededGraph:
    """Prune everything and build the Session-1 investigation graph.

    Returns the :class:`SeededGraph` with stable handles to the seeded nodes.
    """
    import cognee
    from cognee.low_level import setup

    logger.info("build_graph: pruning and seeding fresh investigation")
    await cognee.forget(everything=True)
    await setup()  # (re)create relational tables the storage pipeline needs
    seeded = await build_investigation()
    return seeded


async def build_diamond_graph() -> SeededGraph:
    """Prune everything and build the diamond-dependency investigation graph.

    Same clean-slate sequence as :func:`build_graph`, but seeds the extended graph
    with Conclusion K2 (critically dependent on both E_qa and E_email). Drives the
    two-phase "survive then collapse" demo; see
    :func:`falsify.seed.build_diamond_investigation`.
    """
    import cognee
    from cognee.low_level import setup

    logger.info("build_diamond_graph: pruning and seeding diamond investigation")
    await cognee.forget(everything=True)
    await setup()
    return await build_diamond_investigation()


async def use_backend(
    mode: str = "opensource",
    *,
    url: Optional[str] = None,
    api_key: Optional[str] = None,
) -> str:
    """Route Cognee operations to a backend and return the active mode.

    ``mode="cloud"`` (with a tenant ``url`` + ``api_key``) points every subsequent
    ``remember`` / ``recall`` / ``memify`` / ``forget`` call at a Cognee Cloud tenant
    via :func:`cognee.serve` — making the *same* FALSIFY pipeline demonstrable on the
    Cognee Cloud track without changing any belief logic. Anything else keeps the
    self-hosted (open-source) engines. Best-effort: if the cloud handshake fails we
    log and stay open-source so the demo never hard-fails.
    """
    import cognee

    if mode == "cloud" and url and api_key:
        try:
            await cognee.serve(url=url, api_key=api_key)
            logger.info("FALSIFY backend -> Cognee Cloud (%s)", url)
            return "cloud"
        except Exception as exc:
            logger.warning("cognee.serve failed (%s); staying open-source", exc)
            return "opensource"
    logger.info("FALSIFY backend -> self-hosted (open source)")
    return "opensource"


async def revise(
    new_fact: str,
    *,
    pinned_target_id: Optional[str] = None,
    source_id: str = "session2_fact",
) -> RevisionReport:
    """Run the full belief-revision pipeline for an incoming fact.

    Pipeline (REQUIREMENTS §1.3-§1.5):
        1. detect_contradictions -> which evidence does the fact contradict?
        2. propagate_refutation  -> refute it; cascade invalidation forward.
        3. promote_competing_hypothesis -> demote the losing hypothesis, ignite the rival.
        4. record the new fact as Evidence + a ``supersedes`` edge to the refuted node
           (so the refuted node is retained as a provenance tombstone).
        5. cascade_forget -> hard-delete orphaned dead-ends from graph + vector.

    Args:
        new_fact: the incoming claim.
        pinned_target_id: demo override — refute this evidence id deterministically.
        source_id: provenance id for the materialized new-fact Evidence node.

    Returns:
        A :class:`RevisionReport` describing everything that changed.
    """
    report = RevisionReport(new_fact=new_fact)

    # Route through cognee.memify() pipeline for deep Cognee API integration.
    # Falls back to direct calls if memify is unavailable.
    try:
        report = await revise_via_memify(
            new_fact, pinned_target_id=pinned_target_id, source_id=source_id,
        )
        if not report.revised:
            logger.info("revise: no contradiction found; graph unchanged")
        else:
            logger.info(
                "revise complete (via memify): refuted=%d invalidated=%d forgotten=%d",
                len(report.refuted), len(report.invalidated), len(report.forgotten),
            )
        return report
    except Exception as exc:
        logger.warning("memify pipeline failed (%s); falling back to direct calls", exc)

    # Fallback: direct task calls (same logic, no pipeline wrapper)
    contradictions = await detect_contradictions(new_fact, pinned_target_id=pinned_target_id)
    report.contradictions = contradictions
    if not contradictions:
        logger.info("revise: no contradiction found; graph unchanged")
        return report

    target_ids = [c.target_id for c in contradictions]

    prop = await propagate_refutation(target_ids)
    report.refuted = prop.refuted
    report.invalidated = prop.invalidated
    report.epoch = prop.epoch

    report.hypothesis_actions = await promote_competing_hypothesis(target_ids, prop.epoch)

    report.new_evidence_id = await _record_new_fact(new_fact, contradictions, source_id)

    report.rag_snapshot = await _snapshot_rag(new_fact)

    forget_res = await cascade_forget(prop.affected)
    report.forgotten = forget_res.forgotten
    report.forgotten_labels = forget_res.labels
    report.retained_provenance = forget_res.retained_provenance

    logger.info(
        "revise complete (direct): refuted=%d invalidated=%d forgotten=%d",
        len(report.refuted), len(report.invalidated), len(report.forgotten),
    )
    return report


async def _record_new_fact(
    new_fact: str,
    contradictions: List[Contradiction],
    source_id: str,
) -> Optional[str]:
    """Materialize the new fact as an Evidence node and link supersedes edges.

    The new (alive) evidence ``supersedes`` each refuted evidence node. This both
    records provenance and pins the refuted node as a retained tombstone (an alive
    supersedes-source protects its target from forget — REQUIREMENTS §1.5c).
    """
    from cognee.tasks.storage import add_data_points

    try:
        new_ev = Evidence(
            claim=new_fact,
            source_id=source_id,
            stance="refutes",
            confidence=max((c.confidence for c in contradictions), default=0.9),
        )
        await add_data_points([new_ev])
        for c in contradictions:
            await graph_ops.add_edge(
                str(new_ev.id), str(c.target_id), SUPERSEDES, {"confidence": c.confidence}
            )
        logger.info("recorded new fact %s superseding %d node(s)", new_ev.id, len(contradictions))
        return str(new_ev.id)
    except Exception as exc:
        logger.error("failed to record new fact: %s", exc)
        return None


async def _snapshot_rag(query: str) -> List[Dict]:
    """Capture RAG vector hits before cascade_forget deletes them."""
    ve = graph_ops.get_vector_engine()
    try:
        hits = await ve.search(
            _EVIDENCE_COLLECTION, query_text=query, limit=5, include_payload=True,
        )
    except Exception:
        return []
    results = []
    for h in (hits or []):
        payload = getattr(h, "payload", {}) or {}
        results.append({"id": str(h.id), "payload": payload})
    return results


# --------------------------------------------------------------------------- #
# memify adapter — wraps FALSIFY tasks as a cognee.memify() pipeline
# --------------------------------------------------------------------------- #


async def _task_detect(data: List[Dict[str, Any]], **kwargs) -> List[Dict[str, Any]]:
    """memify extraction task: detect contradictions."""
    c = data[0] if data else {}
    contradictions = await detect_contradictions(
        c["new_fact"], pinned_target_id=c.get("pinned_target_id"),
    )
    c["contradictions"] = contradictions
    return [c]


async def _task_propagate(data: List[Dict[str, Any]], **kwargs) -> List[Dict[str, Any]]:
    """memify enrichment task 1: propagate refutation + promote hypotheses."""
    c = data[0] if data else {}
    contradictions = c.get("contradictions", [])
    if not contradictions:
        return [c]

    target_ids = [con.target_id for con in contradictions]
    prop = await propagate_refutation(target_ids)
    c["propagation"] = prop
    c["hypothesis_actions"] = await promote_competing_hypothesis(target_ids, prop.epoch)
    return [c]


async def _task_record_and_forget(data: List[Dict[str, Any]], **kwargs) -> List[Dict[str, Any]]:
    """memify enrichment task 2: record new fact, snapshot RAG, cascade forget."""
    c = data[0] if data else {}
    contradictions = c.get("contradictions", [])
    if not contradictions:
        return [c]

    new_evidence_id = await _record_new_fact(
        c["new_fact"], contradictions, c.get("source_id", "session2_fact"),
    )
    c["new_evidence_id"] = new_evidence_id
    c["rag_snapshot"] = await _snapshot_rag(c["new_fact"])

    prop = c.get("propagation")
    if prop:
        forget_res = await cascade_forget(prop.affected)
        c["forget_result"] = forget_res

    return [c]


async def revise_via_memify(
    new_fact: str,
    *,
    pinned_target_id: Optional[str] = None,
    source_id: str = "session2_fact",
) -> RevisionReport:
    """Run the belief-revision pipeline through cognee.memify().

    Functionally identical to the direct-call path, but routes through Cognee's
    memify pipeline runner so the revision tasks appear as first-class Cognee
    pipeline stages — demonstrating deep API integration.
    """
    import cognee
    from cognee.modules.pipelines.tasks.task import Task

    pipeline_input = [{
        "new_fact": new_fact,
        "pinned_target_id": pinned_target_id,
        "source_id": source_id,
    }]

    await cognee.memify(
        extraction_tasks=[Task(_task_detect)],
        enrichment_tasks=[
            Task(_task_propagate),
            Task(_task_record_and_forget),
        ],
        data=pipeline_input,
    )

    # Build the report from the mutated context dict
    ctx = pipeline_input[0]
    report = RevisionReport(new_fact=new_fact)
    report.contradictions = ctx.get("contradictions", [])

    prop = ctx.get("propagation")
    if prop:
        report.refuted = prop.refuted
        report.invalidated = prop.invalidated
        report.epoch = prop.epoch

    report.hypothesis_actions = ctx.get("hypothesis_actions", {})
    report.new_evidence_id = ctx.get("new_evidence_id")
    report.rag_snapshot = ctx.get("rag_snapshot")

    forget_res = ctx.get("forget_result")
    if forget_res:
        report.forgotten = forget_res.forgotten
        report.forgotten_labels = forget_res.labels
        report.retained_provenance = forget_res.retained_provenance

    return report


@dataclass
class Scoreboard:
    """The FALSIFY-vs-RAG comparison shown every run."""

    question: str
    falsify_answer: str
    falsify_support: List[str] = field(default_factory=list)
    rag_answer: str = ""
    rag_citations: List[str] = field(default_factory=list)
    stale: bool = False  # True if RAG still cites a refuted node FALSIFY dropped


async def scoreboard(
    question: str,
    seeded: Optional[SeededGraph] = None,
    rag_snapshot: Optional[List[Dict]] = None,
) -> Scoreboard:
    """Compare FALSIFY's revised answer against a plain-RAG baseline.

    FALSIFY answer: derived from the graph, reading truth-state and using only
    hypotheses/evidence still ``alive`` (the promoted frontier hypothesis).

    RAG baseline: a raw vector search over ``Evidence_claim`` with **no** truth
    filter — so it still returns evidence FALSIFY has refuted, and cites the stale
    fact. This asymmetry is the demo's whole point.
    """
    board = Scoreboard(question=question, falsify_answer="(no surviving hypothesis)")

    # ---- FALSIFY: try cognee.recall() first, fall back to graph traversal ----
    recall_succeeded = False
    try:
        import cognee
        from cognee.modules.search.types.SearchType import SearchType

        recall_results = await cognee.recall(
            query_text=question,
            query_type=SearchType.GRAPH_COMPLETION,
            top_k=3,
        )
        if recall_results:
            best = recall_results[0]
            answer_text = getattr(best, "text", None) or str(best)
            board.falsify_answer = answer_text
            board.falsify_support = ["(via cognee.recall GRAPH_COMPLETION)"]
            recall_succeeded = True
            logger.info("scoreboard: used cognee.recall() for FALSIFY answer")
    except Exception as exc:
        logger.info("cognee.recall() unavailable (%s); falling back to graph traversal", exc)

    # Fall back to manual graph traversal (always works, including --demo offline mode)
    nodes, edges = await graph_ops.load_graph()
    node_ids = [nid for nid, _p in nodes]
    truth = await graph_ops.get_truth(node_ids)
    props_by_id = {str(nid): (p or {}) for nid, p in nodes}

    if not recall_succeeded:
        best_hyp, best_score = None, -1.0
        support_edges = [(s, d, p) for (s, d, r, p) in edges if r == SUPPORTS]
        for nid, props in nodes:
            nid = str(nid)
            if "statement" not in props:
                continue
            if _ALIVE not in truth.get(nid, [_ALIVE]):
                continue
            score = 0.0
            alive_support = []
            for (src, dst, ep) in support_edges:
                if str(dst) != nid:
                    continue
                if _ALIVE in truth.get(str(src), [_ALIVE]):
                    score += float(ep.get("weight", 0.5))
                    alive_support.append(graph_ops.node_label(props_by_id.get(str(src), {})))
            if alive_support and score > best_score:
                best_hyp, best_score = nid, score
                board.falsify_answer = graph_ops.node_label(props)
                board.falsify_support = alive_support

    # ---- RAG baseline: use pre-forget snapshot if available, else live search ----
    refuted_ids = {str(nid) for nid in node_ids
                   if TruthState.REFUTED.value in truth.get(str(nid), [])}

    if rag_snapshot is not None:
        for entry in rag_snapshot:
            payload = entry.get("payload", {})
            text = (payload.get("claim") or payload.get("text")
                    or graph_ops.node_label(payload))
            board.rag_citations.append(str(text))
            if entry["id"] in refuted_ids:
                board.stale = True
    else:
        ve = graph_ops.get_vector_engine()
        try:
            hits = await ve.search(
                _EVIDENCE_COLLECTION, query_text=question, limit=5,
                include_payload=True,
            )
        except Exception as exc:
            logger.warning("RAG baseline search failed: %s", exc)
            hits = []
        for h in (hits or []):
            payload = getattr(h, "payload", {}) or {}
            text = (payload.get("claim") or payload.get("text")
                    or graph_ops.node_label(payload))
            board.rag_citations.append(str(text))
            if str(h.id) in refuted_ids:
                board.stale = True

    board.rag_answer = (board.rag_citations[0] if board.rag_citations
                        else "(no vector hits)")

    logger.info("scoreboard: falsify=%r stale_rag=%s", board.falsify_answer, board.stale)
    return board