"""Shared kernel tool execution for chat engine, MCP server, and REST API.""" from __future__ import annotations import os from pathlib import Path from typing import Any, Dict, List, Optional import numpy as np from ..coherence import benchmark_operators, coherence_retention from ..data.public_feeds import PublicSensorFeed from ..data.survey_store import STORE from ..data.wifi_survey import scan_wifi_networks from ..integrations.atlas_bridge import AtlasBridge from ..integrations.gateway_client import GatewayClient from ..integrations.primal_bridge import PrimalBridge from ..integrations.primallang_bridge import PrimalLangBridge from ..kernel import SymbolicRecursionKernel, run_worked_examples from ..equations import LambdaLightfootProtocol, LambdaMode, get_equation, list_equations from ..equations.catalog import catalog_stats from ..equations.lambda_protocol import MeshGraph, sha512_hex from ..verification import PrimalVerificationAgent, calculate_semantic_fatigue from ..verification.kalman import KalmanParams, PrimalKalmanFilter from ..dbits import ( DbitParams, apply_gate, continuous_offset_series, list_gates, list_spine, multi_dbit_trajectory, run_all_proofs, run_proof, spine_summary, trajectory, ) TOOL_NAMES = [ "health", "live_sensors", "sensor_series", "benchmark", "atlas", "worked_examples", "compute", "gateway", "coherence_analysis", "primallang_run", "primallang_examples", "primallang_cymatics", "primallang_fuse_q", "proximity_survey", "wifi_survey", "lambda_protocol", "equation_catalog", "verify_claims", "semantic_fatigue", "primal_kalman", "dbits_proofs", "dbits_simulate", "dbits_gate", "dbits_spine", "dbits_offset", "amyloid_beta", ] def _load_dotenv() -> None: env_path = Path(__file__).resolve().parents[2] / ".env" if not env_path.exists(): return for line in env_path.read_text(encoding="utf-8").splitlines(): if "=" in line and not line.strip().startswith("#"): k, v = line.split("=", 1) os.environ.setdefault(k.strip(), v.strip()) class KernelToolRunner: """Executes symbolic recursion kernel tools.""" def __init__(self): _load_dotenv() self._feed = PublicSensorFeed() self._bridge = PrimalBridge() self._primallang = PrimalLangBridge() self._atlas = AtlasBridge() self._gateway = GatewayClient() self._verifier = PrimalVerificationAgent() def detect_tools(self, message: str) -> List[str]: m = message.lower() tools: List[str] = ["health"] if any(w in m for w in ("sensor", "live", "weather", "temperature", "wind", "pressure", "kp", "feed", "q(t)")): tools.append("live_sensors") if any(w in m for w in ("series", "trajectory", "history", "hours", "timeline", "chart")): tools.append("sensor_series") if any(w in m for w in ("benchmark", "operator", "gradient", "fourier", "wavelet", "compare", "retention")): tools.append("benchmark") if any(w in m for w in ("atlas", "3ia", "attractor", "planck", "memory")): tools.append("atlas") if any(w in m for w in ("example", "forcing", "oscillat", "fractal", "hybrid", "decay")): tools.append("worked_examples") if any(w in m for w in ("dx", "integral", "compute", "meta", "o(f)", "recursion")): tools.append("compute") if any(w in m for w in ("gateway", "psi", "rpo", "stk", "offload")): tools.append("gateway") if any(w in m for w in ("coherence", "collapse", "integrity", "stable", "diagnostic", "semantic")): tools.append("coherence_analysis") if any( w in m for w in ( "phone", "proximity", "survey", "wifi", "wi-fi", "accelerometer", "motion", "orientation", "gps", "mobile", "device sensor", ) ): tools.append("proximity_survey") if any(w in m for w in ("wifi", "wi-fi", "network", "ssid", "signal")): tools.append("wifi_survey") if any( w in m for w in ( "primallang", "primal lang", "primal script", ".primal", "cymatics", "chladni", "quantum state", "alein", ) ): tools.append("primallang_examples") if any(w in m for w in ("528", "440", "741", "hz", "cymatics", "chladni", "freq", "frequency")): tools.append("primallang_cymatics") elif any(w in m for w in ("fuse", "q(t)", "q series", "sensor fusion")): tools.append("primallang_fuse_q") else: tools.append("primallang_run") if any( w in m for w in ( "lambda", "lightfoot-lambda", "lightfoot lambda", "llp", "command sovereignty", "mesh resilience", "sha-512", "sha512", "guardian mode", "blacksite", "harmonic profile", "rie", ) ): tools.append("lambda_protocol") if any(w in m for w in ("equation", "catalog", "latex", "paper formula", "encode", "protocol math")): tools.append("equation_catalog") if any( w in m for w in ( "verify", "verification", "hallucination", "fact check", "fact-check", "true or false", "is this true", "claim", "suspect", ) ): tools.append("verify_claims") if any(w in m for w in ("fatigue", "redundancy", "collapse", "srec", "lexical", "repetition")): tools.append("semantic_fatigue") if any(w in m for w in ("kalman", "innovation", "x_k", "coherence filter")): tools.append("primal_kalman") if any( w in m for w in ( "dbit", "dbits", "dynamic quantum", "lyapunov", "stability box", "gate family", "c_pl", "non-markov", "nonmarkov", ) ): tools.append("dbits_proofs") if any(w in m for w in ("simulat", "trajectory", "scale", "multi")): tools.append("dbits_simulate") if "gate" in m: tools.append("dbits_gate") if "spine" in m or "ontology" in m: tools.append("dbits_spine") if "offset" in m or "attractor" in m or "d0" in m: tools.append("dbits_offset") if any(w in m for w in ("amyloid", "abeta", "aβ", "beta-42", "beta42", "aβ42")): tools.append("amyloid_beta") return list(dict.fromkeys(tools)) async def run( self, name: str, *, script_path: str = "", code: str = "", freq_hz: float = 528.0, category: str = "", hours: int = 12, device_id: str = "default", eq_id: str = "", family: str = "", mode: str = "guardian", message: str = "authorize hold relay", biometric_ok: bool = True, timestamp_ok: bool = True, n_points: int = 200, text: str = "", scores: Optional[List[float]] = None, gate: str = "F9_pl_kernel", steps: int = 150, n_agents: int = 32, proof: str = "", x0: float = 0.5, m0: float = 0.2, t_end: float = 4.0, ) -> Dict[str, Any]: if name == "health": from ..inference.huggingface_client import load_hf_token gw_ok = False try: await self._gateway.health() gw_ok = True except Exception: pass return { "engines": { **self._bridge.engines_available, "primallang_interpreter": self._primallang.available, }, "primallang_root": self._primallang.root_path, "gateway_connected": gw_ok, "huggingface_authenticated": bool(load_hf_token()), "constants": {"mu": 0.16905, "a": 1.0, "b": 0.091, "D_attractor": 149.999}, } if name == "live_sensors": snap = await self._feed.fetch_current() fused = STORE.fused_q(snap.q_t, device_id) q_use = fused["Q_total"] if fused["proximity"]["source"] != "none" else snap.q_t dx = SymbolicRecursionKernel().step(q_use, dt=1.0) return { "timestamp": snap.timestamp, "DT": snap.dt_dev, "DP": snap.dp_dev, "DEM": snap.dem_dev, "DW": snap.dw_dev, "Q_t": snap.q_t, "Q_total": fused["Q_total"], "proximity": fused["proximity"], "dx_instant": dx, "source": snap.source, "survey_url": "/survey", } if name == "sensor_series": series = await self._feed.fetch_series(hours=hours) q = np.array([s.q_t for s in series]) dx = SymbolicRecursionKernel().integrate_series(q, dt=3600.0) primal = self._bridge.from_sensor_snapshots(series, dt=3600.0) return { "points": len(series), "dx_final": float(dx[-1]), "coherence_retention": coherence_retention(q, dx), "atlas_fusion": self._atlas.fuse_with_symbolic_recursion(q, dx), "ewic_control": primal.get("ewic_control"), } if name == "benchmark": rows = benchmark_operators() return {"operators_tested": len(rows), "top_gains": rows[:5]} if name == "atlas": series = await self._feed.fetch_series(hours=6) q = np.array([s.q_t for s in series]) dx = SymbolicRecursionKernel().integrate_series(q, dt=3600.0) return self._atlas.fuse_with_symbolic_recursion(q, dx) if name == "worked_examples": return run_worked_examples(n_points=100) if name == "compute": return {"worked_examples": run_worked_examples(n_points=50), "meta": {"b": 0.091, "mu": 0.16905}} if name == "gateway": try: return {"psi": await self._gateway.compute_psi(), "constants": await self._gateway.constants()} except Exception as e: return {"error": str(e)} if name == "coherence_analysis": series = await self._feed.fetch_series(hours=hours) q = np.array([s.q_t for s in series]) dx = SymbolicRecursionKernel().integrate_series(q, dt=3600.0) score = coherence_retention(q, dx) return { "coherence_retention": score, "recursion_integrity": "PRESERVED" if score > 0.4 else "BREACH_RISK", "dx_final": float(dx[-1]), "diagnosis": "Recursive alignment intact." if score > 0.5 else "Monitor sensor coupling.", } if name == "primallang_examples": return { "available": self._primallang.available, "constants": self._primallang.constants(), "examples": self._primallang.list_examples(category=category)[:20], } if name == "primallang_run": if code.strip(): result = self._primallang.run_code(code) elif script_path.strip(): result = self._primallang.run_script(script_path) else: default = "examples/alein/cymatics/run_41_528hz_chladni.primal" result = self._primallang.run_script(default) return { "ok": result.ok, "stdout": result.stdout, "context": result.context, "error": result.error, "script_path": result.script_path, } if name == "primallang_cymatics": result = self._primallang.run_cymatics(freq_hz=freq_hz) return { "ok": result.ok, "freq_hz": freq_hz, "stdout": result.stdout, "context": result.context, "error": result.error, } if name == "primallang_fuse_q": series = await self._feed.fetch_series(hours=hours) q = [s.q_t for s in series] result = self._primallang.fuse_q_series(q, dt=3600.0) return { "ok": result.ok, "stdout": result.stdout, "context": result.context, "error": result.error, "q_points": len(q), } if name == "proximity_survey": snap = await self._feed.fetch_current() prox = STORE.build_snapshot(device_id) fused = STORE.fused_q(snap.q_t, device_id) return { "fused": fused, "proximity": { "DM": prox.dm_dev, "DO": prox.do_dev, "DWi": prox.dwi_dev, "DGeo": prox.dgeo_dev, "Q_proximity": prox.q_proximity, "phone_readings": prox.phone_readings, "wifi_networks": prox.wifi_networks, "source": prox.source, }, "store": STORE.status(), "survey_url": "http://localhost:8080/survey", "hint": "Open /survey on your phone to stream motion, GPS, and network sensors.", } if name == "wifi_survey": networks = scan_wifi_networks() STORE.set_wifi_scan(networks) return { "count": len(networks), "networks": [ {"ssid": n.ssid, "signal_pct": n.signal_pct, "channel": n.channel, "auth": n.auth} for n in networks[:20] ], "scanned_at": STORE._wifi_scanned_at, } if name == "lambda_protocol": # Live environmental Θ from public sensors when available dt_dev = dp_dev = dem_dev = dw_dev = 0.0 try: snap = await self._feed.fetch_current() dt_dev, dp_dev, dem_dev, dw_dev = snap.dt_dev, snap.dp_dev, snap.dem_dev, snap.dw_dev except Exception: pass mode_s = mode if mode in ("guardian", "override", "blacksite") else "guardian" proto = LambdaLightfootProtocol(mode=LambdaMode(mode_s)) msg = message or "authorize hold relay" mesh = MeshGraph() mesh.add_edge("alpha", "beta", 0.9) mesh.add_edge("beta", "gamma", 0.6) mesh.add_edge("alpha", "gamma", 0.3) st = proto.evaluate( message=msg, trusted_hash=sha512_hex(msg), biometric_ok=biometric_ok, timestamp_ok=timestamp_ok, mesh=mesh, dt_dev=dt_dev, dp_dev=dp_dev, dem_dev=dem_dev, dw_dev=dw_dev, ) return { "protocol": "Lightfoot-Lambda Protocol (LLP)", "master": "L = Φ(S, C, M, H, Θ)", "entry_point": "lambda_protocol", "state": { "mode": st.mode.value, "S": st.S, "C": st.C, "M": st.M, "H": st.H, "Theta": st.theta, "Dx": st.dx, "L_score": st.L_score, "stable": st.stable, "stealth": st.stealth, }, "phi": proto.master_phi(st), "diagnosis": proto.diagnose(), "rie": st.rie, "notes": st.notes, "equations": ["L01", "L02", "L08", "L10", "L15", "L17", "L19", "L20", "L22"], "catalog_stats": catalog_stats(), } if name == "equation_catalog": rows = list_equations(family=family or None) detail = None if eq_id: try: eq = get_equation(eq_id) detail = { "id": eq.id, "name": eq.name, "latex": eq.latex, "family": eq.family, "description": eq.description, "source": eq.source, "callable": eq.eval is not None, } except KeyError as e: detail = {"error": str(e)} return { "stats": catalog_stats(), "equations": rows, "detail": detail, "paper_note": ( "RELEASE STACK: (1) DBITS Dynamic Quantum Bits D01–D24 + 6 proof certificates; " "(2) Lightfoot-Lambda L=Φ(S,C,M,H,Θ); " "(3) Primal Verification Agent; (4) SRC Dx/O(f). " "Paper: Dynamic Quantum Bits (dbits) — Donte Lightfoot 2025." ), } if name == "dbits_proofs": if proof and proof.strip(): return {"single": run_proof(proof.strip()), "paper": "DBITS Donte Lightfoot 2025"} return run_all_proofs() if name == "dbits_simulate": p = DbitParams() x, m = trajectory(p, int(steps), x0=float(x0), m0=float(m0)) scale = multi_dbit_trajectory(n_agents=int(n_agents), steps=min(int(steps), 300)) return { "paper": "DBITS discrete PL + multi-dbit scale", "params": {"delta": p.delta, "lam": p.lam, "kappa": p.kappa, "a_c": p.a_c}, "rho": p.rho(), "in_stability_box": p.in_stability_box(), "single_dbit": { "steps": int(steps), "x_final": float(x[-1]), "m_final": float(m[-1]), "x_series": x[:: max(1, len(x) // 40)].tolist(), "m_series": m[:: max(1, len(m) // 40)].tolist(), }, "multi_dbit": scale, "equations": ["D02", "D03", "D04", "D05", "D21"], "mu": 0.16905, "D0": 149.9992314, } if name == "dbits_gate": return apply_gate(gate or "F9_pl_kernel", x=float(x0), m=float(m0)) if name == "dbits_spine": return { "summary": spine_summary(), "nodes": list_spine(), "gates": list_gates(), } if name == "dbits_offset": return continuous_offset_series( t_end=float(t_end), n_points=min(int(n_points), 100), x0=float(x0) if x0 != 0.5 else 150.0, ) if name == "amyloid_beta": from ..models.amyloid_beta import ( AmyloidBeta42Model, package_inventory, verify_package_integrity, ) inv = package_inventory() integrity = verify_package_integrity() model = AmyloidBeta42Model() run = model.run(use_package=True) # Full circuit is embedded on the model object itself. full_circuit = model.full_circuit return { "model": "Amyloid beta-42", "source": ( "Peer-review package " "Amyloid_beta42_full_spectrum_peer_review_package " "(statevector + samples + full circuit embedded in model)" ), "inventory": inv, "integrity": { "matched": integrity.get("matched"), "checked": integrity.get("checked"), "all_core_match": integrity.get("all_core_match"), }, "circuit_embedded_in_model": True, "circuit_sha256": model.circuit_sha256, "circuit_moments": model.circuit_moments, "full_circuit": full_circuit, "run": run, "note": ( "4-qubit truncated Aβ42-labeled quantum prototype. " "The full circuit is embedded in AmyloidBeta42Model " "(model.full_circuit). Not a full biological model." ), } if name == "verify_claims": payload = text or message or code or "" if not payload.strip(): payload = ( "Hamlet was written by William Shakespeare around 1601. " "Captain John Vance stepped on the sun in 1984." ) report = self._verifier.verify_text(payload) return report.to_dict() if name == "semantic_fatigue": payload = text or message or code or "test input with repeated repeated repeated words words words" st = calculate_semantic_fatigue(payload, [], None) return { "agent": "primal-verification-agent", "lexical_diversity": st.lexical_diversity, "redundancy_score": st.redundancy_score, "cumulative_fatigue": st.cumulative_fatigue, "beta_intent": st.beta_intent, "echo_integrand": st.echo_integrand, "wave_integrator": st.wave_integrator, "is_collapsed": st.is_collapsed, "collapse_factor": st.collapse_factor, "top_repeated_words": st.top_repeated_words, } if name == "primal_kalman": z = scores if scores is not None else [0.95, 0.9, 0.2, 0.85] kf = PrimalKalmanFilter(KalmanParams()) steps = kf.observe_sequence([float(x) for x in z]) return { "agent": "primal-verification-agent", "observations": list(z), "final_xk": kf.xk, "final_yk": kf.yk, "steps": [ {"zk": s.zk, "xk": s.xk, "yk": s.yk, "innovation": s.innovation} for s in steps ], "status_last": kf.status_from_zk(steps[-1].zk) if steps else "UNVERIFIED", } return {"error": f"Unknown tool: {name}"}