tests/test_falsify.py

"""
FALSIFY core-logic tests — belief revision without any API key or database.

Each test builds a small graph via the :class:`~tests.conftest.FakeGraph` fixture,
runs the real cascade/forget code, and asserts on the resulting truth-states. These
cover the correctness cases enumerated in REQUIREMENTS.md §4.3:

    * direct refutation                       (test_direct_refutation)
    * forward cascade invalidates conclusion  (test_forward_cascade_invalidates)
    * diamond / partial refutation            (test_diamond_partial_refutation)
    * non-critical dependency survives         (test_non_critical_dependency_survives)
    * cycle safety (termination)              (test_cycle_safety)
    * surgical forget of orphans only         (test_forget_orphan_keeps_provenance)
    * hypothesis demote/promote               (test_promote_competing_hypothesis)
"""

from __future__ import annotations

import pytest

from falsify.edges import DEPENDS_ON, SUPERSEDES, SUPPORTS
from falsify.models import TruthState
from falsify.tasks.cascade_forget import cascade_forget
from falsify.tasks.propagate_refutation import (
    promote_competing_hypothesis,
    propagate_refutation,
)

pytestmark = pytest.mark.asyncio


async def _alignment(g, nid: str):
    entry = g.truth.get(str(nid))
    return entry["truth_alignment"] if entry else None


async def test_direct_refutation(fake_graph):
    """Refuting an evidence node sets its truth_alignment to ['refuted']."""
    g = fake_graph
    g.add_node("E", claim="the March QA report documents the defect")

    res = await propagate_refutation(["E"])

    assert res.refuted == ["E"]
    assert await _alignment(g, "E") == [TruthState.REFUTED.value]


async def test_forward_cascade_invalidates(fake_graph):
    """A→B→C depends_on chain: refuting the base evidence invalidates the chain.

    Topology (depends_on points Conclusion→Evidence):
        C  --depends_on(critical)-->  B  --depends_on(critical)-->  E(evidence)
    Refuting E must invalidate B, then C.
    """
    g = fake_graph
    g.add_node("E", claim="base evidence")
    g.add_node("B", statement="mid conclusion")
    g.add_node("C", statement="top conclusion")
    g.add_edge("B", "E", DEPENDS_ON, critical=True)
    g.add_edge("C", "B", DEPENDS_ON, critical=True)

    res = await propagate_refutation(["E"])

    assert await _alignment(g, "E") == [TruthState.REFUTED.value]
    assert await _alignment(g, "B") == [TruthState.INVALIDATED.value]
    assert await _alignment(g, "C") == [TruthState.INVALIDATED.value]
    assert set(res.invalidated) == {"B", "C"}


async def test_diamond_partial_refutation(fake_graph):
    """A conclusion with two critical supporters survives losing only one.

    K depends_on E1 (critical) AND E2 (critical). Refuting only E1 must keep K alive
    (E2 still supports it); refuting E2 as well then invalidates K.
    """
    g = fake_graph
    g.add_node("E1", claim="evidence one")
    g.add_node("E2", claim="evidence two")
    g.add_node("K", statement="conclusion on both")
    g.add_edge("K", "E1", DEPENDS_ON, critical=True)
    g.add_edge("K", "E2", DEPENDS_ON, critical=True)

    # First refutation: K keeps an alive critical supporter (E2) -> stays alive.
    await propagate_refutation(["E1"])
    assert await _alignment(g, "E1") == [TruthState.REFUTED.value]
    assert await _alignment(g, "K") == [TruthState.ALIVE.value]

    # Second refutation: K loses its last critical supporter -> invalidated.
    await propagate_refutation(["E2"])
    assert await _alignment(g, "K") == [TruthState.INVALIDATED.value]


async def test_diamond_demo_two_phase(fake_graph):
    """Full --diamond demo scenario: single-dep K dies in phase 1 while the
    diamond K2 survives, then K2 collapses in phase 2.

    Topology mirrors seed.build_diamond_investigation():
        K  --depends_on(critical)--> E_qa                 (single leg)
        K2 --depends_on(critical)--> E_qa                 (diamond leg 1)
        K2 --depends_on(critical)--> E_email              (diamond leg 2)

    This is the discriminating case the naive-cascade strawman gets wrong:
    refuting E_qa must kill K but NOT K2 (E_email still grounds it).
    """
    g = fake_graph
    g.add_node("E_qa", claim="QA report March 2021")
    g.add_node("E_email", claim="supplier email Jan 2021")
    g.add_node("K", statement="knew by March 2021")
    g.add_node("K2", statement="multiple sources confirm pre-recall knowledge")
    g.add_edge("K", "E_qa", DEPENDS_ON, critical=True)
    g.add_edge("K2", "E_qa", DEPENDS_ON, critical=True)
    g.add_edge("K2", "E_email", DEPENDS_ON, critical=True)

    # Phase 1: refute E_qa. K (single leg) dies; K2 survives on E_email.
    res1 = await propagate_refutation(["E_qa"])
    assert await _alignment(g, "E_qa") == [TruthState.REFUTED.value]
    assert await _alignment(g, "K") == [TruthState.INVALIDATED.value]
    assert await _alignment(g, "K2") == [TruthState.ALIVE.value]  # the whole point
    assert "K" in res1.invalidated
    assert "K2" not in res1.invalidated

    # Phase 2: refute E_email. K2 loses its last leg and finally collapses.
    res2 = await propagate_refutation(["E_email"])
    assert await _alignment(g, "E_email") == [TruthState.REFUTED.value]
    assert await _alignment(g, "K2") == [TruthState.INVALIDATED.value]
    assert "K2" in res2.invalidated


async def test_non_critical_dependency_survives(fake_graph):
    """Refuting a NON-critical dependency weakens but does not invalidate."""
    g = fake_graph
    g.add_node("E", claim="soft evidence")
    g.add_node("K", statement="conclusion softly resting on E")
    g.add_edge("K", "E", DEPENDS_ON, critical=False)

    res = await propagate_refutation(["E"])

    assert await _alignment(g, "K") == [TruthState.ALIVE.value]
    assert "K" in res.weakened
    assert "K" not in res.invalidated


async def test_cycle_safety(fake_graph):
    """A cyclic depends_on graph terminates (visited set) and doesn't hang."""
    g = fake_graph
    g.add_node("E", claim="evidence")
    g.add_node("X", statement="X")
    g.add_node("Y", statement="Y")
    # Cycle among conclusions, all critically resting on E and each other.
    g.add_edge("X", "E", DEPENDS_ON, critical=True)
    g.add_edge("Y", "X", DEPENDS_ON, critical=True)
    g.add_edge("X", "Y", DEPENDS_ON, critical=True)

    res = await propagate_refutation(["E"])  # must return, not loop forever

    assert await _alignment(g, "E") == [TruthState.REFUTED.value]
    assert "X" in res.invalidated and "Y" in res.invalidated


async def test_forget_orphan_keeps_provenance(fake_graph):
    """cascade_forget deletes an orphaned invalidated conclusion but keeps the
    refuted evidence that is a supersedes-anchor (provenance tombstone)."""
    g = fake_graph
    g.add_node("E", claim="refuted evidence")
    g.add_node("K", statement="orphaned conclusion")
    g.add_node("NEW", claim="the new fact")
    g.add_edge("K", "E", DEPENDS_ON, critical=True)
    g.add_edge("NEW", "E", SUPERSEDES, confidence=0.9)  # NEW (alive) supersedes E

    # Run the cascade, then forget.
    prop = await propagate_refutation(["E"])
    forget = await cascade_forget(prop.affected)

    # K is orphaned (no alive consumer) -> deleted from both stores.
    assert "K" in forget.forgotten
    assert "K" in g.deleted
    assert g.deleted_from_collections  # vector collections were targeted
    # E is refuted but retained as the supersedes provenance anchor.
    assert "E" in forget.retained_provenance
    assert "E" not in g.deleted


async def test_promote_competing_hypothesis(fake_graph):
    """When A's only support dies, A is superseded and rival B is promoted."""
    g = fake_graph
    g.add_node("E_a", claim="evidence for A")
    g.add_node("E_b", claim="evidence for B")
    g.add_node("A", statement="hypothesis A")
    g.add_node("B", statement="hypothesis B")
    g.add_edge("E_a", "A", SUPPORTS, weight=0.8)
    g.add_edge("E_b", "B", SUPPORTS, weight=0.7)

    # Refute A's evidence, then re-score hypotheses.
    prop = await propagate_refutation(["E_a"])
    actions = await promote_competing_hypothesis(["E_a"], prop.epoch)

    assert actions.get("A") == "superseded"
    assert actions.get("B") == "promoted"
    assert await _alignment(g, "A") == [TruthState.SUPERSEDED.value]