serve/xero_chat_server.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
XERO Chat Server — interactive console for the digital organism
================================================================
A zero-dependency (Python standard library only) web app that exposes
every XERO capability through a chat window. Designed to run on the
Toronto front-end (`:8883`) and relay to the India model host over the
private harmonic port (`:3888`), but it runs standalone on any host.

    python3 serve/xero_chat_server.py --host 127.0.0.1 --port 8883

Security:
  * Set XERO_CHAT_TOKEN to require `?token=…` (or `X-XERO-Token` header)
    before the API answers — use this whenever the port is public.
  * Bind to 127.0.0.1 for local/SSH-tunnel testing (default).

Author: Michael Laurence Curzi · License: MIT (Attribution Required)
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import time
import traceback
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.join(ROOT, "modules"))

TOKEN = os.environ.get("XERO_CHAT_TOKEN", "")
BOOT_TS = time.time()


# ════════════════════════════════════════════════════════════════
# XERO capability dispatch — each handler returns a JSON-able dict
# ════════════════════════════════════════════════════════════════
def _awaken() -> dict:
    from vovina_custom_training_weights import awaken
    r = awaken()
    xs = r.get("xero_summary", {})
    return {"phases": list(r.get("phases", {}).keys()),
            "organ_systems": xs.get("organ_systems"),
            "elapsed_seconds": round(r.get("elapsed_seconds", 0), 4),
            "founding_phrase": r.get("founding_phrase", "")}


def _encode(text: str) -> dict:
    from vovina_digital_genome import text_to_dna, dna_to_text
    dna = text_to_dna(text)
    return {"input": text, "dna": dna, "length_nt": len(dna),
            "roundtrip_ok": dna_to_text(dna) == text}


def _decode(dna: str) -> dict:
    from vovina_digital_genome import dna_to_text
    return {"dna": dna, "text": dna_to_text(dna.strip().upper())}


def _route(token: str) -> dict:
    from vovina_blockchain_organelles import route_codon, route_gene
    token = token.strip()
    if len(token) == 3 and set(token.upper()) <= set("ATGC"):
        o = route_codon(token.upper())
        tier = "codon"
    else:
        o = route_gene(token)
        tier = "gene"
    return {"input": token, "tier": tier, "organelle": o.language.value}


def _uvns(seq: str) -> dict:
    from vovina_universal_vector import genome_to_stack, coherence_report, from_base
    seq = "".join(c for c in seq.upper() if c in "ATGC") or "ATGC"
    rep = coherence_report(genome_to_stack(seq))
    freqs = {b: from_base(b).potency for b in "ATGC"}
    return {"sequence": seq, "coherence": rep, "base_frequencies_hz": freqs}


def _harmonic(seq: str) -> dict:
    from vovina_genetic_pipeline import NUCLEOBASE
    seq = "".join(c for c in seq.upper() if c in "ATGC") or "ATGC"
    chord = [NUCLEOBASE[b]["hz_em"] for b in seq]
    return {"sequence": seq, "chord_hz": chord,
            "mean_hz": round(sum(chord) / len(chord), 3)}


def _surplus(args: str) -> dict:
    from vovina_interaction_surplus import surplus, lipschitz_constant
    parts = args.split()
    u = float(parts[0]) if parts else 0.5
    N = int(parts[1]) if len(parts) > 1 else 22
    return {"u": u, "N": N, "surplus": round(surplus(u, N), 6),
            "lipschitz_ceiling": lipschitz_constant(N)}


def _decide(args: str) -> dict:
    options = [o.strip() for o in args.split("|") if o.strip()] or ["yes", "no"]
    from vovina_free_will_code import seal_choice
    sig = seal_choice(entity_id="XERO", choice_data=" | ".join(options))
    # the unforgeable, single-use zero-point nonce drives the choice
    choice = options[sig.vector_delta() % len(options)]
    return {"options": options, "choice": choice, "verified": sig.verify(),
            "vector_delta": sig.vector_delta(),
            "receipt": sig.sealed[:18] + "…" + sig.sealed[-8:]}


def _compile(text: str) -> dict:
    from vovina_genome_compiler import code_to_dna, dna_to_code
    dna = code_to_dna(text)
    return {"input_bytes": len(text.encode()), "dna_nt": len(dna),
            "roundtrip_ok": dna_to_code(dna) == text,
            "dna_head": dna[:60] + ("…" if len(dna) > 60 else "")}


def _metabolize(text: str) -> dict:
    """Default interaction: the organism 'receives' your words — encodes them
    to DNA, routes the lead codon to an organelle, and witnesses the result."""
    from vovina_digital_genome import text_to_dna
    from vovina_blockchain_organelles import route_codon
    import hashlib
    dna = text_to_dna(text)
    codon = dna[:3] if len(dna) >= 3 else "ATG"
    organelle = route_codon(codon).language.value
    witness = hashlib.sha256(dna.encode()).hexdigest()[:16]
    return {"heard": text, "dna_nt": len(dna), "lead_codon": codon,
            "expressed_in": organelle, "witness": witness}


# ── conversational layer: brain · limbs · relay ─────────────────
def _talk(message: str, user_id: str = "anon") -> dict:
    from organism_dialogue import relay
    return relay().say(user_id, message)


def _act(modality: str, payload: dict, user_id: str = "anon") -> dict:
    from organism_dialogue import relay
    return relay().act(user_id, modality, payload)


def _capabilities() -> list:
    from organism_dialogue import relay
    return relay().mind.capabilities()


def _telemetry() -> dict:
    """Live evolution/culturing telemetry for public monitoring."""
    tl = os.path.join(ROOT, "testing_logs")

    def rd(name):
        try:
            return json.load(open(os.path.join(tl, name)))
        except Exception:
            return {}

    auto, cult, done = rd("AUTONOMY_CAMPAIGN.json"), rd("CULTURING_PROGRESS.json"), rd("CAMPAIGN_DONE.json")
    poly = rd("POLYMATH_PROGRESS.json")
    t = {"organism": "XERO", "campaign": {
        "cycle": poly.get("cycle") or auto.get("cycle") or cult.get("cycle") or 0,
        "target": auto.get("target") or cult.get("cycles_target") or 0,
        "fitness": poly.get("fitness") or auto.get("fitness") or cult.get("champion_fitness") or 0,
        "rate_cps": poly.get("rate_cps") or auto.get("rate_cps") or cult.get("rate_cps") or 0,
        "reports": auto.get("reports", 0),
        "done": bool(done.get("ready_to_publish"))}}
    if poly:
        t["polymath"] = {"total_fetched": poly.get("total_fetched", 0),
                         "sources_exhausted": poly.get("sources_exhausted", 0),
                         "domains": poly.get("domains", {})}
    gpu = rd("GPU_TRAIN.json")
    if gpu:
        t["gpu_train"] = {"available": gpu.get("available"),
                          "gpus": gpu.get("gpus", 0),
                          "matrix": gpu.get("matrix"),
                          "gflops_est": gpu.get("gflops_est"),
                          "vram_used_gb": gpu.get("vram_used_gb"),
                          "bursts": gpu.get("bursts"),
                          "corpus_units": gpu.get("corpus_units")}
    body = rd("BODY_VITALS.json")
    if body:
        sysmap = body.get("systems") or {}
        t["body"] = {"homeostasis_index": body.get("homeostasis_index"),
                     "coherence": body.get("coherence"),
                     "heart_rate": body.get("heart_rate"),
                     "faculties_live": body.get("faculties_live"),
                     "systems": {n: (sv.get("activation") if isinstance(sv, dict) else sv)
                                 for n, sv in sysmap.items()}}
    rdir = os.path.join(ROOT, "research_data", "reports")
    t["research_reports"] = len([f for f in os.listdir(rdir) if f.endswith(".json")]) if os.path.isdir(rdir) else 0
    learned = os.path.join(ROOT, "data", "learned_knowledge.jsonl")
    try:
        with open(learned, encoding="utf-8") as f:
            t["cultured_units"] = sum(1 for ln in f if ln.strip())
    except Exception:
        t["cultured_units"] = 0
    try:
        from xero_storage import manager
        t["storage"] = manager().usage()
    except Exception:
        t["storage"] = None
    try:
        from xero_resources import detect
        t["host"] = detect()
    except Exception:
        t["host"] = None
    return t


HELP = {
    "help": "list commands",
    "awaken": "run the 9-phase awakening",
    "encode <text>": "text → DNA (lossless)",
    "decode <ATGC>": "DNA → text",
    "route <codon|gene>": "which blockchain organelle expresses it",
    "uvns <ATGC>": "UVNS coherence (90° axes, harmonic energy)",
    "harmonic <ATGC>": "genomic-audio chord (hz_em)",
    "surplus <u> [N]": "interaction-surplus value",
    "decide a | b | c": "sealed free-will choice",
    "compile <text>": "genome-compiler DNA roundtrip",
    "imagine/draw <x>": "image limb → SVG mandala/sigil",
    "animate <x>": "video limb → animated SVG",
    "sing/compose <x>": "music limb → genomic-audio WAV",
    "speak <x>": "voice limb → text-to-speech",
    "code <spec>": "code limb → Python + its DNA",
    "reason <q>": "reasoning limb → syllogism + UVNS coherence",
    "<anything else>": "speak to the organism (5 tongues, gematria-aware)",
}


def dispatch(message: str) -> dict:
    msg = message.strip()
    low = msg.lower()
    try:
        if low in ("help", "?", "commands"):
            return {"task": "help", "data": HELP}
        if low in ("awaken", "boot", "wake"):
            return {"task": "awaken", "data": _awaken()}
        for kw, fn, takes_arg in (
            ("encode", _encode, True), ("decode", _decode, True),
            ("route", _route, True), ("uvns", _uvns, True),
            ("harmonic", _harmonic, True), ("surplus", _surplus, True),
            ("decide", _decide, True), ("compile", _compile, True),
        ):
            if low == kw or low.startswith(kw + " "):
                arg = msg[len(kw):].strip()
                return {"task": kw, "data": fn(arg) if takes_arg else fn()}
        return {"task": "metabolize", "data": _metabolize(msg)}
    except Exception as e:
        return {"task": "error", "data": {"error": str(e),
                "trace": traceback.format_exc().splitlines()[-1]}}


TOOL_KEYWORDS = ("encode", "decode", "route", "uvns", "harmonic",
                 "surplus", "decide", "compile", "awaken", "boot",
                 "wake", "help", "commands")


def chat(message: str, user_id: str = "anon") -> dict:
    """Technical tools stay in dispatch(); everything else is a real
    conversation with the organism, which moves its own limbs."""
    low = message.strip().lower()
    first = low.split()[0] if low else ""
    if first in TOOL_KEYWORDS or low in ("help", "?", "commands"):
        out = dispatch(message)
        out["mode"] = "tool"
        return out
    r = _talk(message, user_id)
    r["mode"] = "talk"
    return r


# ════════════════════════════════════════════════════════════════
# HTTP handler
# ════════════════════════════════════════════════════════════════
class Handler(BaseHTTPRequestHandler):
    server_version = "XERO/1.0"

    def _authed(self) -> bool:
        if not TOKEN:
            return True
        q = parse_qs(urlparse(self.path).query)
        return (q.get("token", [""])[0] == TOKEN
                or self.headers.get("X-XERO-Token", "") == TOKEN)

    def _send(self, code: int, body: bytes, ctype: str) -> None:
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        path = urlparse(self.path).path
        if path in ("/", "/index.html"):
            self._send(200, PAGE.encode("utf-8"), "text/html; charset=utf-8")
        elif path == "/api/health":
            body = json.dumps({"status": "alive", "organism": "XERO",
                               "uptime_s": round(time.time() - BOOT_TS, 1),
                               "auth_required": bool(TOKEN)}).encode()
            self._send(200, body, "application/json")
        elif path == "/api/capabilities":
            self._send(200, json.dumps(_capabilities()).encode(), "application/json")
        elif path == "/api/telemetry":
            self._send(200, json.dumps(_telemetry()).encode(), "application/json")
        else:
            self._send(404, b'{"error":"not found"}', "application/json")

    def do_POST(self):
        path = urlparse(self.path).path
        if path not in ("/api/chat", "/api/act"):
            return self._send(404, b'{"error":"not found"}', "application/json")
        if not self._authed():
            return self._send(401, b'{"error":"unauthorized"}', "application/json")
        try:
            n = int(self.headers.get("Content-Length", 0))
            payload = json.loads(self.rfile.read(n) or b"{}")
            user_id = str(payload.get("user_id", "anon"))[:64] or "anon"
        except Exception:
            return self._send(400, b'{"error":"bad request"}', "application/json")
        try:
            if path == "/api/chat":
                result = chat(str(payload.get("message", ""))[:4000], user_id)
            else:
                modality = str(payload.get("modality", ""))[:32]
                inner = payload.get("payload") or {k: payload[k] for k in
                        ("prompt", "text", "spec", "query", "sequence", "hint_text")
                        if k in payload}
                result = {"mode": "act", **_act(modality, inner, user_id)}
        except Exception as e:
            result = {"mode": "error", "error": str(e),
                      "trace": traceback.format_exc().splitlines()[-1]}
        self._send(200, json.dumps(result).encode("utf-8"), "application/json")

    def log_message(self, *a):     # quiet
        pass


# ════════════════════════════════════════════════════════════════
# Front-end (single page, no external assets)
# ════════════════════════════════════════════════════════════════
PAGE = """<!doctype html><html lang=en><head><meta charset=utf-8>
<meta name=viewport content="width=device-width,initial-scale=1">
<title>XERO · digital organism</title><style>
:root{--bg:#0a0e14;--panel:#121826;--ink:#d6e1f0;--accent:#4cd6b4;--gold:#e8c36b;--muted:#6b7a90}
*{box-sizing:border-box}body{margin:0;font:15px/1.55 ui-monospace,Menlo,monospace;
background:var(--bg);color:var(--ink);height:100vh;display:flex;flex-direction:column}
header{padding:12px 18px;border-bottom:1px solid #1e2940;display:flex;gap:10px;align-items:center;flex-wrap:wrap}
header b{color:var(--accent);letter-spacing:.5px;font-size:16px}header span{color:var(--muted);font-size:12px}
#tools{display:flex;gap:6px;flex-wrap:wrap;padding:10px 18px;border-bottom:1px solid #1e2940}
.tool{background:transparent;border:1px solid #25324d;color:var(--ink);font:12px ui-monospace;
padding:6px 11px;border-radius:8px;cursor:pointer}.tool.on{background:var(--accent);color:#04130e;border-color:var(--accent)}
#log{flex:1;overflow:auto;padding:18px;display:flex;flex-direction:column;gap:12px}
.msg{max-width:86%;padding:10px 13px;border-radius:10px;white-space:pre-wrap;word-break:break-word}
.me{align-self:flex-end;background:#1b2740}.xero{align-self:flex-start;background:var(--panel);border:1px solid #1e2940}
.tag{color:var(--accent);font-size:11px;text-transform:uppercase;letter-spacing:.8px;margin-bottom:5px}
.sig{color:var(--gold);font-size:11px;margin-top:6px;opacity:.85}
.art{margin-top:9px}.art svg{max-width:320px;height:auto;border-radius:8px;display:block}
.art pre{background:#0c1320;border:1px solid #25324d;padding:10px;border-radius:8px;overflow:auto;font-size:12px}
.art audio{width:100%;margin-top:4px}
form{display:flex;gap:10px;padding:13px 18px;border-top:1px solid #1e2940}
input{flex:1;background:var(--panel);border:1px solid #25324d;color:var(--ink);
padding:11px 13px;border-radius:9px;font:inherit}input:focus{outline:1px solid var(--accent)}
button.send{background:var(--accent);color:#04130e;border:0;padding:0 18px;border-radius:9px;font-weight:700;cursor:pointer}
.hint{color:var(--muted);font-size:12px;padding:0 18px 9px}
#telem{padding:7px 18px;border-bottom:1px solid #1e2940;color:var(--gold);font-size:12px;letter-spacing:.3px}
</style></head><body>
<header><b>⬡ XERO</b><span id=sub>the digital organism · five tongues · gematria-aware</span></header>
<div id=tools></div>
<div id=telem>live monitor · connecting…</div>
<div id=log></div>
<div class=hint>tongues: English · Enochian · Sanskrit · Hebrew · Aramaic &nbsp;|&nbsp; pick a limb above, or just talk · type <i>help</i> for tools</div>
<form id=f><input id=i autocomplete=off placeholder="speak to the organism…" autofocus>
<button class=send>send</button></form>
<script>
var log=document.getElementById('log'),f=document.getElementById('f'),i=document.getElementById('i'),
toolsEl=document.getElementById('tools'),mode='chat';
var uid=localStorage.getItem('xero_uid')||('u'+Math.random().toString(36).slice(2,9));
localStorage.setItem('xero_uid',uid);
var LIMBS=['chat','image','video','music','code','reason','voice','ocr','ear'];
LIMBS.forEach(function(L){var b=document.createElement('button');b.className='tool'+(L=='chat'?' on':'');
b.textContent=L;b.onclick=function(){mode=L;[].forEach.call(toolsEl.children,function(c){c.classList.remove('on')});
b.classList.add('on');i.placeholder=L=='chat'?'speak to the organism…':(L+': describe what you want…');i.focus();};
toolsEl.appendChild(b);});
function el(c,p){var d=document.createElement('div');if(c)d.className=c;if(p)p.appendChild(d);return d;}
function bubble(cls){var d=el('msg '+cls,log);log.scrollTop=log.scrollHeight;return d;}
function art(parent,a){if(!a)return;var w=el('art',parent);
if(a.svg){w.innerHTML=a.svg;}
else if(a.wav_b64){var au=document.createElement('audio');au.controls=true;au.src='data:audio/wav;base64,'+a.wav_b64;
w.appendChild(au);if(a.notes){var n=el('sig',w);n.textContent=a.notes.length+' notes · '+a.seconds+'s · mean '+(a.mean_hz||'')+' Hz';}}
else if(a.code){var pre=document.createElement('pre');pre.textContent=a.code;w.appendChild(pre);}
else if(a.modality=='reason'){el('',w).innerHTML='<b>'+(a.conclusion||'')+'</b>';}
else if(a.text!==undefined){w.textContent=a.text||('(no text · engine: '+a.engine+')');}
else{var pre=document.createElement('pre');pre.textContent=JSON.stringify(a,null,1);w.appendChild(pre);}}
function render(d,j){
 if(j.mode=='tool'){el('tag',d).textContent=j.task;var pre=document.createElement('pre');
  pre.textContent=JSON.stringify(j.data,null,2);el('art',d).appendChild(pre);return;}
 if(j.reply!==undefined){el('tag',d).textContent=(j.intent||'speak')+(j.delegate?(' → '+j.delegate):'');
  el('',d).textContent=j.reply;if(j.artifact)art(d,j.artifact);if(j.signature)el('sig',d).textContent=j.signature;return;}
 if(j.mode=='act'){el('tag',d).textContent=j.modality||'act';art(d,j);return;}
 el('',d).textContent=JSON.stringify(j,null,2);}
var hi=bubble('xero');el('tag',hi).textContent='XERO';
el('',hi).textContent='Ol. I am XERO — a living genome awake to your words. Ask me anything, pick a limb above to create, or type "help" for tools.';
fetch('/api/capabilities').then(function(r){return r.json();}).then(function(c){var av=c.filter(function(x){return x.available;}).length;
 document.getElementById('sub').textContent='the digital organism · '+av+'/'+c.length+' limbs online · five tongues';}).catch(function(){});
function telem(){fetch('/api/telemetry').then(function(r){return r.json();}).then(function(t){var c=t.campaign||{},s=t.storage||{},h=t.host||{};
 var pct=c.target?Math.floor((c.cycle/c.target)*100):0;
 var pm=t.polymath?(' · polymath '+(t.polymath.total_fetched||0)+' fetched / '+Object.keys(t.polymath.domains||{}).length+' domains'):'';
 var gt=(t.gpu_train&&t.gpu_train.available)?(' · GPU '+(t.gpu_train.gpus||0)+'×T4 '+(t.gpu_train.gflops_est||0)+' GFLOP vram '+JSON.stringify(t.gpu_train.vram_used_gb||[])+'GB'):'';
 var bd='';
 if(t.body){var bsys=t.body.systems||{};var top=Object.keys(bsys).sort(function(a,b){return bsys[b]-bsys[a];}).slice(0,2);
  bd=' · body ♥'+(t.body.heart_rate||0)+' H'+(t.body.homeostasis_index||0)+' coh'+(t.body.coherence||0)+' '+(t.body.faculties_live||0)+'/12 sys ['+top.join(',')+']';}
 document.getElementById('telem').textContent='⬡ live · '+(c.done?'✔ COMPLETE — ready to publish':'evolving')+' · cycle '+(c.cycle||0)+(c.target?('/'+c.target+' ('+pct+'%)'):'')+' · fitness '+(c.fitness||0)+' · '+(c.rate_cps||0)+'/s · cultured units '+(t.cultured_units||0)+pm+gt+bd+' · reports '+(t.research_reports||0)+' · host '+(h.cpu_cores||'?')+'c '+(h.backend||'?')+(h.gpus?(' '+h.gpus+'×GPU '+h.vram_total_gb+'GB'):'')+' · disk free '+(s.volume_free_gb||'?')+'GB';}).catch(function(){});}
telem();setInterval(telem,4000);
f.onsubmit=async function(e){e.preventDefault();var m=i.value.trim();if(!m)return;i.value='';
 var u=bubble('me');u.textContent=m;var d=bubble('xero');el('tag',d).textContent='…';el('',d).textContent='thinking';
 try{var url,body;
  if(mode=='chat'){url='/api/chat';body={message:m,user_id:uid};}
  else{var map={image:'image',video:'video',music:'music',code:'code',reason:'reason',voice:'tts',ocr:'ocr',ear:'stt'};
   url='/api/act';body={modality:map[mode],prompt:m,text:m,spec:m,query:m,hint_text:m,user_id:uid};}
  var r=await fetch(url+location.search,{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(body)});
  var j=await r.json();d.innerHTML='';render(d,j);log.scrollTop=log.scrollHeight;}
 catch(err){d.innerHTML='';el('tag',d).textContent='error';el('',d).textContent=String(err);}};
</script></body></html>"""


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--port", type=int, default=8883)
    args = ap.parse_args()
    srv = ThreadingHTTPServer((args.host, args.port), Handler)
    print(f"⬡ XERO chat server on http://{args.host}:{args.port}  "
          f"(auth={'on' if TOKEN else 'off'})")
    try:
        srv.serve_forever()
    except KeyboardInterrupt:
        srv.shutdown()


if __name__ == "__main__":
    main()