backend/hypergraph/engine.py

"""rustworkx-backed Living Mycelial Hypergraph."""
import rustworkx as rx
from typing import Dict
from ..schema import HypergraphDelta


class Hypergraph:
    def __init__(self):
        self.g = rx.PyDiGraph(multigraph=True)
        self._idx: Dict[str, int] = {}

    def apply_delta(self, delta: HypergraphDelta):
        for n in delta.nodes_added:
            if n.node_id in self._idx:
                continue
            i = self.g.add_node({
                "node_id": n.node_id, "node_type": n.node_type, "label": n.label,
                "payload": n.payload, "embedding_hint": n.embedding_hint,
            })
            self._idx[n.node_id] = i
        for u in delta.nodes_updated:
            if u.node_id in self._idx:
                d = self.g[self._idx[u.node_id]]
                d.update(u.fields_changed)
        for e in delta.edges_added:
            s = self._idx.get(e.source_node_id)
            t = self._idx.get(e.target_node_id)
            if s is None or t is None:
                continue
            self.g.add_edge(s, t, {
                "edge_id": e.edge_id, "edge_type": e.edge_type,
                "weight": e.weight, "payload": e.payload,
            })

    def conflict_centrality(self) -> Dict[str, float]:
        try:
            bc = rx.betweenness_centrality(self.g)
            return {self.g[i]["node_id"]: v for i, v in enumerate(bc)}
        except Exception:
            return {}

    def context_summary(self, max_nodes: int = 12) -> str:
        items = []
        for i in list(self.g.node_indexes())[-max_nodes:]:
            n = self.g[i]
            items.append(f"{n['label']}({n['node_type']})")
        return "; ".join(items)

    def node_count(self) -> int: return self.g.num_nodes()
    def edge_count(self) -> int: return self.g.num_edges()