cognee-hackathon-project-main/falsify/models.py

"""
FALSIFY belief-graph data models.

Every node is a Cognee ``DataPoint`` subclass, so the same object serializes into
the graph DB (Ladybug) and, via its ``Embeddable`` field, into the vector DB
(LanceDB). The vector collection auto-created for a class is ``"{ClassName}_{field}"``
(e.g. ``Evidence_claim`` — the collection the contradiction prefilter searches).

Lifecycle / truth-state note
----------------------------
The *authoritative, persistent* belief state of a node lives on the graph node as
``truth_alignment`` (a list) + ``truth_epoch`` (int), written through the graph
engine (``set_node_truth_state``), not as pydantic model fields. Those props are
what ``recall()`` filters on and what survives a process restart.

The ``truth_state`` model field below is a convenience mirror of the node's initial
state at ingest time (defaults to ``ALIVE``). Do not treat it as the source of
truth after a memify run — always read back with ``get_node_truth_state``.

Dedup across sessions
---------------------
``Hypothesis``, ``Evidence`` and ``Assertion`` mark identity fields with ``Dedup()``.
DataPoint derives a stable UUID5 id from those fields (``DataPoint.id_for``), so
re-adding the same belief in a later session updates the existing node instead of
creating a duplicate — essential for a Session-2 contradiction to land on the exact
node built in Session 1.
"""

from datetime import datetime, timezone
from enum import Enum
from typing import Annotated, List, Optional

from pydantic import Field

from cognee.infrastructure.engine import DataPoint, Dedup, Embeddable, LLMContext


def _now_iso() -> str:
    """Return the current UTC time as an ISO-8601 string (used for node timestamps)."""
    return datetime.now(timezone.utc).isoformat()


class TruthState(str, Enum):
    """Canonical belief-lifecycle states for a FALSIFY graph node.

    The values are the exact strings written into the node's ``truth_alignment``
    list, so a state can be compared directly against what ``get_node_truth_state``
    returns (e.g. ``get_node_truth_state([n])[n]["truth_alignment"] == [TruthState.REFUTED]``).

    States:
    - ``ALIVE``: default; the node participates in recall context.
    - ``REFUTED``: Evidence directly contradicted by a newer fact — the entry point
      of a forward cascade.
    - ``SUPERSEDED``: a node replaced/demoted by a newer competing node; retained as
      provenance, excluded from recall.
    - ``INVALIDATED``: a Conclusion whose critical supporting Evidence chain was
      refuted (forward-cascade victim).
    - ``FORGOTTEN``: an orphaned dead-end scheduled for surgical delete (transient;
      the node is then hard-removed from both graph and vector stores).

    Note: the FALSIFY spec's minimal enum names ALIVE/REFUTED/SUPERSEDED/FORGOTTEN.
    ``INVALIDATED`` is added here because the forward-propagation algorithm
    (REQUIREMENTS §1.2/§1.4) needs a distinct state for cascade-victim Conclusions
    versus directly-refuted Evidence.
    """

    ALIVE = "alive"
    REFUTED = "refuted"
    SUPERSEDED = "superseded"
    INVALIDATED = "invalidated"
    FORGOTTEN = "forgotten"


# --------------------------------------------------------------------------- #
# Edge relationship-name constants (passed as ``relationship_name`` to add_edge)
# --------------------------------------------------------------------------- #

# Conclusion -> Evidence. THE forward-propagation rail. edge_properties={"critical": bool}
DEPENDS_ON = "depends_on"

# Evidence -> Hypothesis. Evidence corroborates a hypothesis. edge_properties={"weight": float}
SUPPORTS = "supports"

# Evidence -> Hypothesis. Evidence contradicts a hypothesis. edge_properties={"weight": float}
# (REQUIREMENTS §1.1 names this edge "refutes"; ``REFUTES`` is provided as an alias.)
CONTRADICTS = "refutes"
REFUTES = CONTRADICTS

# Evidence(new) -> Evidence(old). New fact overrides an old evidence node.
# edge_properties={"confidence": float}
SUPERSEDES = "supersedes"


class InvestigationQuestion(DataPoint):
    """Root node of a belief graph: the research question under investigation.

    Example: "Did Company X know about the defect before the recall?" Everything
    else (hypotheses, evidence, conclusions) hangs off this question via
    ``question_id``.
    """

    question: Annotated[str, Embeddable(), LLMContext()]
    truth_state: TruthState = TruthState.ALIVE
    confidence: float = 1.0
    timestamp: str = Field(default_factory=_now_iso)
    source_id: Optional[str] = None

    metadata: dict = {
        "index_fields": ["question"],
        "identity_fields": ["question"],
    }


class Hypothesis(DataPoint):
    """A candidate explanation competing to answer the InvestigationQuestion.

    Hypotheses gain/lose standing through ``supports``/``refutes`` Evidence edges.
    When a hypothesis' only supporting Evidence is refuted, it is demoted to
    ``SUPERSEDED`` and the rival with the strongest surviving support is promoted.
    """

    statement: Annotated[str, Embeddable(), Dedup(), LLMContext()]
    question_id: str
    status: str = "alive"
    prior: float = 0.5
    truth_state: TruthState = TruthState.ALIVE
    confidence: float = 0.5
    timestamp: str = Field(default_factory=_now_iso)
    source_id: Optional[str] = None

    metadata: dict = {
        "index_fields": ["statement"],
        "identity_fields": ["question_id", "statement"],
    }


class Evidence(DataPoint):
    """A factual claim bearing on one or more hypotheses.

    Evidence is the contradiction entry point: the ``Evidence_claim`` vector
    collection is what the two-gate detector prefilters, and a ``REFUTED`` Evidence
    node is the seed of every forward cascade. ``asserted_at`` is used as the
    tie-break when deciding which of two competing claims supersedes the other
    (newer wins).
    """

    claim: Annotated[str, Embeddable(), Dedup(), LLMContext()]
    source_id: str
    quote: str = ""
    stance: str = "supports"  # "supports" | "refutes"
    asserted_at: str = Field(default_factory=_now_iso)
    truth_state: TruthState = TruthState.ALIVE
    confidence: float = 0.5
    timestamp: str = Field(default_factory=_now_iso)

    metadata: dict = {
        "index_fields": ["claim"],
        "identity_fields": ["source_id", "claim"],
    }


class Conclusion(DataPoint):
    """A derived finding that rests on one or more Evidence nodes.

    A Conclusion ``depends_on`` the Evidence it is built from (edge carries
    ``critical: bool``). When a *critical* dependency is refuted and no alive
    critical alternative remains, the Conclusion is ``INVALIDATED`` by the forward
    cascade; if it then has no surviving consumer it is ``FORGOTTEN`` (hard-deleted).
    """

    statement: Annotated[str, Embeddable(), LLMContext()]
    confidence: float = 0.5
    depends_on_ids: List[str] = Field(default_factory=list)
    truth_state: TruthState = TruthState.ALIVE
    timestamp: str = Field(default_factory=_now_iso)
    source_id: Optional[str] = None

    metadata: dict = {
        "index_fields": ["statement"],
        "identity_fields": ["statement"],
    }


class Assertion(DataPoint):
    """A raw, unclassified incoming claim — e.g. the new fact pasted in Session 2.

    An Assertion is the pre-belief form of an incoming statement before the detector
    decides whether it contradicts/supersedes existing Evidence and materializes a
    proper ``Evidence`` node. It carries the same lifecycle scaffolding as the other
    nodes so it can be reasoned over uniformly.
    """

    text: Annotated[str, Embeddable(), Dedup(), LLMContext()]
    truth_state: TruthState = TruthState.ALIVE
    confidence: float = 0.5
    timestamp: str = Field(default_factory=_now_iso)
    source_id: Optional[str] = None

    metadata: dict = {
        "index_fields": ["text"],
        "identity_fields": ["text"],
    }