main.py

#!/usr/bin/env python3
"""
FALSIFY — the AI research copilot that *revises*, not forgets.

Run this to watch belief revision happen on a real Cognee knowledge graph:

    python main.py            # full run (uses your LLM to judge the contradiction)
    python main.py --demo     # deterministic: pins the contradiction so the cascade
                              #   always runs, even without/with a flaky LLM key
    python main.py --keep     # don't prune existing memory first (advanced)

The story (Company X recall investigation)
------------------------------------------
Session 1 builds a belief graph with two competing hypotheses:
    A — "X knew via the March 2021 QA report"        (supported by evidence E_qa)
    B — "X knew via a January 2021 supplier email"   (supported by evidence E_email)
and a Conclusion K that *depends on* E_qa.

Session 2 drops ONE contradicting fact: "the March QA report was back-dated."
FALSIFY refutes E_qa, cascades the refutation forward (K collapses), promotes B as
the new frontier, and surgically forgets the orphaned conclusion — writing the
disbelief onto the graph so it survives a restart. A plain-RAG baseline, which has no
notion of truth-state, keeps citing the refuted March report. That contrast is the
whole point: *AI revised, not forgot.*
"""

from __future__ import annotations

import argparse
import asyncio
import os
import sys

# Load .env before importing cognee/falsify so provider config is in place.
try:
    from dotenv import load_dotenv

    load_dotenv()
except Exception:  # python-dotenv is optional; env may be set another way
    pass

# Importing falsify sets single-user Cognee env defaults (access control off, cache on).
import falsify  # noqa: F401  (import-time side effects)
from falsify.seed import NEW_FACT, QUESTION_TEXT


C = {
    "b": "\033[1m", "dim": "\033[2m", "g": "\033[92m", "r": "\033[91m",
    "y": "\033[93m", "c": "\033[96m", "x": "\033[0m",
}
if os.environ.get("NO_COLOR") or not sys.stdout.isatty():
    C = {k: "" for k in C}


def banner(text: str) -> None:
    print(f"\n{C['b']}{C['c']}{'━' * 64}{C['x']}")
    print(f"{C['b']}{C['c']}  {text}{C['x']}")
    print(f"{C['b']}{C['c']}{'━' * 64}{C['x']}")


def _has_llm_key() -> bool:
    """True if some LLM API key is configured (OpenAI-compatible or otherwise)."""
    for var in ("LLM_API_KEY", "OPENAI_API_KEY"):
        if os.environ.get(var):
            return True
    return False


def _print_no_key_help() -> None:
    print(
        f"""
{C['y']}{C['b']}No LLM API key found.{C['x']}

FALSIFY needs an OpenAI-compatible API key to (a) embed claims for the vector
prefilter and (b) judge contradictions. Set it up:

  1. cp .env.template .env
  2. edit .env and set:
       LLM_API_KEY="your_key_here"
       LLM_MODEL="gpt-4o-mini"          # or any OpenAI-compatible model
       # For a non-OpenAI endpoint (OpenRouter, vLLM, LM Studio, Groq):
       #   LLM_PROVIDER="custom"
       #   LLM_ENDPOINT="https://your-endpoint/v1"
  3. re-run:  python main.py

{C['dim']}Tip: `python main.py --demo` runs FULLY OFFLINE — fastembed handles
embeddings locally and the contradiction is pinned, so no API key is needed at all.
A key is only required for LIVE mode, where the LLM judges the contradiction.{C['x']}
"""
    )


async def run(demo: bool, keep: bool) -> int:
    from falsify.falsify import build_graph, revise, scoreboard
    from falsify.utils import get_belief_summary, print_graph_state, visualize_belief_graph

    banner("FALSIFY — belief-revision research copilot")
    print(f"  Research question: {C['b']}{QUESTION_TEXT}{C['x']}")
    print(f"  Mode: {'DEMO (deterministic contradiction pin)' if demo else 'LIVE (LLM judge)'}")

    # ---- Session 1: build the belief graph -------------------------------
    banner("SESSION 1 — build the investigation")
    if keep:
        from cognee.low_level import setup

        from falsify.seed import build_investigation

        await setup()
        seeded = await build_investigation()
    else:
        seeded = await build_graph()
    await print_graph_state("BEFORE — both hypotheses stand, Conclusion K rests on E_qa")

    # ---- Session 2: drop the contradicting fact --------------------------
    banner("SESSION 2 — a new fact arrives")
    print(f"  {C['y']}New fact:{C['x']} {NEW_FACT}\n")

    pinned = seeded.refuted_target_id if demo else None
    report = await revise(NEW_FACT, pinned_target_id=pinned)

    if not report.revised:
        print(f"  {C['y']}No contradiction was confirmed — graph unchanged.{C['x']}")
        print(f"  {C['dim']}(Try `python main.py --demo` to force the cascade deterministically.){C['x']}")
    else:
        print(f"  {C['r']}✗ refuted:{C['x']} {len(report.refuted)} evidence node(s)")
        print(f"  {C['r']}✗ invalidated:{C['x']} {len(report.invalidated)} conclusion(s)")
        for hid, action in report.hypothesis_actions.items():
            if action == "superseded":
                mark = f"{C['r']}↓ superseded{C['x']}"
            else:
                mark = f"{C['g']}↑ promoted (new frontier){C['x']}"
            print(f"     hypothesis {hid[:8]} → {mark}")
        for fid in report.forgotten:
            label = report.forgotten_labels.get(fid, fid)
            print(f"  {C['dim']}🗑 forgotten (deleted from graph + vector):{C['x']} {label}")
        if report.retained_provenance:
            print(f"  {C['dim']}⚑ kept as red provenance:{C['x']} {len(report.retained_provenance)} node(s)")

    await print_graph_state("AFTER — refutation cascaded, B ignites, orphan forgotten")

    # ---- The scoreboard: FALSIFY vs plain RAG ----------------------------
    banner("SCOREBOARD — FALSIFY (revised) vs plain RAG (stale)")
    board = await scoreboard(
        QUESTION_TEXT, seeded,
        rag_snapshot=report.rag_snapshot if report.revised else None,
    )
    print(f"  {C['g']}{C['b']}FALSIFY :{C['x']} {board.falsify_answer}")
    if board.falsify_support:
        print(f"            {C['dim']}supported by: {', '.join(board.falsify_support)}{C['x']}")
    rag_tag = f"{C['r']}[STALE — still cites a refuted fact]{C['x']}" if board.stale else ""
    print(f"  {C['y']}{C['b']}RAG     :{C['x']} {board.rag_answer} {rag_tag}")
    print(f"\n  {C['b']}→ AI revised, not forgot.{C['x']}")

    # ---- Cross-session proof: reload belief state fresh ------------------
    banner("CROSS-SESSION PROOF — reopen memory, beliefs stay revised")
    summary = await get_belief_summary()
    print(f"  Persisted belief state (re-read from graph): {summary}")
    print(f"  {C['dim']}Truth-state lives on the graph nodes, so a brand-new process sees the")
    print(f"  revised graph — the refuted branch never comes back.{C['x']}")

    # ---- Visualization ---------------------------------------------------
    out = await visualize_belief_graph("output/graph.html", title="FALSIFY — Company X investigation")
    if out:
        banner("VISUALIZATION")
        print(f"  Interactive belief graph written to: {C['b']}{os.path.abspath(out)}{C['x']}")
        print(f"  {C['dim']}Open it in a browser — red = refuted, grey = invalidated, green = alive.{C['x']}")

    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description="FALSIFY belief-revision demo")
    parser.add_argument("--demo", action="store_true",
                        help="pin the contradiction deterministically (LLM-independent cascade)")
    parser.add_argument("--keep", action="store_true",
                        help="do not prune existing memory before building")
    args = parser.parse_args()

    if not _has_llm_key() and not args.demo:
        _print_no_key_help()
        # Live mode needs an LLM key to judge the contradiction. Exit cleanly (0) with
        # setup help rather than a stack trace. (`--demo` runs fully offline below.)
        return 0

    try:
        return asyncio.run(run(demo=args.demo, keep=args.keep))
    except KeyboardInterrupt:
        print("\ninterrupted.")
        return 130


if __name__ == "__main__":
    raise SystemExit(main())