symbolic_recursion/server.py

"""
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)