cognee-hackathon-project-main/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 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_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

    @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 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)

    # 1) detect
    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]

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

    # 3) promote competing hypothesis
    report.hypothesis_actions = await promote_competing_hypothesis(target_ids, prop.epoch)

    # 4) record the new fact + supersedes edge (keeps the refuted node as provenance)
    report.new_evidence_id = await _record_new_fact(new_fact, contradictions, source_id)

    # 5) forget orphaned dead-ends
    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: 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


@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) -> 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: alive-filtered graph answer ----
    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}

    # The winning hypothesis = alive hypothesis with the most alive supporting evidence weight.
    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:  # crude: hypotheses/conclusions carry 'statement'
            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: raw vector search, no truth filter ----
    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 = []

    refuted_ids = {nid for nid in node_ids if TruthState.REFUTED.value in truth.get(str(nid), [])}
    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