cognee-hackathon-project-main/falsify/tasks/propagate_refutation.py

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