code/encompassment_engine.py

"""
Encompassment Engine - K6 as the CEO of knowledge.

K6 is not a counter. K6 is the central index that:
  1. Registers every cell's pattern fingerprint
  2. Computes centrality of each cell in the knowledge graph
  3. Decides when births are warranted (new patterns)
  4. Guides retrieval of relevant cells
  5. Serves as the "summary dimension" of K1-K5
"""
from collections import defaultdict
from typing import Dict, List, Optional
import math


class EncompassmentEngine:
    def __init__(self, max_index_size=10000):
        self.pattern_index: Dict[str, List[str]] = defaultdict(list)
        self.centrality: Dict[str, float] = {}
        self.cell_metadata: Dict[str, dict] = {}
        self.birth_history: List[dict] = []
        self.max_index_size = max_index_size
        self._total_registrations = 0

    def register(self, cell_id: str, fp: str, other_K: List[float]):
        if not fp:
            return
        if cell_id in self.cell_metadata:
            old_fp = self.cell_metadata[cell_id].get("fp")
            if old_fp == fp:
                self._recompute_centrality(cell_id)
                return
            if old_fp and cell_id in self.pattern_index.get(old_fp, []):
                self.pattern_index[old_fp].remove(cell_id)

        self.pattern_index[fp].append(cell_id)
        K1, K2, K3, K4, K5 = other_K[:5]
        self.cell_metadata[cell_id] = {
            "fp": fp, "K1": K1, "K3": K3, "K5": K5, "K2": K2, "K4": K4,
        }
        self._recompute_centrality(cell_id)
        self._total_registrations += 1

        if len(self.pattern_index) > self.max_index_size:
            self._prune()

    def _recompute_centrality(self, cell_id: str):
        meta = self.cell_metadata.get(cell_id)
        if not meta:
            self.centrality[cell_id] = 0.0
            return
        fp = meta["fp"]
        n_sharing = len(self.pattern_index.get(fp, []))
        quality = (meta["K1"] + meta["K3"] + meta["K5"]) / 3.0
        spread = min(1.0, math.log1p(n_sharing) / math.log1p(20))
        c = 0.6 * quality + 0.4 * spread
        self.centrality[cell_id] = max(0.0, min(1.0, c))

    def _prune(self):
        keep = {}
        for fp, members in self.pattern_index.items():
            if len(members) >= 2:
                keep[fp] = members
            else:
                for m in members:
                    if self.centrality.get(m, 0.0) > 0.7:
                        keep[fp] = members
                        break
        self.pattern_index = defaultdict(list, keep)

    def K6(self, cell_id: str) -> float:
        return float(self.centrality.get(cell_id, 0.0))

    def K6_array(self, cell_ids: List[str]) -> List[float]:
        return [self.K6(c) for c in cell_ids]

    def should_birth(self, fp: str, hamming_fn, threshold: float = 0.55) -> bool:
        if not fp:
            return False
        if fp in self.pattern_index and len(self.pattern_index[fp]) > 0:
            return False
        for existing_fp in self.pattern_index:
            try:
                sim = hamming_fn(fp, existing_fp)
            except Exception:
                continue
            if sim > threshold:
                return False
        return True

    def record_birth(self, cell_id: str, parent_id: str, fp: str, step: int):
        self.birth_history.append({
            "cell_id": cell_id, "parent_id": parent_id,
            "fp": fp, "step": step,
        })

    def guide_retrieval(self, query_fp: str, hamming_fn, top_k: int = 5) -> List[str]:
        if not query_fp:
            return []
        scores = []
        for fp, members in self.pattern_index.items():
            try:
                sim = hamming_fn(query_fp, fp)
            except Exception:
                sim = 0.0
            if sim < 0.1:
                continue
            for cid in members:
                c = self.centrality.get(cid, 0.0)
                scores.append((sim * (0.5 + 0.5 * c), cid))
        scores.sort(reverse=True)
        return [cid for _, cid in scores[:top_k]]

    def summary(self) -> dict:
        n_patterns = len(self.pattern_index)
        n_cells = len(self.cell_metadata)
        if self.centrality:
            c_vals = list(self.centrality.values())
            c_min, c_max = min(c_vals), max(c_vals)
            c_mean = sum(c_vals) / len(c_vals)
        else:
            c_min = c_max = c_mean = 0.0
        return {
            "n_patterns": n_patterns,
            "n_cells_indexed": n_cells,
            "centrality_min": round(c_min, 4),
            "centrality_max": round(c_max, 4),
            "centrality_mean": round(c_mean, 4),
            "births_total": len(self.birth_history),
            "total_registrations": self._total_registrations,
        }


if __name__ == "__main__":
    def fake_hamming(a, b):
        if not a or not b:
            return 0.0
        n = min(len(a), len(b))
        return sum(1 for i in range(n) if a[i] == b[i]) / n

    e = EncompassmentEngine()
    e.register("cell-1", "aaaaaaaaaa", [0.9, 0, 0.2, 0, 0.9])
    e.register("cell-2", "bbbbbbbbbb", [0.8, 0, 0.3, 0, 0.85])
    e.register("cell-3", "aaaaaaaaaa", [0.95, 0, 0.1, 0, 0.95])
    print("K6 cell-1:", e.K6("cell-1"))
    print("K6 cell-2:", e.K6("cell-2"))
    print("K6 cell-3:", e.K6("cell-3"))
    print("should_birth(cccc):", e.should_birth("cccccccccc", fake_hamming, threshold=0.55))
    print("should_birth(aaaa):", e.should_birth("aaaaaaaaaa", fake_hamming, threshold=0.55))
    print("retrieval:", e.guide_retrieval("aaaaaaaaaa", fake_hamming, top_k=2))
    print("summary:", e.summary())