""" Symbolic Recursion Coherence API — compute gateway with public data feeds. Run: uvicorn symbolic_recursion.server:app --host 0.0.0.0 --port 8080 """ from __future__ import annotations import hashlib import logging from contextlib import asynccontextmanager from datetime import datetime, timezone from typing import Any, Dict, List, Optional import numpy as np from pathlib import Path import os from fastapi import FastAPI, Header, HTTPException, Request from fastapi.responses import FileResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field 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 .inference.chat_engine import CoherenceChatEngine from .tools.kernel_tools import KernelToolRunner from .inference.huggingface_client import discover_working_model, get_active_model, hf_whoami, load_hf_token from .bots.slack import router as slack_router from .kernel import MetaOperator, SymbolicRecursionKernel, run_worked_examples @asynccontextmanager async def lifespan(app: FastAPI): if load_hf_token(): model = await discover_working_model() if model: logging.getLogger("uvicorn.error").info(f"HF chat model ready: {model}") yield app = FastAPI( title="Symbolic Recursion Coherence Kernel", description="Primal Logic sovereign kernel — Dx(t) and O(f)(t) with public sensor feeds", version="0.1.0", lifespan=lifespan, ) _feed = PublicSensorFeed() _bridge = PrimalBridge() _primallang = PrimalLangBridge() _atlas = AtlasBridge() _gateway = GatewayClient() _chat = CoherenceChatEngine() _tools = KernelToolRunner() _STATIC = Path(__file__).resolve().parent / "static" app.mount("/static", StaticFiles(directory=str(_STATIC)), name="static") app.include_router(slack_router) @app.get("/") async def root_dashboard(): return RedirectResponse("/dashboard") @app.get("/dashboard") async def dashboard(): return FileResponse(_STATIC / "dashboard.html") @app.get("/chat") async def chat_page(): return FileResponse(_STATIC / "chat.html") @app.get("/survey") async def survey_page(): return FileResponse(_STATIC / "survey.html") class ChatRequest(BaseModel): message: str session_id: str = "default" class ChatClearRequest(BaseModel): session_id: str = "default" class ComputeRequest(BaseModel): a: float = 1.0 b: float = 0.091 mu: float = 0.16905 dt: float = 0.01 q_values: Optional[List[float]] = None class LocationRequest(BaseModel): latitude: float = 38.6270 longitude: float = -90.1994 hours: int = Field(default=24, ge=1, le=168) @app.get("/health") async def health() -> Dict[str, Any]: engines = { **_bridge.engines_available, "primallang_interpreter": _primallang.available, } gateway_ok = False try: await _gateway.health() gateway_ok = True except Exception: pass return { "status": "ok", "kernel": "symbolic-recursion", "timestamp": datetime.now(timezone.utc).isoformat(), "engines": engines, "primallang_root": _primallang.root_path, "gateway_connected": gateway_ok, "mcp": "python -m symbolic_recursion.mcp.server", } @app.get("/api/system/verify") async def system_verify() -> Dict[str, Any]: """One-shot verification of all SRC subsystems for the dashboard.""" checks: List[Dict[str, Any]] = [] def add(name: str, ok: bool, detail: str = "", **extra: Any) -> None: checks.append({"name": name, "ok": ok, "detail": detail, **extra}) engines = {**_bridge.engines_available, "primallang_interpreter": _primallang.available} add("Kernel API", True, "symbolic-recursion online") add("Primal Trading", engines.get("primal_trading", False)) add("PrimalLang Physics", engines.get("primallang_physics", False)) add("PrimalLang Interpreter", engines.get("primallang_interpreter", False)) try: await _gateway.health() add("Physics Gateway", True, "localhost:3000") except Exception as e: add("Physics Gateway", False, str(e)[:80]) hf_ok = bool(load_hf_token()) add("HuggingFace Token", hf_ok, "inference" if hf_ok else "set HF_TOKEN in .env") try: active = get_active_model() add("HF Chat Model", hf_ok, active or "none") except Exception: add("HF Chat Model", False, "discovery failed") add("MCP Server", True, "python -m symbolic_recursion.mcp.server") add("Telegram Bot", bool(os.getenv("TELEGRAM_BOT_TOKEN")), "TELEGRAM_BOT_TOKEN") add("Slack Bot", bool(os.getenv("SLACK_BOT_TOKEN")), "SLACK_BOT_TOKEN + signing secret") try: snap = await _feed.fetch_current() add("Live Sensors", True, f"Q(t)={snap.q_t:.3f}") except Exception as e: add("Live Sensors", False, str(e)[:80]) prox = STORE.build_snapshot() add( "Proximity Survey", prox.source != "none" or STORE.status()["wifi_networks"] > 0, f"source={prox.source}, readings={prox.phone_readings}", survey_url="/survey", ) wifi = scan_wifi_networks() add("WiFi Scan", len(wifi) > 0, f"{len(wifi)} networks") cym = _primallang.run_cymatics(freq_hz=528.0) add("PrimalLang Cymatics", cym.ok, cym.error or "528 Hz OK") rows = benchmark_operators() add("Operator Benchmarks", len(rows) >= 10, f"{len(rows)} operators") return { "timestamp": datetime.now(timezone.utc).isoformat(), "all_ok": all(c["ok"] for c in checks if c["name"] not in ("Telegram Bot", "Slack Bot", "Physics Gateway")), "checks": checks, "links": { "chat": "/chat", "survey": "/survey", "dashboard": "/dashboard", "gateway": os.getenv("GATEWAY_URL", "http://localhost:3000"), "hf_space": "https://dontelightfoot-symbolic-recursive-engine.hf.space", }, } @app.get("/v1/system/constants") async def constants() -> Dict[str, float]: return { "a_default": 1.0, "b_default": 0.091, "mu": 0.16905, "D_attractor": 149.999, "S_ratio": 23.0983417, } @app.post("/api/compute/dx") async def compute_dx(req: ComputeRequest) -> Dict[str, Any]: if not req.q_values: raise HTTPException(400, "q_values required") kernel = SymbolicRecursionKernel(a=req.a, mu=req.mu) q = np.asarray(req.q_values) dx = kernel.integrate_series(q, dt=req.dt) return { "dx_final": float(dx[-1]), "trajectory": dx.tolist(), "status": "STABLE", } @app.post("/api/compute/meta") async def compute_meta(req: ComputeRequest) -> Dict[str, Any]: if not req.q_values: raise HTTPException(400, "q_values required (used as f(t))") meta = MetaOperator(b=req.b, mu=req.mu) f = np.asarray(req.q_values) of = meta.apply_series(f, dt=req.dt) collapse = meta.collapse_to_attractor(f, dt=req.dt) return { "of_final": float(of[-1]), "collapse": collapse, "trajectory": of.tolist(), "status": "LIPSCHITZ_BOUND_OK", } @app.get("/api/examples/worked") async def worked_examples() -> Dict[str, Any]: return run_worked_examples() @app.get("/api/benchmarks") async def benchmarks() -> Dict[str, Any]: rows = benchmark_operators() gains = [r["gain_pct"] for r in rows if isinstance(r["gain_pct"], (int, float))] return { "table": rows, "summary": { "operators_tested": len(rows), "mean_gain_pct": round(float(np.mean(gains)), 1) if gains else 0, "max_gain_pct": round(float(np.max(gains)), 1) if gains else 0, "min_gain_pct": round(float(np.min(gains)), 1) if gains else 0, }, } @app.post("/api/feed/live") async def live_feed(loc: LocationRequest) -> Dict[str, Any]: _feed.latitude = loc.latitude _feed.longitude = loc.longitude snap = await _feed.fetch_current() kernel = SymbolicRecursionKernel() dx = kernel.step(snap.q_t, dt=1.0) return { "snapshot": { "timestamp": snap.timestamp, "DT": snap.dt_dev, "DP": snap.dp_dev, "DEM": snap.dem_dev, "DW": snap.dw_dev, "Q_t": snap.q_t, "source": snap.source, }, "dx_instant": dx, } @app.post("/api/feed/series") async def feed_series(loc: LocationRequest) -> Dict[str, Any]: _feed.latitude = loc.latitude _feed.longitude = loc.longitude series = await _feed.fetch_series(hours=loc.hours) primal_result = _bridge.from_sensor_snapshots(series, dt=3600.0) q_arr = np.array([s.q_t for s in series]) kernel = SymbolicRecursionKernel() dx = kernel.integrate_series(q_arr, dt=3600.0) atlas_fusion = _atlas.fuse_with_symbolic_recursion(q_arr, dx) trust_anchor = hashlib.sha512(str(dx[-1]).encode()).hexdigest() return { "points": len(series), "source": series[0].source if series else "none", "timestamps": [s.timestamp for s in series], "q_series": q_arr.tolist(), "dx_series": dx.tolist(), "primal_bridge": primal_result, "atlas_fusion": atlas_fusion, "coherence_retention": coherence_retention(q_arr, dx), "trust_anchor_sha512": trust_anchor, } @app.get("/api/gateway/status") async def gateway_status() -> Dict[str, Any]: try: health = await _gateway.health() consts = await _gateway.constants() return {"connected": True, "health": health, "constants": consts} except Exception as e: return {"connected": False, "error": str(e)} @app.post("/api/gateway/offload") async def gateway_offload(t: float = 3.5, x0: float = 150.0) -> Dict[str, Any]: try: psi = await _gateway.compute_psi(t=t, x0=x0) rpo = await _gateway.compute_rpo() stk = await _gateway.stk_simulate(steps=1000) return {"psi": psi, "rpo": rpo, "stk": stk} except Exception as e: raise HTTPException(502, f"Gateway offload failed: {e}") from e @app.get("/api/inference/status") async def inference_status() -> Dict[str, Any]: import os hf = await hf_whoami() active = get_active_model() if load_hf_token() and active == os.getenv("HF_CHAT_MODEL", "meta-llama/Llama-3.1-8B-Instruct"): discovered = await discover_working_model() if discovered: active = discovered return { "huggingface": hf, "token_configured": bool(load_hf_token()), "chat_model": active, "embed_model": os.getenv("HF_EMBED_MODEL", "sentence-transformers/all-MiniLM-L6-v2"), "providers": ["huggingface", "gemini", "local-kernel"], } @app.post("/api/chat") async def chat_inference(req: ChatRequest) -> Dict[str, Any]: try: result = await _chat.chat(req.message, session_id=req.session_id) # Public chat surface: prose only — never raw kernel/Kalman/fatigue dumps return { "reply": result.reply, "tools_used": result.tools_used, "model": result.model, "session_id": result.session_id, "semantic_coherence": result.semantic_coherence, } except ValueError as e: raise HTTPException(400, str(e)) from e except Exception as e: raise HTTPException(500, f"Inference error: {e}") from e @app.get("/api/chat/history") async def chat_history(session_id: str = "default") -> Dict[str, Any]: return {"session_id": session_id, "messages": _chat.get_history(session_id)} @app.post("/api/chat/clear") async def chat_clear(req: ChatClearRequest) -> Dict[str, str]: _chat.clear_session(req.session_id) return {"status": "cleared", "session_id": req.session_id} class PrimalLangRunRequest(BaseModel): script_path: str = "" code: str = "" class PrimalLangCymaticsRequest(BaseModel): freq_hz: float = 528.0 @app.get("/api/primallang/status") async def primallang_status() -> Dict[str, Any]: return { "available": _primallang.available, "root": _primallang.root_path, "constants": _primallang.constants(), "example_count": len(_primallang.list_examples()), } @app.get("/api/primallang/examples") async def primallang_examples(category: str = "") -> Dict[str, Any]: return { "examples": _primallang.list_examples(category=category), "category_filter": category or None, } @app.post("/api/primallang/run") async def primallang_run(req: PrimalLangRunRequest) -> Dict[str, Any]: if req.code.strip(): result = _primallang.run_code(req.code) elif req.script_path.strip(): result = _primallang.run_script(req.script_path) else: raise HTTPException(400, "Provide script_path or code") return { "ok": result.ok, "stdout": result.stdout, "context": result.context, "error": result.error, "script_path": result.script_path, } @app.post("/api/primallang/cymatics") async def primallang_cymatics(req: PrimalLangCymaticsRequest) -> Dict[str, Any]: result = _primallang.run_cymatics(freq_hz=req.freq_hz) return { "ok": result.ok, "freq_hz": req.freq_hz, "stdout": result.stdout, "context": result.context, "error": result.error, } @app.post("/api/primallang/fuse-q") async def primallang_fuse_q(loc: LocationRequest) -> Dict[str, Any]: _feed.latitude = loc.latitude _feed.longitude = loc.longitude series = await _feed.fetch_series(hours=loc.hours) q = [s.q_t for s in series] result = _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), } _webhook_log: List[Dict[str, Any]] = [] @app.post("/api/webhook/hf") async def huggingface_webhook( request: Request, x_webhook_secret: Optional[str] = Header(default=None), ) -> Dict[str, Any]: """Receive Hugging Face repo-update webhooks. Set HF_WEBHOOK_SECRET in .env.""" expected = os.getenv("HF_WEBHOOK_SECRET", "") if expected and x_webhook_secret != expected: raise HTTPException(401, "Invalid webhook secret") payload = await request.json() event = payload.get("event", {}) repo = event.get("repo", {}) entry = { "received_at": datetime.now(timezone.utc).isoformat(), "action": event.get("action"), "repo": repo.get("name"), "repo_type": repo.get("type"), "author": event.get("author", {}).get("name"), } _webhook_log.append(entry) if len(_webhook_log) > 50: _webhook_log.pop(0) # On Space push: warm inference + health check actions = [] if repo.get("type") == "space" and event.get("action") in ("update", "create"): try: h = await health() actions.append({"health_check": h.get("status")}) except Exception as e: actions.append({"health_check_error": str(e)}) return {"status": "ok", "event": entry, "actions": actions} @app.get("/api/webhook/hf/log") async def webhook_log() -> Dict[str, Any]: return {"events": _webhook_log[-20:]} class PhoneSurveyRequest(BaseModel): device_id: str = "default" readings: List[Dict[str, Any]] @app.post("/api/survey/phone") async def survey_phone(req: PhoneSurveyRequest) -> Dict[str, Any]: n = STORE.add_phone_batch(req.device_id, req.readings) snap = await _feed.fetch_current() fused = STORE.fused_q(snap.q_t, req.device_id) return {"ingested": n, "fused": fused, "wifi_count": STORE.status()["wifi_networks"]} @app.post("/api/survey/wifi") async def survey_wifi() -> Dict[str, Any]: 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} for n in networks[:15] ], } @app.get("/api/survey/status") async def survey_status(device_id: str = "default") -> Dict[str, Any]: snap = await _feed.fetch_current() fused = STORE.fused_q(snap.q_t, device_id) prox = STORE.build_snapshot(device_id) return { "store": STORE.status(), "proximity": { "DM": prox.dm_dev, "DO": prox.do_dev, "DWi": prox.dwi_dev, "DGeo": prox.dgeo_dev, "Q_proximity": prox.q_proximity, "source": prox.source, }, "fused": fused, "survey_url": "/survey", } @app.get("/api/tools/{tool_name}") async def run_kernel_tool(tool_name: str, hours: int = 12, device_id: str = "default") -> Dict[str, Any]: if tool_name not in ( "health", "live_sensors", "sensor_series", "benchmark", "atlas", "worked_examples", "compute", "gateway", "coherence_analysis", "primallang_examples", "primallang_run", "primallang_cymatics", "primallang_fuse_q", "proximity_survey", "wifi_survey", ): raise HTTPException(404, f"Unknown tool: {tool_name}") return await _tools.run(tool_name, hours=hours)