""" Forward refutation propagation — the core of FALSIFY's belief revision. When a piece of Evidence is refuted, the conclusions that *depend on it* can no longer stand. This module walks the dependency structure and flips the truth-state of every node that transitively rests on the refuted evidence, then re-scores the competing hypotheses. Why this can't be done by RAG ----------------------------- Vector similarity has no notion of "this fact supports that conclusion three hops away." Refutation propagation is *graph traversal over typed edges* — it is exactly the thing a knowledge graph can do and an embedding index cannot. This is FALSIFY's differentiator and maps directly to the hackathon's "Best Use of Cognee" criterion. Direction of travel (critical detail) ------------------------------------- The ``depends_on`` edge points **Conclusion -> Evidence** (a conclusion depends on the evidence it rests on). So to find what *breaks* when Evidence ``E`` is refuted, we look for ``depends_on`` edges whose **target** is ``E``; their **sources** are the dependent Conclusions. We then recurse: a newly-invalidated Conclusion may itself be the target of further ``depends_on`` edges. Correctness cases handled (REQUIREMENTS §4.3) -------------------------------------------- * **Cycle safety** — a ``visited`` set guarantees termination on cyclic graphs. * **Critical vs non-critical** — only a ``critical: true`` dependency can invalidate a conclusion. A non-critical dependency being refuted decays confidence but the conclusion stays ``alive``. * **Diamond / partial refutation** — a conclusion with several critical supporters is invalidated only when it loses its **last** alive critical supporter. If an alternative critical support is still alive, the conclusion survives (and the refuted evidence is *retained*, because it still feeds a live node). """ from __future__ import annotations import logging from dataclasses import dataclass, field from typing import Dict, List, Optional, Set from falsify import graph_ops from falsify.edges import ( DEPENDENCY_EDGE_TYPES, DEPENDS_ON, SUPPORTS, is_critical_dependency, ) from falsify.models import TruthState logger = logging.getLogger("falsify.propagate") # Truth states that count as "dead" for the purpose of dependency support. _DEAD_STATES = {TruthState.REFUTED.value, TruthState.INVALIDATED.value, TruthState.FORGOTTEN.value} @dataclass class PropagationResult: """Outcome of a refutation cascade. Attributes: refuted: evidence node ids set to ``refuted`` (the cascade seeds). invalidated: conclusion node ids set to ``invalidated`` by the cascade. weakened: conclusion ids whose confidence decayed but stayed ``alive`` (a non-critical dependency was refuted). epoch: the revision epoch stamped on every state change in this cascade. affected: convenience union of refuted + invalidated ids (the death set candidates for :mod:`falsify.tasks.cascade_forget`). """ refuted: List[str] = field(default_factory=list) invalidated: List[str] = field(default_factory=list) weakened: List[str] = field(default_factory=list) epoch: int = 0 @property def affected(self) -> List[str]: return list(dict.fromkeys(self.refuted + self.invalidated)) async def _next_epoch() -> int: """Return a monotonically increasing revision epoch. We derive it from the current maximum ``truth_epoch`` present on any node so the counter survives restarts (state is persisted on nodes). Falls back to 1. """ try: nodes, _edges = await graph_ops.load_graph() max_epoch = 0 for _nid, props in nodes: ep = props.get("truth_epoch") if isinstance(ep, int) and ep > max_epoch: max_epoch = ep return max_epoch + 1 except Exception as exc: # pragma: no cover - defensive logger.debug("epoch derivation failed (%s); defaulting to 1", exc) return 1 async def propagate_refutation( refuted_evidence_ids: List[str], epoch: Optional[int] = None, ) -> PropagationResult: """Refute the given evidence and cascade the consequence forward. Args: refuted_evidence_ids: evidence node ids directly contradicted by a new fact. epoch: optional explicit revision epoch; if omitted a fresh one is derived. Returns: A :class:`PropagationResult` describing what changed. All state changes are persisted on the graph nodes via ``set_node_truth_state`` (so they survive a process restart — the basis of cross-session belief revision). Algorithm — grounded least-fixpoint justification ------------------------------------------------- A conclusion is *justified* only if it has a **critical** ``depends_on`` support chain that bottoms out in a still-alive node. We therefore: 1. Mark each seed evidence ``refuted``. 2. Build the "dead" set = seeds plus anything already refuted / invalidated / superseded from prior revisions. 3. Compute the GROUNDED set as a least fixpoint: a node is grounded if it is not dead and either (a) it has no critical ``depends_on`` edges (a base node — evidence, or a conclusion resting only on non-critical support) or (b) at least one of its critical dependencies is itself grounded. Iterate to convergence. 4. Every *conclusion* (a node that is the source of a ``depends_on`` edge) that is currently alive but **not** grounded is ``invalidated``. 5. A conclusion that survives (stays grounded) yet lost some dependency to the dead set is merely ``weakened`` (confidence decayed). This single formulation is correct for chains, diamonds (survives while any critical alternative is grounded), partial/non-critical refutation, **and cycles** (a mutually-supporting loop with no grounded base is not justified, so it collapses) — the fixpoint terminates because ``grounded`` only ever grows. """ result = PropagationResult(epoch=epoch if epoch is not None else await _next_epoch()) seeds = [str(e) for e in refuted_evidence_ids if e] if not seeds: logger.info("propagate_refutation called with no seeds; nothing to do") return result nodes, edges = await graph_ops.load_graph() node_ids = [str(nid) for nid, _p in nodes] # 1) seed refutations (persisted) for ev_id in seeds: await graph_ops.set_state(ev_id, TruthState.REFUTED, result.epoch) await graph_ops.set_weight(ev_id, 0.0) result.refuted.append(ev_id) logger.info("refuted evidence %s", ev_id) # 2) dead set = seeds + already-dead-from-prior-revisions truth = await graph_ops.get_truth(node_ids) dead: Set[str] = set(seeds) for nid in node_ids: alignment = truth.get(nid, [TruthState.ALIVE.value]) if any(s in _DEAD_STATES or s == TruthState.SUPERSEDED.value for s in alignment): dead.add(nid) # Dependency structure: node -> critical / all depends_on targets. conclusions: Set[str] = set() critical_targets: Dict[str, Set[str]] = {} all_targets: Dict[str, Set[str]] = {} for src, dst, rel, props in edges: if rel not in DEPENDENCY_EDGE_TYPES: continue s, d = str(src), str(dst) conclusions.add(s) all_targets.setdefault(s, set()).add(d) if is_critical_dependency(rel, props): critical_targets.setdefault(s, set()).add(d) def _has_critical(n: str) -> bool: return bool(critical_targets.get(n)) # 3) grounded least fixpoint grounded: Set[str] = {nid for nid in node_ids if nid not in dead and not _has_critical(nid)} changed = True while changed: changed = False for c in conclusions: if c in grounded or c in dead: continue if critical_targets.get(c, set()) & grounded: grounded.add(c) changed = True # 4) invalidate currently-alive conclusions that lost grounding for c in conclusions: if c in grounded: continue alignment = truth.get(c, [TruthState.ALIVE.value]) if TruthState.ALIVE.value not in alignment: continue # already dead in a prior revision; don't re-report await graph_ops.set_state(c, TruthState.INVALIDATED, result.epoch) await graph_ops.set_weight(c, 0.0) result.invalidated.append(c) logger.info("invalidated conclusion %s (lost grounded critical support)", c) # 5) weaken survivors that lost some dependency to the dead set for c in conclusions: if c not in grounded: continue if all_targets.get(c, set()) & dead: result.weakened.append(c) await graph_ops.set_weight(c, 0.3) logger.info("weakened conclusion %s (lost a dependency but stays grounded)", c) logger.info( "propagation done: refuted=%d invalidated=%d weakened=%d epoch=%d", len(result.refuted), len(result.invalidated), len(result.weakened), result.epoch, ) return result async def promote_competing_hypothesis( refuted_evidence_ids: List[str], epoch: int, ) -> Dict[str, str]: """Demote hypotheses whose support just died; promote the strongest survivor. A hypothesis is ``superseded`` when every ``supports`` Evidence pointing at it is now dead. Among the hypotheses still holding at least one alive ``supports`` edge, the one with the greatest summed support ``weight`` is promoted (its feedback weight is boosted) and becomes the new frontier answer. Returns a dict mapping hypothesis id -> action (``"superseded"`` / ``"promoted"``). """ actions: Dict[str, str] = {} _nodes, edges = await graph_ops.load_graph() # Collect hypotheses that are the target of any supports edge. supports_edges = [(s, d, p) for (s, d, r, p) in edges if r == SUPPORTS] hypothesis_ids = {str(d) for (_s, d, _p) in supports_edges} if not hypothesis_ids: return actions # Determine current dead evidence set (seeds + anything already refuted/invalidated). all_ids = list({str(s) for (s, _d, _p) in supports_edges} | {str(e) for e in refuted_evidence_ids}) truth = await graph_ops.get_truth(all_ids) def _is_dead(node_id: str) -> bool: alignment = truth.get(str(node_id), [TruthState.ALIVE.value]) return any(state in _DEAD_STATES for state in alignment) or str(node_id) in { str(e) for e in refuted_evidence_ids } # Score each hypothesis by its surviving support. live_support: Dict[str, float] = {} for hyp_id in hypothesis_ids: total = 0.0 for (src, dst, props) in supports_edges: if str(dst) != hyp_id: continue if _is_dead(src): continue total += float(props.get("weight", 0.5)) live_support[hyp_id] = total # Demote hypotheses with zero surviving support. for hyp_id, score in live_support.items(): if score <= 0.0: await graph_ops.set_state(hyp_id, TruthState.SUPERSEDED, epoch) await graph_ops.set_weight(hyp_id, 0.0) actions[hyp_id] = "superseded" logger.info("superseded hypothesis %s (no surviving support)", hyp_id) # Promote the strongest surviving hypothesis, if any. survivors = {h: s for h, s in live_support.items() if s > 0.0} if survivors: winner = max(survivors, key=survivors.get) await graph_ops.set_weight(winner, 1.0) actions[winner] = "promoted" logger.info("promoted hypothesis %s (support=%.2f) as new frontier", winner, survivors[winner]) return actions