#!/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=<model or default> and
    CHAIRMAN_EMIT_USAGE=1.

    Shared by /api/chat and /api/conversations/<id>/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/<name>, 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"<html><body style='background:#050505;color:#eee;font-family:monospace;padding:40px'>"
                    b"chat UI landing shortly, ops view at <a href='/ops' style='color:#c8b560'>/ops</a>"
                    b"</body></html>", "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()