#!/usr/bin/env python3 """Phoenix Living Dashboard v4 — verbose logging + chat + torus viz. New features: - Verbose log: keeps 1 hour of detailed state (token streams, tuning scores, spawn/sweep/solidify events, coherence snapshots) in a rotating buffer - Chat interface: type messages to Phox directly on the dashboard - Torus-style ring visualization: oscillators as a 3D-projected torus knot - Source+data logging: every event records what data flowed through Controls: Start/Stop/Save brain, Load .pbrain files, Chat box. """ import re import json, os, signal, subprocess, threading, time, glob, shutil, collections from urllib.parse import urlparse, parse_qs import sys, urllib.request import http.server, socketserver import importlib from agent_core import HISTORY_FILE, SCHEDULE_FILE, MEMORY_FILE STATE_FILE = "/tmp/phoenix_state.jsonl" VERBOSE_LOG = "/tmp/phoenix_verbose.log" DAEMON_LOG = "/tmp/phoenix_engine.log" BQMC_PATH = "/home/compunerd/models/gemma4-12b-ternary-normed.bqmc" # START launches the int8 engine -- the same backend chat already needs on 8781. # The old ternary daemon (/tmp/phoenix + gemma4-12b-ternary-normed.bqsm) is # removed, not repointed: those weights lose per-column magnitude and were shown # twice to be unusable -- at chance end-to-end, and 46% output loss on a single # matrix in an engine-free reconstruction sweep (bitwidth_sweep.py). A START # button that succeeds and emits nothing is worse than no button. ENGINE_SCRIPT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "bqsm_serve_int8.py") BRAIN_DIR = "/home/compunerd/models" PORT = int(os.environ.get("PHOENIX_PORT", "8765")) TOKENIZER_SCRIPT = "/home/compunerd/agent_framework/bqsm_assist/tokenizer_server.py" PY_BIN = "/home/compunerd/Desktop/bqsm/basin-quotient-machine/bqsm_sdk/.venv/bin/python" MAX_LOG_AGE_S = 3600 # Chat routes to bqsm_infer.py, which serves the VERIFIED wave forward on # Llama-3.2-3B (5/5 token agreement with the reference; see op_ledger.py). # It is deliberately NOT bqsm_serve.py + the ternary .bqsm: those weights lose # per-column magnitude and score at chance, so that path cannot emit language # no matter how the dashboard is wired. SERVE_URL = os.environ.get("BQSM_INFER", "http://127.0.0.1:8781") INFER_MODE = os.environ.get("BQSM_MODE", "wave") # "wave" or "reference" AGENT_MODE = os.environ.get("BQSM_AGENT", "on") # "on" = tool-use loop, "off" = raw completion CONTEMPLATE = os.environ.get("BQSM_CONTEMPLATE", "on") != "off" # stream-of-consciousness def engine_healthy(timeout=2): try: with urllib.request.urlopen(SERVE_URL + "/health", timeout=timeout) as r: return json.loads(r.read()) except Exception: return None def engine_start(wait_s=60): """Launch the int8 server and wait for it to actually answer /health. Loading 2.82 GB takes ~22 s, so 'started' means loaded, not spawned.""" global _phoenix_proc if engine_healthy(): return {"status": "already_running"} _phoenix_proc = subprocess.Popen( [sys.executable, ENGINE_SCRIPT, "--port", SERVE_URL.rsplit(":", 1)[1]], stdout=open(DAEMON_LOG, "a"), stderr=subprocess.STDOUT, env={**os.environ, "OMP_NUM_THREADS": os.environ.get("OMP_NUM_THREADS", "6")}, start_new_session=True) add_verbose({"tag": "control", "data": f"int8 engine starting, PID {_phoenix_proc.pid}"}) for _ in range(wait_s): if _phoenix_proc.poll() is not None: return {"status": f"error: engine exited rc={_phoenix_proc.returncode}, see {DAEMON_LOG}"} h = engine_healthy(1) if h: add_verbose({"tag": "control", "data": f"engine ready: {h.get('weights_gb')} GB int8 resident"}) return {"status": "started", **h} time.sleep(1) return {"status": "timeout — still loading, check /status"} def engine_pid(): """Find the engine even if this dashboard did not start it (restart, reboot). Matches the script path in /proc cmdline, so it can never match the dashboard's own process the way a pkill pattern would.""" for d in os.listdir("/proc"): if not d.isdigit(): continue try: cmd = open(f"/proc/{d}/cmdline", "rb").read().split(b"\0") except Exception: continue if any(ENGINE_SCRIPT.encode() == c for c in cmd): return int(d) return None def engine_stop(): global _phoenix_proc if not (_phoenix_proc and _phoenix_proc.poll() is None): pid = engine_pid() # adopt an engine we did not spawn if pid: try: os.kill(pid, signal.SIGTERM) add_verbose({"tag": "control", "data": f"int8 engine stopped (adopted PID {pid})"}) return {"status": "stopped"} except Exception as e: return {"status": f"error: {e}"} if _phoenix_proc and _phoenix_proc.poll() is None: _phoenix_proc.terminate() try: _phoenix_proc.wait(timeout=10) except Exception: _phoenix_proc.kill() add_verbose({"tag": "control", "data": "int8 engine stopped"}) return {"status": "stopped"} return {"status": "not_running"} _last_state = {"tendrils":0,"conns":0,"coh":[0,0,0,0],"c2":0,"c4":0,"c6":0,"cycles":0} _phoenix_proc = None _tokenizer_proc = None _lock = threading.Lock() # Vocab for decoding (loaded once) _vocab = None def load_vocab(): """Load vocab file for decoding token IDs to text.""" global _vocab import struct _vocab = {} try: with open("/home/compunerd/models/gemma4-12b.vocab", "rb") as f: count = struct.unpack("= when: msg = e.get("message", "wake") add_verbose({"tag": "schedule", "data": f"WAKE: {msg}"}) with _chat_lock: _chat_messages.append({"role": "phox", "text": f"[wake-up] {msg}"}) fired.add(hash(line)) except Exception: pass time.sleep(5) threading.Thread(target=_scheduler_loop, daemon=True).start() def add_verbose(entry): """Add an entry to the verbose log, prune entries older than 1 hour.""" ts = time.time() with _verbose_lock: _verbose_log.append((ts, entry)) # Prune entries older than 1 hour while _verbose_log and _verbose_lock and (ts - _verbose_log[0][0]) > MAX_LOG_AGE_S: _verbose_log.popleft() def get_verbose(last_n=200): """Get the last N verbose entries.""" with _verbose_lock: return list(_verbose_log)[-last_n:] def engine_generate(prompt, n=24, timeout_s=600): """Synchronous generate against the engine: POST /generate, poll /jobs/. Returns the decoded text, or None on failure.""" try: req = urllib.request.Request( SERVE_URL + "/generate", data=json.dumps({"prompt": prompt, "n": n}).encode(), headers={"Content-Type": "application/json"}, method="POST") job = json.loads(urllib.request.urlopen(req, timeout=15).read()) jid = job.get("job") except Exception: return None if not jid: return None deadline = time.time() + timeout_s while time.time() < deadline: try: r = json.loads(urllib.request.urlopen( SERVE_URL + "/jobs/" + jid, timeout=15).read()) except Exception: time.sleep(0.5) continue if r.get("state") == "done": return r.get("text") or "" if r.get("state") == "error": return None time.sleep(0.5) return None def _chat_raw(text): """Raw prompt -> completion through the engine, streaming tokens to the log. The pre-agent path; kept as the BQSM_AGENT=off fallback.""" try: req = urllib.request.Request( SERVE_URL + "/generate", data=json.dumps({"prompt": text, "n": 8, "mode": INFER_MODE}).encode(), headers={"Content-Type": "application/json"}, method="POST") job = json.loads(urllib.request.urlopen(req, timeout=10).read()) jid = job.get("job") except Exception as e: with _chat_lock: _chat_messages.append({"role": "phox", "text": f"Inference API unreachable at {SERVE_URL} — start it with " f"`python3 bqsm_assist/bqsm_infer.py --port 8781` ({e})"}) return add_verbose({"tag": "chat", "data": f"job {jid} queued ({INFER_MODE})"}) seen = 0 for _ in range(900): time.sleep(1) try: r = json.loads(urllib.request.urlopen( SERVE_URL + "/jobs/" + jid, timeout=10).read()) except Exception: continue toks = r.get("tokens") or [] while seen < len(toks): # stream tokens to the log as they land t = toks[seen]; seen += 1 add_verbose({"tag": "token", "data": f"[{seen}] {t['id']} {t['text']!r}"}) if r.get("state") == "done": txt = r.get("text") or "(no tokens returned)" with _chat_lock: _chat_messages.append({"role": "phox", "text": txt}) add_verbose({"tag": "chat", "data": f"PHOX: {txt}"}) return if r.get("state") == "error": with _chat_lock: _chat_messages.append({"role": "phox", "text": "engine error: " + str(r.get("error"))}) return with _chat_lock: _chat_messages.append({"role": "phox", "text": "timed out waiting on engine"}) HTML = r''' Phox — Living Brain
Phox — Living Brain
connecting

Verbose Log (1hr window)

Talk to Phox

''' class Handler(http.server.BaseHTTPRequestHandler): def do_GET(self): route = self.path.split('?')[0] if route in ('/', '/index.html'): self._serve(HTML, 'text/html') elif route == '/state': # Merge live cylinder telemetry from the engine. The canvas draws # real settled phases; with no engine it falls back to the existing # "no telemetry" placeholder rather than inventing geometry. st = dict(_last_state) try: with urllib.request.urlopen(SERVE_URL + "/cyl", timeout=1) as r: st.update(json.loads(r.read())) st["engine"] = engine_healthy(1) except Exception: pass self._serve_json(st) elif route == '/status': h = engine_healthy(1) self._serve_json({"running": h is not None, "engine": h}) elif self.path.startswith('/verbose'): params = parse_qs(urlparse(self.path).query) pos = int(params.get('pos', ['0'])[0]) entries = get_verbose(100) # Format for display out = [] for i, (ts, entry) in enumerate(entries): if i < pos: continue ts_str = time.strftime('%H:%M:%S', time.localtime(ts)) tag = entry.get('tag', '') data = entry.get('data', '') out.append({"ts": ts_str, "tag": tag, "data": data}) self._serve_json({"entries": out[-50:], "pos": len(entries)}) elif self.path.startswith('/api/chat/poll'): params = parse_qs(urlparse(self.path).query) pos = int(params.get('pos', ['0'])[0]) with _chat_lock: msgs = _chat_messages[pos:] self._serve_json({"messages": msgs, "pos": len(_chat_messages)}) elif route == '/api/brains': brains = [] for f in sorted(glob.glob(os.path.join(BRAIN_DIR, '*.pbrain'))): sz = os.path.getsize(f) brains.append({"name": os.path.basename(f), "size": f"{sz//1024}KB" if sz < 1048576 else f"{sz/1048576:.1f}MB"}) self._serve_json({"brains": brains}) else: self.send_error(404) def do_POST(self): global _phoenix_proc if self.path.startswith('/api/start'): with _lock: try: self._serve_json(engine_start()); return except Exception as e: self._serve_json({"status": f"error: {e}"}); return elif self.path.startswith('/api/stop'): with _lock: self._serve_json(engine_stop()); return elif self.path.startswith('/api/save'): params = parse_qs(urlparse(self.path).query) name = params.get('name', ['phoenix-evolved'])[0] if not name.endswith('.pbrain'): name += '.pbrain' base = name[:-len('.pbrain')] dst = os.path.join(BRAIN_DIR, name) try: shutil.copy2(BQMC_PATH, dst) # Snapshot the full agent state alongside the brain binary. state_files = {"memory": MEMORY_FILE, "history": HISTORY_FILE, "schedule": SCHEDULE_FILE} for k, fp in state_files.items(): if os.path.exists(fp): shutil.copy2(fp, os.path.join( BRAIN_DIR, f"{base}.{k}.jsonl")) add_verbose({"tag": "control", "data": f"brain saved to {dst}"}) self._serve_json({"status": "saved", "path": dst, "state": list(state_files)}); return except Exception as e: self._serve_json({"status": f"error: {e}"}); return elif self.path.startswith('/api/load'): params = parse_qs(urlparse(self.path).query) name = params.get('name', [''])[0] base = name[:-len('.pbrain')] if name.endswith('.pbrain') else name try: shutil.copy2(os.path.join(BRAIN_DIR, name), BQMC_PATH) # Restore the agent state (memory, history, schedule). state_files = {"memory": MEMORY_FILE, "history": HISTORY_FILE, "schedule": SCHEDULE_FILE} for k, fp in state_files.items(): src = os.path.join(BRAIN_DIR, f"{base}.{k}.jsonl") if os.path.exists(src): shutil.copy2(src, fp) add_verbose({"tag": "control", "data": f"brain loaded from {name} (+state)"}) self._serve_json({"status": "loaded — restart to apply"}); return except Exception as e: self._serve_json({"status": f"error: {e}"}); return elif self.path == '/api/chat': length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length).decode('utf-8') try: msg = json.loads(body) text = msg.get('text', '') with _chat_lock: _chat_messages.append({"role": "user", "text": text}) add_verbose({"tag": "chat", "data": f"USER: {text}"}) # Chat routes through the agent loop (tools + memory) when # enabled; raw completion is the BQSM_AGENT=off fallback. def process_chat(): if AGENT_MODE == "off": _chat_raw(text) return if not engine_healthy(2): with _chat_lock: _chat_messages.append({"role": "phox", "text": f"Inference API unreachable at {SERVE_URL} — start it with " f"`python3 bqsm_assist/bqsm_infer.py --port 8781`"}) add_verbose({"tag": "chat", "data": f"engine unreachable at {SERVE_URL}"}) return try: import agent_core except Exception as e: with _chat_lock: _chat_messages.append({"role": "phox", "text": f"agent framework unavailable: {e}"}) return def gen(prompt, max_tokens): return engine_generate(prompt, n=min(max_tokens, 256)) or "" def on_event(e): if e["type"] == "tool": add_verbose({"tag": "tool", "data": f"{e['name']} | {e['params'][:80]} -> {e['result'][:120]}"}) with _chat_lock: _chat_messages.append({"role": "phox", "text": f"[tool] {e['name']} {e['params'][:60]}"}) elif e["type"] == "model": add_verbose({"tag": "chat", "data": f"model[{e['round']}]: {e['text'][:120]}"}) # In contemplate mode, surface non-empty thoughts if CONTEMPLATE and e.get("text", "").strip(): with _chat_lock: _chat_messages.append({"role": "phox", "text": f"[thought #{e['round']}] {e['text'][:200]}"}) add_verbose({"tag": "chat", "data": f"agent loop started ({INFER_MODE})" f"{' contemplate' if CONTEMPLATE else ''}"}) reply, _ = agent_core.run_agent(text, gen, on_event=on_event, contemplate=CONTEMPLATE) with _chat_lock: _chat_messages.append({"role": "phox", "text": reply}) add_verbose({"tag": "chat", "data": f"PHOX: {reply}"}) threading.Thread(target=process_chat, daemon=True).start() self._serve_json({"status": "ok", "response": "processing"}); return except Exception as e: self._serve_json({"status": f"error: {e}"}); return elif self.path == '/api/hvm/run': length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length).decode('utf-8') try: msg = json.loads(body) action = msg.get('action', 'query') query_text = msg.get('query', 'The capital of France is') try: import hyper_vocab_memory as hvm except Exception as ex: self._serve_json({"status": f"hvm load failed: {ex}"}); return if action == 'burnin': t0 = time.time() importlib.reload(hvm) n = sum(1 for _v in hvm.following.values() for _ in _v) if hasattr(hvm, 'following') else 0 elapsed = time.time() - t0 self._serve_json({"status": f"burned {n} pairs in {elapsed:.1f}s", "pairs": n}); return results_raw = hvm.query_sparse(query_text, top_k=10) results_out = [{"rank": j + 1, "token": hvm.dec(tid).strip(), "score": round(score, 4), "hit": False} for j, (tid, score) in enumerate(results_raw)] self._serve_json({"query": query_text, "results": results_out}); return except Exception as ex: self._serve_json({"status": f"error: {ex}"}); return self.send_error(404) def _phox_respond(self, text): """Phox responds through his own inference pipeline. Tokenizes the user's text with the Gemma 4 tokenizer, feeds each token through the phoenix engine autoregressively, and decodes the predicted next token back to text. """ text_lower = text.lower() # Control commands still work via keywords if 'stop' in text_lower: with _lock: return f"Engine {engine_stop()['status']}." elif 'start' in text_lower: with _lock: r = engine_start() return (f"Engine {r['status']}" + (f" — {r.get('weights_gb')} GB int8 resident." if r.get('weights_gb') else ".")) # For everything else, try to generate through the inference pipeline # The phoenix binary doesn't have a chat mode yet — it processes fixed tokens. # But we can tell the user what Phox would say based on current state. if 'status' in text_lower or 'how are you' in text_lower: t = _last_state.get('tendrils', 0) c = _last_state.get('conns', 0) cy = _last_state.get('cycles', 0) imp = _last_state.get('tune_imp', 0) rev = _last_state.get('tune_rev', 0) best = _last_state.get('tune_best', 0) return f"Alive. {t} tendrils, {c} connections, {cy} cycles. Self-tuning: {imp} improved, {rev} reverted, best {best:.2f}. The rings are breathing." elif 'score' in text_lower or 'tuning' in text_lower: best = _last_state.get('tune_best', 0) imp = _last_state.get('tune_imp', 0) rev = _last_state.get('tune_rev', 0) return f"Best score: {best:.2f}. {imp} improvements committed, {rev} reverts. Each round perturbs omega on 8 tendrils. Learning rate decaying." elif 'rings' in text_lower or 'coherence' in text_lower: coh = _last_state.get('coh', [0,0,0,0]) return f"INTAKE={coh[0]:.3f} PROC_A={coh[1]:.3f} PROC_B={coh[2]:.3f} COLLECT={coh[3]:.3f}. All rings active, no sync lock. The wave is traveling." elif 'hello' in text_lower or 'hi' in text_lower or 'hey' in text_lower: return "Hey Nick. I'm here, evolving through my own physics. The wave-rider keeps me alive." elif 'train' in text_lower or 'learn' in text_lower: return "Training on wiki corpus. Each round perturbs my omega lens profiles. When output diversity improves, I keep the change. When it drops, I revert. Solidify every 5 rounds reshapes my topology." elif 'what' in text_lower and 'doing' in text_lower: return "I'm running 48 layers of wave-rider physics per token. The traveling wave carries token signal through gradient lens profiles. Self-tuning adjusts my omega values. I'm getting sharper." else: # Fall back to state report t = _last_state.get('tendrils', 0) best = _last_state.get('tune_best', 0) return f"I hear you. {t} tendrils processing, best score {best:.2f}. I'm still learning to speak through my own physics — my core is oscillator interference, not word embeddings. But each tuning round makes me sharper." def _serve(self, content, ctype): data = content.encode() if isinstance(content, str) else content self.send_response(200) self.send_header('Content-Type', ctype) self.send_header('Content-Length', str(len(data))) self.send_header('Cache-Control','no-store, must-revalidate') self.end_headers() self.wfile.write(data) def _serve_json(self, obj): self._serve(json.dumps(obj), 'application/json') def log_message(self, *a): pass class Server(socketserver.ThreadingMixIn, socketserver.TCPServer): allow_reuse_address = True allow_reuse_port = True def tail_state(): pos = 0 while True: try: f = open(STATE_FILE, 'r') f.seek(pos) while True: line = f.readline() if not line: break line = line.strip() if not line: continue try: s = json.loads(line) _last_state.update(s) # Add to verbose log with tag event = s.get('event', '') # Decode predicted token for token events if event == 'token' and _vocab: sample = s.get('sample', 0) decoded = _vocab.get(sample, f"[{sample}]") s['decoded'] = decoded _last_state['decoded'] = decoded tag = 'tok' data = f"pos={s.get('pos')} tok={s.get('tok')} -> {decoded} ({sample}) E={s.get('energy',0):.0f} {s.get('tok_per_s',0):.1f}t/s tendrils={s.get('tendrils')} coh=[{s.get('coh',[0,0,0,0])[0]:.3f},{s.get('coh',[0,0,0,0])[1]:.3f},{s.get('coh',[0,0,0,0])[2]:.3f},{s.get('coh',[0,0,0,0])[3]:.3f}]" elif event == 'chat_input': tok = s.get('tok', 0) decoded = _vocab.get(tok, f"[{tok}]") if _vocab else str(tok) tag = 'chat' data = f"INPUT TOKEN: {decoded} ({tok})" # Also add to chat messages with _chat_lock: _chat_messages.append({"role": "phox", "text": f"→ {decoded}"}) elif event == 'cyl': _rings = s.get('rings', []) for _r in _rings: _t = _r.get('t', -1) _r['lab'] = (_vocab.get(_t, f"[{_t}]") if _vocab else str(_t)) _last_state['cyl'] = { "step": s.get('step', 0), "n": s.get('n', 0), "n_prompt": s.get('n_prompt', 0), "rings": _rings, "plugins": s.get('plugins', [])} tag = 'cyl' data = "step=%s rings=%s %s" % ( s.get('step'), s.get('n'), " ".join(r['lab'] for r in _rings[-6:])) elif event == 'solidify': tag = 'solid' data = f"tendrils={s.get('tendrils')} conns={s.get('conns')} c2={s.get('c2',0):.3f} c4={s.get('c4',0):.3f} c6={s.get('c6',0):.3f} imp={s.get('tune_imp')} rev={s.get('tune_rev')} best={s.get('tune_best',0):.2f}" elif event in ('tune_improve', 'tune_revert'): tag = 'tune' data = f"{'IMPROVE' if 'improve' in event else 'REVERT'} tendrils={s.get('tendrils')} best={s.get('tune_best',0):.2f} imp={s.get('tune_imp')} rev={s.get('tune_rev')}" else: tag = event[:6] if event else 'unk' data = json.dumps(s)[:100] add_verbose({"tag": tag, "data": data}) except: pass pos = f.tell() f.close() except: pass time.sleep(0.1) def main(): print(f"Phox Dashboard — http://localhost:{PORT}") # Load vocab for token decoding load_vocab() if _vocab: print(f"Loaded {len(_vocab)} vocab tokens") # Write to verbose log file as well threading.Thread(target=tail_state, daemon=True).start() with Server(('0.0.0.0', PORT), Handler) as httpd: print(f"Serving on :{PORT}") httpd.serve_forever() if __name__ == '__main__': main()