falsify/graph_ops.py

"""
Low-level graph operations for FALSIFY.

This module is the single, verified surface between FALSIFY's belief logic and
Cognee's graph/vector engines. Every engine call used elsewhere in the project goes
through a helper here, so the (few) places that touch Cognee internals are auditable.

Verified Cognee API shapes (grep-confirmed against /workspaces/cognee — see REQUIREMENTS.md §0):

    ge = await get_graph_engine()                     # ASYNC factory
    nodes, edges = await ge.get_graph_data()          # ([(id, props)], [(src, dst, rel, props)])
    nodes, edges = await ge.get_neighborhood(ids, depth=, edge_types=)
    await ge.add_edge(from_node, to_node, relationship_name, edge_properties={})
    await ge.set_node_truth_state({id: {"truth_alignment": [...], "truth_epoch": N}})
    state = await ge.get_node_truth_state([ids])      # {id: {"truth_alignment": [...], ...}}
    await ge.set_node_feedback_weights({id: 0.0})
    await ge.delete_nodes([ids])

    ve = get_vector_engine()                          # SYNC factory
    await ve.delete_data_points(collection_name, [uuids])
    hits = await ve.search(collection_name, query_text=, limit=, include_payload=)
"""

from __future__ import annotations

import logging
from typing import Any, Dict, List, Optional, Set, Tuple

from cognee.infrastructure.databases.graph import get_graph_engine
from cognee.infrastructure.databases.vector import get_vector_engine

from falsify import events
from falsify.models import TruthState

logger = logging.getLogger("falsify.graph_ops")

# A parsed edge: (source_id, target_id, relationship_name, properties)
Edge = Tuple[str, str, str, Dict[str, Any]]
# A parsed node: (node_id, properties)
GraphNode = Tuple[str, Dict[str, Any]]


async def load_graph() -> Tuple[List[GraphNode], List[Edge]]:
    """Return the full graph as ``(nodes, edges)``.

    Nodes are ``(id, props)`` tuples; edges are ``(src, dst, rel, props)`` tuples.
    IDs are normalized to ``str`` so they can be used as dict keys regardless of
    whether the adapter returns ``UUID`` or ``str``.
    """
    ge = await get_graph_engine()
    raw_nodes, raw_edges = await ge.get_graph_data()

    nodes: List[GraphNode] = [(str(nid), props or {}) for nid, props in raw_nodes]

    edges: List[Edge] = []
    for e in raw_edges:
        # Adapters return 4-tuples (src, dst, rel, props); be defensive about arity.
        if len(e) >= 4:
            src, dst, rel, props = e[0], e[1], e[2], e[3]
        elif len(e) == 3:
            src, dst, rel, props = e[0], e[1], e[2], {}
        else:  # pragma: no cover - unexpected adapter shape
            continue
        edges.append((str(src), str(dst), str(rel), props or {}))
    return nodes, edges


async def get_truth(node_ids: List[str]) -> Dict[str, List[str]]:
    """Return ``{node_id: truth_alignment_list}`` for the given ids.

    A node with no stored truth state is treated as ``["alive"]`` (the default).
    """
    if not node_ids:
        return {}
    ge = await get_graph_engine()
    ids = [str(n) for n in node_ids]
    try:
        raw = await ge.get_node_truth_state(ids)
    except Exception as exc:  # adapter may not have state yet
        logger.debug("get_node_truth_state failed (%s); defaulting to alive", exc)
        raw = {}
    out: Dict[str, List[str]] = {}
    for nid in ids:
        entry = (raw or {}).get(nid) or (raw or {}).get(str(nid))
        alignment = (entry or {}).get("truth_alignment") if entry else None
        out[nid] = list(alignment) if alignment else [TruthState.ALIVE.value]
    return out


async def is_alive(node_id: str) -> bool:
    """True iff the node's truth_alignment currently contains ``alive``."""
    state = await get_truth([str(node_id)])
    return TruthState.ALIVE.value in state.get(str(node_id), [TruthState.ALIVE.value])


async def set_state(node_id: str, state: TruthState, epoch: int) -> None:
    """Persist ``truth_alignment=[state]`` + ``truth_epoch=epoch`` on a node.

    Truth state is stored ON the graph node, so it survives a process restart — the
    basis of FALSIFY's cross-session persistence guarantee.
    """
    ge = await get_graph_engine()
    value = state.value if isinstance(state, TruthState) else str(state)
    try:
        await ge.set_node_truth_state(
            {str(node_id): {"truth_alignment": [value], "truth_epoch": int(epoch)}}
        )
    except Exception as exc:
        logger.error("set_node_truth_state failed for %s -> %s: %s", node_id, value, exc)
        raise
    # Reaching here means the write succeeded (the except re-raises). Tell any live
    # UI so the node can animate its state change; no-op under the CLI.
    await events.emit_state_change(str(node_id), value, int(epoch))


async def set_weight(node_id: str, weight: float) -> None:
    """Set a node's feedback weight (confidence/health signal). Best-effort."""
    ge = await get_graph_engine()
    try:
        await ge.set_node_feedback_weights({str(node_id): float(weight)})
    except Exception as exc:  # non-fatal: weight is a secondary signal
        logger.debug("set_node_feedback_weights failed for %s: %s", node_id, exc)


async def add_edge(src: str, dst: str, rel: str, props: Optional[Dict[str, Any]] = None) -> None:
    """Add a directed edge ``src --rel--> dst`` with optional properties."""
    ge = await get_graph_engine()
    await ge.add_edge(str(src), str(dst), rel, props or {})


async def delete_from_both_stores(node_ids: List[str], collections: List[str]) -> int:
    """Hard-delete nodes from the graph AND their rows from vector collections.

    Returns the number of node ids deleted. Vector deletion is attempted per
    collection and is best-effort (a node may not live in every collection).
    """
    if not node_ids:
        return 0
    ids = [str(n) for n in node_ids]

    ge = await get_graph_engine()
    try:
        await ge.delete_nodes(ids)
    except Exception as exc:
        logger.error("delete_nodes failed: %s", exc)
        raise
    # Graph delete succeeded — animate each node's forget-dissolve in any live UI.
    for nid in ids:
        await events.emit_forgotten(nid)

    ve = get_vector_engine()
    for collection in collections:
        try:
            await ve.delete_data_points(collection, ids)
        except Exception as exc:  # collection may not exist / id not present
            logger.debug("delete_data_points(%s) best-effort skip: %s", collection, exc)
    return len(ids)


# --------------------------------------------------------------------------- #
# Adjacency helpers (built from a single load_graph() snapshot)
# --------------------------------------------------------------------------- #


def dependents_of(evidence_id: str, edges: List[Edge], rel: str) -> List[str]:
    """Return source ids of ``rel`` edges pointing INTO ``evidence_id``.

    For ``depends_on`` (Conclusion -> Evidence) this yields the Conclusions that
    depend on the given Evidence — i.e. the forward-cascade victims.
    """
    tid = str(evidence_id)
    return [src for (src, dst, r, _p) in edges if r == rel and str(dst) == tid]


def incoming(node_id: str, edges: List[Edge], rel: str) -> List[Tuple[str, Dict[str, Any]]]:
    """Return ``[(source_id, props)]`` for ``rel`` edges pointing into ``node_id``."""
    tid = str(node_id)
    return [(src, p) for (src, dst, r, p) in edges if r == rel and str(dst) == tid]


def outgoing(node_id: str, edges: List[Edge], rel: str) -> List[Tuple[str, Dict[str, Any]]]:
    """Return ``[(target_id, props)]`` for ``rel`` edges leaving ``node_id``."""
    sid = str(node_id)
    return [(dst, p) for (src, dst, r, p) in edges if r == rel and str(src) == sid]


def node_label(props: Dict[str, Any]) -> str:
    """Best-effort human label for a node from its properties."""
    for key in ("statement", "claim", "question", "text", "name"):
        if props.get(key):
            return str(props[key])
    return props.get("id", "?")


async def flush_and_release() -> None:
    """Checkpoint the graph WAL and evict the engine from Cognee's cache.

    After this call, the next graph operation re-opens the on-disk store from
    scratch — so a subsequent read is a genuine *cold* read, which is what makes
    the web UI's persistence proof (``GET /api/verify``) honest rather than a
    reflection of a warm in-memory handle. Every step is best-effort: on an
    adapter that lacks ``checkpoint`` (e.g. the test FakeGraph), or if Cognee's
    eviction internals move, we log and continue rather than break the demo.
    """
    ge = await get_graph_engine()
    try:
        if hasattr(ge, "checkpoint"):
            await ge.checkpoint()
    except Exception as exc:
        logger.debug("checkpoint best-effort skip: %s", exc)
    try:
        from cognee.infrastructure.databases.graph.get_graph_engine import (
            evict_graph_engine,
        )
        from cognee.infrastructure.databases.graph.config import get_graph_config

        evict_graph_engine(**get_graph_config().to_hashable_dict())
    except Exception as exc:
        logger.debug("evict_graph_engine best-effort skip: %s", exc)