#!/usr/bin/env python3
"""
PRACTICALITY SYSTEM 28.0 — MODULAR ORACLE ENGINE
══════════════════════════════════════════════════════════════════════════
Architecture:
- practicality_core.py : Tensor Math, Interval Arithmetic, Adam Optimizer
- practicality_axioms.py : Ray Tracer, UCB1 Bandit, Hypothesis Generation
- practicality_oracle.py : Ground Truth Verification, Heisenberg Quantum Sim
- app.py (This file) : Orchestrator, LLM Bridge, AXL Parser, Gradio UI
"""
import os, time, threading, json, re, textwrap, copy
import traceback, html, itertools, math
import requests # Explicitly declared
from collections import deque, defaultdict
from typing import Dict, List, Tuple, Any, Optional, Set
from dataclasses import dataclass, field
import sympy as sp
from sympy.parsing.sympy_parser import parse_expr
import gradio as gr
from google import genai
from google.genai import types
# Import our decoupled engines
import practicality_core as core
import practicality_axioms as axioms
import practicality_oracle as oracle
DEFAULT_GEMINI_API_KEY = os.environ.get("AI_Key", "")
GEMINI_MODEL = "gemini-2.5-flash"
print(f"[SYSTEM 28.0] Orchestrator Initialized. Compute: {core.DEVICE.type.upper()}")
# ══════════════════════════════════════════════════════════════════════
# SECTION 1: GLOBAL STATE & UI LOCKS
# ══════════════════════════════════════════════════════════════════════
STATE_LOCK = threading.Lock()
FINGERPRINT_LOCK = threading.Lock()
FINGERPRINT_STORE: Dict[str, List[Dict]] = defaultdict(list)
GLOBAL_SOLVE_STATE = {
"is_running": False, "phase": "IDLE", "best_ce": float('inf'),
"total_fired": 0, "best_ray": "—",
"recent_traces": deque(maxlen=25), "high_level_logs": [],
"answer": "", "raw_results_json": "", "infeasibility_report": "",
"minimize_result": "",
}
def _update_state(**kwargs):
with STATE_LOCK:
for k, v in kwargs.items(): GLOBAL_SOLVE_STATE[k] = v
def _add_log(line: str):
with STATE_LOCK: GLOBAL_SOLVE_STATE["high_level_logs"].append(line)
print(f"[LOG] {line}")
# ══════════════════════════════════════════════════════════════════════
# SECTION 2: AXL / PSL PARSER (The Bridge)
# ══════════════════════════════════════════════════════════════════════
@dataclass
class AXLInvariant:
name: str; expr: str; tolerance: float = 1e-3; mode: str = "eq"
compiled_func: Any = None; syms_used: List[str] = field(default_factory=list)
def compile(self, variables: List[str]):
self.expr = self.expr.replace("^", "**")
syms = {v: sp.Symbol(v) for v in variables}
parsed = parse_expr(self.expr, local_dict=syms)
if getattr(parsed, 'is_Equality', False) or getattr(parsed, 'is_Relational', False):
parsed = parsed.lhs - parsed.rhs
for s in list(parsed.free_symbols):
if str(s) not in variables: parsed = parsed.subs(s, 1.0)
self.syms_used = [str(s) for s in parsed.free_symbols if str(s) in variables]
self.compiled_func = sp.lambdify([sp.Symbol(v) for v in self.syms_used], parsed, modules="math")
@dataclass
class AXLProblemDef:
name: str; description: str; axioms: Set[str]; variables: List[Dict]; constraints: List[Dict]
scopes: List[Dict]; anchors: List[AXLInvariant]; observations: Dict[str, float] = field(default_factory=dict)
minimize_var: str = ""
class AXLtoPSLCompiler:
def compile(self, prob: AXLProblemDef) -> str:
lines = [f"SPACE {prob.name}"]; emitted_vars = set()
for v in prob.variables:
lo, hi = v["lo"], v["hi"]
if v["name"] in prob.observations: val = prob.observations[v["name"]]; lo = hi = val
is_int_flag = " INT" if v.get("is_int", False) else ""
lines.append(f"VAR {v['name']} {lo} {hi}{is_int_flag}")
emitted_vars.add(v["name"])
for obs_name, obs_val in prob.observations.items():
if obs_name not in emitted_vars:
is_int_flag = " INT" if float(obs_val).is_integer() else ""
lines.append(f"VAR {obs_name} {obs_val} {obs_val}{is_int_flag}")
emitted_vars.add(obs_name)
for c in prob.constraints:
wt = c.get("weight", 1.0); expr = str(c['expr']).replace("^", "**")
lines.append(f"{c['kind']} {expr} WEIGHT {wt}")
return "\n".join(lines)
AXL_COMPILER = AXLtoPSLCompiler()
def parse_template_as_axl(template_text: str) -> AXLProblemDef:
lines = template_text.strip().split('\n')
variables, constraints, anchors, observations = [], [], [], {}
minimize_var, name = "", "template"
for line in lines[:6]:
if "HYPOTHESIS" in line.upper():
name = re.sub(r'[^a-zA-Z0-9_]', '_', line.split(":", 1)[-1].strip())[:28]
break
for line in lines:
stripped = line.strip()
if not stripped or stripped.startswith('#'): continue
if stripped.upper().startswith("MINIMIZE:"):
minimize_var = stripped.split(":", 1)[1].strip()
continue
var_match = re.match(r'^([a-zA-Z_0-9,\s]+)\s+in\s+\[([^\]]+)\]\s*(INT)?', stripped, re.IGNORECASE)
if var_match:
vnames = [v.strip() for v in var_match.group(1).split(',')]
parts = [p.strip() for p in var_match.group(2).split(',')]
lo, hi = float(parts[0]), float(parts[1])
is_int = bool(var_match.group(3))
for vn in vnames:
if vn: variables.append({"name": vn, "lo": lo, "hi": hi, "is_int": is_int})
continue
con_match = re.match(r'^(EQ|GEQ|LEQ)(?:\s+weight\s*=\s*([0-9.]+))?\s*:\s*(.+)$', stripped, re.IGNORECASE)
if con_match:
kind, wt, expr = con_match.group(1).upper(), float(con_match.group(2) or 1.0), con_match.group(3).replace('^', '**')
constraints.append({"kind": kind, "weight": wt, "expr": expr})
continue
anc_match = re.match(r'^Anchor:\s*(.+)$', stripped, re.IGNORECASE)
if anc_match:
anchors.append(AXLInvariant(name=f"anc_{len(anchors)}", expr=anc_match.group(1).replace('^', '**')))
continue
obs_match = re.match(r'^Observation:\s*([a-zA-Z_0-9]+)\s*=\s*([0-9.e+\-]+)', stripped, re.IGNORECASE)
if obs_match:
observations[obs_match.group(1)] = float(obs_match.group(2))
return AXLProblemDef(
name=name, description=template_text[:100], axioms=set(),
variables=variables, constraints=constraints, scopes=[],
anchors=anchors, observations=observations, minimize_var=minimize_var
)
def build_core_problem(axl_def: AXLProblemDef) -> core.Problem:
psl_text = AXL_COMPILER.compile(axl_def)
bounds, int_vars = {}, set()
for v in axl_def.variables:
bounds[v['name']] = (v['lo'], v['hi'])
if v.get('is_int'): int_vars.add(v['name'])
for obs_k, obs_v in axl_def.observations.items():
bounds[obs_k] = (obs_v, obs_v)
var_names = list(bounds.keys())
compiled_constraints = []
kind_map = {'EQ': 'equality', 'GEQ': 'inequality', 'LEQ': 'inequality'}
dir_map = {'EQ': 'eq', 'GEQ': 'geq', 'LEQ': 'leq'}
for c in axl_def.constraints:
k = c['kind'].upper()
mc = core.compile_mc(kind_map[k], c['expr'], dir_map[k], var_names, c.get('weight', 1.0))
compiled_constraints.append(mc)
prob = core.Problem(
pid=axl_def.name, variables=var_names, bounds=bounds,
compiled_constraints=compiled_constraints, int_vars=int_vars,
minimize_var=axl_def.minimize_var
)
return prob
# ══════════════════════════════════════════════════════════════════════
# SECTION 3: EXTERNAL / INTERNAL LLM BRIDGE & ROBUST JSON EXTRACTION
# ══════════════════════════════════════════════════════════════════════
def _extract_json_from_text(text: str) -> str:
"""Robust, lexer-style JSON extractor that handles LaTeX groupings."""
text = text.strip()
if not text:
return ""
blocks = re.findall(r'```(?:json)?\s*(\{.*?\})\s*```', text, re.DOTALL | re.IGNORECASE)
for b in blocks:
cleaned = b.strip()
if any(k in cleaned for k in ['"variables"', '"constraints"', '"hypothesis_markdown"', '"name"']):
return cleaned
if blocks:
return blocks[0].strip()
candidates = []
for match in re.finditer(r'\{', text):
start_idx = match.start()
depth = 0
for i in range(start_idx, len(text)):
if text[i] == '{':
depth += 1
elif text[i] == '}':
depth -= 1
if depth == 0:
candidates.append(text[start_idx:i+1])
break
for cand in candidates:
cleaned = cand.strip()
if any(k in cleaned for k in ['"variables"', '"constraints"', '"hypothesis_markdown"', '"name"']):
return cleaned
if candidates:
return max(candidates, key=len).strip()
return text
def call_external_ai(url: str, model: str, system_prompt: str, prompt: str) -> str:
url = url.rstrip("/")
try:
res = requests.post(f"{url}/api/chats", timeout=10)
res.raise_for_status()
chat_id = res.json()["id"]
payload = {"model": model, "prompt": prompt, "system_prompt": system_prompt}
stream_res = requests.post(f"{url}/api/chats/{chat_id}/stream", json=payload, stream=True, timeout=300)
stream_res.raise_for_status()
for _ in stream_res.iter_content(chunk_size=8192): pass
chat_res = requests.get(f"{url}/api/chats/{chat_id}", timeout=10)
chat_res.raise_for_status()
chat_data = chat_res.json()
try: requests.delete(f"{url}/api/chats/{chat_id}", timeout=10)
except: pass
messages = chat_data.get("messages", [])
if not messages: raise ValueError("No messages returned from external AI")
last_msg = messages[-1]
if last_msg.get("role") != "assistant": raise ValueError("Last message is not from assistant")
return last_msg.get("content", "")
except Exception as e:
raise RuntimeError(f"External AI Call Failed: {str(e)}")
def encode_problem(text: str, api_key: str, use_ext: bool, ext_url: str, ext_model: str) -> AXLProblemDef:
if use_ext:
_add_log(f"Calling External AI Encoder ({ext_model})...")
raw = call_external_ai(ext_url, ext_model, oracle.ENCODER_PROMPT, text)
else:
_add_log("Calling Internal Gemini Encoder...")
client = genai.Client(api_key=api_key)
config = types.GenerateContentConfig(
system_instruction=[types.Part.from_text(text=oracle.ENCODER_PROMPT)],
response_mime_type="application/json", temperature=0.1)
resp = client.models.generate_content(
model=GEMINI_MODEL,
contents=[types.Content(role="user", parts=[types.Part.from_text(text=text)])],
config=config)
raw = resp.text
raw = raw.strip()
if not raw:
raise ValueError("The AI model returned an empty response.")
json_str = _extract_json_from_text(raw)
if not json_str:
raise ValueError(f"Could not extract JSON from AI response.\nRaw output: {raw[:200]}...")
try:
d = json.loads(json_str)
except json.JSONDecodeError as e:
raise ValueError(f"JSON Parse Error: {e}\nExtracted String: {json_str[:200]}")
if "error" in d: raise ValueError(f"Isomorphism Invalid: {d['error']} - {d.get('reason', '')}")
anchors = [AXLInvariant(name=a["name"], expr=a["expr"]) for a in d.get("anchors", [])]
variables = [{"name": v["name"], "lo": float(v.get("lo", -1000.0)), "hi": float(v.get("hi", 1000.0)),
"is_int": bool(v.get("is_int", False))} for v in d["variables"]]
return AXLProblemDef(
name=d["name"], description=d.get("description", ""), axioms=set(),
variables=variables, constraints=d.get("constraints", []),
scopes=[], anchors=anchors, observations=d.get("observations", {})
)
def collapse_solution(payload: dict, api_key: str, use_ext: bool, ext_url: str, ext_model: str) -> str:
if use_ext:
_add_log(f"Calling External AI Collapser ({ext_model})...")
raw = call_external_ai(ext_url, ext_model, oracle.COLLAPSER_PROMPT, json.dumps(payload))
else:
_add_log("Calling Internal Gemini Collapser...")
client = genai.Client(api_key=api_key)
config = types.GenerateContentConfig(
system_instruction=[types.Part.from_text(text=oracle.COLLAPSER_PROMPT)],
temperature=0.2)
resp = client.models.generate_content(
model=GEMINI_MODEL,
contents=[types.Content(role="user", parts=[types.Part.from_text(text=json.dumps(payload))])],
config=config)
raw = resp.text
raw = raw.strip()
if not raw:
return "The AI model returned an empty response during collapse."
json_str = _extract_json_from_text(raw)
try:
data = json.loads(json_str)
return data.get("hypothesis_markdown", "No markdown generated.")
except Exception as e:
return f"JSON Decode Error from Collapser: {e}\n\nRaw Output:\n{raw}"
# ══════════════════════════════════════════════════════════════════════
# SECTION 4: THE ORCHESTRATOR LOOP (Adversary Pre-Flight Removed)
# ══════════════════════════════════════════════════════════════════════
def global_solve_task(original_text: str, api_key: str, template_name: str = None,
use_ext: bool = False, ext_url: str = "", ext_model: str = ""):
_update_state(
is_running=True, phase="ENCODING", high_level_logs=[], answer="",
best_ce=float('inf'), total_fired=0, best_ray="—",
recent_traces=deque(maxlen=25), raw_results_json="",
infeasibility_report="", minimize_result=""
)
try:
if template_name:
_add_log(f"Loading Template: {template_name}")
template_text = oracle.STRUCTURAL_HYPOTHESIS_TEMPLATES[template_name]
axl_def = parse_template_as_axl(template_text)
else:
axl_def = encode_problem(original_text, api_key, use_ext, ext_url, ext_model)
_add_log(f"`{axl_def.name}` | {len(axl_def.variables)} vars")
if axl_def.minimize_var: _add_log(f" MINIMIZE: {axl_def.minimize_var}")
is_factorization = any(v['name'] in ['p', 'q', 't', 's', 'd'] for v in axl_def.variables)
_update_state(phase="RAY TRACING")
base_prob = build_core_problem(axl_def)
init_b, init_ce = axioms._mprt_sample(base_prob, 1000)
_add_log("Searching continuous relaxation...")
templates = axioms.generate_templates(base_prob, axioms.compute_sketch(base_prob))
tracer = axioms.SequenceRayTracer(
log_callback=_add_log,
update_state_callback=lambda **kwargs: _update_state(**kwargs)
)
binding, ce, traces, best_ray_name = tracer.trace(
base_prob, init_b, init_ce, templates, max_rays=1500
)
_add_log(f"Deduction Engine Complete. Best CE = {ce:.6f}")
g1_pass = ce < core.SOLVE_THRESHOLD
g2_pass = True
if is_factorization and len(axl_def.observations) > 0:
N_target = int(list(axl_def.observations.values())[0])
# Call ground_truth_verify via the oracle module import
gt_passed, gt_msg = oracle.ground_truth_verify(binding, N_target)
_add_log(f"GROUND TRUTH: {gt_msg}")
if not gt_passed:
_update_state(infeasibility_report=f"GROUND TRUTH FAILED: {gt_msg}")
g2_pass = False
_update_state(phase="COLLAPSING", best_ce=ce)
payload = {
"solved": g1_pass and g2_pass,
"raw_binding": {k: core.safe_round(v, 4) for k, v in binding.items()},
"infeasibility_report": GLOBAL_SOLVE_STATE["infeasibility_report"]
}
ans = collapse_solution(payload, api_key, use_ext, ext_url, ext_model)
_update_state(answer=ans, raw_results_json=json.dumps(payload, indent=2), phase="DONE")
_add_log("Pipeline Complete.")
except Exception as e:
tb_str = traceback.format_exc()
print(tb_str)
error_msg = f"💥 FATAL ERROR ({type(e).__name__}): {str(e)}"
_add_log(error_msg)
md_error = f"## 💥 SYSTEM CRASH\n\n**Error Type:** `{type(e).__name__}`\n**Message:** {str(e)}\n\n### Traceback\n```python\n{tb_str}\n```"
_update_state(phase="ERROR", answer=md_error)
finally:
_update_state(is_running=False)
def trigger_solve(problem_text, api_key, template_name=None, use_ext=False, ext_url="", ext_model=""):
if GLOBAL_SOLVE_STATE["is_running"] or not problem_text.strip(): return
threading.Thread(target=global_solve_task, args=(problem_text, api_key, template_name, use_ext, ext_url, ext_model), daemon=True).start()
def trigger_hyp_solve(template_name, api_key, use_ext=False, ext_url="", ext_model=""):
if GLOBAL_SOLVE_STATE["is_running"]: return
threading.Thread(target=global_solve_task, args=("", api_key, template_name, use_ext, ext_url, ext_model), daemon=True).start()
def run_quantum_demo():
_update_state(is_running=True, phase="QUANTUM SIMULATION", answer="Running Physics Simulator...\nCheck console for output.", high_level_logs=[])
def _run():
try:
oracle.run_shors_demo(15)
oracle.run_shors_demo(35)
oracle.benchmark_heisenberg_vs_schrodinger(20)
_update_state(phase="DONE", answer="Quantum simulation complete. Output logged to console.")
except Exception as e:
tb_str = traceback.format_exc()
print(tb_str)
_add_log(f"💥 QUANTUM DEMO CRASH: {str(e)}")
_update_state(phase="ERROR", answer=f"## 💥 QUANTUM CRASH\n\n```python\n{tb_str}\n```")
finally:
_update_state(is_running=False)
threading.Thread(target=_run, daemon=True).start()
# ══════════════════════════════════════════════════════════════════════
# SECTION 5: GRADIO UI
# ══════════════════════════════════════════════════════════════════════
CSS = """
body{background:#090909;color:#e0e0e0;font-family:monospace;}
.solver-log{font-size:0.75em;background:#0d0d0d;border:1px solid #181818;}
.answer-box{font-size:0.85em;background:#0d0d0d;border:1px solid #222;padding:10px;}
.infeas-banner{background:#1a0000;border:1px solid #5C0000;color:#FF6B6B;padding:8px;border-radius:4px;font-size:0.8em;margin:4px 0;}
"""
def build_live_html():
with STATE_LOCK:
phase = GLOBAL_SOLVE_STATE["phase"]
ce = GLOBAL_SOLVE_STATE["best_ce"]
logs = GLOBAL_SOLVE_STATE["high_level_logs"]
infeas = GLOBAL_SOLVE_STATE["infeasibility_report"]
phase_colors = {"DONE": "#4CAF50", "ERROR": "#F44336", "ADVERSARY PRE-FLIGHT": "#FF9800"}
phase_color = phase_colors.get(phase, "#26C6DA" if phase != "IDLE" else "#888")
log_html = ""
for l in logs[-15:]:
safe_l = html.escape(str(l)).replace('\n', '
')
if "FATAL ERROR" in safe_l or "CRASH" in safe_l:
log_html += f"- {safe_l}
"
else:
log_html += f"- {safe_l}
"
log_html += "
"
infeas_html = f"🚫 {html.escape(infeas[:300])}
" if infeas else ""
return (f""
f"
● Phase: {phase}
"
f"
{log_html}
"
f"{infeas_html}"
f"