Update app.py

app.py

@@ -1,14 +1,42 @@
 #!/usr/bin/env python3
 """
-PRACTICALITY SYSTEM 18.0 — NAKED EVALUATION + GRADIENT ORACLE
+PRACTICALITY SYSTEM 19.1 — THE NEURO-SYMBOLIC ENZYME
 ═══════════════════════════════════════════════════════════════════
-THE RIGOROUS NEURO-SYMBOLIC ENGINE:
-  - Axioms are mathematical compilers. They generate bindings AND strict `pinned_vars` masks.
-  - Variable Filling is strictly MASKED. PyTorch is allowed to fill the continuous blanks, 
-    but it is mathematically barred from altering the variables pinned by the Axiom.
-  - `loss.backward()` is used as a Diagnostic Oracle. High-gradient variables reveal 
-    structural tension, feeding back into the discrete Markov sequence.
-  - No global safety nets. No "Clever Hans" effect. The trace is the true isomorphism.
+AIM: 
+  To extract a true, LLM-readable structural isomorphism of a mathematical problem. 
+  Previous iterations suffered from the "Clever Hans" effect: a powerful continuous 
+  optimizer (Adam) was quietly solving the problem and letting irrelevant discrete 
+  logic (Axioms) take the credit. 
+  
+  The goal of 19.1 is strict credit assignment. If the sequence trace says 
+  [SYMMETRIC → MONOTONE_PRODUCT → COMPOSITE] solved the problem, it must be a 
+  mathematical guarantee that those geometric manipulations isolated the exact 
+  basin of attraction.
+
+THE CURRENT SYSTEM (1.5M Scale):
+  We stripped the safety nets and replaced them with a "Naked Evaluation" framework, 
+  augmented by biological design patterns (Enzyme catalysis & Allosteric feedback).
+
+IMPLEMENTATION AT A GLANCE:
+  1. NAKED EVALUATION (The Masked Compiler): 
+     Axioms no longer output fuzzy suggestions. They compile into strict mathematical 
+     masks (`pinned_vars`). PyTorch is allowed to fill in the continuous blanks using 
+     Adam, but the gradients for pinned variables are strictly zeroed (`X.grad *= mask`). 
+     PyTorch is locked in a cage; it cannot undo the Axiom's structural claim.
+
+  2. BIDIRECTIONAL SEARCH (The Active Site):
+     Instead of just forward-chaining, we spawn `RAY_TARGET` seeds by parsing the 
+     target PSL anchors (the final goal state). The system builds the logical chain 
+     from both the substrate (initial variables) and the product (anchor states), 
+     meeting in the middle.
+
+  3. INTELLIGENT CHAINING (Allosteric Feedback):
+     We use PyTorch's `loss.backward()` NOT as an optimizer, but as a Diagnostic Oracle. 
+     If a hypothesis fails, the gradients reveal exact structural tension. 
+       - 1 variable in tension = Isolated failure  → Chain heavily biases Point Axioms (ATOMIC)
+       - 2 variables in tension= Relational failure→ Chain heavily biases Pair Axioms (SYMMETRIC)
+       - 3+ variables in tension= Systemic failure → Chain heavily biases Global Axioms (CONSERVED)
+     The continuous math directly biases the discrete Markov transition matrix.
 """
 
 import time, random, math, threading, warnings
@@ -26,7 +54,7 @@ import gradio as gr
 warnings.filterwarnings("ignore")
 USE_GPU = torch.cuda.is_available()
 DEVICE  = torch.device("cuda" if USE_GPU else "cpu")
-print(f"[SYSTEM 18.0] Compute: {DEVICE.type.upper()} | Engine: Masked Deduction + Gradient Diagnostics")
+print(f"[SYSTEM 19.1] Compute: {DEVICE.type.upper()} | Engine: Neuro-Symbolic Enzyme (1.5M Scale)")
 
 # ══════════════════════════════════════════════════════════════════════
 # SECTION 1: CONSTANTS
@@ -34,15 +62,16 @@ print(f"[SYSTEM 18.0] Compute: {DEVICE.type.upper()} | Engine: Masked Deduction
 SOLVE_THRESHOLD      = 0.05
 VERIFY_G1_THRESHOLD  = 0.08
 VERIFY_G2_TOLERANCE  = 1e-3
-HC4_VOLUME_FLOOR     = 0.20
-DEDUCE_ADAM_STEPS    = 30    # GPU variable-filling steps (strictly masked)
+DEDUCE_ADAM_STEPS    = 25    # Masked variable filling only
 N_MPRT_EXPLORE       = 1000  # Initial bounding box sampling only
 PROJECTION_CACHE: Dict = {}
-MAX_TOTAL_RAYS       = 25000 
-RAY_BATCH_SIZE       = 256   
+
+# MASSIVE SCALE PARAMETERS
+MAX_TOTAL_RAYS       = 1_500_000 
+RAY_BATCH_SIZE       = 4096   
 CE_EARN_RATIO        = 0.90
 SCOUT_CE_THRESHOLD   = 0.5
-BRANCH_WIDTH         = 5
+BRANCH_WIDTH         = 4
 
 # ══════════════════════════════════════════════════════════════════════
 # SECTION 2: INTERVAL ARITHMETIC (HC4)
@@ -118,7 +147,7 @@ def _hc4(box,constraints):
     return cur
 
 # ══════════════════════════════════════════════════════════════════════
-# SECTION 3: PSL 2.0 PARSER & COMPILER (Unchanged, Fast)
+# SECTION 3: PSL 2.0 PARSER & COMPILER 
 # ══════════════════════════════════════════════════════════════════════
 @dataclass
 class PSLVar:        name:str; lo:float; hi:float
@@ -299,27 +328,6 @@ class Problem:
         self.compiled_constraints=[compile_mc(c.kind,c.expr,c.direction,self.variables,c.weight,c.scope,c.branches) for c in self.constraints]
         self.var_idx = {v: i for i, v in enumerate(self.variables)}
         
-    def constraint_energy(self,b:Dict[str,float]) -> float:
-        total=0.0
-        for mc in self.compiled_constraints:
-            if getattr(mc,'weight',1.0)==0.0: continue
-            if mc.kind=="or_eq":
-                vals=[]
-                for bmc in mc.branches:
-                    if bmc.fast_pt:
-                        try: vals.append(abs(float(bmc.fast_pt(*[b.get(v,0.0) for v in bmc.syms_used]))))
-                        except: pass
-                total+=(min(vals)*mc.weight) if vals else 1.0
-            else:
-                if mc.fast_pt is None: total+=1.0; continue
-                try:
-                    val=float(mc.fast_pt(*[b.get(v,0.0) for v in mc.syms_used]))
-                    total+=(abs(val) if mc.kind=="equality" else max(0.0,-val) if mc.direction=="geq" else max(0.0,val))*mc.weight
-                except: total+=1.0
-        for v,(lo,hi) in self.bounds.items():
-            bv=b.get(v,0.0); total+=max(0.0,lo-bv)+max(0.0,bv-hi)
-        return total
-
     def tensor_energy(self, X: torch.Tensor) -> torch.Tensor:
         is_batched = (X.dim() == 2)
         total = torch.zeros(X.shape[0] if is_batched else 1, device=DEVICE, dtype=torch.float32)
@@ -356,10 +364,8 @@ class L9Certificate: residual_ce:float; dominant_vars:List[str]; dominant_exprs:
 
 def _batched_deduce_and_evaluate(problem: Problem, hyps: List['Hypothesis']) -> List[Tuple[Dict, float, List[str]]]:
     """
-    The Core Engine of 18.0.
-    Takes Axiom hypotheses (bindings + strict pinned constraints).
-    Uses PyTorch to fill in unpinned continuous variables (variable filling).
-    Uses loss.backward() to find the exact structural tensions (Oracle).
+    Naked Evaluation: Axioms map masks. HC4 bounds. Adam fills ONLY continuous gaps. 
+    PyTorch loss.backward() acts purely as a gradient oracle to feed the Markov Chain.
     """
     if not hyps: return []
     B = len(hyps); V = len(problem.variables)
@@ -410,7 +416,6 @@ def _batched_deduce_and_evaluate(problem: Problem, hyps: List['Hypothesis']) ->
     results = []
     for i in range(B):
         final_b = {problem.variables[j]: float(X_vals[i, j]) for j in range(V)}
-        # Oracle: which variables have the most structural tension right now?
         var_tensions = {problem.variables[j]: float(grads[i, j]) for j in range(V)}
         dom_vars = sorted(var_tensions.keys(), key=lambda v: -var_tensions[v])[:3]
         results.append((final_b, float(ce_vals[i]), dom_vars))
@@ -484,10 +489,9 @@ class StructuralModel: best_binding:Dict[str,float]; best_ce:float; failure_patt
 
 def _make_hyp(hid,binding,h_type,claim,derivation,pinned,free,conf): return Hypothesis(hid=hid,binding=binding,h_type=h_type,claim=claim,derivation=derivation,pinned_vars=pinned,free_vars=free,confidence=conf)
 
-# 18.0 Constructors: They return rigorous structural masks (`pinned_vars`)
+# Strict Axiomatic Structural Compilers
 def _hyp_continuous(p,bt,a,m): return [_make_hyp("cnt",bt.binding,"continuous","Manifold",[],{},p.variables,0.45)]
 def _hyp_discrete(p,bt,a,m): 
-    # Use internal HC4 deduction specifically on scope boundaries
     b = dict(bt.binding); pinned = {}
     for sn in p.scope_order:
         svars = p.scope_vars.get(sn, [])
@@ -505,7 +509,6 @@ def _hyp_quadratic(p,bt,a,m):
     hyps=[]; b=dict(bt.binding)
     if bt.l9 and bt.l9.dominant_vars:
         for v in bt.l9.dominant_vars[:2]: 
-            # Pin the exact dominant variable to its current spot to force others to align around it
             hyps.append(_make_hyp(f"qua_{v}",b,"quadratic",f"Root({v})",[],{v:b[v]},p.variables,0.70))
     return hyps
 
@@ -576,7 +579,6 @@ def _hyp_atomic(p,bt,a,m):
         if hyps: break
     return hyps
 
-# Pass-through exploratory/diagnostic axioms (empty pinned_vars)
 def _hyp_metric(p,bt,a,m): return [_make_hyp("met",bt.binding,"metric","Radial",[],{},p.variables,0.75)]
 def _hyp_ordered(p,bt,a,m): return [_make_hyp("ord",bt.binding,"ordered","OrderBal",[],{},p.variables,0.67)]
 def _hyp_monotone(p,bt,a,m): return [_make_hyp("mon",bt.binding,"monotone","MonoPath",[],{},p.variables,0.67)]
@@ -610,10 +612,58 @@ AXIOM_CONSTRUCTORS={
 }
 
 # ══════════════════════════════════════════════════════════════════════
-# SECTION 6: TRACER + SEEDER (Gradient Diagnostic integration)
+# SECTION 6: TRACER + SEEDER (Intelligent & Bidirectional Search)
 # ══════════════════════════════════════════════════════════════════════
 class CreativeSeeder:
-    def tension_seeds(self, max_n:int=1000, dom_vars:List[str]=None):
+    def target_seeds(self, anchors: List[AXLInvariant]) -> List[AxiomRay]:
+        """BIDIRECTIONAL SEARCH: Spawn Axioms based explicitly on the PSL goal state."""
+        seeds = []
+        for a in anchors:
+            if "*" in a.expr and "**" not in a.expr:
+                seeds.append(AxiomRay([Axiom.BILINEAR], f"TGT_BIL_{a.name}", "target"))
+                seeds.append(AxiomRay([Axiom.MONOTONE_PRODUCT], f"TGT_MPR_{a.name}", "target"))
+            elif "**2" in a.expr:
+                seeds.append(AxiomRay([Axiom.QUADRATIC], f"TGT_QUA_{a.name}", "target"))
+                seeds.append(AxiomRay([Axiom.METRIC], f"TGT_MET_{a.name}", "target"))
+            elif "+" in a.expr:
+                seeds.append(AxiomRay([Axiom.CONSERVED], f"TGT_CSV_{a.name}", "target"))
+        return seeds
+
+    def intelligent_branch(self, ray, out_baton, remaining_axioms, branch_width):
+        """
+        ALLOSTERIC FEEDBACK: Uses the PyTorch Gradient Oracle (dominant_vars) 
+        to bias the discrete Markov transition matrix.
+        """
+        dom_vars = out_baton.l9.dominant_vars if out_baton.l9 else []
+        n_dom = len(dom_vars)
+
+        # Map Axioms to the type of structural tension they resolve
+        isolated   = {Axiom.ATOMIC, Axiom.EXTREMAL, Axiom.MUTABLE, Axiom.DUALITY, Axiom.LOCALITY}
+        relational = {Axiom.SYMMETRIC, Axiom.BILINEAR, Axiom.MONOTONE_PRODUCT, Axiom.ORDERED, Axiom.METRIC, Axiom.INJECTIVE}
+        systemic   = {Axiom.CONSERVED, Axiom.NETWORK, Axiom.HOLISM, Axiom.COMPOSITE, Axiom.ENTROPY}
+
+        scored_axioms = []
+        for ax in remaining_axioms:
+            weight = 1.0 # Base probability
+            # Boost probability heavily if the Axiom class matches the Gradient topology
+            if n_dom == 1 and ax in isolated: weight += 5.0
+            elif n_dom == 2 and ax in relational: weight += 5.0
+            elif n_dom >= 3 and ax in systemic: weight += 5.0
+            
+            scored_axioms.append((weight, ax))
+
+        # Weighted stochastic selection
+        scored_axioms.sort(key=lambda x: x[0] * random.random(), reverse=True)
+
+        children = []
+        for _, axiom in scored_axioms[:branch_width]:
+            child = ray.extend(axiom, "branch", new_prior=out_baton.ce)
+            if child: 
+                child.baton = out_baton
+                children.append(child)
+        return children
+
+    def tension_seeds(self, max_n:int=1000):
         results=[]
         for (a,c) in ADJACENT_CONTRADICTIONS:
             for b1 in ALL_AXIOMS:
@@ -661,16 +711,20 @@ class CreativeSeeder:
             if child: child.ray_id=f"RES_{child.ray_id}"; results.append(child)
         return results
 
+CREATIVE_SEEDER = CreativeSeeder()
+
 class SequenceRayTracer:
     def trace(self, axl_def, base_problem):
-        all_traces=[]; baton_registry={}; successful_rays=[]
+        hero_traces = [] # Memory protection: only store important traces
+        baton_registry={}; successful_rays=[]
+        
         init_b, init_ce = _mprt_sample(base_problem, {v:IV(*base_problem.bounds[v]) for v in base_problem.variables}, N_MPRT_EXPLORE)
         seed_baton = Baton(binding=init_b, ce=init_ce, ray_id="SEED", l9=L9Certificate(init_ce, base_problem.variables[:3], []))
         
         queues={
             "core": [AxiomRay([a], f"S{i}", "seed", ce_prior=init_ce) for i,a in enumerate(ALL_AXIOMS)],
-            "explore": CreativeSeeder().tension_seeds(200) + CreativeSeeder().random_scouts(100),
-            "genetic": []
+            "explore": CREATIVE_SEEDER.tension_seeds(200) + CREATIVE_SEEDER.random_scouts(100),
+            "genetic": CREATIVE_SEEDER.target_seeds(axl_def.anchors) # Inverse chain seeds
         }
         for q in queues.values(): random.shuffle(q)
         queue_order = ["core", "explore", "genetic"]; q_idx = 0
@@ -693,26 +747,22 @@ class SequenceRayTracer:
             
             t0 = time.time(); historical_batons = list(baton_registry.values())
             
-            # Step 1: Constructors generate naked hypotheses + masks
             batch_hyps = []
             for i, ray in enumerate(batch_rays):
                 p_baton = ray.baton or seed_baton
                 hyps = AXIOM_CONSTRUCTORS[ray.sequence[-1]](base_problem, p_baton, historical_batons, model)
                 if not hyps: hyps = [_make_hyp("fb", p_baton.binding, "fallback", "Pass", [], {}, base_problem.variables, 0.1)]
-                batch_hyps.append((i, hyps[0])) # Take top hyp for simplicity
+                batch_hyps.append((i, hyps[0]))
                 
-            # Step 2: Masked Deduction + Diagnostic Oracle
             evaluated_results = _batched_deduce_and_evaluate(base_problem, [h for _, h in batch_hyps])
             
             solved = False
             elapsed_ms = (time.time() - t0) * 1000 / len(batch_rays)
             
-            # Step 3: Verify and Route
             for (ray_idx, hyp), (final_b, final_ce, dom_vars) in zip(batch_hyps, evaluated_results):
                 ray = batch_rays[ray_idx]
                 parent_ce_val = (ray.baton or seed_baton).ce
                 
-                # Verify Gate (For anchors)
                 g1_pass = final_ce < VERIFY_G1_THRESHOLD
                 g2_pass = True
                 for inv in axl_def.anchors:
@@ -725,11 +775,13 @@ class SequenceRayTracer:
                     except: g2_pass=False
                 
                 overall_pass = g1_pass and g2_pass
-                all_traces.append(HypothesisTrace(ray.ray_id, total_fired-len(batch_rays)+ray_idx, ray.name, ray.ray_type, "Eval", parent_ce_val, final_ce, None, overall_pass, elapsed_ms))
+                
+                improved = final_ce < best_ce
+                if improved or overall_pass or (random.random() < 0.001): # Save heroes + tiny sample
+                    hero_traces.append(HypothesisTrace(ray.ray_id, total_fired-len(batch_rays)+ray_idx, ray.name, ray.ray_type, "Eval", parent_ce_val, final_ce, None, overall_pass, elapsed_ms))
                 
                 out_baton = Baton(binding=final_b, ce=final_ce, l9=L9Certificate(final_ce, dom_vars, []))
                 
-                improved = final_ce < best_ce
                 if improved:
                     best_ce=final_ce; best_binding=final_b; stall_counter=0
                     baton_registry[ray.ray_id] = out_baton; successful_rays.append(ray)
@@ -740,25 +792,28 @@ class SequenceRayTracer:
                 scout_earned = (ray.ray_type == "scout" and final_ce < SCOUT_CE_THRESHOLD)
                 
                 if earned or scout_earned:
-                    new_children = []; remaining = [a for a in ALL_AXIOMS if a != ray.sequence[-1] and frozenset([ray.sequence[-1], a]) not in _CONTRA_SET and a not in ray.sequence]
-                    random.shuffle(remaining)
-                    for axiom in remaining[:BRANCH_WIDTH]:
-                        child = ray.extend(axiom, "branch", new_prior=final_ce)
-                        if child: child.baton = out_baton; new_children.append(child)
+                    new_children = []
+                    remaining = [a for a in ALL_AXIOMS if a != ray.sequence[-1] and frozenset([ray.sequence[-1], a]) not in _CONTRA_SET and a not in ray.sequence]
+                    
+                    # ════ INTELLIGENT CHAINING (ALLOSTERIC FEEDBACK) ════
+                    # The gradient oracle biases the next structural move
+                    new_children.extend(
+                        CREATIVE_SEEDER.intelligent_branch(ray, out_baton, remaining, BRANCH_WIDTH)
+                    )
                     
-                    inv = CreativeSeeder().invert(ray, final_ce)
+                    inv = CREATIVE_SEEDER.invert(ray, final_ce)
                     if inv: new_children.append(inv)
                     
                     if len(successful_rays) >= 2:
                         candidates = [r for r in successful_rays if r.ray_id != ray.ray_id]
                         if candidates:
                             partner = random.choice(candidates); ba = baton_registry.get(ray.ray_id, out_baton); bb = baton_registry.get(partner.ray_id, out_baton)
-                            new_children.extend(CreativeSeeder().recombine(ray, partner, ba, bb)[:3])
+                            new_children.extend(CREATIVE_SEEDER.recombine(ray, partner, ba, bb)[:3])
                     
-                    new_children.extend(CreativeSeeder().resonance_extensions(ray, final_ce, model.resonant_pairs))
+                    new_children.extend(CREATIVE_SEEDER.resonance_extensions(ray, final_ce, model.resonant_pairs))
                     
                     for child in new_children:
-                        if child.ray_type in ("seed", "branch"): queues["core"].append(child)
+                        if child.ray_type in ("seed", "branch", "target"): queues["core"].append(child)
                         elif child.ray_type in ("scout", "tension"): queues["explore"].append(child)
                         elif child.ray_type in ("recombine", "invert"): queues["genetic"].append(child)
                         
@@ -766,7 +821,7 @@ class SequenceRayTracer:
             
             if solved: break
             
-        return best_binding, best_ce, all_traces
+        return best_binding, best_ce, hero_traces, total_fired
 
 # ══════════════════════════════════════════════════════════════════════
 # SECTION 7: AXL PROBLEMS
@@ -813,7 +868,7 @@ def background_worker():
         base_prob = build_base_problem(axl_def)
         
         t0 = time.time()
-        binding, ce, history = tracer.trace(axl_def, base_prob)
+        binding, ce, history, total_fired = tracer.trace(axl_def, base_prob)
         ms = round((time.time() - t0) * 1000)
         
         with STATE_LOCK:
@@ -822,7 +877,8 @@ def background_worker():
                 "ce": ce,
                 "solved": ce < SOLVE_THRESHOLD,
                 "ms": ms,
-                "traces": len(history)
+                "traces_saved": len(history),
+                "total_fired": total_fired
             })
             if len(RUN_LOG) > 30: RUN_LOG.pop()
 
@@ -830,13 +886,13 @@ def toggle_engine():
     global IS_RUNNING
     IS_RUNNING = not IS_RUNNING
     if IS_RUNNING: threading.Thread(target=background_worker, daemon=True).start()
-    return "⏹ STOP 18.0" if IS_RUNNING else "▶ START 18.0"
+    return "⏹ STOP 19.1" if IS_RUNNING else "▶ START 19.1"
 
 def refresh_dashboard():
     with STATE_LOCK:
         if not RUN_LOG: return "<i>No runs yet. Click Start Engine.</i>"
         html = "<table style='width:100%; border-collapse:collapse; text-align:left;'>"
-        html += "<tr style='border-bottom:1px solid #444'><th>Problem</th><th>CE</th><th>Status</th><th>Speed</th><th>Traces Fired</th></tr>"
+        html += "<tr style='border-bottom:1px solid #444'><th>Problem</th><th>CE</th><th>Status</th><th>Speed</th><th>Total Rays Fired</th></tr>"
         for r in RUN_LOG:
             color = "#4CAF50" if r["solved"] else "#FF9800"
             html += f"<tr style='border-bottom:1px solid #222'>"
@@ -844,14 +900,14 @@ def refresh_dashboard():
             html += f"<td style='padding:5px; color:{color}; font-weight:bold'>{r['ce']:.5f}</td>"
             html += f"<td style='padding:5px; color:{color}'>{'SOLVED' if r['solved'] else 'SEARCHING'}</td>"
             html += f"<td style='padding:5px; color:#888'>{r['ms']}ms</td>"
-            html += f"<td style='padding:5px; color:#888'>{r['traces']} rays</td></tr>"
+            html += f"<td style='padding:5px; color:#888'>{r['total_fired']:,}</td></tr>"
         html += "</table>"
         return html
 
 with gr.Blocks(theme=gr.themes.Monochrome(text_size="sm")) as demo:
-    gr.Markdown(f"# ⚗ Practicality 18.0 — Naked Evaluation + Oracle")
-    gr.Markdown(f"**Compute Node:** `{DEVICE.type.upper()}` | **Engine:** `Masked Deduction + Gradient Diagnostics`")
-    btn_toggle = gr.Button("▶ START 18.0", variant="primary")
+    gr.Markdown(f"# ⚗ Practicality 19.1 — Neuro-Symbolic Enzyme")
+    gr.Markdown(f"**Compute Node:** `{DEVICE.type.upper()}` | **Engine:** `Allosteric Feedback + Bidirectional Search (1.5M Scale)`")
+    btn_toggle = gr.Button("▶ START 19.1", variant="primary")
     html_output = gr.HTML(refresh_dashboard())
     
     btn_toggle.click(toggle_engine, inputs=None, outputs=btn_toggle)