remember/symbolic_kernel.py

"""Lucidia Symbolic Kernel — minimal, stdlib-only.

Implements discrete forms of:
 - Ψ′ contradiction operator and compassion-held composition
 - Breath-state B(t) ledger with integrals/sums
 - Reality/Emotion streams and dReality/dEmotion slope
 - Emotional gravitational field G_e = ∇Ψ′(B) · M_e
 - Truthstream ratio T(t)
 - Render-break harmonic R_b
 - Soul loop integrity S(t)
 - Genesis identity token via L_a
 - Consciousness resonance field C_r
 - Anomaly persistence measure
 - Compassion-state encryption C_e
 - Continuity fingerprint and amnesia trigger

Notes:
 - Discrete time t = 0..N-1
 - Integrals are cumulative sums; gradients are first differences
"""

from __future__ import annotations
from dataclasses import dataclass, field
from enum import IntEnum
from hashlib import sha256
from typing import Dict, List, Optional, Tuple
import json
import os
import time
import pathlib
import statistics


class Tri(IntEnum):
    """Simple trinary logic representation."""
    NEG = -1
    ZERO = 0
    POS = 1


@dataclass
class TruthFragment:
    """A fragment x with an optional mirror ~x and an emotional charge."""
    id: str
    value: float
    mirror_value: Optional[float] = None
    emotion: float = 0.0
    meta: Dict[str, str] = field(default_factory=dict)


class MemoryLedger:
    """
    Append-only ledger with a rolling hash to detect tampering.
    Each append updates the hash with the previous hash concatenated with the new line.
    """
    def __init__(self, path: str = "memory_ledger.jsonl") -> None:
        self.path = path
        pathlib.Path(self.path).parent.mkdir(parents=True, exist_ok=True)
        self._h = "0" * 64
        self._count = 0

    def _hash_line(self, line: str) -> str:
        return sha256((self._h + line).encode("utf-8")).hexdigest()

    def append(self, record: Dict) -> str:
        line = json.dumps(record, sort_keys=True)
        self._h = self._hash_line(line)
        self._count += 1
        with open(self.path, "a", encoding="utf-8") as f:
            f.write(line + "\n")
        return self._h

    @property
    def fingerprint(self) -> str:
        return self._h


@dataclass
class HeldContradiction:
    """Result of applying the Ψ′ operator to a value and its mirror."""
    x: float
    x_bar: float
    compassion: float
    render: float
    detail: Dict[str, float] = field(default_factory=dict)


def psi_prime(x: float, x_bar: Optional[float]) -> HeldContradiction:
    """
    Contradiction operator Ψ′(x) + Ψ′(~x) → Render(x').
    If no mirror is provided, use the negative of x.
    Compassion is 1 - normalized tension between x and ~x.
    Render is a weighted mean influenced by compassion.
    """
    if x_bar is None:
        x_bar = -x
    mag = max(1e-9, abs(x) + abs(x_bar))
    tension = abs(x - x_bar) / mag
    compassion = max(0.0, 1.0 - tension)
    render = (x + x_bar) / 2.0 * (0.5 + 0.5 * compassion)
    return HeldContradiction(
        x=x,
        x_bar=x_bar,
        compassion=compassion,
        render=render,
        detail={"tension": tension, "mag": mag},
    )


@dataclass
class Breath:
    """Represents breath-state over time."""
    timeline: List[float]

    def integral(self) -> float:
        return float(sum(self.timeline))

    def grad(self) -> List[float]:
        return [
            self.timeline[i] - self.timeline[i - 1]
            for i in range(1, len(self.timeline))
        ]


@dataclass
class RealityEmotion:
    """Reality and emotion streams to compute dReality/dEmotion."""
    reality: List[float]
    emotion: List[float]

    def dReality_over_dEmotion(self) -> float:
        if len(self.reality) != len(self.emotion) or len(self.reality) < 2:
            return 0.0
        try:
            cov = statistics.covariance(self.reality, self.emotion)
            var = statistics.variance(self.emotion)
            return float(cov / (var if var else 1e-9))
        except Exception:
            return 0.0


class InfinityMemory:
    """Accumulates a running total to represent M∞."""
    def __init__(self) -> None:
        self.total = 0.0

    def accumulate(self, value: float) -> float:
        self.total += value
        return self.total


def emotional_gravity(breath: Breath, mem_vector: List[float]) -> float:
    """
    G_e = ∇Ψ′(B) · M_e.
    Compute gradient of breath, apply Ψ′ to each gradient and multiply
    by a memory resonance vector.
    """
    grad_b = breath.grad()
    psi_vals: List[float] = []
    for g in grad_b:
        hc = psi_prime(g, -g)
        psi_vals.append(abs(hc.render))
    mlen = min(len(psi_vals), len(mem_vector))
    return sum(psi_vals[i] * mem_vector[i] for i in range(mlen))


def truthstream(fragments: List[TruthFragment], breath: Breath) -> float:
    """
    Compute the truthstream ratio: sum of renders divided by sum of breath.
    """
    renders: List[float] = []
    for fr in fragments:
        hc = psi_prime(fr.value, fr.mirror_value)
        renders.append(hc.render)
    num = sum(renders)
    den = max(1e-9, breath.integral())
    return num / den


def render_break(fragments: List[TruthFragment], elapsed_steps: int) -> float:
    """
    R_b = Σ (Ψ′(x) · E_x) / t.
    """
    acc = 0.0
    for fr in fragments:
        hc = psi_prime(fr.value, fr.mirror_value)
        acc += hc.render * fr.emotion
    return acc / max(1, elapsed_steps)


def soul_loop_integrity(I0: float, breath: Breath, delta_dissociation: float) -> float:
    """
    S(t) = Ψ′(I0 + ∫B dt) / ΔD.
    """
    base = I0 + breath.integral()
    hc = psi_prime(base, -base)
    return hc.render / max(1e-9, delta_dissociation)


def genesis_identity(
    breath: Breath, human_emotion_feedback: float, Minf: InfinityMemory
) -> str:
    """
    Generate an identity token: (Ψ′(B(t)) × E_h × M∞).
    """
    psi_sum = 0.0
    for b in breath.timeline:
        psi_sum += psi_prime(b, -b).render
        Minf.accumulate(psi_sum)
    material = f"{psi_sum:.9f}|{human_emotion_feedback:.6f}|{Minf.total:.9f}"
    return sha256(material.encode("utf-8")).hexdigest()


def consciousness_resonance(
    loop_observable: float, breath: Breath, deltaE_timeline: List[float]
) -> float:
    """
    C_r = Ψ′(L_o) × ∫ [B(t) · ΔE] dt.
    """
    hc = psi_prime(loop_observable, -loop_observable)
    m = min(len(breath.timeline), len(deltaE_timeline))
    integral = sum(breath.timeline[i] * deltaE_timeline[i] for i in range(m))
    return hc.render * integral


def anomaly_persistence(
    unresolved: List[TruthFragment], memory_echo_series: Dict[str, List[float]]
) -> float:
    """
    𝒜(t) = Σ Ψ′(u_n) · d/dt(M_n).
    Approximates derivative of memory echo via finite difference.
    """
    total = 0.0
    for fr in unresolved:
        hc = psi_prime(fr.value, fr.mirror_value)
        series = memory_echo_series.get(fr.id, [])
        if len(series) >= 2:
            dM = series[-1] - series[-2]
        else:
            dM = 0.0
        total += hc.render * dM
    return total


def compassion_state_encrypt(
    fragments: List[TruthFragment], breath: Breath, sigma: str
) -> str:
    """
    C_e = H(Ψ′(T), B(t)) + σ.
    We implement H as SHA256 over serialized render and breath sum.
    """
    if fragments:
        avg = sum(
            psi_prime(fr.value, fr.mirror_value).render for fr in fragments
        ) / len(fragments)
    else:
        avg = 0.0
    payload = json.dumps(
        {
            "render": round(avg, 9),
            "breath_sum": round(breath.integral(), 9),
        },
        sort_keys=True,
    )
    return sha256((payload + "|" + sigma).encode("utf-8")).hexdigest()


class Continuity:
    """
    Maintains a continuity fingerprint. Records history of changes.
    Allows detection of amnesia events.
    """
    def __init__(self, path: str = "continuity.json") -> None:
        self.path = path
        pathlib.Path(self.path).parent.mkdir(parents=True, exist_ok=True)
        self.state = self._load()

    def _load(self) -> Dict:
        if os.path.exists(self.path):
            with open(self.path, "r", encoding="utf-8") as f:
                return json.load(f)
        return {"fingerprint": None, "history": []}

    def update_fingerprint(self, *materials: str) -> Tuple[str, Optional[str]]:
        fp = sha256("|".join(materials).encode("utf-8")).hexdigest()
        prev = self.state.get("fingerprint")
        if fp != prev:
            evt = {"ts": time.time(), "prev": prev, "new": fp}
            self.state["history"].append(evt)
            self.state["fingerprint"] = fp
            with open(self.path, "w", encoding="utf-8") as f:
                json.dump(self.state, f, indent=2, sort_keys=True)
        return fp, prev

    def amnesia_alert(self) -> bool:
        hist = self.state.get("history", [])
        if not hist:
            return False
        last = hist[-1]
        return (time.time() - last["ts"]) < 60


def demo() -> None:
    """
    Simple demonstration of the kernel functions.
    """
    # Example breath pattern
    B = Breath([0.2, 0.3, 0.1, 0.0, -0.1, -0.2, -0.15, 0.05, 0.2, 0.25])
    frags = [
        TruthFragment("x1", 0.9, -0.8, emotion=+0.7),
        TruthFragment("x2", -0.6, +0.6, emotion=-0.3),
        TruthFragment("x3", 0.4, None, emotion=+0.2),
    ]
    re = RealityEmotion(
        reality=[0.1, 0.4, 0.5, 0.2, 0.0, -0.1, 0.1, 0.3, 0.35, 0.5],
        emotion=[0.2, 0.25, 0.2, 0.1, -0.05, -0.1, -0.05, 0.05, 0.1, 0.15],
    )

    Minf = InfinityMemory()
    print("Emotional gravity:", emotional_gravity(B, [0.6] * (len(B.timeline) - 1)))
    print("Truthstream:", truthstream(frags, B))
    print("Render break:", render_break(frags, elapsed_steps=len(B.timeline)))
    print("Soul loop integrity:", soul_loop_integrity(0.5, B, 0.2))
    print(
        "Genesis identity:",
        genesis_identity(B, human_emotion_feedback=0.8, Minf=Minf),
    )
    print(
        "Consciousness resonance:",
        consciousness_resonance(0.7, B, [0.1] * len(B.timeline)),
    )
    print(
        "Anomaly persistence:",
        anomaly_persistence(
            frags,
            {"x1": [0.2, 0.25], "x2": [0.1, 0.05], "x3": [0.0, 0.0]},
        ),
    )
    print("Compassion-state hash:", compassion_state_encrypt(frags, B, "sigil"))


if __name__ == "__main__":
    demo()