v2 complete: NGC graft, causal energy, auto-expanding codebook, benchmark integration

tensegrity/v2/causal_energy.py

@@ -0,0 +1,180 @@
+"""
+Causal Energy: Pearl's SCMs as energy terms in the unified landscape.
+
+Each SCM contributes a prediction error to the total energy:
+    E_causal(M_k) = Σ_v ||z_v - f_v(z_pa(v))||²
+
+Where:
+    z_v = observed value of variable v
+    f_v(z_pa(v)) = structural equation's prediction from parents
+    pa(v) = parents of v in the causal DAG
+
+Multiple SCMs compete. The model with lowest causal energy provides
+the best explanation. This replaces the v1 causal arena's log-likelihood
+comparison with a unified energy-based comparison.
+
+The causal energy connects to the NGC energy through shared variables:
+if a causal variable maps to an NGC layer's abstract state, then the
+NGC prediction error and the causal prediction error are literally
+the same quantity at different scales of description.
+"""
+
+import numpy as np
+from typing import Dict, List, Optional, Any, Tuple
+from tensegrity.causal.scm import StructuralCausalModel
+
+
+class CausalEnergyTerm:
+    """
+    Computes causal prediction error energy for an SCM.
+    
+    Given observations of some variables, computes how well
+    the SCM's structural equations predict them.
+    """
+    
+    def __init__(self, scm: StructuralCausalModel, precision: float = 1.0):
+        self.scm = scm
+        self.precision = precision
+    
+    def energy(self, observations: Dict[str, int]) -> float:
+        """
+        Compute causal prediction error energy.
+        
+        E = Σ_v (1/2σ²) ||obs_v - predicted_v||²
+        
+        Where predicted_v = E[V | parents of V observed]
+        """
+        total_energy = 0.0
+        order = self.scm.topological_order()
+        
+        for var in order:
+            if var not in observations:
+                continue
+            
+            mech = self.scm.mechanisms[var]
+            parent_vals = {p: observations.get(p, 0) for p in mech.parents}
+            
+            # Expected value under the CPT
+            cpt = mech.cpt
+            config_idx = mech.parent_config_index(parent_vals)
+            probs = cpt[:, config_idx]
+            
+            # Prediction = expected value index
+            expected = np.sum(probs * np.arange(len(probs)))
+            observed = float(observations[var])
+            
+            # Squared prediction error
+            error = (observed - expected) ** 2
+            total_energy += 0.5 * self.precision * error
+        
+        return total_energy
+    
+    def prediction(self, observations: Dict[str, int], 
+                   target: str) -> np.ndarray:
+        """Predict distribution over target given observed parents."""
+        mech = self.scm.mechanisms.get(target)
+        if mech is None:
+            return np.array([1.0])
+        
+        parent_vals = {p: observations.get(p, 0) for p in mech.parents}
+        config_idx = mech.parent_config_index(parent_vals)
+        return mech.cpt[:, config_idx]
+
+
+class CausalArenaV2:
+    """
+    v2 causal arena: SCMs compete via energy, not log-likelihood.
+    
+    Each model is wrapped in a CausalEnergyTerm. The model with
+    lowest energy wins. The tension is the ratio of energies
+    (or equivalently, the softmax distribution over models).
+    
+    This integrates with the unified energy landscape:
+      E_total = E_perception(NGC) + E_memory(Hopfield) + E_causal(arena)
+    
+    Where E_causal = min_k E_causal(M_k) — we use the best model's energy.
+    """
+    
+    def __init__(self, precision: float = 1.0, beta: float = 1.0):
+        """
+        Args:
+            precision: Causal prediction error precision
+            beta: Inverse temperature for model selection softmax
+        """
+        self.models: Dict[str, CausalEnergyTerm] = {}
+        self.beta = beta
+        self.precision = precision
+        self._history: List[Dict[str, float]] = []
+    
+    def register(self, scm: StructuralCausalModel):
+        """Add a competing causal model."""
+        self.models[scm.name] = CausalEnergyTerm(scm, self.precision)
+    
+    def compete(self, observations: Dict[str, int]) -> Dict[str, Any]:
+        """
+        All models compute their causal energy on the observation.
+        Returns energies, posteriors, and tension.
+        """
+        energies = {}
+        for name, term in self.models.items():
+            energies[name] = term.energy(observations)
+        
+        if not energies:
+            return {"winner": None, "tension": 1.0, "energies": {}}
+        
+        # Softmax over negative energies (lower energy = higher weight)
+        vals = np.array(list(energies.values()))
+        neg_e = -self.beta * vals
+        neg_e -= neg_e.max()
+        weights = np.exp(neg_e)
+        weights /= weights.sum()
+        
+        posteriors = dict(zip(energies.keys(), weights.tolist()))
+        
+        # Tension = normalized entropy
+        probs = weights[weights > 0]
+        if len(probs) > 1:
+            entropy = -np.sum(probs * np.log(probs))
+            tension = float(entropy / np.log(len(probs)))
+        else:
+            tension = 0.0
+        
+        winner = min(energies, key=energies.get)
+        best_energy = energies[winner]
+        
+        result = {
+            "winner": winner,
+            "tension": tension,
+            "posteriors": posteriors,
+            "energies": energies,
+            "best_energy": best_energy,
+        }
+        self._history.append(energies)
+        
+        return result
+    
+    def best_energy(self, observations: Dict[str, int]) -> float:
+        """Get the energy of the best-fitting model."""
+        result = self.compete(observations)
+        return result.get("best_energy", 0.0)
+    
+    def update_models(self, observations: Dict[str, int]):
+        """Update all models' parameters from observation (Dirichlet counting)."""
+        for name, term in self.models.items():
+            term.scm.update_from_data([observations])
+    
+    @property
+    def tension(self) -> float:
+        """Current tension (from last competition)."""
+        if not self._history:
+            return 1.0
+        last = self._history[-1]
+        vals = np.array(list(last.values()))
+        neg_e = -self.beta * vals
+        neg_e -= neg_e.max()
+        w = np.exp(neg_e)
+        w /= w.sum()
+        w = w[w > 0]
+        if len(w) > 1:
+            return float(-np.sum(w * np.log(w)) / np.log(len(w)))
+        return 0.0

tensegrity/v2/fhrr.py

@@ -58,11 +58,16 @@ class FHRRCodebook:
         self._labels: Dict[str, int] = {}
     
     def register(self, label: str) -> int:
-        """Register a named symbol, return its index."""
+        """Register a named symbol, return its index. Auto-expands if full."""
         if label not in self._labels:
             idx = len(self._labels)
             if idx >= self.n_symbols:
-                raise ValueError(f"Codebook full ({self.n_symbols} symbols)")
+                # Auto-expand: generate more random vectors
+                rng = np.random.RandomState(hash(label) % 2**31)
+                new_phases = rng.uniform(0, 2 * np.pi, size=(256, self.dim))
+                new_vecs = np.exp(1j * new_phases).astype(np.complex64)
+                self.vectors = np.concatenate([self.vectors, new_vecs], axis=0)
+                self.n_symbols += 256
             self._labels[label] = idx
         return self._labels[label]
     
@@ -140,7 +145,7 @@ class FHRREncoder:
     def __init__(self, dim: int = 2048,
                  n_position_moduli: int = 3,
                  position_range: int = 100000,
-                 n_features: int = 256,
+                 n_features: int = 4096,
                  n_roles: int = 32):
         """
         Args:

tensegrity/v2/graft.py

@@ -0,0 +1,316 @@
+"""
+v2 Graft: NGC prediction errors → per-step logit biases during LLM decoding.
+
+This bridges the gap between the manifold approach (continuous constraint
+propagation inside the decode loop) and Tensegrity's causal reasoning
+(epistemically grounded beliefs about what's true).
+
+At each decode step:
+  1. The generated tokens so far are encoded as an FHRR sequence
+  2. The NGC circuit settles on this observation (minimizing VFE)
+  3. The prediction error at each NGC layer is computed
+  4. These errors are projected into vocabulary space as logit biases
+
+The projection works because:
+  - Layer 0 errors (sensory) → token-level constraints (word choice)
+  - Layer 1 errors (hidden) → phrase-level constraints (coherence)
+  - Layer L errors (abstract) → semantic constraints (topic, logic)
+
+Each layer's projection is a fixed random matrix (no learning needed
+at the graft interface — all learning happens inside the NGC circuit).
+
+Convergence gating:
+  - Only emit bias when NGC has settled (energy delta < threshold)
+  - Scale bias by inverse entropy (confident beliefs → strong bias)
+  - Never worse than base: ungated fallback to native LLM logits
+"""
+
+import numpy as np
+from typing import Dict, List, Optional, Callable, Set, Tuple
+import math
+import logging
+
+logger = logging.getLogger(__name__)
+
+# Lazy torch import
+torch = None
+def _ensure_torch():
+    global torch
+    if torch is None:
+        import importlib
+        torch = importlib.import_module('torch')
+
+
+class NGCLogitsProcessor:
+    """
+    HuggingFace LogitsProcessor that runs NGC settling at each decode step.
+    
+    This is the v2 equivalent of TensegrityLogitsProcessor, but instead of
+    projecting flat hypothesis posteriors, it projects hierarchical prediction
+    errors from the NGC circuit.
+    
+    The manifold ran ~47 internal steps per decode step until coherence > 0.96.
+    We do the same: the NGC circuit settles until energy delta < threshold,
+    then projects its state into logit space.
+    """
+    
+    supports_continuous_batching = False  # Stateful
+    
+    def __init__(self,
+                 field,  # UnifiedField instance
+                 tokenizer,
+                 vocab_projections: Optional[List[np.ndarray]] = None,
+                 scale: float = 1.0,
+                 energy_gate: float = 0.1,
+                 max_settle_steps: int = 30,
+                 max_bias: float = 5.0):
+        """
+        Args:
+            field: UnifiedField instance (owns NGC + FHRR + Hopfield)
+            tokenizer: HuggingFace tokenizer
+            vocab_projections: Per-NGC-layer projection matrices to vocab space.
+                             If None, generated randomly (fixed, not learned).
+            scale: Overall bias magnitude multiplier
+            energy_gate: Only emit bias when NGC energy change < this per step
+            max_settle_steps: NGC settling budget per decode step
+            max_bias: Clamp per-token bias magnitude
+        """
+        _ensure_torch()
+        
+        self.field = field
+        self.tokenizer = tokenizer
+        self.scale = scale
+        self.energy_gate = energy_gate
+        self.max_settle_steps = max_settle_steps
+        self.max_bias = max_bias
+        
+        self.vocab_size = tokenizer.vocab_size
+        
+        # Build per-layer projection matrices: NGC layer dim → vocab_size
+        # These are fixed random projections, not learned
+        if vocab_projections is not None:
+            self.projections = vocab_projections
+        else:
+            self.projections = self._build_projections()
+        
+        # Tracking
+        self._step_count = 0
+        self._emissions = 0
+        self._total_settle_steps = 0
+    
+    def _build_projections(self) -> List[np.ndarray]:
+        """
+        Build random projection matrices from NGC error space to vocab space.
+        
+        Higher layers get stronger projection weights (semantic > surface).
+        Layer weights: [1.0, 2.0, 4.0, ...] (doubling per level).
+        """
+        projections = []
+        rng = np.random.RandomState(7777)
+        
+        for ell, size in enumerate(self.field.ngc.layer_sizes):
+            # Random projection: (vocab_size, layer_size)
+            # Scaled by 1/sqrt(layer_size) for variance normalization
+            # Higher layers get more weight
+            layer_weight = 2.0 ** ell
+            P = rng.randn(self.vocab_size, size).astype(np.float64)
+            P *= layer_weight / np.sqrt(size)
+            projections.append(P)
+        
+        return projections
+    
+    def _tokens_to_observation(self, input_ids) -> np.ndarray:
+        """
+        Convert generated tokens so far into an FHRR observation vector,
+        then project to NGC sensory space.
+        
+        Uses the last N tokens as a sequence encoding.
+        """
+        # Decode last 16 tokens to text
+        ids = input_ids[0].tolist()
+        recent_ids = ids[-16:]  # Last 16 tokens
+        text = self.tokenizer.decode(recent_ids, skip_special_tokens=True)
+        tokens = text.lower().split()
+        
+        if not tokens:
+            return np.zeros(self.field.obs_dim, dtype=np.float64)
+        
+        # Encode as FHRR sequence → project to NGC observation space
+        fhrr_vec = self.field.encoder.encode_sequence(tokens)
+        obs_vec = self.field._fhrr_to_obs(fhrr_vec)
+        
+        return obs_vec
+    
+    def _error_to_bias(self) -> np.ndarray:
+        """
+        Project NGC prediction errors into vocabulary space.
+        
+        bias = Σ_ℓ P_ℓ · error_ℓ
+        
+        Where P_ℓ is the fixed random projection for layer ℓ,
+        and error_ℓ is the precision-weighted prediction error.
+        
+        Low-level errors → token-level biases (surface form)
+        High-level errors → semantic biases (topic/logic)
+        """
+        bias = np.zeros(self.vocab_size, dtype=np.float64)
+        
+        for ell in range(self.field.ngc.n_layers):
+            error = self.field.ngc.layers[ell].error
+            if np.linalg.norm(error) < 1e-10:
+                continue
+            
+            # Project error into vocab space
+            layer_bias = self.projections[ell] @ error
+            bias += layer_bias
+        
+        # Normalize by number of layers
+        bias /= max(self.field.ngc.n_layers, 1)
+        
+        return bias
+    
+    def __call__(self, input_ids, scores):
+        """
+        Called at each decode step by model.generate().
+        
+        1. Convert generated tokens → FHRR observation
+        2. Settle NGC circuit on this observation
+        3. If converged: project prediction errors into logit biases
+        4. If not: pass through unmodified
+        """
+        self._step_count += 1
+        
+        # Convert tokens to observation
+        obs = self._tokens_to_observation(input_ids)
+        
+        # Settle NGC
+        settle_result = self.field.ngc.settle(obs, steps=self.max_settle_steps)
+        self._total_settle_steps += self.max_settle_steps
+        
+        # Check convergence: did the energy stabilize?
+        energy_trace = settle_result["energy_trace"]
+        if len(energy_trace) >= 2:
+            energy_delta = abs(energy_trace[-1] - energy_trace[-2])
+            converged = energy_delta < self.energy_gate
+        else:
+            converged = False
+        
+        if not converged:
+            return scores  # Graceful fallback — native LLM behavior
+        
+        # Query Hopfield memory with abstract state (top NGC layer)
+        abstract = self.field.ngc.get_abstract_state(level=-1)
+        retrieved, mem_energy = self.field.memory.retrieve(abstract, steps=3)
+        
+        # Compute bias from prediction errors
+        bias = self._error_to_bias()
+        
+        # Scale by inverse energy (lower energy = more confident = stronger bias)
+        current_energy = settle_result["final_energy"]
+        confidence = 1.0 / (1.0 + current_energy)  # Sigmoid-like scaling
+        bias *= self.scale * confidence
+        
+        # Clamp
+        np.clip(bias, -self.max_bias, self.max_bias, out=bias)
+        
+        # Convert to torch and apply
+        bias_tensor = torch.tensor(bias, device=scores.device, dtype=scores.dtype)
+        
+        self._emissions += 1
+        
+        return scores + bias_tensor.unsqueeze(0)
+    
+    @property
+    def statistics(self):
+        return {
+            "decode_steps": self._step_count,
+            "emissions": self._emissions,
+            "emission_rate": self._emissions / max(self._step_count, 1),
+            "total_settle_steps": self._total_settle_steps,
+            "avg_settle_per_decode": self._total_settle_steps / max(self._step_count, 1),
+            "ngc_energy": self.field.ngc.total_energy,
+            "memory_patterns": self.field.memory.n_patterns,
+        }
+
+
+class V2ScoringBridge:
+    """
+    Bridge between v2 architecture and the benchmark harness.
+    
+    Converts a TaskSample's choices into FHRR observations,
+    runs the NGC circuit on each, and scores choices by
+    prediction error: lower error = better fit = higher score.
+    
+    This replaces v1's flat Bayesian posterior scoring with
+    hierarchical predictive coding scoring.
+    """
+    
+    def __init__(self, field=None, obs_dim: int = 128, 
+                 hidden_dims: Optional[List[int]] = None):
+        from tensegrity.v2.field import UnifiedField
+        
+        self.field = field or UnifiedField(
+            obs_dim=obs_dim,
+            hidden_dims=hidden_dims or [64, 16],
+            fhrr_dim=1024,
+            hopfield_beta=0.05,
+            ngc_settle_steps=20,
+            ngc_learning_rate=0.005,
+        )
+    
+    def score_choices(self, prompt: str, choices: List[str]) -> Tuple[List[float], float]:
+        """
+        Score each choice via v2 predictive coding.
+        
+        For each choice:
+          1. Encode prompt as FHRR → settle NGC (establish context beliefs)
+          2. Encode prompt+choice as FHRR → settle NGC (observe with choice)
+          3. Score = negative prediction error (lower error = better fit)
+        
+        Returns:
+            (scores, entropy) where scores[i] = score for choice i
+        """
+        # First, establish context by observing the prompt
+        prompt_tokens = prompt.lower().split()[:32]  # Cap at 32 tokens
+        if prompt_tokens:
+            self.field.observe(prompt_tokens, input_type="tokens")
+        
+        # Score each choice by prediction error
+        scores = []
+        for choice in choices:
+            choice_tokens = (prompt + " " + choice).lower().split()[-32:]
+            
+            # Create a fresh copy of the NGC state for counterfactual scoring
+            # (we don't want scoring one choice to affect scoring another)
+            saved_layers = [
+                (l.z.copy(), l.z_bar.copy(), l.error.copy()) 
+                for l in self.field.ngc.layers
+            ]
+            
+            # Observe the choice
+            fhrr_vec = self.field.encoder.encode_sequence(choice_tokens)
+            obs = self.field._fhrr_to_obs(fhrr_vec)
+            settle_result = self.field.ngc.settle(obs, steps=10)
+            
+            # Score = negative energy (lower energy = better explanation)
+            score = -settle_result["final_energy"]
+            scores.append(score)
+            
+            # Restore NGC state
+            for i, (z, z_bar, err) in enumerate(saved_layers):
+                self.field.ngc.layers[i].z = z
+                self.field.ngc.layers[i].z_bar = z_bar
+                self.field.ngc.layers[i].error = err
+        
+        # Entropy of softmax(scores) for confidence estimation
+        scores_arr = np.array(scores)
+        shifted = scores_arr - scores_arr.max()
+        probs = np.exp(shifted) / np.exp(shifted).sum()
+        entropy = float(-np.sum(probs * np.log(probs + 1e-16)) / np.log(max(len(probs), 2)))
+        
+        return scores, entropy
+    
+    def reset(self):
+        """Reset the field's NGC state between samples."""
+        self.field.ngc._initialized = False
+        self.field.ngc.layers = []

tests/test_v2_bench.py

@@ -0,0 +1,113 @@
+"""
+Test v2 scoring bridge against benchmarks.
+"""
+import sys
+sys.path.insert(0, '/app')
+import numpy as np
+np.random.seed(42)
+
+
+def test_v2_scoring():
+    """Test v2 NGC-based scoring on benchmark samples."""
+    print("=" * 60)
+    print("TEST: v2 NGC Scoring vs v1 Baseline on Sample Tasks")
+    print("=" * 60)
+    
+    from tensegrity.v2.graft import V2ScoringBridge
+    from tensegrity.bench.tasks import load_task_samples
+    
+    bridge = V2ScoringBridge(obs_dim=128, hidden_dims=[64, 16])
+    
+    tasks = ["copa", "sciq", "arc_challenge"]
+    
+    for task_name in tasks:
+        try:
+            samples = load_task_samples(task_name, max_samples=30)
+        except Exception as e:
+            print(f"\n  {task_name}: SKIP ({e})")
+            continue
+        
+        correct = 0
+        total = 0
+        
+        for sample in samples:
+            bridge.reset()
+            scores, entropy = bridge.score_choices(sample.prompt, sample.choices)
+            pred = int(np.argmax(scores))
+            if pred == sample.gold:
+                correct += 1
+            total += 1
+        
+        acc = correct / max(total, 1)
+        print(f"\n  {task_name}: {correct}/{total} = {acc:.1%}")
+    
+    print(f"\n  ✓ v2 scoring bridge functional")
+    return True
+
+
+def test_causal_energy():
+    """Test the causal energy term."""
+    print("\n" + "=" * 60)
+    print("TEST: Causal Energy Arena v2")
+    print("=" * 60)
+    
+    from tensegrity.causal.scm import StructuralCausalModel
+    from tensegrity.v2.causal_energy import CausalArenaV2
+    
+    # Two competing models
+    m_correct = StructuralCausalModel("correct")
+    m_correct.add_variable("X", n_values=3)
+    m_correct.add_variable("Y", n_values=3, parents=["X"])
+    
+    m_wrong = StructuralCausalModel("wrong")
+    m_wrong.add_variable("X", n_values=3)
+    m_wrong.add_variable("Y", n_values=3)  # No causal link
+    
+    # Train correct model on data where X causes Y
+    data = m_correct.sample(100)
+    m_correct.update_from_data(data)
+    m_wrong.update_from_data(data)
+    
+    arena = CausalArenaV2(precision=1.0, beta=2.0)
+    arena.register(m_correct)
+    arena.register(m_wrong)
+    
+    # Test on 20 observations
+    test_data = m_correct.sample(20)
+    winners = []
+    for obs in test_data:
+        result = arena.compete(obs)
+        winners.append(result["winner"])
+        arena.update_models(obs)
+    
+    correct_wins = sum(1 for w in winners if w == "correct")
+    print(f"  Correct model wins: {correct_wins}/{len(winners)}")
+    print(f"  Final tension: {arena.tension:.3f}")
+    
+    # Energy comparison
+    last_result = arena.compete(test_data[-1])
+    print(f"  Last energies: {last_result['energies']}")
+    print(f"  Last posteriors: {last_result['posteriors']}")
+    
+    print(f"  ✓ Causal energy arena functional")
+    return True
+
+
+if __name__ == "__main__":
+    tests = [
+        ("v2 Scoring", test_v2_scoring),
+        ("Causal Energy", test_causal_energy),
+    ]
+    
+    print("\n" + "█" * 60)
+    print("  v2 Integration Tests")
+    print("█" * 60)
+    
+    for name, fn in tests:
+        try:
+            fn()
+        except Exception as e:
+            print(f"\n  ✗ {name} FAILED: {e}")
+            import traceback; traceback.print_exc()
+    
+    print()