pl/core.py

"""
pl/core.py — Propagation Logic Core
=====================================

P / G → Q

The single primitive operator of Propagation Logic.

A loaded pattern P = (v_P, L_P) propagates through gradient field G
in context C = (Γ_C, θ_C) to produce updated pattern Q.

Everything in this module is derived from that operator.
Nothing is assumed that is not forced by the mechanism.

Key insight (PL v13, Section 2.6):
    G is not a different kind of thing from P.
    G is P occupying the gradient role contextually.
    All G is P post-boundary-imposition.
    There is no view from outside.
    This file is also P / G → Q.
"""

from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union
import math


# =============================================================================
# LOADED PATTERN
# =============================================================================

@dataclass
class Pattern:
    """
    P = (v_P, L_P)

    Definition 2.1 (PL v13):
        P = (v_P, H_P) with v_P ∈ V, H_P a propagation history.
        V is the carrier set.
        L_P = |H_P| ≥ 0 is the informational load.

    v_P : designation component — what the pattern currently designates.
          In logic: 0 or 1.
          In arithmetic: a number.
          In language: a token or sentence.
          In physics: a field value.

    L_P : informational load — magnitude of accumulated propagation history.
          L_P = 0 : seed state. No history. Propagates freely.
          L_P > 0 : loaded. Has been through gradient fields.
                    The more it has been through, the harder to propagate further.

    history : the full qualitative record (L_P is its magnitude).
              Conceptually rich; computationally we track both.
    """
    v: Any
    L: float = 0.0
    history: List[str] = field(default_factory=list)

    def __repr__(self) -> str:
        h = " → ".join(self.history[-3:]) if self.history else "∅"
        return f"P(v={self.v!r}, L={self.L:.3f}, hist=[{h}])"

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Pattern):
            return False
        return self.v == other.v and abs(self.L - other.L) < 1e-9

    def __hash__(self) -> int:
        return hash((repr(self.v), round(self.L, 9)))

    def is_seed(self) -> bool:
        """L_P = 0: no propagation history."""
        return self.L == 0.0

    def copy_with(self, v=None, L=None, history=None) -> "Pattern":
        return Pattern(
            v=v if v is not None else self.v,
            L=L if L is not None else self.L,
            history=history if history is not None else self.history.copy(),
        )


def seed(v: Any) -> Pattern:
    """Convenience: create a seed pattern with no history."""
    return Pattern(v=v, L=0.0)


# =============================================================================
# CONTEXT
# =============================================================================

@dataclass
class Context:
    """
    C = (Γ_C, θ_C)

    Definition 2.2 (PL v13):
        A context is a pair C = (Γ_C, θ_C) where:
            Γ_C : the set of gradient fields available in C
            θ_C : the coherence threshold

    Key derived quantities:
        support(P, C) = min(L_P, θ_C)
        demand(P, C)  = max(0, L_P − θ_C)

        demand = 0  → coherent (pattern is fully supported)
        demand > 0  → incoherent (pattern needs more context than available)

    Theorem 2.1 (the propagation rate theorem):
        Among incoherent patterns, rate ∝ 1/L_P.
        Simpler patterns propagate faster.
        This is Zipf's law, natural selection, why e^x is its own derivative —
        all one theorem.
    """
    gradients: List["Gradient"]
    theta: float = 1.0
    name: str = "C"

    def support(self, p: Pattern) -> float:
        """support(P, C) = min(L_P, θ_C)"""
        return min(p.L, self.theta)

    def demand(self, p: Pattern) -> float:
        """demand(P, C) = max(0, L_P − θ_C)"""
        return max(0.0, p.L - self.theta)

    def is_coherent(self, p: Pattern) -> bool:
        """demand = 0: pattern is fully supported by context."""
        return self.demand(p) == 0.0

    def is_valid(self, p: Pattern, designated: Set = None) -> bool:
        """
        valid = designated AND coherent.
        designated: values that count as "true" in this carrier.
        Default: {1, True} (classical logic convention).
        """
        if designated is None:
            designated = {1, True, 1.0}
        return p.v in designated and self.is_coherent(p)

    def propagation_rate(self, p: Pattern) -> float:
        """
        Theorem 2.1: rate = 1/L_P for incoherent patterns.
        Rate = inf for coherent patterns (already there).
        """
        if self.is_coherent(p):
            return float("inf")
        return 1.0 / p.L if p.L > 0 else float("inf")


# =============================================================================
# GRADIENT
# =============================================================================

class Gradient:
    """
    G — a gradient field.

    G is P occupying the gradient role contextually (PL v13, Section 2.6).
    Every G has loaded history. Every G was constituted by prior propagation.
    G can become P in a higher-order event.

    The propagation event:
        P / G → Q
        v_Q = transform(v_P)          — designation changes
        L_Q = L_P + cost(P)           — load accumulates
        H_Q = H_P + [G.name]          — history records this gradient

    cost: by default 1.0 per propagation step.
          Zero-cost gradients (identity, observation) can be specified.
          Variable-cost gradients can depend on P.
    """

    def __init__(
        self,
        name: str,
        transform: Callable[[Any], Any],
        cost: Union[float, Callable[["Pattern"], float]] = 1.0,
        domain: Optional[Set] = None,
        description: str = "",
    ):
        self.name = name
        self._transform = transform
        self._cost_fn = (cost if callable(cost) else (lambda p, c=cost: c))
        self.domain = domain          # None = any carrier
        self.description = description

    def transform(self, v: Any) -> Any:
        """Apply the designation transformation."""
        return self._transform(v)

    def cost(self, p: Pattern) -> float:
        """Load cost for propagating this pattern through this gradient."""
        return self._cost_fn(p)

    def propagate(self, p: Pattern) -> Pattern:
        """
        P / G → Q

        The primitive operation. Everything else is derived from this.
        """
        v_Q = self._transform(p.v)
        L_Q = p.L + self._cost_fn(p)
        hist_Q = p.history + [self.name]
        return Pattern(v=v_Q, L=L_Q, history=hist_Q)

    def __call__(self, p: Pattern) -> Pattern:
        return self.propagate(p)

    def __repr__(self) -> str:
        return f"G[{self.name}]"

    def is_closed_on(self, carrier: Set) -> Tuple[bool, List]:
        """
        Does this gradient keep the carrier closed?
        Returns (is_closed, violations)
        where violations = [(v_in, v_out_of_carrier), ...]
        """
        violations = []
        for v in carrier:
            out = self._transform(v)
            if out not in carrier:
                violations.append((v, out))
        return len(violations) == 0, violations

    def fixed_points(self, carrier: Set) -> List:
        """Values v ∈ V where G(v) = v."""
        return [v for v in carrier if self._transform(v) == v]

    def orbit(self, v: Any, max_steps: int = 64) -> List:
        """
        The orbit of v under repeated application of G.
        Returns the full cycle if found, truncated at max_steps otherwise.

        If the orbit escapes the carrier (transform raises ValueError),
        returns the visited list so far — indicating no cycle within V.
        This is the correct result for closure violations:
        orbits that exit V have no cycle within V.
        """
        visited = [v]
        current = v
        for _ in range(max_steps):
            try:
                current = self._transform(current)
            except (ValueError, KeyError):
                # Orbit has escaped the carrier — no cycle within V.
                return visited
            if current == v:
                return visited   # full cycle
            visited.append(current)
        return visited           # truncated (no cycle found within max_steps)

    def cycle_length(self, v: Any, max_steps: int = 64) -> Optional[int]:
        """
        Length of the orbit cycle. None if no cycle found.
        None is also correct when the orbit escapes the carrier (closure violation).
        """
        o = self.orbit(v, max_steps)
        if not o:
            return None
        current = o[-1]
        try:
            next_v = self._transform(current)
        except (ValueError, KeyError):
            return None   # orbit escapes carrier, no cycle
        if next_v == o[0]:
            return len(o)
        return None


# =============================================================================
# GRADIENT FAMILIES
# Standard gradient families for known carriers.
# Each family is itself a Pattern in a higher-order carrier.
# =============================================================================

# ── Classical Logic: V = {0, 1} ────────────────────────────────────────────

def G_neg() -> Gradient:
    """
    Classical negation on {0, 1}.
    G_neg flips the designation but does NOT change the load.
    Negating "all dogs bark" costs the same as asserting it.
    """
    return Gradient(
        name="neg",
        transform=lambda v: 1 - v,
        domain={0, 1},
        description="Classical negation: v → 1 - v",
    )

def G_and() -> Gradient:
    """
    Conjunction as a binary gradient.
    Requires tuple input (v_P, v_Q); returns conjunction value.
    """
    return Gradient(
        name="and",
        transform=lambda v: int(v[0] and v[1]) if isinstance(v, tuple) else v,
        domain=None,
        description="Classical conjunction",
    )

def G_or() -> Gradient:
    """Disjunction."""
    return Gradient(
        name="or",
        transform=lambda v: int(v[0] or v[1]) if isinstance(v, tuple) else v,
        domain=None,
        description="Classical disjunction",
    )

def G_id() -> Gradient:
    """
    Identity: zero-cost, leaves everything unchanged.
    The simplest gradient. Fixed point of the gradient-family gradient family.
    """
    return Gradient(
        name="id",
        transform=lambda v: v,
        cost=0.0,
        description="Identity: v → v, zero cost",
    )

# ── Fuzzy Logic: V = {0, 0.5, 1} ───────────────────────────────────────────

def G_fuzzy_neg() -> Gradient:
    """
    Fuzzy negation: v → 1 - v.
    Same formula as classical neg, but on a richer carrier.
    On V={0,0.5,1}: 0.5 is a fixed point. Excluded middle fails at 0.5.
    This is DERIVED from the carrier structure, not assumed.
    """
    return Gradient(
        name="fuzzy_neg",
        transform=lambda v: 1 - v,
        domain={0, 0.5, 1},
        description="Fuzzy negation: v → 1-v on {0, 0.5, 1}",
    )

def G_lukasiewicz_and() -> Gradient:
    """Łukasiewicz conjunction: max(0, v[0] + v[1] - 1)."""
    return Gradient(
        name="luk_and",
        transform=lambda v: max(0, v[0] + v[1] - 1) if isinstance(v, tuple) else v,
        description="Łukasiewicz conjunction",
    )

# ── Arithmetic: V = ℕ, ℤ, ℚ, ℝ ─────────────────────────────────────────────

def G_succ() -> Gradient:
    """
    Successor: n → n + 1.
    On ℕ: always closed.
    No fixed points (nothing maps to itself under +1).
    """
    return Gradient(
        name="succ",
        transform=lambda v: v + 1,
        description="Successor: v → v + 1",
    )

def G_pred() -> Gradient:
    """
    Predecessor: n → n - 1.
    On ℕ = {0, 1, 2, ...}: NOT closed (0 → -1 ∉ ℕ).
    Closure violation FORCES extension to ℤ.
    This is how ℕ → ℤ is derived, not assumed.
    """
    return Gradient(
        name="pred",
        transform=lambda v: v - 1,
        description="Predecessor: v → v - 1 (forces ℕ→ℤ extension)",
    )

def G_double() -> Gradient:
    """v → 2v. On ℤ: closed. Reveals even/odd structure."""
    return Gradient(
        name="double",
        transform=lambda v: 2 * v,
        description="Doubling: v → 2v",
    )

def G_halve() -> Gradient:
    """
    v → v/2.
    On ℤ: NOT closed for odd integers (1/2 ∉ ℤ).
    Forces extension to ℚ.
    This is how ℤ → ℚ is derived.
    """
    return Gradient(
        name="halve",
        transform=lambda v: v / 2,
        description="Halving: v → v/2 (forces ℤ→ℚ extension)",
    )

def G_sqrt() -> Gradient:
    """
    v → √v.
    On ℚ⁺: NOT closed (√2 ∉ ℚ).
    Forces extension to ℝ.
    This is how ℚ → ℝ is derived.
    """
    return Gradient(
        name="sqrt",
        transform=lambda v: v ** 0.5,
        description="Square root: v → √v (forces ℚ→ℝ extension)",
    )

def G_neg_sqrt() -> Gradient:
    """
    v → √(-v) for v < 0, else √v.
    On ℝ: NOT closed for negative values.
    Forces extension to ℂ.
    This is how ℝ → ℂ is derived.
    """
    return Gradient(
        name="neg_sqrt",
        transform=lambda v: complex(0, (-v)**0.5) if v < 0 else v**0.5,
        description="Negative sqrt: forces ℝ→ℂ extension",
    )

# ── Modular arithmetic ──────────────────────────────────────────────────────

def G_mod(n: int, op: str = "add1") -> Gradient:
    """Modular arithmetic on ℤ/nℤ."""
    ops = {
        "add1": lambda v: (v + 1) % n,
        "add2": lambda v: (v + 2) % n,
        "neg":  lambda v: (-v) % n,
        "double": lambda v: (2 * v) % n,
    }
    if op not in ops:
        raise ValueError(f"Unknown op {op!r}. Choose from {list(ops)}")
    return Gradient(
        name=f"mod{n}_{op}",
        transform=ops[op],
        domain=set(range(n)),
        description=f"Mod-{n} {op}",
    )

# ── Custom gradient ─────────────────────────────────────────────────────────

def G_custom(name: str, mapping: Dict[Any, Any], cost: float = 1.0) -> Gradient:
    """
    Build a gradient from an explicit mapping.
    Any carrier, any domain. The engine derives what it forces.
    This is the 'novel carrier' entry point for boundary condition extrapolation.
    """
    def _transform(v: Any) -> Any:
        if v not in mapping:
            raise ValueError(
                f"G[{name}]: value {v!r} not in mapping {set(mapping.keys())}. "
                f"Carrier extension may be required."
            )
        return mapping[v]

    return Gradient(
        name=name,
        transform=_transform,
        cost=cost,
        domain=set(mapping.keys()),
        description=f"Custom gradient with mapping {mapping}",
    )


# =============================================================================
# GRADIENT FAMILY REGISTRY
# Standard (V, Γ) configurations and what they force.
# Used by the engine and the data generator.
# =============================================================================

KNOWN_SYSTEMS = {
    "classical_logic": {
        "description": "Classical two-valued logic",
        "carrier": {0, 1},
        "gradients": lambda: [G_neg(), G_id()],
        "designated": {1},
        "forced": ["closure", "involution", "excluded_middle", "double_negation"],
    },
    "three_valued_logic": {
        "description": "Three-valued (Łukasiewicz) logic",
        "carrier": {0, 0.5, 1},
        "gradients": lambda: [G_fuzzy_neg(), G_id()],
        "designated": {1},
        "forced": ["closure", "middle_value_fixed", "excluded_middle_fails"],
    },
    "natural_numbers": {
        "description": "ℕ with successor (closed) and predecessor (not closed)",
        "carrier": set(range(10)),    # finite sample; full ℕ is unbounded
        "gradients": lambda: [G_succ(), G_id()],
        "designated": {1},
        "forced": ["closure_under_succ", "no_fixed_points_succ"],
    },
    "integers_forced": {
        "description": "ℕ ∪ predecessor → forces ℤ",
        "carrier": {0, 1, 2, 3, 4},
        "gradients": lambda: [G_pred(), G_id()],
        "designated": {1},
        "forced": ["closure_violation", "negative_extension_forced"],
    },
    "rationals_forced": {
        "description": "ℤ ∪ halving → forces ℚ",
        "carrier": {-2, -1, 0, 1, 2, 3, 4},
        "gradients": lambda: [G_halve(), G_id()],
        "designated": {1},
        "forced": ["closure_violation", "rational_extension_forced"],
    },
    "Z4": {
        "description": "Cyclic group ℤ/4ℤ",
        "carrier": {0, 1, 2, 3},
        "gradients": lambda: [G_mod(4, "add1"), G_id()],
        "designated": {0},
        "forced": ["closure", "uniform_4_cycle", "no_fixed_points"],
    },
    "Z2": {
        "description": "Cyclic group ℤ/2ℤ (bit flip)",
        "carrier": {0, 1},
        "gradients": lambda: [G_mod(2, "add1"), G_id()],
        "designated": {0},
        "forced": ["closure", "involution"],
    },
}


# =============================================================================
# PROPAGATION CHAIN
# Records a sequence of P / G → Q steps.
# This is the training unit for the mechanism-first model.
# =============================================================================

@dataclass
class PropagationChain:
    """
    A recorded sequence of propagation steps.

    Step 0: P_0 / G_0 → P_1
    Step 1: P_1 / G_1 → P_2
    ...
    Step n: P_n / G_n → P_{n+1}

    This is the fundamental training example:
    not prose about the mechanism, but the mechanism running.
    """
    steps: List[Tuple[Pattern, Gradient, Pattern]] = field(default_factory=list)
    context: Optional[Context] = None
    carrier: Optional[Set] = None

    def add(self, p_in: Pattern, g: Gradient, p_out: Pattern):
        self.steps.append((p_in, g, p_out))

    def run(self, initial: Pattern, gradients: List[Gradient]) -> "PropagationChain":
        """Run a chain of propagation steps from initial pattern."""
        chain = PropagationChain(context=self.context, carrier=self.carrier)
        current = initial
        for g in gradients:
            next_p = g.propagate(current)
            chain.add(current, g, next_p)
            current = next_p
        return chain

    def as_text(self) -> str:
        """
        Render as training text.
        This is the format the LM learns to predict.
        """
        lines = []
        if self.carrier:
            lines.append(f"CARRIER: {sorted(self.carrier, key=str)}")
        for i, (p_in, g, p_out) in enumerate(self.steps):
            lines.append(
                f"STEP {i}: P(v={p_in.v!r}, L={p_in.L:.1f}) / G[{g.name}] "
                f"→ Q(v={p_out.v!r}, L={p_out.L:.1f})"
            )
        return "\n".join(lines)

    def demand_profile(self) -> List[float]:
        """
        How does demand change across the chain?
        Increasing demand: moving away from coherence (incoherence accumulating).
        Decreasing demand: moving toward coherence (gradient is solving something).
        """
        if not self.context:
            return []
        return [self.context.demand(p_out) for _, _, p_out in self.steps]

    def __len__(self) -> int:
        return len(self.steps)

    def __repr__(self) -> str:
        return f"Chain({len(self.steps)} steps)"