""" 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