#!/usr/bin/env python3 """Chairman Command dashboard server. Stdlib only. Run: python3 dashboard/server.py Serves the chat-first dashboard/index.html at /, the ops room at /ops (dashboard/ops.html), and the JSON API on port 8919 (127.0.0.1). PORT env var overrides the port (default 8919). """ import json import os import re import shlex import subprocess import sys import threading import time import urllib.error import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlparse, parse_qs # ---------------------------------------------------------------- paths ROOT = Path("/Users/19keys/sovereign-empire-sweep") CA = ROOT / "chairman-agent" DASH = CA / "dashboard" INBOX = CA / "inbox" DONE = CA / "done" LOG_DIR = CA / "logs" NEO_DIR = CA / "neo" QUEUE_DIR = NEO_DIR / "queue" SUBAGENT_RUNNER = CA / "subagents" / "subagent.py" DEPLOY_SCRIPT = NEO_DIR / "deploy-agents-to-neo.sh" MEMORY_FILE = ROOT / "sovereign-wiki" / "memory" / "CHAIRMAN-MEMORY.md" DASHBOARD_LOG = LOG_DIR / "dashboard.log" CONV_DIR = DASH / "conversations" CONV_TRASH = CONV_DIR / "trash" for d in (DASH, INBOX, DONE, LOG_DIR, QUEUE_DIR, CONV_DIR, CONV_TRASH): d.mkdir(parents=True, exist_ok=True) PORT = int(os.environ.get("PORT", "8919")) PYTHON = sys.executable or "python3" # ---------------------------------------------------------------- neo NEO_HOSTS = ["macmini-ts", "macmini"] TELEGRAM_CHAT_ID = "457766643" NEO_CACHE = {"t": 0.0, "data": None} NEO_LOCK = threading.Lock() RECOVER_CMDS = { # exact NEO-RECOVERY.md playbook commands, run over ssh "restart_gateway": ( "/bin/launchctl bootout gui/501/ai.openclaw.gateway 2>/dev/null; " "/usr/bin/pkill -9 -f openclaw; sleep 3; " "env PATH=/opt/homebrew/bin:/usr/bin:/bin /usr/bin/nohup " "/opt/homebrew/bin/openclaw gateway > /tmp/openclaw-gateway.log 2>&1 &" ), "warm_models": ( 'curl -s -X POST http://localhost:11434/api/generate ' '-d "{\\"model\\":\\"gemma4\\",\\"prompt\\":\\"x\\",\\"keep_alive\\":-1,\\"stream\\":false}" > /dev/null; ' 'curl -s -X POST http://localhost:11434/api/generate ' '-d "{\\"model\\":\\"qwen3:8b\\",\\"prompt\\":\\"x\\",\\"keep_alive\\":-1,\\"stream\\":false}" > /dev/null; ' 'echo warmed' ), "test_telegram": ( "env PATH=/opt/homebrew/bin /opt/homebrew/bin/openclaw agent " f"--channel telegram --to '{TELEGRAM_CHAT_ID}' " "--message 'Status check — respond in English.' --deliver" ), } # ---------------------------------------------------------------- agents SUBAGENTS = [ {"id": "brand-prophet", "name": "Brand Prophet", "tagline": "Reads culture, protects the brand"}, {"id": "capital-allocator", "name": "Capital Allocator", "tagline": "Decides where money and time go"}, {"id": "cultural-forecaster", "name": "Cultural Forecaster", "tagline": "Spots waves before they break"}, {"id": "orchestrator", "name": "Orchestrator", "tagline": "Routes work across the agent org"}, ] CHAT_AGENTS = ["chairman"] + [a["id"] for a in SUBAGENTS] # ---------------------------------------------------------------- models # Authoritative registry. Pricing is $/MTok. API key spend is real credits. MODELS = [ {"id": "claude-haiku-4-5-20251001", "label": "Haiku 4.5", "in_per_mtok": 1.00, "out_per_mtok": 5.00, "note": "Cheap and fast. Simple asks.", "badge": "CHEAPEST"}, {"id": "claude-sonnet-4-6", "label": "Sonnet 4.6", "in_per_mtok": 3.00, "out_per_mtok": 15.00, "note": "Solid work. Blocked until the Anthropic balance is topped up.", "badge": ""}, {"id": "claude-sonnet-5", "label": "Sonnet 5", "in_per_mtok": 3.00, "out_per_mtok": 15.00, "note": "Near-Opus coding and agent work. Intro pricing 2/10 through Aug 31.", "badge": "NEW"}, {"id": "claude-opus-4-8", "label": "Opus 4.8", "in_per_mtok": 5.00, "out_per_mtok": 25.00, "note": "Heavy strategy. Long autonomous runs.", "badge": ""}, {"id": "claude-fable-5", "label": "Fable 5", "in_per_mtok": 10.00, "out_per_mtok": 50.00, "note": "The hardest calls only. Can refuse security topics. " "Needs 30 day data retention on the org.", "badge": "APEX"}, # Non-Claude providers. Full OpenAI function-calling tool loop in # agent.py (bash, files, web_search, text_19keys), proven live per # model through this dashboard on 2026-07-02. # Ids and pricing verified live against provider APIs 2026-07-02. {"id": "openrouter/openai/gpt-5.5", "label": "ChatGPT (GPT-5.5)", "in_per_mtok": 5.00, "out_per_mtok": 30.00, "note": "OpenAI flagship via OpenRouter. Full tool access.", "badge": "OPENAI"}, {"id": "openrouter/z-ai/glm-5.2", "label": "GLM-5.2", "in_per_mtok": 0.93, "out_per_mtok": 3.00, "note": "Neo's builder brain, via OpenRouter. Full tool access.", "badge": "DEFAULT"}, {"id": "deepseek-v4-flash", "label": "DeepSeek Flash", "in_per_mtok": 0.10, "out_per_mtok": 0.40, "note": "Cheapest brain in the stack. Full tool access.", "badge": "BUDGET"}, {"id": "deepseek-v4-pro", "label": "DeepSeek Pro", "in_per_mtok": 0.30, "out_per_mtok": 1.20, "note": "Strong one-shot reasoning at a fraction of the price. " "Full tool access.", "badge": ""}, ] MODEL_IDS = {m["id"]: m for m in MODELS} # Anthropic API balance exhausted 2026-07-02; GLM-5.2 is the working default # until the org tops up. After top-up flip to claude-sonnet-5 (same price as # 4.6, strictly better) — agent.py's CHAIRMAN_MODEL fallback already is. DEFAULT_MODEL = "openrouter/z-ai/glm-5.2" USAGE_FILE = DASH / "usage.jsonl" USAGE_LOCK = threading.Lock() USAGE_MARK = "__CHAIRMAN_USAGE__" CONNECTOR_ENV = ROOT / "content-os" / "config" / ".env" # system prompt char count, measured once per server process (subprocess # import so this process never touches agent.py / prompt.py env handling) PROMPT_CHARS_CACHE = {"chars": None} # ---------------------------------------------------------------- helpers def now_iso(): return time.strftime("%Y-%m-%dT%H:%M:%S") def iso(ts): return time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(ts)) def slugify(s): s = re.sub(r"[^a-z0-9]+", "-", s.lower()).strip("-") return s[:60] or "task" def run(cmd, timeout=60, cwd=None, env=None): """subprocess.run wrapper that never raises; returns (rc, combined output).""" try: p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, cwd=cwd, env=env) out = (p.stdout or "") + (("\n" + p.stderr) if p.stderr else "") return p.returncode, out.strip() except subprocess.TimeoutExpired: return 124, f"timed out after {timeout}s" except FileNotFoundError as e: return 127, f"not found: {e}" except Exception as e: # noqa: BLE001 return 1, f"{type(e).__name__}: {e}" def ssh(host, remote_cmd, timeout=30): return run( ["ssh", "-o", "ConnectTimeout=4", "-o", "BatchMode=yes", host, remote_cmd], timeout=timeout, ) def probe_neo(force=False): """ssh probe, cached 30s. Returns {reachable, host, detail, checked_at}.""" with NEO_LOCK: if not force and NEO_CACHE["data"] and time.time() - NEO_CACHE["t"] < 30: return NEO_CACHE["data"] data = None for host in NEO_HOSTS: rc, out = ssh(host, "echo ok", timeout=12) if rc == 0 and "ok" in out: data = {"reachable": True, "host": host, "detail": f"ssh {host} ok", "checked_at": now_iso()} break if data is None: data = {"reachable": False, "host": None, "detail": "ssh failed on macmini-ts and macmini", "checked_at": now_iso()} NEO_CACHE["data"] = data NEO_CACHE["t"] = time.time() return data def tailscale_last_seen(): for ts_bin in ("/opt/homebrew/bin/tailscale", "/usr/local/bin/tailscale", "/Applications/Tailscale.app/Contents/MacOS/Tailscale"): if Path(ts_bin).exists(): rc, out = run([ts_bin, "status"], timeout=8) if rc != 0: return None for line in out.splitlines(): cols = line.split() # exact hostname match only — "keyss-mac-mini-4" is a stale old node if len(cols) >= 2 and cols[1] == "keyss-mac-mini": i = line.find("last seen") if i != -1: return line[i:].split(",")[0].strip() tail = line.split("macOS", 1) return tail[1].strip(" -\t") if len(tail) == 2 else line.strip() return None return None def heartbeat_loaded(): rc, _ = run(["launchctl", "list", "com.chairman.heartbeat"], timeout=10) return rc == 0 def read_json_body(handler): try: n = int(handler.headers.get("Content-Length") or 0) raw = handler.rfile.read(n) if n else b"" return json.loads(raw.decode("utf-8")) if raw else {} except Exception: # noqa: BLE001 return None # ---------------------------------------------------------------- chat runtime def split_usage(out): """Split a runtime stdout into (reply_text, usage_dict_or_None). The runtime (agent.py with CHAIRMAN_EMIT_USAGE=1) prints one line "__CHAIRMAN_USAGE__ {json}". Strip it from the reply wherever it appears; tolerate absence or malformed JSON -> usage None. """ usage = None kept = [] for line in (out or "").splitlines(): s = line.strip() if s.startswith(USAGE_MARK): try: parsed = json.loads(s[len(USAGE_MARK):].strip()) if isinstance(parsed, dict): usage = parsed except Exception: # noqa: BLE001 pass continue kept.append(line) return "\n".join(kept).strip(), usage def usage_ints(usage): vals = {} for k in ("input_tokens", "output_tokens", "cache_read_input_tokens", "cache_creation_input_tokens"): try: vals[k] = int(usage.get(k) or 0) except (TypeError, ValueError): vals[k] = 0 return vals def cost_estimate(usage_vals, model_id): """$ estimate for one reply using registry pricing for model_id.""" m = MODEL_IDS.get(model_id) or MODEL_IDS[DEFAULT_MODEL] cost = (usage_vals["input_tokens"] * m["in_per_mtok"] + usage_vals["output_tokens"] * m["out_per_mtok"] + usage_vals["cache_read_input_tokens"] * m["in_per_mtok"] * 0.1 + usage_vals["cache_creation_input_tokens"] * m["in_per_mtok"] * 1.25 ) / 1_000_000 return round(cost, 6) def append_usage_line(entry): with USAGE_LOCK: with USAGE_FILE.open("a", encoding="utf-8") as f: f.write(json.dumps(entry) + "\n") def agent_chat(agent, prompt, model=None): """One chat turn against an agent runtime. Returns (rc, reply, usage). usage is the parsed __CHAIRMAN_USAGE__ dict from the runtime stdout (the line is stripped from the reply) or None if the runtime emitted none. The subprocess env carries CHAIRMAN_MODEL= and CHAIRMAN_EMIT_USAGE=1. Shared by /api/chat and /api/conversations//message so both hit the exact same runtimes: chairman -> agent.py --chat, others -> subagent.py. Testing override: set env CHAIRMAN_CHAT_CMD to any command string and it runs INSTEAD of the real runtime, with the built prompt appended as the final argv element. Working examples: CHAIRMAN_CHAT_CMD="printf %s" echoes the prompt back CHAIRMAN_CHAT_CMD="/bin/sh -c 'echo test reply'" fixed canned reply Note: a bare CHAIRMAN_CHAT_CMD="printf 'test reply'" exits 1 under /usr/bin/printf because the appended prompt arg has no format character to land in, so the server reports it as an agent error. Use the forms above. The command is split with shlex.split into an argv list, never handed to a shell. """ override = os.environ.get("CHAIRMAN_CHAT_CMD", "").strip() if override: cmd = shlex.split(override) + [prompt] elif agent == "chairman": cmd = [PYTHON, str(CA / "agent.py"), "--chat", prompt] else: if not SUBAGENT_RUNNER.exists(): return 127, f"subagent runner not found: {SUBAGENT_RUNNER}", None cmd = [PYTHON, str(SUBAGENT_RUNNER), agent, "--chat", prompt] env = {**os.environ, "CHAIRMAN_MODEL": model or DEFAULT_MODEL, "CHAIRMAN_EMIT_USAGE": "1"} rc, out = run(cmd, timeout=360, cwd=str(CA), env=env) reply, usage = split_usage(out) return rc, reply, usage # ---------------------------------------------------------------- conversations # \Z (not $) so a trailing newline can never sneak through the id check CONV_ID_RE = re.compile(r"[a-z0-9-]+\Z") CONV_HISTORY_MSGS = 12 CONV_HISTORY_CHARS = 8000 TITLE_MAX = 46 # serializes load->mutate->save on conversation files so concurrent posts # can't drop each other's messages (agent runtime calls stay OUTSIDE it) CONV_LOCK = threading.Lock() def valid_cid(cid): return bool(CONV_ID_RE.fullmatch(cid)) def conv_path(cid): return CONV_DIR / f"{cid}.json" def load_conv(cid): p = conv_path(cid) if not p.is_file(): return None try: return json.loads(p.read_text(encoding="utf-8")) except Exception: # noqa: BLE001 return None def save_conv(conv): """Persist atomically: unique tmp then os.replace. Tmp name includes pid + random hex so two concurrent writers can never truncate each other's tmp file mid-write (a shared tmp path would let os.replace publish a half-written file). """ p = conv_path(conv["id"]) tmp = p.parent / f".{p.name}.{os.getpid()}.{os.urandom(4).hex()}.tmp" try: tmp.write_text(json.dumps(conv, ensure_ascii=False, indent=1), encoding="utf-8") os.replace(tmp, p) finally: if tmp.exists(): # only on a failure between write and replace try: tmp.unlink() except OSError: pass def make_title(text): """First user message trimmed to TITLE_MAX chars on a word boundary.""" t = " ".join(text.split()) if len(t) <= TITLE_MAX: return t cut = t[:TITLE_MAX] if " " in cut: cut = cut.rsplit(" ", 1)[0] return cut def build_conv_prompt(history, message): """Prompt for one turn. history = messages BEFORE the new user message. First message in a conversation goes to the runtime bare. After that the runtime gets the last CONV_HISTORY_MSGS messages, Keys:/You: prefixed, capped at CONV_HISTORY_CHARS total with the oldest dropped first. """ if not history: return message lines = [] for m in history[-CONV_HISTORY_MSGS:]: prefix = "Keys:" if m.get("role") == "user" else "You:" lines.append(f"{prefix} {m.get('text', '')}") while len(lines) > 1 and len("\n".join(lines)) > CONV_HISTORY_CHARS: lines.pop(0) convo = "\n".join(lines) if len(convo) > CONV_HISTORY_CHARS: # single message bigger than the cap convo = convo[-CONV_HISTORY_CHARS:] return ("Conversation so far (you are mid-conversation, continue naturally):\n" f"{convo}\n\nKeys' new message: {message}") # ---------------------------------------------------------------- usage/context def read_usage_entries(): """All parseable lines of usage.jsonl, oldest first. Missing file -> [].""" if not USAGE_FILE.exists(): return [] entries = [] try: lines = USAGE_FILE.read_text(encoding="utf-8", errors="replace").splitlines() except Exception: # noqa: BLE001 return [] for line in lines: line = line.strip() if not line: continue try: e = json.loads(line) except Exception: # noqa: BLE001 continue if isinstance(e, dict): entries.append(e) return entries def usage_agg(entries): def i(e, k): try: return int(e.get(k) or 0) except (TypeError, ValueError): return 0 def f(e, k): try: return float(e.get(k) or 0) except (TypeError, ValueError): return 0.0 return { "cost_est": round(sum(f(e, "cost_est") for e in entries), 6), "input_tokens": sum(i(e, "input_tokens") for e in entries), "output_tokens": sum(i(e, "output_tokens") for e in entries), "messages": len(entries), } def history_chars_of(messages): """Char count of the history block build_conv_prompt would send next turn.""" if not messages: return 0 lines = [] for m in messages[-CONV_HISTORY_MSGS:]: prefix = "Keys:" if m.get("role") == "user" else "You:" lines.append(f"{prefix} {m.get('text', '')}") while len(lines) > 1 and len("\n".join(lines)) > CONV_HISTORY_CHARS: lines.pop(0) convo = "\n".join(lines) if len(convo) > CONV_HISTORY_CHARS: convo = convo[-CONV_HISTORY_CHARS:] return len(convo) def system_prompt_chars(): """len(SYSTEM_PROMPT) measured in a throwaway subprocess, cached forever. Never import agent.py or prompt.py into this process (they read env and keys). Falls back to prompt.py file size if the subprocess fails. """ if PROMPT_CHARS_CACHE["chars"] is not None: return PROMPT_CHARS_CACHE["chars"] rc, out = run( [PYTHON, "-c", "import sys; sys.path.insert(0, '/Users/19keys/sovereign-empire-sweep/chairman-agent'); " "from prompt import SYSTEM_PROMPT; print(len(SYSTEM_PROMPT))"], timeout=10) chars = None if rc == 0: try: chars = int(out.strip().splitlines()[-1]) except (ValueError, IndexError): chars = None if chars is None: try: chars = (CA / "prompt.py").stat().st_size except OSError: chars = 0 PROMPT_CHARS_CACHE["chars"] = chars return chars # ---------------------------------------------------------------- connectors def connector_category(name): n = name.upper() if "ANTHROPIC" in n or "OPENAI" in n or "OPENROUTER" in n: return "ai" if "TELEGRAM" in n or "SENDBLUE" in n: return "messaging" if "SUPABASE" in n: return "database" if "STRIPE" in n: return "payments" if "ELEVEN" in n: return "voice" return "other" def mask_secret(value): """SECURITY ABSOLUTE: never return a full value from the env file.""" if len(value) >= 14: return value[:3] + "..." + value[-4:] return "set" def scan_connectors(): items = [] if CONNECTOR_ENV.exists(): try: lines = CONNECTOR_ENV.read_text(errors="replace").splitlines() except Exception: # noqa: BLE001 lines = [] for line in lines: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) k = k.strip() v = v.strip().strip('"').strip("'") if not k: continue items.append({ "name": k, "masked": mask_secret(v) if v else "", "present": bool(v), "category": connector_category(k), }) neo = probe_neo() items.append({ "name": "Neo (Mac Mini ssh)", "masked": neo["host"] or "offline", "present": bool(neo["reachable"]), "category": "other", }) items.append({ "name": "Claude Max CLI (Neo tiers)", "masked": "subscription", "present": True, "category": "ai", }) try: crm_creds() crm_ok = True except Exception: # noqa: BLE001 crm_ok = False items.append({ "name": "Sovereign CRM (/api/crm)", "masked": "wired" if crm_ok else "creds missing", "present": crm_ok, "category": "database", }) return items # ---------------------------------------------------------------- crm # Sovereign CRM read proxy (Supabase project mkbnyejnzstqgyvqoqau). # Creds are parsed from CONNECTOR_ENV at request time and NEVER logged or # echoed. All aggregates come from the prebuilt crm_* RPCs because PostgREST # aggregate functions are disabled on this project (PGRST123). Counts use # HEAD + Prefer: count=exact and read Content-Range, the proven pattern. CRM_TTL = 300 # seconds; force=1 bypasses CRM_CACHE = {} # key -> {"t": epoch, "data": dict} CRM_LOCK = threading.Lock() class CrmError(Exception): """CRM proxy failure carrying a safe, key-free message.""" def crm_creds(): """(rest_base_url, api_key) parsed fresh from the content-os .env. Prefers SUPABASE_SERVICE_KEY (required for the exact-count HEAD reads since the 2026-07-02 RLS lockdown removed anon table access); falls back to SUPABASE_ANON_KEY, which still covers the read RPCs. SUPABASE_URL in that file already ends with /rest/v1/; normalize anyway. Raises CrmError when the file or either key is missing. Never log values. """ url = key = anon = "" try: lines = CONNECTOR_ENV.read_text(errors="replace").splitlines() except OSError: raise CrmError(f"env file not readable: {CONNECTOR_ENV}") for line in lines: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) k, v = k.strip(), v.strip().strip('"').strip("'") if k == "SUPABASE_URL": url = v elif k == "SUPABASE_SERVICE_KEY": key = v elif k == "SUPABASE_ANON_KEY": anon = v key = key or anon if not url or not key: raise CrmError( "SUPABASE_URL or SUPABASE_SERVICE_KEY/SUPABASE_ANON_KEY missing " "from content-os/config/.env") if "/rest/v1" not in url: url = url.rstrip("/") + "/rest/v1/" if not url.endswith("/"): url += "/" return url, key def crm_http(path, method="GET", body=None, prefer=None, timeout=20): """One PostgREST call. Returns (status, headers dict, raw bytes). Error messages carry only the path (no query string) and the response snippet, never the key. """ base, anon = crm_creds() req = urllib.request.Request(base + path.lstrip("/"), method=method) req.add_header("apikey", anon) req.add_header("Authorization", f"Bearer {anon}") if prefer: req.add_header("Prefer", prefer) data = None if body is not None: data = json.dumps(body).encode("utf-8") req.add_header("Content-Type", "application/json") safe = path.split("?")[0] try: with urllib.request.urlopen(req, data=data, timeout=timeout) as resp: return resp.status, dict(resp.headers), resp.read() except urllib.error.HTTPError as e: snippet = "" try: snippet = e.read()[:300].decode("utf-8", "replace") except Exception: # noqa: BLE001 pass raise CrmError(f"supabase {method} {safe} failed: HTTP {e.code} {snippet}".strip()) except CrmError: raise except Exception as e: # noqa: BLE001 raise CrmError(f"supabase {method} {safe} failed: {type(e).__name__}: {e}") def crm_count(table, filters=""): """Exact row count via the crm_table_counts RPC. Direct HEAD + count=exact reads stopped working for the anon key with the 2026-07-02 RLS lockdown (no table grants); the SECURITY DEFINER RPC returns the same exact counts. The one filtered count the dashboard uses (cwk identities) has its own key in the RPC payload. """ counts = crm_rpc("crm_table_counts", {}) if isinstance(counts, list): counts = counts[0] if counts else {} key = table if filters: if table == "crm_platform_identities" and "platform=eq.cwk" in filters: key = "cwk_identities" else: raise CrmError(f"crm_table_counts cannot answer filtered count: {table} {filters}") v = counts.get(key) if isinstance(counts, dict) else None if v is None: raise CrmError(f"crm_table_counts has no entry for {key}") return int(v) def crm_rpc(name, body=None, timeout=25): """POST /rpc/, parsed JSON back.""" _, _, raw = crm_http(f"rpc/{name}", method="POST", body=body or {}, timeout=timeout) try: return json.loads(raw.decode("utf-8")) except Exception: # noqa: BLE001 raise CrmError(f"rpc {name} returned unparseable JSON") def crm_cached(key, force, fetch): """(data, cached_bool) with a CRM_TTL cache per key. fetch() may raise CrmError; the network call runs outside the lock.""" with CRM_LOCK: e = CRM_CACHE.get(key) if not force and e and time.time() - e["t"] < CRM_TTL: return e["data"], True data = fetch() with CRM_LOCK: CRM_CACHE[key] = {"t": time.time(), "data": data} return data, False def crm_fetch_overview(): """Headline numbers. One RPC (crm_dashboard_stats) + 4 exact counts.""" stats = crm_rpc("crm_dashboard_stats", {}) if isinstance(stats, list): # tolerate a single-row wrap stats = stats[0] if stats else {} if not isinstance(stats, dict): raise CrmError("crm_dashboard_stats returned an unexpected shape") def num(k): v = stats.get(k) try: return float(v) if v is not None else None except (TypeError, ValueError): return None contacts = crm_count("crm_contacts") identities = crm_count("crm_platform_identities") purchases = crm_count("crm_purchases") interactions = crm_count("crm_interactions") paying = int(stats.get("paying_contacts") or 0) hot1000 = int(stats.get("hot_1000_count") or 0) super10k = int(stats.get("super_10000_count") or 0) total_c = int(stats.get("total_contacts") or contacts) lead = stats.get("tiers") or {} lead_rev = stats.get("tier_revenue") or {} platforms = sorted( ({"platform": str(k), "count": int(v or 0)} for k, v in (stats.get("platforms") or {}).items()), key=lambda x: x["count"], reverse=True) # Three Tribes hierarchy mapped onto the real columns (lead_tier, # hot_1000, paying). tags[] and cognitive_type exist on only 29 contacts # so they cannot drive this yet. Counts are non-overlapping and sum to # total contacts. tiers = [ {"name": "Empire Builders", "count": hot1000, "definition": ("Hot 1000: top spenders by lifetime value (hot_1000 flag). " f"Peers and whales. lead_tier sovereign holds {int(lead.get('sovereign') or 0)} of them.")}, {"name": "Practitioners", "count": max(paying - hot1000, 0), "definition": "Bought at least once (LTV above zero) and not in the Hot 1000. Buyers to ascend."}, {"name": "Philosophers", "count": max(total_c - paying, 0), "definition": "On the list with no purchase yet. Lurkers to convert with content and offers."}, ] return { "totals": { "contacts": contacts, "identities": identities, "purchases": purchases, "interactions": interactions, "purchasers": paying or None, "revenue_total": num("total_revenue"), "revenue_gross": num("gross_revenue"), "revenue_90d": None, }, "platforms": platforms, "tiers": tiers, "lead_tiers": [ {"name": t, "count": int(lead.get(t) or 0), "revenue": float(lead_rev.get(t) or 0)} for t in ("sovereign", "hot", "warm", "cold") ], "hot_lists": {"hot_1000": hot1000, "super_10000": super10k}, "avg_ltv": num("avg_ltv"), "notes": [ "revenue_total is the sum of contacts.ltv_cents in dollars via crm_dashboard_stats; revenue_gross is platform level gross.", "revenue_90d is null: crm_purchases is a partial mostly historical ledger (about 85 rows in the last 90 days), so recent sales velocity is not truthfully derivable from it.", "purchasers means contacts with LTV above zero (paying_contacts from crm_dashboard_stats), not distinct rows in crm_purchases.", "Three Tribes ride on lead_tier, hot_1000 and paying counts; tags and cognitive_type are populated on only 29 contacts so they cannot drive the hierarchy yet.", ], "updated": now_iso(), } def crm_fetch_top(limit): """Ranked customer list via crm_top_contacts_detail (curated: filters B2B suspects, adds geo, real product names, paging). ltv is in dollars.""" rows = crm_rpc("crm_top_contacts_detail", {"p_limit": limit, "p_offset": 0}) if not isinstance(rows, list): raise CrmError("crm_top_contacts_detail returned an unexpected shape") return { "contacts": rows, "count": len(rows), "source": "rpc crm_top_contacts_detail", "updated": now_iso(), } def crm_fetch_ready(): """Readiness scorecard: can the Chairman actually work this list.""" checks = [] def add(name, ok, detail): checks.append({"name": name, "ok": bool(ok), "detail": detail}) try: crm_creds() add("creds_present", True, "SUPABASE_URL and SUPABASE_SERVICE_KEY (anon fallback) parsed from content-os/config/.env at request time, values never logged") except CrmError as e: add("creds_present", False, str(e)) try: ov, _ = crm_cached("overview", False, crm_fetch_overview) t = ov["totals"] rt = t.get("revenue_total") rev = f"${rt:,.2f} lifetime revenue" if isinstance(rt, (int, float)) else "revenue unavailable" add("overview_query", int(t.get("contacts") or 0) > 0, f"{t['contacts']:,} contacts, {t['identities']:,} identities, {rev} live via crm_dashboard_stats") except CrmError as e: add("overview_query", False, str(e)) try: top, _ = crm_cached("top:5", False, lambda: crm_fetch_top(5)) first = (top.get("contacts") or [{}])[0] who = first.get("full_name") or first.get("email") or "unknown" try: ltv = float(first.get("ltv") or 0) except (TypeError, ValueError): ltv = 0.0 add("top_query", bool(top.get("contacts")), f"crm_top_contacts_detail live, number one {who} at ${ltv:,.2f} LTV") except CrmError as e: add("top_query", False, str(e)) add("chairman_can_query", True, "the panel and chat agents hit this server at /api/crm/overview, /api/crm/top and /api/crm/ready; " "the schema and tested queries are documented in the CRM recon") try: n = crm_count("crm_platform_identities", "platform=eq.cwk") add("cwk_bridge", n > 0, f"cognitive-wealth-results writes live via crm_upsert_contact: {n} cwk identities in the CRM") except CrmError as e: add("cwk_bridge", False, str(e)) add("ziion_bridge", False, "ziion-mvp captures nothing today: every CTA is a plain anchor, no form storage. " "Wire the passport form to crm_upsert_contact with platform ziion, copying the CWK pattern") # Live probe: try a write on crm_contacts with the ANON key (impossible # filter, so nothing can actually change). 401/403 = locked down (good). # 2xx = the 2026-07-02 RLS lockdown has regressed. try: anon = "" for line in CONNECTOR_ENV.read_text(errors="replace").splitlines(): line = line.strip() if line.startswith("SUPABASE_ANON_KEY="): anon = line.split("=", 1)[1].strip().strip('"').strip("'") if not anon: raise CrmError("SUPABASE_ANON_KEY missing from env for probe") base, _ = crm_creds() req = urllib.request.Request( base + "crm_contacts?id=eq.00000000-0000-0000-0000-000000000000", method="DELETE") req.add_header("apikey", anon) req.add_header("Authorization", f"Bearer {anon}") try: with urllib.request.urlopen(req, timeout=15) as resp: add("write_access", False, f"anon key can still write crm_contacts (HTTP {resp.status}); " "the RLS lockdown has regressed — re-apply " "content-os/crm/migrations/2026-07-02-rls-lockdown/02-lockdown.sql") except urllib.error.HTTPError as e: if e.code in (401, 403): add("write_access", True, "anon key write to crm tables denied (HTTP %d): RLS lockdown of " "2026-07-02 is holding; writes are service_role only" % e.code) else: add("write_access", False, f"unexpected probe response HTTP {e.code}") except Exception as e: # noqa: BLE001 add("write_access", False, f"anon write probe failed to run: {e}") ok_n = sum(1 for c in checks if c["ok"]) return {"checks": checks, "score": f"{ok_n}/{len(checks)}", "updated": now_iso()} # ---------------------------------------------------------------- handler class H(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, fmt, *args): try: with open(DASHBOARD_LOG, "a") as f: f.write(f"{now_iso()} {self.address_string()} {fmt % args}\n") except Exception: # noqa: BLE001 pass def _json(self, o, code=200): b = json.dumps(o).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(b))) self.end_headers() self.wfile.write(b) def _bytes(self, b, ct, code=200): self.send_response(code) self.send_header("Content-Type", ct) self.send_header("Content-Length", str(len(b))) self.end_headers() self.wfile.write(b) # ------------------------------------------------------------ GET def do_GET(self): # noqa: N802 try: u = urlparse(self.path) q = parse_qs(u.query) if u.path in ("/", "/index.html"): idx = DASH / "index.html" if idx.is_file(): return self._bytes(idx.read_bytes(), "text/html; charset=utf-8") return self._bytes( b"" b"chat UI landing shortly, ops view at /ops" b"", "text/html; charset=utf-8") if u.path == "/ops": ops = DASH / "ops.html" if ops.is_file(): return self._bytes(ops.read_bytes(), "text/html; charset=utf-8") return self._json({"error": "ops.html not found"}, 404) if u.path == "/api/conversations": items = [] for p in CONV_DIR.glob("*.json"): if not p.is_file(): continue try: c = json.loads(p.read_text(encoding="utf-8")) except Exception: # noqa: BLE001 continue mt = p.stat().st_mtime items.append({ "id": c.get("id") or p.stem, "title": c.get("title") or "", "agent": c.get("agent") or "chairman", "updated": c.get("updated") or iso(mt), "count": len(c.get("messages") or []), "_mtime": mt, }) items.sort(key=lambda x: x["_mtime"], reverse=True) for it in items: it.pop("_mtime", None) return self._json({"items": items}) if u.path.startswith("/api/conversations/"): cid = u.path[len("/api/conversations/"):] if not valid_cid(cid): return self._json({"error": "invalid conversation id"}, 400) conv = load_conv(cid) if conv is None: return self._json({"error": "conversation not found"}, 404) return self._json({ "id": conv.get("id") or cid, "title": conv.get("title") or "", "agent": conv.get("agent") or "chairman", "model": conv.get("model") or DEFAULT_MODEL, "messages": conv.get("messages") or [], }) if u.path == "/api/models": return self._json({"models": MODELS, "default_model": DEFAULT_MODEL}) if u.path == "/api/usage": cid = (q.get("conversation") or [""])[0] if cid: if not valid_cid(cid): return self._json({"error": "invalid conversation id"}, 400) conv = load_conv(cid) if conv is None: return self._json({"error": "conversation not found"}, 404) msg_usages = [m.get("usage") for m in (conv.get("messages") or []) if isinstance(m.get("usage"), dict)] return self._json({"conversation": usage_agg(msg_usages)}) entries = read_usage_entries() today_key = time.strftime("%Y-%m-%d") week_key = time.strftime( "%Y-%m-%d", time.localtime(time.time() - 7 * 86400)) today = [e for e in entries if str(e.get("ts") or "")[:10] == today_key] week = [e for e in entries if str(e.get("ts") or "")[:10] >= week_key] by_model = {} for e in entries: m = str(e.get("model") or "unknown") b = by_model.setdefault(m, {"model": m, "cost_est": 0.0, "messages": 0}) try: b["cost_est"] += float(e.get("cost_est") or 0) except (TypeError, ValueError): pass b["messages"] += 1 models_out = sorted(by_model.values(), key=lambda x: x["cost_est"], reverse=True) for b in models_out: b["cost_est"] = round(b["cost_est"], 6) recent = [{"ts": e.get("ts") or "", "agent": e.get("agent") or "", "model": e.get("model") or "", "cost_est": e.get("cost_est") or 0} for e in entries[-20:]][::-1] return self._json({ "today": usage_agg(today), "week": usage_agg(week), "by_model": models_out, "recent": recent, }) if u.path == "/api/context": cid = (q.get("conversation") or [""])[0] if not cid: return self._json({ "system_prompt_tokens_est": 0, "history_chars": 0, "history_cap": CONV_HISTORY_CHARS, "history_pct": 0.0, "message_count": 0, "est_next_turn_input_tokens": 0, }) if not valid_cid(cid): return self._json({"error": "invalid conversation id"}, 400) conv = load_conv(cid) if conv is None: return self._json({"error": "conversation not found"}, 404) msgs = conv.get("messages") or [] sp_chars = system_prompt_chars() h_chars = history_chars_of(msgs) return self._json({ "system_prompt_tokens_est": sp_chars // 4, "history_chars": h_chars, "history_cap": CONV_HISTORY_CHARS, "history_pct": round(h_chars / CONV_HISTORY_CHARS * 100, 1), "message_count": len(msgs), "est_next_turn_input_tokens": (sp_chars + h_chars + 2000) // 4, }) if u.path == "/api/connectors": items = scan_connectors() return self._json({ "connectors": items, "counts": { "present": sum(1 for c in items if c["present"]), "total": len(items), }, }) if u.path == "/api/status": neo = probe_neo() hb_log = LOG_DIR / "heartbeat.log" mem_exists = MEMORY_FILE.exists() return self._json({ "chairman": { "model": os.environ.get("CHAIRMAN_MODEL", "claude-sonnet-4-6"), "heartbeat_loaded": heartbeat_loaded(), "last_heartbeat": iso(hb_log.stat().st_mtime) if hb_log.exists() else None, "inbox_count": len(list(INBOX.glob("*.md"))), "done_count": len(list(DONE.glob("*.md"))), }, "neo": { "reachable": neo["reachable"], "detail": neo["detail"], "checked_at": neo["checked_at"], }, "memory": { "updated": iso(MEMORY_FILE.stat().st_mtime) if mem_exists else None, "size_kb": round(MEMORY_FILE.stat().st_size / 1024, 1) if mem_exists else 0.0, }, "queue_count": len([p for p in QUEUE_DIR.iterdir() if p.is_file()]), }) if u.path == "/api/inbox": items = [] for p in sorted(INBOX.glob("*.md"), key=lambda p: p.stat().st_mtime, reverse=True): st = p.stat() try: preview = p.read_text(errors="replace")[:300] except Exception: # noqa: BLE001 preview = "" items.append({"name": p.name, "size": st.st_size, "mtime": iso(st.st_mtime), "preview": preview}) return self._json({"items": items}) if u.path == "/api/done": items = [{"name": p.name, "mtime": iso(p.stat().st_mtime)} for p in sorted(DONE.glob("*.md"), key=lambda p: p.stat().st_mtime, reverse=True)] return self._json({"items": items}) if u.path == "/api/memory": if not MEMORY_FILE.exists(): return self._json({"error": f"memory file not found: {MEMORY_FILE}"}, 404) return self._json({"markdown": MEMORY_FILE.read_text(errors="replace")}) if u.path == "/api/logs": name = (q.get("name") or [""])[0] if name not in ("heartbeat", "chat", "webhook"): return self._json({"error": "name must be heartbeat, chat or webhook"}, 400) try: lines = max(1, min(int((q.get("lines") or ["100"])[0]), 5000)) except ValueError: return self._json({"error": "lines must be a number"}, 400) f = LOG_DIR / f"{name}.log" if not f.exists(): return self._json({"text": ""}) text = "\n".join(f.read_text(errors="replace").splitlines()[-lines:]) return self._json({"text": text}) if u.path == "/api/subagents": deployed = SUBAGENT_RUNNER.exists() return self._json({"agents": [dict(a, deployed_local=deployed) for a in SUBAGENTS]}) if u.path == "/api/neo/status": neo = probe_neo() return self._json({ "reachable": neo["reachable"], "host": neo["host"], "detail": neo["detail"], "last_seen": tailscale_last_seen(), }) if u.path == "/api/crm/overview": force = (q.get("force") or [""])[0] in ("1", "true") try: data, cached = crm_cached("overview", force, crm_fetch_overview) except CrmError as e: return self._json({"error": str(e)}, 502) return self._json(dict(data, cached=cached)) if u.path == "/api/crm/top": force = (q.get("force") or [""])[0] in ("1", "true") try: limit = int((q.get("limit") or ["25"])[0]) except ValueError: return self._json({"error": "limit must be a number"}, 400) limit = max(1, min(limit, 100)) try: data, cached = crm_cached(f"top:{limit}", force, lambda: crm_fetch_top(limit)) except CrmError as e: return self._json({"error": str(e)}, 502) return self._json(dict(data, cached=cached)) if u.path == "/api/crm/ready": force = (q.get("force") or [""])[0] in ("1", "true") try: data, cached = crm_cached("ready", force, crm_fetch_ready) except CrmError as e: return self._json({"error": str(e)}, 502) return self._json(dict(data, cached=cached)) # static files under dashboard/ (path-traversal guarded; the # os.sep suffix stops sibling-dir prefix matches like dashboard-x) f = (DASH / u.path.lstrip("/")).resolve() if str(f).startswith(str(DASH) + os.sep) and f.is_file(): ct = {"html": "text/html; charset=utf-8", "css": "text/css", "js": "application/javascript", "json": "application/json", "png": "image/png", "svg": "image/svg+xml"}.get( f.suffix.lstrip("."), "application/octet-stream") return self._bytes(f.read_bytes(), ct) return self._json({"error": "not found"}, 404) except BrokenPipeError: pass except Exception as e: # noqa: BLE001 try: self._json({"error": f"{type(e).__name__}: {e}"}, 500) except Exception: # noqa: BLE001 pass # ------------------------------------------------------------ POST def do_POST(self): # noqa: N802 try: u = urlparse(self.path) d = read_json_body(self) if d is None: return self._json({"error": "body must be valid JSON"}, 400) if u.path == "/api/inbox": title = (d.get("title") or "").strip() body = d.get("body") or "" if not title: return self._json({"error": "title required"}, 400) name = f"{time.strftime('%Y%m%d-%H%M')}-{slugify(title)}.md" (INBOX / name).write_text(f"# {title}\n\n{body}\n") return self._json({"ok": True, "name": name}) if u.path == "/api/chat": message = (d.get("message") or "").strip() agent = (d.get("agent") or "chairman").strip() if not message: return self._json({"error": "message required"}, 400) if agent not in CHAT_AGENTS: return self._json( {"error": f"unknown agent: {agent}. valid: {', '.join(CHAT_AGENTS)}"}, 400) rc, out, _ = agent_chat(agent, message) if rc == 127 and "subagent runner not found" in out: return self._json({"error": out}, 404) if rc != 0: return self._json({"error": out[-2000:] or f"agent exited {rc}"}, 500) return self._json({"reply": out}) if u.path == "/api/conversations": agent = (d.get("agent") or "chairman").strip() if agent not in CHAT_AGENTS: return self._json( {"error": f"unknown agent: {agent}. valid: {', '.join(CHAT_AGENTS)}"}, 400) model = (d.get("model") or "").strip() if model and model not in MODEL_IDS: return self._json( {"error": f"unknown model: {model}. valid: " f"{', '.join(m['id'] for m in MODELS)}"}, 400) cid = time.strftime("%Y%m%d-%H%M%S") + "-" + os.urandom(2).hex() conv = {"id": cid, "title": "", "agent": agent, "model": model or DEFAULT_MODEL, "created": now_iso(), "updated": now_iso(), "messages": []} save_conv(conv) return self._json({"id": cid}) if u.path.startswith("/api/conversations/") and ( u.path.endswith("/message") or u.path.endswith("/messages") ): suffix = "/messages" if u.path.endswith("/messages") else "/message" cid = u.path[len("/api/conversations/"):-len(suffix)] if not valid_cid(cid): return self._json({"error": "invalid conversation id"}, 400) message = (d.get("message") or "").strip() if not message: return self._json({"error": "message required"}, 400) req_model = (d.get("model") or "").strip() if req_model and req_model not in MODEL_IDS: return self._json( {"error": f"unknown model: {req_model}. valid: " f"{', '.join(m['id'] for m in MODELS)}"}, 400) # append the user message under the lock so concurrent posts # to the same conversation can't drop each other's writes with CONV_LOCK: conv = load_conv(cid) if conv is None: return self._json({"error": "conversation not found"}, 404) agent = (d.get("agent") or conv.get("agent") or "chairman").strip() if agent not in CHAT_AGENTS: return self._json( {"error": f"unknown agent: {agent}. valid: {', '.join(CHAT_AGENTS)}"}, 400) if req_model: # persist the switch before running conv["model"] = req_model model = conv.get("model") or DEFAULT_MODEL history = list(conv.get("messages") or []) prompt = build_conv_prompt(history, message) conv.setdefault("messages", []).append( {"role": "user", "agent": agent, "text": message, "ts": now_iso()}) if not conv.get("title"): conv["title"] = make_title(message) title = conv["title"] conv["updated"] = now_iso() save_conv(conv) # user message survives even if the agent call fails # agent runtime call runs OUTSIDE the lock (can take 360s) rc, out, usage = agent_chat(agent, prompt, model=model) if rc == 127 and "subagent runner not found" in out: return self._json({"error": out}, 404) if rc != 0: return self._json({"error": out[-2000:] or f"agent exited {rc}"}, 500) usage_obj = None if usage: vals = usage_ints(usage) model_used = str(usage.get("model") or model) if model_used not in MODEL_IDS: model_used_for_pricing = model else: model_used_for_pricing = model_used usage_obj = dict( vals, model=model_used, cost_est=cost_estimate(vals, model_used_for_pricing)) with CONV_LOCK: # RELOAD before appending the reply — the in-memory copy is # stale after the agent call; writing it back would erase # anything another request appended meanwhile conv = load_conv(cid) or conv reply_msg = {"role": "assistant", "agent": agent, "text": out, "ts": now_iso()} if usage_obj: reply_msg["usage"] = usage_obj conv.setdefault("messages", []).append(reply_msg) conv["updated"] = now_iso() title = conv.get("title") or title save_conv(conv) if usage_obj: try: append_usage_line({ "ts": now_iso(), "conversation": cid, "agent": agent, "model": usage_obj["model"], "input_tokens": usage_obj["input_tokens"], "output_tokens": usage_obj["output_tokens"], "cache_read_input_tokens": usage_obj["cache_read_input_tokens"], "cache_creation_input_tokens": usage_obj["cache_creation_input_tokens"], "cost_est": usage_obj["cost_est"], }) except Exception: # noqa: BLE001 never fail the reply on ledger IO pass return self._json({"reply": out, "title": title or "", "usage": usage_obj}) if u.path == "/api/heartbeat": rc, out = run([PYTHON, str(CA / "agent.py"), "--heartbeat"], timeout=600, cwd=str(CA)) return self._json({"ok": rc == 0, "tail": out[-1500:]}) if u.path == "/api/neo/dispatch": message = (d.get("message") or "").strip() agent = (d.get("agent") or "").strip() if not message: return self._json({"error": "message required"}, 400) neo = probe_neo() if neo["reachable"]: remote = ("env PATH=/opt/homebrew/bin /opt/homebrew/bin/openclaw agent " f"--channel telegram --to '{TELEGRAM_CHAT_ID}' " f"--message {shlex.quote(message)} --deliver") rc, out = ssh(neo["host"], remote, timeout=90) return self._json({"ok": rc == 0, "delivered": rc == 0, "output": out[-2000:]}) name = f"{time.strftime('%Y%m%d-%H%M%S')}-{slugify(agent) if agent else 'direct'}.md" (QUEUE_DIR / name).write_text( f"# queued dispatch\nagent: {agent or 'direct'}\nqueued_at: {now_iso()}\n\n{message}\n") return self._json({"ok": True, "delivered": False, "queued": True, "name": name}) if u.path == "/api/neo/deploy": if not DEPLOY_SCRIPT.exists(): return self._json({"error": f"deploy script not found: {DEPLOY_SCRIPT}"}, 404) rc, out = run(["bash", str(DEPLOY_SCRIPT), "--apply"], timeout=300, cwd=str(CA)) return self._json({"ok": rc == 0, "output": out[-4000:]}) if u.path == "/api/neo/recover": action = (d.get("action") or "").strip() valid = ("probe", "restart_gateway", "warm_models", "test_telegram", "flush_queue") if action not in valid: return self._json({"error": f"action must be one of: {', '.join(valid)}"}, 400) if action == "probe": neo = probe_neo(force=True) return self._json({"ok": neo["reachable"], "output": json.dumps(neo)}) if action == "flush_queue": if not DEPLOY_SCRIPT.exists(): return self._json({"error": f"deploy script not found: {DEPLOY_SCRIPT}"}, 404) rc, out = run(["bash", str(DEPLOY_SCRIPT), "--flush-queue"], timeout=300, cwd=str(CA)) return self._json({"ok": rc == 0, "output": out[-4000:]}) neo = probe_neo() if not neo["reachable"]: return self._json({"ok": False, "output": "Neo unreachable. Check the Mac Mini is on."}) rc, out = ssh(neo["host"], RECOVER_CMDS[action], timeout=120) return self._json({"ok": rc == 0, "output": out[-4000:]}) return self._json({"error": "not found"}, 404) except BrokenPipeError: pass except Exception as e: # noqa: BLE001 try: self._json({"error": f"{type(e).__name__}: {e}"}, 500) except Exception: # noqa: BLE001 pass # ------------------------------------------------------------ DELETE def do_DELETE(self): # noqa: N802 try: u = urlparse(self.path) if u.path.startswith("/api/conversations/"): cid = u.path[len("/api/conversations/"):] if not valid_cid(cid): return self._json({"error": "invalid conversation id"}, 400) p = conv_path(cid) if not p.is_file(): return self._json({"error": "conversation not found"}, 404) dest = CONV_TRASH / p.name if dest.exists(): # keep both, never overwrite dest = CONV_TRASH / f"{cid}-{int(time.time())}.json" p.rename(dest) # soft delete: moved to trash/, never hard-deleted return self._json({"ok": True}) return self._json({"error": "not found"}, 404) except BrokenPipeError: pass except Exception as e: # noqa: BLE001 try: self._json({"error": f"{type(e).__name__}: {e}"}, 500) except Exception: # noqa: BLE001 pass if __name__ == "__main__": print(f"👑 Chairman Command → http://localhost:{PORT}", flush=True) ThreadingHTTPServer(("127.0.0.1", PORT), H).serve_forever()