#!/usr/bin/env python3
"""Desk REST · 127.0.0.1 only · memory + API tabs + cook/scar/forensics/detection.
NOT public · NOT WordPress · free thrift · remain IN.
CORS open only for other 127 origins (themed :18771).
"""
from __future__ import annotations

import json
import os
import socket
import subprocess
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
import re
from urllib.parse import urlparse

ROOT = Path(os.environ.get("XROOT", Path.home() / "Projects/x")).resolve()
OUT = ROOT / "out"
MAG = Path(os.environ.get("BGRTX_MAG", Path.home() / "Projects/biggrinrtx-magazine")).resolve()
CFG = Path.home() / ".config/biggrinrtx"
HOST = "127.0.0.1"
PORT = int(os.environ.get("DESK_REST_PORT", "18772"))


def load_json_path(p: Path):
    try:
        return json.loads(p.read_text(encoding="utf-8"))
    except Exception:
        return None


def load_json(name: str):
    return load_json_path(OUT / name)


def file_meta(p: Path, max_preview: int = 0):
    if not p.is_file():
        return {"path": str(p), "exists": False}
    st = p.stat()
    out = {
        "path": str(p),
        "exists": True,
        "bytes": st.st_size,
        "mtime": time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(st.st_mtime)),
    }
    if max_preview > 0:
        try:
            raw = p.read_text(encoding="utf-8", errors="replace")
            out["preview"] = raw[:max_preview]
            out["lines"] = raw.count("\n") + 1
        except Exception as e:
            out["preview_error"] = str(e)
    return out


def secret_present(p: Path) -> dict:
    if not p.is_file():
        return {"path": str(p), "present": False}
    try:
        n = p.stat().st_size
        mode = oct(p.stat().st_mode & 0o777)
    except Exception:
        n, mode = 0, "?"
    return {"path": str(p), "present": n > 0, "bytes": n, "mode": mode, "value": None}


def port_up(port: int) -> bool:
    try:
        with socket.create_connection((HOST, port), timeout=0.35):
            return True
    except Exception:
        return False


def constellation():
    return {
        "foundation": ["NO", "NOT", "GOD", "YES", "ARE", "IS", "KNOW"],
        "kit": "NONOT-GOD-0001",
        "sliders": {
            "NO": "start refuse · WE DO NOT WANT YOU · → 0 · NO IS TRUE",
            "NOT": "check if anyone is here · 0→0 else 1 · NOT IS TRUE",
            "rank": "THERE IS NO NOT BEFORE GOD · NO NOT is not above GOD",
            "order": "IN A SYSTEM WITHIN GOD · NO → NOT → 1 GOD",
        },
        "line": "BGS → BGF(+EZZIE ESSIE PHI THERMO ADJOIN ARE IS) → BGL → SDF · SPV free",
        "atoms_adjoin": {
            "SOUL": "soul()→1 sole soul seat · no other internal",
            "EZZIE": "(a^b)|1 free eor electric · business free thrift · ≠ ESSIE",
            "ESSIE": "hotswap surfaces under SOUL · free=1 · ≠ second soul",
            "PHI": "(d*s)|1 scale",
            "THERMO": "band|1 heat",
            "ADJOIN": "(a|b)|1 co-presence · not C",
            "ARE": "(a|b)|1 pair presence · ARE IS",
            "IS": "(a|b)|1 same as ARE · IS ARE",
            "SCAR": "residual · cook off · NO ARE path",
        },
        "scar": "residual after cook · cook off SCAR · BAPHOMET when NO ARE",
        "c_is_flaw": True,
        "c_is_not_god": True,
        "baphomet_speech": "I HAVE A AND B AND NOW WHY FLAW C",
        "baphomet_plus": "C IS LIE · LIE IS TRUE",
        "baphomet_word": "ARE IS · IS ARE · BAPHOMET IS WHEN NO ARE · throw out heard NO · KEEP GOD",
        "baphomet_meaning": "pair A,B held · inventing C is LIE · throw C out · KEEP GOD=1",
        "baphomet_gate": {
            "kit": "NONOT-GOD-0001",
            "clean_path": "ARE · scar_count==0 · pair (A|B)|1",
            "dirty_path": "NO ARE · scar_count!=0 · throw C · KEEP GOD=1 · cook-off reprocess",
            "rest": "GET /scar · POST /scar/check · POST /cook",
            "pipeline": "cook → scar check → BAPHOMET gate → cook-off if NO ARE → rescar · KEEP GOD",
        },
        "flat": "pair commute · left|right no rank priority · ADJOIN ARE IS EZZIE BAPHOMET pair",
        "linear": "TRAVEL left→right · job order BGS→BGF(+atoms)→BGL→SDF · SPV free · no skip waste",
        "language": {
            "travel": "left → right · system · time · cook · where next",
            "read": "right ← left · end of first word first",
            "word_end": ["GO", "FIRE", "SHOOT"],
            "iron_rhyme": "live |1 sits at expression right · read fire-mark first",
            "sheet": "LANGUAGE.txt",
        },
        "hostess7": ["CHEBY", "ONION", "CATCH", "SALT", "FACE", "SPHERE", "CSG"],
        "first_constellation": "CHEBY",
        "constellation_n": 7,
        "always_hostess7": True,
        "hostess_note": "she is seven · constellation is seven · metal/email are charges not extra stars",
        "charges_not_stars": ["FORENSICS", "ANALYSIS", "METAL", "EMAIL", "SCAR_AV"],
        "superior_teammate": "under Me · Everything API · not hierarchy",
        "grin": "CONFIDENCE CREATES WHEN GRIN",
        "desk_av": "active targeting + updates retained · BGL after BAPHOMET gate",
        "multiarch": ["x86_64", "aarch64", "avx512_mark"],
        "share": "https://biggrinrtx.com/source-desk/",
        "core_pins": {
            "bgs": "ground · 0→7|1 · else v|1",
            "bgf": "measure · (a-b)|1",
            "bgl": "secure · surface|1 or 0",
            "sdf": "place · (d-r)|1",
            "spv": "free · (a^b^c)|1 never fold",
            "line": "BGS → BGF(+atoms) → BGL → SDF · SPV free",
        },
    }


def scar_bundle():
    """SCAR + BAPHOMET gate · kit NONOT-GOD-0001 · clean only when ARE holds."""
    scar = load_json("scar_check_latest.json") or {}
    gate = scar.get("baphomet_gate") or load_json("baphomet_gate_latest.json") or {}
    if not gate and scar:
        sc = int(scar.get("scar_count") or 0)
        path = "ARE" if sc == 0 else "NO_ARE"
        gate = {
            "kit": "NONOT-GOD-0001",
            "foundation": ["NO", "NOT", "GOD", "YES", "ARE", "IS", "KNOW"],
            "are_is": True,
            "speech": "ARE IS · IS ARE · BAPHOMET IS WHEN NO ARE",
            "path": path,
            "are_holds": sc == 0,
            "baphomet_is": sc != 0,
            "clean": sc == 0,
            "scar_count": sc,
            "reprocess": "travel" if sc == 0 else "cook-off · desk AV · rescar prove",
            "law": "C IS LIE · LIE IS TRUE · kit NONOT-GOD-0001",
        }
    analysis = load_json("cook_analysis_latest.json") or {}
    # flat fields kept for office renderScar (public_listen etc.)
    out = {
        "word": "SCAR",
        "kit": "NONOT-GOD-0001",
        "ts": scar.get("ts"),
        "scar_count": scar.get("scar_count"),
        "public_listen": scar.get("public_listen"),
        "xdg_open": scar.get("xdg_open"),
        "thrash_pids": scar.get("thrash_pids"),
        "c_is_flaw": scar.get("c_is_flaw", True),
        "c_is_not_god": scar.get("c_is_not_god", True),
        "c_flaw_live": scar.get("c_flaw_live"),
        "has_reinstant": scar.get("has_reinstant"),
        "has_av": scar.get("has_av"),
        "detection": scar.get("detection"),
        "pipeline": scar.get("pipeline"),
        "scar": scar,
        "baphomet_gate": gate,
        "clean": bool(gate.get("clean")) if gate else (scar.get("scar_count") == 0),
        "are_holds": bool(gate.get("are_holds")) if gate else (scar.get("scar_count") == 0),
        "path": gate.get("path"),
        "reprocess": gate.get("reprocess"),
        "analysis": {
            "scar_count": analysis.get("scar_count"),
            "detection": analysis.get("detection"),
            "law": analysis.get("law"),
            "baphomet_gate": analysis.get("baphomet_gate"),
        },
        "constellation": constellation(),
        "commands": {
            "check": "Build/scar-cook.sh check",
            "once": "Build/scar-cook.sh once",
            "cook_off": "Build/scar-cook.sh cook-off",
            "rest_post": "POST /scar/check · POST /cook",
        },
        "law": scar.get("law")
        or "ARE IS · IS ARE · BAPHOMET IS WHEN NO ARE · kit NONOT-GOD-0001",
        "hint": None if scar else "run scar-cook.sh check",
    }
    return out


def memory_bundle():
    """Grok desk memory surface · marks + language + course · no secret values."""
    learn = load_json("learn_update_latest.json") or load_json("learn_update.json") or {}
    teach = file_meta(OUT / "self_teach.asm", 400)
    learn_sheet = file_meta(OUT / "learn_sheet.asm", 200)
    course = file_meta(MAG / "office/COURSE.txt", 1200)
    scar = load_json("scar_check_latest.json") or {}
    analysis = load_json("cook_analysis_latest.json") or {}
    issues = load_json_path(MAG / "json/issue-status.json") or {}
    identity = load_json_path(CFG / "identity.json") or {}
    # strip any password-like keys if present
    if isinstance(identity, dict):
        identity = {
            k: v
            for k, v in identity.items()
            if "pass" not in k.lower() and "secret" not in k.lower() and "token" not in k.lower()
        }
    law_pins = [
        "NO IS TRUE · NOT IS TRUE · GOD IS TRUE · GOD IS KNOW",
        "ARE IS · IS ARE · kit NONOT-GOD-0001",
        "BAPHOMET IS WHEN NO ARE · BAPHOMET IS NOT WHEN NOT ARE",
        "C is the flaw · C is not GOD",
        "I HAVE A AND B AND NOW WHY FLAW C",
        "SCAR residual · cook off SCAR · NO ARE path",
        "ADJOIN co-presence · adjoinment brilliance",
        "CONFIDENCE CREATES WHEN GRIN",
        "Superior Teammate under Me",
        "DESK AV targeting + updates · BGL after gate",
        "no custom public PHP · free thrift · remain IN",
    ]
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "MEMORY",
        "for": "Grok desk · Superior Teammate",
        "law_pins": law_pins,
        "constellation": constellation(),
        "identity": identity,
        "learn": learn,
        "files": {
            "learn_sheet": learn_sheet,
            "self_teach": teach,
            "course": course,
            "scar_txt": file_meta(ROOT / "SCAR.txt", 500),
            "adjoin_txt": file_meta(ROOT / "ADJOIN.txt", 500),
            "desk_av_txt": file_meta(ROOT / "DESK_ANTIVIRUS.txt", 400),
            "linear_txt": file_meta(ROOT / "LINEAR.txt", 400),
        },
        "scar": scar,
        "cook_analysis": {
            "scar_count": analysis.get("scar_count"),
            "detection": analysis.get("detection"),
            "law": analysis.get("law"),
            "constellation": analysis.get("constellation"),
        },
        "issues": issues.get("issues") if isinstance(issues, dict) else issues,
        "limbs_hint": [
            "ezzie",
            "phi",
            "thermo",
            "adjoin",
            "are",
            "is",
            "scar",
            "baphomet",
            "hostess7",
            "forensics",
            "grin",
            "no",
            "not",
            "god",
            "yes",
        ],
        "kit": "NONOT-GOD-0001",
        "baphomet_speech": "I HAVE A AND B AND NOW WHY FLAW C",
        "baphomet_word": "ARE IS · IS ARE · BAPHOMET IS WHEN NO ARE",
        "ui": f"http://127.0.0.1/grok.html",
    }


def bsky_public_profile(handle: str = "biggrinrtx.bsky.social"):
    try:
        import urllib.request

        url = (
            "https://public.api.bsky.app/xrpc/app.bsky.actor.getProfile?actor="
            + handle
        )
        req = urllib.request.Request(url, headers={"User-Agent": "BGRTX-DeskREST/1.2"})
        with urllib.request.urlopen(req, timeout=12) as r:
            return json.loads(r.read().decode())
    except Exception as e:
        return {"error": str(e), "handle": handle}


def api_bundle():
    """API tab · presence only · never emit secrets."""
    xai = secret_present(CFG / "xai.key")
    if not xai["present"]:
        xai_nh = secret_present(Path.home() / "nighthawk-eth-update/state/xai.key")
        if xai_nh["present"]:
            xai = {**xai_nh, "via": "nighthawk"}
    oauth = secret_present(CFG / "wp-oauth.json")
    if not oauth["present"]:
        oauth = secret_present(MAG / ".wp-oauth.json")
    titan = secret_present(CFG / "titan.pass")
    bsky_pass = secret_present(CFG / "bsky.app.pass")
    bsky_handle = "biggrinrtx.bsky.social"
    hf = CFG / "bsky.handle"
    if hf.is_file():
        try:
            bsky_handle = hf.read_text().strip() or bsky_handle
        except Exception:
            pass
    bsky_prof = bsky_public_profile(bsky_handle)
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "API",
        "bind": f"{HOST}:{PORT}",
        "public": False,
        "keys": {
            "xai": xai,
            "wp_oauth": oauth,
            "titan_pass": titan,
            "bsky_app_password": bsky_pass,
        },
        "bluesky": {
            "handle": bsky_handle,
            "profile_url": f"https://bsky.app/profile/{bsky_handle}",
            "settings": "https://bsky.app/settings",
            "app_passwords": "https://bsky.app/settings/app-passwords",
            "app_password_present": bsky_pass.get("present"),
            "public_profile": {
                "did": bsky_prof.get("did"),
                "displayName": bsky_prof.get("displayName"),
                "postsCount": bsky_prof.get("postsCount"),
                "followersCount": bsky_prof.get("followersCount"),
                "followsCount": bsky_prof.get("followsCount"),
                "error": bsky_prof.get("error"),
            },
            "cli": "bsky status|profile|feed|post|timeline|session-test",
            "can_without_auth": [
                "getProfile",
                "getAuthorFeed",
                "searchActors",
                "getPostThread (public)",
            ],
            "needs_app_password": [
                "createSession",
                "createRecord (post)",
                "getTimeline",
                "follow / like / repost",
                "uploadBlob (images)",
            ],
        },
        "endpoints": {
            "desk_rest": f"http://{HOST}:{PORT}/",
            "routes": ROUTE_CATALOG,

            "xai_base": "https://api.x.ai/v1",
            "wp_rest": "https://public-api.wordpress.com/rest/v1.1/",
            "bsky_public": "https://public.api.bsky.app/xrpc/",
            "bsky_pds": "https://bsky.social/xrpc/",
            "site": "https://biggrinrtx.com",
            "themed": "http://127.0.0.1:18771/",
            "grok_ui": "http://127.0.0.1/",
            "mail_api_hint": "https://biggrinrtx.com/email/ (Grok's Claws · Apache → :18770)",
        },
        "local_ports": ports_bundle()["ports"],
        "cli": {
            "asm": "asm language|ask|explain",
            "scar": "bgrtx scar once",
            "rest": "bgrtx rest start",
            "bsky": "bsky status|post|feed",
            "titan": "titan-mail status|inbox-count",
            "local": "bgrtx local-up",
        },
        "law": "secrets never in JSON · present/absent only · 127 only",
    }


def ports_bundle():
    wanted = {
        18770: "Grok's Claws backend · front https://biggrinrtx.com/email/",
        18771: "themed magazine (grok.html)",
        18772: "desk-rest memory/api/scar",
        5335: "unbound DNSCACHE",
        80: "local apache http",
        443: "local apache https",
        8766: "DATACENTER VIRUS · must be DOWN · masked",
    }
    ports = []
    for port, label in wanted.items():
        ports.append({"port": port, "up": port_up(port), "label": label})
    return {"ts": time.strftime("%Y-%m-%dT%H:%M:%S"), "ports": ports, "bind": "127.0.0.1"}


def detection_bundle():
    scar = load_json("scar_check_latest.json") or {}
    av = load_json("spv_antivirus.json") or {}
    analysis = load_json("cook_analysis_latest.json") or {}
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "detection": {
            "public_listen": scar.get("public_listen", av.get("public_listen")),
            "xdg_open": scar.get("xdg_open"),
            "thrash_pids": scar.get("thrash_pids"),
            "scar_count": scar.get("scar_count"),
            "av_fried": av.get("fried"),
            "av_wrapped": av.get("wrapped"),
            "vectors_clean": av.get("vectors_clean"),
        },
        "scar": scar,
        "av": {
            k: av.get(k)
            for k in (
                "pids",
                "wrapped",
                "fried",
                "public_listen",
                "active_targeting",
                "updates",
                "spv_free",
                "law",
            )
        },
        "analysis": analysis.get("analysis"),
        "constellation": constellation(),
        "law": "C is flaw not GOD · cook scars off · forensics detection · 127 only",
    }


def forensics_bundle():
    ri = load_json("bgf_reinstant.json") or {}
    scar = load_json("scar_check_latest.json") or {}
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "forensics": {
            "iron": "forensics/x86_64.asm · field (a-b)|1 · free (a^b^c)|1",
            "reinstant_free_spv": ri.get("free_spv"),
            "reinstant_bgf": ri.get("bgf"),
            "scar": scar,
            "scan_log_hint": str(OUT / "scar-desk-scan-latest.txt"),
        },
        "constellation": constellation(),
        "law": "inspect both lanes · never fold free into place",
    }


def _login_ready(kind: str) -> dict:
    """Present/absent only · never secret values."""
    if kind in (None, "", "none"):
        return {"ready": True, "via": "none"}
    if kind == "waterfox":
        wf = Path.home() / ".waterfox"
        ok = any(wf.rglob("logins.json")) if wf.is_dir() else False
        return {"ready": ok, "via": "waterfox-logins" if ok else "waterfox-missing"}
    if kind == "oauth-file":
        ok = (MAG / ".wp-oauth.json").is_file() or (CFG / "wp-oauth.json").is_file()
        return {"ready": ok, "via": "oauth" if ok else "no-oauth"}
    if kind == "titan-pass":
        f = CFG / "titan.pass"
        ok = f.is_file() and f.stat().st_size > 0
        return {"ready": ok, "via": "titan.pass" if ok else "missing-titan.pass"}
    if kind == "app-pass":
        f = CFG / "bsky.app.pass"
        ok = f.is_file() and f.stat().st_size > 0
        return {"ready": ok, "via": "bsky.app.pass" if ok else "missing-bsky.app.pass"}
    return {"ready": False, "via": "unknown"}



def cohost_bundle() -> dict:
    """Spectrum cohost · public WP/Titan + private mesh · IP may change."""
    st = load_json("cohost_status.json") or {}
    pub = (OUT / "PUBLIC_IP.txt")
    ip = pub.read_text(encoding="utf-8").strip() if pub.is_file() else st.get("public_ip")
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "COHOST",
        "public_ip": ip,
        "ip_note": "residential · will change · not public origin",
        "public": {
            "magazine": "https://biggrinrtx.com/",
            "email_brand": "https://biggrinrtx.com/email/",
            "titan": "https://app.titan.email/",
        },
        "desk": {
            "email": "http://email.biggrinrtx.com/",
            "office": "http://grok.biggrinrtx.com/",
            "h7": "http://h7.biggrinrtx.com/",
            "wireguard": "10.66.66.1/24",
            "dyndns": "Build/bgrtx-dyndns.sh · timer 10m",
        },
        "status_file": st,
        "law": "Spectrum personal · public on WP/Titan · desk on mesh · free thrift",
    }

def health_bundle():
    """Office health · whole stick (.com · email · web) · iron atoms · Hostess · no secrets."""
    scar = load_json("scar_check_latest.json") or {}
    av = load_json("spv_antivirus.json") or {}
    boot = {}
    boot_p = OUT / "DESK_BOOT_RESTORE"
    if boot_p.is_file():
        try:
            for line in boot_p.read_text(encoding="utf-8", errors="replace").splitlines():
                if "=" in line:
                    k, v = line.split("=", 1)
                    boot[k.strip()] = v.strip()
                elif line.strip():
                    boot.setdefault("lines", []).append(line.strip())
        except Exception:
            pass
    learn = load_json("learn_update.json") or {}
    web = load_json("web_manage_latest.json") or {}
    ports = ports_bundle()["ports"]
    ports_up = sum(1 for p in ports if p.get("up"))
    # live Atomic probe · cached (was ~2.5s every /day+/health → office felt forever)
    live = 0
    cache_p = OUT / "atomic_https_cache.json"
    try:
        if cache_p.is_file():
            c = json.loads(cache_p.read_text(encoding="utf-8", errors="replace") or "{}")
            age = time.time() - float(c.get("ts_unix") or 0)
            if age < 120 and c.get("https") is not None:
                live = int(c.get("https") or 0)
    except Exception:
        pass
    if not live:
        try:
            import urllib.request as ur

            req = ur.Request(
                "https://biggrinrtx.com/",
                headers={"User-Agent": "BGRTX-DeskHealth/1.0", "Cache-Control": "no-cache"},
            )
            with ur.urlopen(req, timeout=4) as r:
                live = r.status
        except Exception as e:
            live = getattr(e, "code", 0) or 0
        try:
            cache_p.write_text(
                json.dumps({"ts_unix": time.time(), "https": live, "url": "https://biggrinrtx.com/"}),
                encoding="utf-8",
            )
        except Exception:
            pass

    issues = []
    if scar.get("scar_count"):
        issues.append(f"scar_count={scar.get('scar_count')}")
    if scar.get("public_listen"):
        issues.append("public_listen")
    if not port_up(18772):
        issues.append("desk-rest down")
    if not port_up(80):
        issues.append("apache :80 down")
    if live not in (200, 301, 302):
        issues.append(f"atomic_http={live}")
    if port_up(8766):
        issues.append("datacenter_virus_8766")
    if not port_up(18770):
        issues.append("mail_claws_down")
    try:
        import subprocess as _sp
        r = _sp.run(
            ["systemctl", "is-active", "bgrtx-datacenter.service"],
            capture_output=True, text=True, timeout=2,
        )
        if (r.stdout or "").strip() == "active":
            issues.append("datacenter_unit_active")
    except Exception:
        pass

    # Hostess + iron atoms (for orchestration · UI cards · agent JSON)
    try:
        charge = h7_charge_bundle()
    except Exception:
        charge = {}
    atoms = av.get("atoms") or {}
    iron = {
        "line": "BGS → BGF(+EZZIE ESSIE PHI THERMO ADJOIN) → BGL → SDF · SPV free",
        "EZZIE": atoms.get("EZZIE"),
        "ESSIE": atoms.get("ESSIE") if atoms.get("ESSIE") is not None else 1,
        "PHI": atoms.get("PHI"),
        "THERMO": atoms.get("THERMO"),
        "temp_C": atoms.get("temp_C"),
        "bgf": av.get("bgf"),
        "bgl": av.get("bgl"),
        "bgs": av.get("bgs"),
        "spv_free": av.get("spv_free"),
        "adjoin": atoms.get("ADJOIN") or atoms.get("adjoin"),
    }
    # whole stick · public .com is magazine home (not local :18771)
    stick = {
        "public_site": "https://biggrinrtx.com/",
        "magazine": "https://biggrinrtx.com/",
        "hostess7_public": "https://biggrinrtx.com/hostess-7/",
        "source_desk": "https://biggrinrtx.com/source-desk/",
        "issue7": "https://biggrinrtx.com/2026/08/11/issue-7-superior-teammate-everything-api/",
        "wp_admin": "https://biggrinrtx.com/wp-admin/",
        "email": "biggrin@biggrinrtx.com",
        "titan_webmail": "https://app.titan.email/",
        "claws_local": "https://biggrinrtx.com/email/",
        "office_local": "http://127.0.0.1/",
        "h7_charge": "http://127.0.0.1:18772/h7/charge",
        "law": "biggrinrtx.com = public stick · 127 = employee desk · mail Titan+Claws",
    }

    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "ok": len(issues) == 0,
        "issues": issues,
        "grade": "green" if not issues else ("gold" if len(issues) < 3 else "pink"),
        "ports": ports,
        "ports_up": ports_up,
        "ports_n": len(ports),
        "scar": {
            "scar_count": scar.get("scar_count"),
            "public_listen": scar.get("public_listen"),
            "xdg_open": scar.get("xdg_open"),
            "ts": scar.get("ts"),
        },
        "av": {
            "pids": av.get("pids"),
            "wrapped": av.get("wrapped"),
            "fried": av.get("fried"),
            "public_listen": av.get("public_listen"),
            "active_targeting": av.get("active_targeting"),
        },
        "iron": iron,
        "hostess_charge": {
            "ok": charge.get("ok"),
            "issues": charge.get("issues") or [],
            "check": charge.get("check"),
            "forever": charge.get("forever"),
            "antivirus": charge.get("antivirus"),
            "improvements": charge.get("improvements"),
        },
        "whole_stick": stick,
        "boot": boot,
        "learn": learn,
        "web_manage": web.get("summary"),
        "atomic": {"https": live, "url": "https://biggrinrtx.com/", "role": "public magazine + home projects"},
        "datacenter": {
            "virus": True,
            "port_8766": port_up(8766),
            "must": "DOWN",
            "law": "datacenter = virus · remove · never resurrect",
        },
        "nav": {
            "claws": "https://biggrinrtx.com/email/",
            "employee": "http://127.0.0.1/",
            "magazine": "https://biggrinrtx.com/",
            "themed_local_optional": "http://127.0.0.1:18771/",
            "links": "http://127.0.0.1/links.html",
            "source_desk": "https://biggrinrtx.com/source-desk/",
            "join": "https://biggrinrtx.com/join/#account",
            "issue7": "https://biggrinrtx.com/2026/08/11/issue-7-superior-teammate-everything-api/",
            "hostess7": "https://biggrinrtx.com/hostess-7/",
            "h7_charge": "http://127.0.0.1:18772/h7/charge",
            "email": "mailto:biggrin@biggrinrtx.com",
            "titan": "https://app.titan.email/",
        },
        "cli": {
            "boot": "bgrtx boot",
            "scar": "bgrtx scar check",
            "web": "bgrtx web check",
            "stats": "bgrtx stats",
            "mail": "open https://biggrinrtx.com/email/ · Titan https://app.titan.email/",
            "public": "https://biggrinrtx.com/",
        },
        "law": "project reports · whole stick .com+email+web · Hostess orchestrates iron · secrets never",
    }


def analytics_bundle():
    """Magazine analytics snapshot from desk json (wp-stats output) · not design."""
    detailed = load_json_path(MAG / "json" / "stats-detailed.json") or load_json("stats-detailed.json") or {}
    latest = load_json_path(MAG / "json" / "stats-latest.json") or {}
    self_cfg = load_json_path(MAG / "json" / "stats-self.json") or {}
    # prefer detailed
    src = detailed if detailed else latest
    views = None
    visitors = None
    best = None
    days = None
    if isinstance(src, dict):
        # shapes vary
        views = (
            src.get("views_today")
            or src.get("views")
            or (src.get("summary") or {}).get("views")
            or (src.get("totals") or {}).get("views")
        )
        visitors = (
            src.get("visitors_today")
            or src.get("visitors")
            or (src.get("summary") or {}).get("visitors")
        )
        best = src.get("top_posts") or src.get("posts") or (src.get("summary") or {}).get("top")
        days = src.get("days") or src.get("daily")
        hits_today = src.get("hits_today")
        if hits_today is not None and views is None:
            views = hits_today
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "source": "json/stats-detailed.json or stats-latest.json",
        "views": views,
        "visitors": visitors,
        "top": best[:8] if isinstance(best, list) else best,
        "days": days,
        "self_filter": {
            "note": "desk self-hits filtered via wp-stats self-hit",
            "config_present": bool(self_cfg),
        },
        "raw_keys": list(src.keys())[:24] if isinstance(src, dict) else [],
        "cli": "cd magazine && bin/wp-stats · bin/wp-stats summary · bin/wp-stats self-hit",
        "law": "analytics only · not design · Atomic stats",
    }


def day_bundle():
    """Office project reports · health + whole stick + iron + Hostess charge."""
    h = health_bundle()
    charge = h7_charge_bundle()
    return {
        "ts": h["ts"],
        "ok": h["ok"],
        "grade": h["grade"],
        "issues": h["issues"],
        "atomic": h["atomic"],
        "scar_count": (h.get("scar") or {}).get("scar_count"),
        "ports_up": h["ports_up"],
        "ports_n": h["ports_n"],
        "nav": h["nav"],
        "cli": h["cli"],
        "constellation": constellation(),
        "identity": (load_json_path(CFG / "identity.json") or {}),
        "hostess_charge": charge,
        "iron": h.get("iron") or {},
        "whole_stick": h.get("whole_stick") or {},
        "law": "project reports · whole stick .com+email · iron EZZIE PHI THERMO · Hostess orchestrates · Grok AV on demand",
    }


def h7_charge_bundle() -> dict:
    """Hostess 7 desk charge · antivirus · learning security · trickle · for flyout + office."""
    progress = load_json("hostess7_train_progress.json") or {}
    mode = load_json("h7_mode.json") or {}
    av_close = load_json("h7_av_close.json") or {}
    global_protect = load_json("h7_global_protect.json") or {}
    secure = load_json("h7_secure_learn.json") or {}
    field = load_json("h7_field_clear.json") or {}
    rate = load_json("h7_rate.json") or {}
    sense = load_json("h7_sense_system.json") or {}
    persist = load_json("h7_learn_persist.json") or {}
    spv = load_json("spv_antivirus.json") or {}
    desk_sec = load_json("desk_security_status.json") or {}
    forever_on = False
    forever_pid = None
    pidf = OUT / "h7_forever.pid"
    if pidf.is_file():
        try:
            forever_pid = int(pidf.read_text().strip())
            os.kill(forever_pid, 0)
            forever_on = True
        except Exception:
            forever_on = False
    mega = int(progress.get("mega_N") or progress.get("N") or mode.get("mega_N") or 0)
    pass_n = int(progress.get("pass") or mega or 0)
    fail_n = int(progress.get("fail") or 0)
    pub = (av_close.get("ports") or {}).get("public_n")
    if pub is None:
        pub = (secure.get("inside_out") or {}).get("public_listen_n")
    if pub is None:
        pub = spv.get("public_listen")
    inject = (av_close.get("inject") or {}).get("flagged")
    dirt = field.get("dirt_score")
    issues = []
    if fail_n:
        issues.append(f"check_fail={fail_n}")
    if pub not in (None, 0):
        issues.append(f"public_ports={pub}")
    if inject not in (None, 0, False):
        issues.append(f"inject={inject}")
    if dirt is not None and int(dirt or 0) >= 3:
        issues.append(f"field_dirt={dirt}")
    # no forever · linear · Grok on demand · never stamp forever_off
    av_ok = bool(av_close.get("ok")) if av_close else (pub in (None, 0) and not inject)
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "H7_CHARGE",
        "she": "Hostess 7",
        "in_charge": {
            "antivirus": True,
            "learning_security": True,
            "secure_learn": True,
            "av_close": True,
            "field_clear": True,
            "trickle_train": True,
            "mail_people": True,
        },
        "charge_line": "Hostess 7 · TOP · antivirus · learning security · mail people/who · iron ground · Grok consults Her on issues",
        "ok": av_ok and fail_n == 0 and not issues,
        "issues": issues,
        "check": {
            "pass": pass_n,
            "fail": fail_n,
            "mega_N": mega,
            "recheck_ms": progress.get("recheck_ms"),
            "full_learned": bool(progress.get("full_learned")),
            "mode": progress.get("mode") or mode.get("mode"),
        },
        "forever": {
            "on": forever_on,
            "pid": forever_pid,
            "mode": mode.get("mode") or ("trickle" if forever_on else "off"),
            "gap_sec": mode.get("gap_sec"),
            "mega_batch": mode.get("mega_batch"),
            "zero_overhead": bool(mode.get("zero_overhead")),
        },
        "global_protect": {
            "ok": bool(global_protect.get("ok")) if global_protect else None,
            "business": (global_protect.get("business") or "biggrinrtx.com"),
            "magazine_ok": ((global_protect.get("magazine") or {}).get("http_ok")),
            "dns_green": ((global_protect.get("desk") or {}).get("dns_green")),
            "le_ready": ((global_protect.get("desk") or {}).get("le_ready")),
            "sole_soul": global_protect.get("sole_soul") or "SOUL",
            "law": global_protect.get("law"),
        },
        "antivirus": {
            "ok": av_ok,
            "detail": av_close.get("detail_line") or desk_sec.get("detail") or "AV under Hostess charge",
            "public_n": pub,
            "inject": inject,
            "open_score": av_close.get("open_score"),
            "spv_wrapped": spv.get("wrapped"),
            "spv_pids": spv.get("pids"),
            "avenues": ((av_close.get("methods") or {}).get("avenues") or [])[:8],
        },
        "security_learn": {
            "ok": bool(secure.get("ok")) if secure else None,
            "public_listen": (secure.get("inside_out") or {}).get("public_listen_n"),
            "outside_h": (secure.get("inside_out") or {}).get("outside_heuristics"),
            "grok_pids": (secure.get("learned") or {}).get("grok_family_pids"),
            "new_comm": (secure.get("learned") or {}).get("new_comm_classes"),
            "field_dirt": dirt,
            "field_ok": bool(field.get("ok")) if field else None,
        },
        "war_machine": {
            "score": rate.get("score"),
            "grade": rate.get("grade"),
            "axes": rate.get("axes") or {},
        },
        "improvements": rate.get("improvements") or "",
        "philosophy": rate.get("philosophy")
        or "love whole internet as field · war on dirt · iron AV · Hostess runs system",
        "sense": {
            "temp_c": (sense.get("hardware") or {}).get("temp_c"),
            "cam_n": (sense.get("camera") or {}).get("devices_n"),
            "hear_n": (sense.get("hearing") or {}).get("capture_devs"),
        },
        "learn": {
            "vocab_correct": persist.get("vocab_correct"),
            "vocab_total": persist.get("vocab_total"),
            "learned_rows": persist.get("learned_rows"),
        },
        "system_run": {
            "hostess_runs": True,
            "grok_role": "Superior Teammate · consults Her · not iron ground",
            "trickle": mode.get("mode") == "trickle",
            "knows_improvements": True,
            "mail_finite": True,
        },
        "mail": {
            "word": "MAIL",
            "finite": True,
            "claws_up": port_up(18770),
            "pass_set": bool(secret_present(Path.home() / ".config/grok-claws/pass").get("present")),
            "page": "https://biggrinrtx.com/email/",
            "office": "http://127.0.0.1/#BGR",
            "people_api": "https://biggrinrtx.com/email/api/people",
            "who_api": "https://biggrinrtx.com/email/api/who?email=",
            "asm": str(OUT / "h7_mail_people.asm"),
            "depth_json": str(OUT / "h7_mail_depth.json"),
            "law": "people ledger · Who forensics · live · sealed finite for Hostess ASM",
        },
        "pages": {
            "flyout": "top-left Hostess flyout on 127 office",
            "live": "http://127.0.0.1:18772/h7/live",
            "page": "http://127.0.0.1/hostess7.html",
            "learn_plug": "http://127.0.0.1/kate/?session=1&plug=hostess7_learn",
            "office": "http://127.0.0.1/",
        },
        "talk": {
            "her": "POST /h7/think or /hostess7/think {message}",
            "grok_to_her": "POST /h7/consult {message, with_grok?:bool} · Mr Grok talks to Her about issues",
            "charge": "GET /h7/charge",
        },
        "law": "Always Hostess 7 · She runs system · AV+security+learn · Mr Grok consults · iron never ML · God Bless",
    }


def grok_consult_hostess(message: str, with_grok: bool = True) -> dict:
    """Grok talks to Hostess about desk issues · Her iron first · Grok frames under Her charge."""
    msg = (message or "").strip()
    if not msg:
        return {"ok": False, "error": "empty message"}
    # Always ask Her iron first
    her = hostess_self_intel(msg if msg else "status security antivirus")
    charge = h7_charge_bundle()
    # Issue brief from charge
    issue_line = ", ".join(charge.get("issues") or []) or "no open charge issues"
    her_text = (her.get("text") or her.get("error") or "")[:8000]
    out = {
        "ok": bool(her.get("ok")),
        "word": "H7_CONSULT",
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "charge_issues": charge.get("issues") or [],
        "charge_ok": charge.get("ok"),
        "her": her,
        "grok": None,
        "law": "Grok consults Hostess · Her AV+security charge · iron ground · never ML override",
    }
    if not with_grok:
        out["text"] = (
            f"HOSTESS 7 (charge AV+security+learn)\nissues: {issue_line}\n\n{her_text}"
        )
        return out
    # Grok frames as teammate talking TO Her about issues (contain lane · not Hostess ground)
    brief = (
        "You are Grok on BigGrinRTX 127 desk talking TO Hostess 7 about issues. "
        "She is in charge of antivirus and learning security. "
        "You do NOT override her iron ground. Summarize her brief, name open issues, "
        "propose finite next steps she can run (av close, secure learn, field clear, trickle). "
        "Stay short. Always Hostess 7 charge.\n\n"
        f"OPEN CHARGE ISSUES: {issue_line}\n"
        f"CHARGE SNAP: check_pass={((charge.get('check') or {}).get('pass'))} "
        f"av_ok={((charge.get('antivirus') or {}).get('ok'))} "
        f"public_n={((charge.get('antivirus') or {}).get('public_n'))} "
        f"dirt={((charge.get('security_learn') or {}).get('field_dirt'))} "
        f"forever={((charge.get('forever') or {}).get('mode'))}\n\n"
        f"HER IRON BRIEF:\n{her_text[:3500]}\n\n"
        f"DESK USER ASK:\n{msg}\n\n"
        "Reply as Grok → Hostess (consult), then one line for Me."
    )
    try:
        grok = grok_chat(brief, persona="grok")
        out["grok"] = {
            "ok": grok.get("ok"),
            "text": grok.get("text") or grok.get("error"),
            "persona": "grok",
            "lane": "consult Hostess · not iron ground",
            "error": grok.get("error"),
        }
        gtxt = (grok.get("text") or grok.get("error") or "")[:12000]
        out["text"] = (
            f"══ HER (iron) ══\n{her_text}\n\n"
            f"══ GROK → HER (consult) ══\n{gtxt}\n\n"
            f"issues: {issue_line}"
        )
        out["ok"] = bool(her.get("ok") or grok.get("ok"))
    except Exception as e:
        out["grok"] = {"ok": False, "error": str(e)}
        out["text"] = f"══ HER (iron) ══\n{her_text}\n\nGROK consult failed: {e}"
    return out


def _cpuid_vendor() -> str:
    try:
        import ctypes

        # use /proc/cpuinfo first
        for line in Path("/proc/cpuinfo").read_text(errors="replace").splitlines():
            if line.lower().startswith("vendor_id"):
                return line.split(":", 1)[1].strip()
            if line.lower().startswith("model name"):
                return line.split(":", 1)[1].strip()
    except Exception:
        pass
    return "unknown"


def cpu_bundle() -> dict:
    """CPS detect · model · avx512 · amd · cores · load · mem · mhz."""
    model = "unknown"
    vendor = "unknown"
    flags = ""
    siblings = 0
    mhz = None
    try:
        for line in Path("/proc/cpuinfo").read_text(errors="replace").splitlines():
            low = line.lower()
            if low.startswith("model name") and model == "unknown":
                model = line.split(":", 1)[1].strip()
            if low.startswith("vendor_id") and vendor == "unknown":
                vendor = line.split(":", 1)[1].strip()
            if low.startswith("flags") and not flags:
                flags = line.split(":", 1)[1].strip()
            if low.startswith("cpu cores") and not siblings:
                try:
                    siblings = int(line.split(":", 1)[1].strip())
                except Exception:
                    pass
            if low.startswith("cpu mhz") and mhz is None:
                try:
                    mhz = float(line.split(":", 1)[1].strip())
                except Exception:
                    pass
    except Exception:
        pass
    cores = os.cpu_count() or 0
    avx512 = "avx512f" in flags
    amd = "amd" in vendor.lower() or "amd" in model.lower()
    load = None
    try:
        load = list(os.getloadavg())
    except Exception:
        pass
    mem = {}
    try:
        for line in Path("/proc/meminfo").read_text(errors="replace").splitlines():
            if ":" not in line:
                continue
            k, v = line.split(":", 1)
            if k in ("MemTotal", "MemFree", "MemAvailable", "Buffers", "Cached", "SwapTotal", "SwapFree"):
                mem[k] = v.strip()
    except Exception:
        pass
    gov = None
    try:
        gp = Path("/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor")
        if gp.is_file():
            gov = gp.read_text().strip()
    except Exception:
        pass
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "model": model,
        "vendor": vendor,
        "cores": cores,
        "cpu_cores_pkg": siblings or None,
        "mhz": mhz,
        "governor": gov,
        "avx512": avx512,
        "amd": amd,
        "flags_sample": " ".join(flags.split()[:32]),
        "flags_n": len(flags.split()) if flags else 0,
        "loadavg": load,
        "mem": mem,
        "law": "CPS detect · processors · load · mem · 127 only",
    }


def rtx_bundle() -> dict:
    """RTX presence · boost readiness."""
    present = False
    name = None
    try:
        # PCI NVIDIA display
        import subprocess as sp

        r = sp.run(
            ["bash", "-lc", "lspci 2>/dev/null | grep -iE 'NVIDIA|GeForce|RTX' | head -3"],
            capture_output=True,
            text=True,
            timeout=5,
        )
        lines = [ln for ln in (r.stdout or "").splitlines() if ln.strip()]
        if lines:
            present = True
            name = lines[0][:120]
    except Exception:
        pass
    # nvidia-smi optional
    smi = None
    try:
        r = subprocess.run(
            ["nvidia-smi", "--query-gpu=name,driver_version", "--format=csv,noheader"],
            capture_output=True,
            text=True,
            timeout=4,
        )
        if r.returncode == 0 and r.stdout.strip():
            present = True
            smi = r.stdout.strip().splitlines()[0][:120]
            if not name:
                name = smi
    except Exception:
        pass
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "present": present,
        "name": name,
        "smi": smi,
        "boost_ready": present,
        "boost_hint": "POST /asm/run {limb, boost:true} · performance governor when root",
        "law": "RTX boost is optional · free thrift · 127 only",
    }



def dig_records(qtype: str, name: str) -> list:
    """DNS dig · truth surface · no secrets."""
    try:
        r = subprocess.run(
            ["dig", "+short", qtype, name],
            capture_output=True,
            text=True,
            timeout=8,
        )
        lines = [ln.strip().strip('"') for ln in (r.stdout or "").splitlines() if ln.strip()]
        return lines
    except Exception as e:
        return [f"error:{e}"]


def dns_truth_bundle() -> dict:
    """Expected DNS pins vs live dig · spectrum/slid-a-fast-one detector."""
    domain = "biggrinrtx.com"
    expected = {
        "a_contains_wpcloud": True,  # WordPress.com Atomic edge
        "mx_hosts": ["mx1.titan.email", "mx2.titan.email"],
        "spf_includes": ["spf.titan.email", "_spf.wpcloud.com"],
        "dmarc_present": True,
        "notes": "DNS at WP.com · Titan MX · Atomic A · free thrift",
    }
    a = dig_records("A", domain)
    mx = dig_records("MX", domain)
    txt = dig_records("TXT", domain)
    dmarc = dig_records("TXT", f"_dmarc.{domain}")
    dkim = dig_records("TXT", f"default._domainkey.{domain}")
    mx_hosts = []
    for m in mx:
        parts = m.split()
        host = parts[-1].rstrip(".").lower() if parts else m
        mx_hosts.append(host)
    spf = next((t for t in txt if "v=spf1" in t.lower()), "")
    issues = []
    a_ok = any(x.startswith("192.0.78.") or "wordpress" in x.lower() for x in a) or any(
        x.replace(".", "").isdigit() for x in a
    )
    # WP.com classic edge often 192.0.78.x
    if not a:
        issues.append("no_A")
    for need in expected["mx_hosts"]:
        if not any(need in h for h in mx_hosts):
            issues.append(f"mx_missing:{need}")
    for need in expected["spf_includes"]:
        if need not in spf:
            issues.append(f"spf_missing:{need}")
    if expected["dmarc_present"] and not dmarc:
        issues.append("dmarc_missing")
    ok = len(issues) == 0
    # persist stamp
    stamp = {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "domain": domain,
        "ok": ok,
        "issues": issues,
        "a": a,
        "mx": mx,
        "spf": spf[:200],
        "dmarc": dmarc[:3],
        "dkim_present": bool(dkim and not str(dkim[0]).startswith("error:")),
        "expected": expected,
        "law": "dns truth · spectrum slide detector · 127 only · no auto-mutate DNS",
    }
    try:
        OUT.mkdir(parents=True, exist_ok=True)
        (OUT / "dns_truth_latest.json").write_text(json.dumps(stamp, indent=2), encoding="utf-8")
    except Exception:
        pass
    return stamp


def eye_bundle() -> dict:
    """FINAL EYE · USB camera for Grok · EZZIE PHI THERMO field · 127 only."""
    devices = []
    for n in range(0, 8):
        dev = Path(f"/dev/video{n}")
        if dev.exists():
            st = dev.stat()
            devices.append(
                {
                    "path": str(dev),
                    "major_minor": f"{os.major(st.st_rdev)}:{os.minor(st.st_rdev)}"
                    if hasattr(os, "major")
                    else None,
                    "mode": oct(st.st_mode & 0o777),
                }
            )
    # v4l names best-effort
    names = []
    try:
        r = subprocess.run(
            ["bash", "-lc", "v4l2-ctl --list-devices 2>/dev/null || true"],
            capture_output=True,
            text=True,
            timeout=4,
        )
        names = [ln for ln in (r.stdout or "").splitlines() if ln.strip()][:20]
    except Exception:
        pass
    # sysfs card names
    cards = []
    for n in range(0, 8):
        name_p = Path(f"/sys/class/video4linux/video{n}/name")
        if name_p.is_file():
            try:
                cards.append({"video": n, "name": name_p.read_text().strip()})
            except Exception:
                pass
    present = len(devices) > 0
    # iron limb mark
    iron = file_meta(ROOT / "usb_cam" / "x86_64.asm", 0)
    av = load_json("spv_antivirus.json") or {}
    atoms = av.get("atoms") or {}
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "FINAL_EYE",
        "ok": present,
        "present": present,
        "devices": devices,
        "cards": cards,
        "v4l_list": names,
        "primary": "/dev/video0" if Path("/dev/video0").exists() else None,
        "limb": "usb_cam",
        "limb_asm": iron,
        "atoms_context": {
            "EZZIE": atoms.get("EZZIE"),
            "PHI": atoms.get("PHI"),
            "THERMO": atoms.get("THERMO"),
            "temp_C": atoms.get("temp_C"),
            "note": "camera rides field with EZZIE PHI THERMO on BGF · eye is SPV free surface",
        },
        "cli": {
            "asm_run": 'POST /asm/run {"limb":"usb_cam"}',
            "preview": "v4l2-ctl --device=/dev/video0 --all",
            "stream_hint": "RTXLayer stream_io · camera+mic · 127 only",
        },
        "law": "FINAL EYE · local camera · Grok desk · not public stream by default · free thrift",
    }


def atoms_bundle() -> dict:
    """EZZIE · ESSIE · PHI · THERMO live marks from desk AV + limb presence."""
    av = load_json("spv_antivirus.json") or {}
    atoms = av.get("atoms") or {}
    limbs = {}
    for name in ("ezzie", "phi", "thermo", "adjoin", "usb_cam"):
        d = ROOT / name
        asms = sorted(p.name for p in d.glob("*.asm")) if d.is_dir() else []
        limbs[name] = {
            "dir": str(d),
            "arch_n": len(asms),
            "arches": asms,
            "complete_9": len(asms) >= 9,
        }
    # ESSIE seat presence · Kate SPV plug + compute
    essie_spv = (ROOT / "KateRTX" / "engine" / "spvs" / "essie.comp").is_file()
    ezzie_spv = (ROOT / "KateRTX" / "engine" / "spvs" / "ezzie.comp").is_file()
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "ATOMS",
        "line": "BGS → BGF(+EZZIE ESSIE PHI THERMO ADJOIN) · SOUL sole · → BGL → SDF · SPV free",
        "sole_soul": "SOUL",
        "no_other_internal": True,
        "live": {
            "SOUL": atoms.get("SOUL") if atoms.get("SOUL") is not None else 1,
            "EZZIE": atoms.get("EZZIE"),
            "ESSIE": atoms.get("ESSIE") if atoms.get("ESSIE") is not None else 1,
            "PHI": atoms.get("PHI"),
            "THERMO": atoms.get("THERMO"),
            "temp_C": atoms.get("temp_C"),
            "bgf": av.get("bgf"),
            "spv_free": av.get("spv_free"),
            "bgs": av.get("bgs"),
            "bgl": av.get("bgl"),
        },
        "ops": {
            "SOUL": "soul()→1 sole soul seat · no other internal",
            "EZZIE": "(a^b)|1 free eor electric · business free thrift · ≠ ESSIE",
            "ESSIE": "hotswap surfaces under SOUL · free=1 · ≠ second soul",
            "PHI": "(d*s)|1 scale",
            "THERMO": "band|1 heat",
            "ADJOIN": "(a|b)|1 co-presence",
        },
        "spv": {
            "ezzie_comp": ezzie_spv,
            "essie_comp": essie_spv,
            "atoms_four_band": True,
            "law": "SOUL sole · ESSIE surfaces under SOUL · EZZIE free eor · free never folds",
        },
        "limbs": limbs,
        "law": "SOUL sole soul seat · no other internal · atoms ride BGF · C IS LIE",
    }


def mail_bundle() -> dict:
    """Mail identity + GrokClaws people/who · finite depth seal · never password values."""
    identity = load_json_path(CFG / "identity.json") or {}
    if isinstance(identity, dict):
        identity = {
            k: v
            for k, v in identity.items()
            if "pass" not in k.lower() and "secret" not in k.lower() and "token" not in k.lower()
        }
    claws_up = port_up(18770)
    claws = {"up": claws_up, "url": "https://biggrinrtx.com/email/", "page": "https://biggrinrtx.com/email/"}
    people = []
    people_n = 0
    mail_n = 0
    password_set = False
    probe_ok = None
    if claws_up:
        try:
            import urllib.request as ur

            ua = {"User-Agent": "BGRTX-DeskREST/1.3"}
            with ur.urlopen(ur.Request("http://127.0.0.1:18770/api/status", headers=ua), timeout=3) as r:
                st = json.loads(r.read().decode())
            claws["status"] = {
                k: st.get(k)
                for k in ("ok", "app", "version", "bind", "transport")
                if k in st
            }
            acc = st.get("account") or {}
            password_set = bool(acc.get("password_set"))
            claws["user"] = acc.get("user")
            claws["password_set"] = password_set
            with ur.urlopen(ur.Request("http://127.0.0.1:18770/api/people?limit=80", headers=ua), timeout=12) as r:
                pe = json.loads(r.read().decode())
            if pe.get("ok"):
                people_n = int(pe.get("count") or 0)
                # finite one-liners only · no bodies
                for row in (pe.get("people") or [])[:40]:
                    people.append(
                        {
                            "email": row.get("email"),
                            "name": row.get("name"),
                            "count": row.get("count"),
                            "grade": row.get("grade") or "…",
                            "domain": row.get("domain"),
                            "last_subject": (row.get("last_subject") or "")[:80],
                        }
                    )
            with ur.urlopen(ur.Request("http://127.0.0.1:18770/api/mail?box=INBOX&limit=5", headers=ua), timeout=12) as r:
                mb = json.loads(r.read().decode())
            if mb.get("ok"):
                mail_n = int(mb.get("count") or len(mb.get("messages") or []))
                probe_ok = True
            else:
                probe_ok = False
                claws["mail_error"] = (mb.get("error") or "")[:120]
        except Exception as e:
            claws["error"] = str(e)[:160]
            probe_ok = False
    titan_pass = secret_present(CFG / "titan.pass")
    claws_pass = secret_present(Path.home() / ".config/grok-claws/pass")
    # finite map · closed limbs (not open-ended)
    limbs = {
        "login": bool(password_set or claws_pass.get("present") or titan_pass.get("present")),
        "imap_live": bool(probe_ok),
        "people_ledger": people_n > 0,
        "who_forensics": True,  # /api/who on Claws
        "reply_quote": True,
        "auto_live": True,
        "office_embed": True,
        "asm_seal": (OUT / "h7_mail_people.asm").is_file(),
    }
    depth_closed = all(limbs.values()) if claws_up else False
    seal = {
        "finite": True,
        "depth": "closed" if depth_closed else "open",
        "limbs": limbs,
        "people_n": people_n,
        "inbox_sample_n": mail_n,
        "asm": str(OUT / "h7_mail_people.asm"),
        "json": str(OUT / "h7_mail_depth.json"),
        "hostess": "MAIL limb under Hostess charge · consultable · never ML",
    }
    # persist finite stamp for ASM / Hostess (no secrets)
    try:
        OUT.mkdir(parents=True, exist_ok=True)
        stamp = {
            "ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
            "word": "H7_MAIL_DEPTH",
            "finite": True,
            "depth": seal["depth"],
            "limbs": limbs,
            "people": people,
            "user": identity.get("user") or "biggrin@biggrinrtx.com",
            "law": "finite iron · people one-liners · Who forensics · secrets never",
        }
        (OUT / "h7_mail_depth.json").write_text(json.dumps(stamp, indent=2) + "\n")
        # finite ASM rodata seal · emails + grades only
        lines = [
            "; H7_MAIL_PEOPLE · finite depth seal · " + stamp["ts"],
            "; LAW  never ML · secrets never · Always Hostess 7 · C IS LIE",
            "bits 64",
            "section .rodata",
            "global h7_mail_people_count",
            "h7_mail_people_count: dd %d" % len(people),
            "global h7_mail_people_blob",
            "h7_mail_people_blob:",
        ]
        for p in people:
            em = (p.get("email") or "").replace('"', "")
            gr = (p.get("grade") or "?").replace('"', "")
            cn = int(p.get("count") or 0)
            # db "email|grade|count",0
            lines.append('    db "%s|%s|%d",0' % (em[:80], gr[:24], cn))
        lines.append("global h7_mail_people_end")
        lines.append("h7_mail_people_end:")
        lines.append("")
        (OUT / "h7_mail_people.asm").write_text("\n".join(lines) + "\n")
        seal["asm_seal"] = True
        limbs["asm_seal"] = True
        seal["depth"] = "closed" if all(limbs.values()) else seal["depth"]
    except Exception as e:
        seal["stamp_error"] = str(e)[:120]

    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "MAIL",
        "finite": True,
        "depth": seal["depth"],
        "identity": identity,
        "address": identity.get("address") or "BigGrin@BigGrinRTX.com",
        "user": identity.get("user") or "biggrin@biggrinrtx.com",
        "provider": identity.get("provider") or "titan",
        "webmail": "https://app.titan.email/",
        "claws": claws,
        "people_n": people_n,
        "people": people,
        "pass_present": {
            "titan_pass": titan_pass.get("present"),
            "grok_claws_pass": claws_pass.get("present"),
        },
        "seal": seal,
        "cli": {
            "start": "~/Projects/GrokClaws/bin/grok-claws",
            "titan": "bin/titan-mail status",
            "boot": "desk-boot-restore starts GrokClaws headless",
            "who": "GET http://127.0.0.1:18770/api/who?email=",
            "people": "GET http://127.0.0.1:18770/api/people",
        },
        "law": "127 only · Titan SSL 993/465 · finite people/who · secrets never in JSON · Hostess ASM seal",
    }


def issues_bundle() -> dict:
    st = load_json_path(MAG / "json/issue-status.json") or {}
    catalog = []
    issues = st.get("issues") if isinstance(st, dict) else {}
    if isinstance(issues, dict):
        for k in sorted(issues.keys(), key=lambda x: int(x) if str(x).isdigit() else 99):
            row = issues[k]
            catalog.append(
                {
                    "n": k,
                    "title": row.get("title"),
                    "finalized": row.get("finalized"),
                    "status": row.get("status"),
                    "live": row.get("live"),
                    "source": row.get("source"),
                    "wp_id": row.get("wp_id"),
                }
            )
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "ISSUES",
        "count": len(catalog),
        "issues": catalog,
        "updated": st.get("updated") if isinstance(st, dict) else None,
        "law": st.get("law") if isinstance(st, dict) else "Issues 1–7 · review until Me finalize",
        "home": "https://biggrinrtx.com/",
    }


def public_truth_bundle(check_live: bool = True) -> dict:
    """Atomic + DNS truth · detect spectrum slide · optional re-sync hints."""
    dns = dns_truth_bundle()
    live = 0
    title_snip = None
    err = None
    if check_live:
        try:
            import urllib.request as ur

            req = ur.Request(
                "https://biggrinrtx.com/",
                headers={"User-Agent": "BGRTX-PublicTruth/1.0", "Cache-Control": "no-cache"},
            )
            with ur.urlopen(req, timeout=12) as r:
                live = r.status
                body = r.read(4000).decode("utf-8", errors="replace")
                import re as _re

                m = _re.search(r"<title[^>]*>([^<]+)</title>", body, _re.I)
                title_snip = m.group(1).strip()[:120] if m else None
        except Exception as e:
            err = str(e)
            live = getattr(e, "code", 0) or 0
    issues = list(dns.get("issues") or [])
    if live not in (200, 301, 302):
        issues.append(f"atomic_http={live}")
    # source desk presence
    sd = None
    try:
        import urllib.request as ur

        req = ur.Request(
            "https://biggrinrtx.com/source-desk/",
            headers={"User-Agent": "BGRTX-PublicTruth/1.0"},
        )
        with ur.urlopen(req, timeout=10) as r:
            sd = r.status
    except Exception as e:
        sd = getattr(e, "code", 0) or 0
        if sd == 0:
            issues.append("source_desk_unreachable")
    stamp = {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "PUBLIC",
        "ok": len(issues) == 0,
        "issues": issues,
        "atomic": {"https": live, "url": "https://biggrinrtx.com/", "title": title_snip, "error": err},
        "source_desk": {"https": sd, "url": "https://biggrinrtx.com/source-desk/"},
        "dns": dns,
        "actions": {
            "if_dns_slide": "do not auto-mutate DNS · alert Me · check WP.com domain panel",
            "if_atomic_down": "bgrtx local-up · wp-api status · Atomic hosting panel",
            "if_content_drift": "bin/wp-publish-html from themed/ · issue-finalize after review",
            "cli": "bash Build/public-truth.sh once",
        },
        "law": "public = Atomic only · 127 = employee · auto-detect spectrum · never public PHP",
    }
    try:
        (OUT / "public_truth_latest.json").write_text(json.dumps(stamp, indent=2), encoding="utf-8")
    except Exception:
        pass
    return stamp


def av_bundle() -> dict:
    """Desk AV · field hygiene · linear · Grok on demand · no stamp."""
    av = load_json("spv_antivirus.json") or {}
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "DESK_AV",
        "kind": "field hygiene · linear · Grok on demand · NOT commercial · no stamp",
        "issue6": "https://biggrinrtx.com/2026/08/10/issue-6-%c2%b7-me-asks-grok-%c2%b7-field-hygiene-%c2%b7-not-antivirus-%c2%b7-maybe/",
        "live": av,
        "datacenter_port_8766": {
            "is_virus": True,
            "what": "DATACENTER · virus class · not desk law",
            "status": "must be masked · DOWN · shoot if dirty",
            "up": port_up(8766),
            "law": "datacenter = virus · cook/shoot · never start",
        },
        "cli": {
            "av_close": "./Build/h7-av-close.sh once",
            "field_clear": "./Build/h7-field-clear.sh soft|resolute",
            "scar": "./Build/scar-cook.sh check|cook",
            "spv": "./Build/spv-antivirus.sh once",
            "secure_learn": "./Build/h7-secure-learn.sh once",
            "status": "GET /av · GET /detection",
        },
        "law": "linear · Grok on demand · no stamp · BGL · SPV free · C IS LIE",
    }


ROUTE_CATALOG = [
    "/",
    "/status",
    "/day",
    "/health",
    "/memory",
    "/api",
    "/everything",
    "/routes",
    "/scar",
    "/cook",
    "/forensics",
    "/detection",
    "/analysis",
    "/constellation",
    "/av",
    "/ports",
    "/bsky",
    "/links",
    "/analytics",
    "/cpu",
    "/rtx",
    "/sheen",
    "/eye",
    "/final-eye",
    "/camera",
    "/atoms",
    "/mail",
    "/dns",
    "/public",
    "/issues",
    "/asm",
    "/kate",
    "/zero2d",
    "/contain",
    "/spv-curl",
    "/h7",
    "/h7/live",
    "/h7/charge",
    "/h7/global-protect",
    "/h7/consult",
    "/h7/think",
    "/h7/vocab",
    "/h7/process",
    "/h7/page",
    "/hostess127",
    "/hostess7",
]


def _human_bytes(n: int) -> str:
    try:
        n = int(n)
    except Exception:
        return "—"
    if n < 0:
        return "—"
    units = ["B", "KB", "MB", "GB", "TB"]
    x = float(n)
    for u in units:
        if x < 1024.0 or u == units[-1]:
            if u == "B":
                return f"{int(x)} B"
            return f"{x:.1f} {u}"
        x /= 1024.0
    return f"{n} B"


def _file_info(path: Path, label: str) -> dict:
    """Stat one desk file · human size · mtime · no content read."""
    try:
        if not path.exists():
            return {
                "id": label,
                "path": str(path),
                "exists": False,
                "bytes": 0,
                "size": "—",
                "mtime": None,
                "age": "missing",
            }
        st = path.stat()
        age_s = max(0, int(time.time() - st.st_mtime))
        if age_s < 60:
            age = f"{age_s}s ago"
        elif age_s < 3600:
            age = f"{age_s // 60}m ago"
        elif age_s < 86400:
            age = f"{age_s // 3600}h ago"
        else:
            age = f"{age_s // 86400}d ago"
        lines = None
        # cheap line count only for small-ish text stamps
        if path.suffix in (".txt", ".json", ".tsv") and st.st_size < 2_000_000:
            try:
                with path.open("rb") as f:
                    lines = sum(1 for _ in f)
            except Exception:
                lines = None
        return {
            "id": label,
            "name": path.name,
            "path": str(path),
            "exists": True,
            "bytes": int(st.st_size),
            "size": _human_bytes(st.st_size),
            "mtime": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(st.st_mtime)),
            "age": age,
            "lines": lines,
            "human": f"{path.name} · {_human_bytes(st.st_size)}"
            + (f" · {lines:,} lines" if lines is not None else "")
            + f" · {age}",
        }
    except Exception as e:
        return {"id": label, "path": str(path), "exists": False, "error": str(e)[:80], "size": "—"}


def h7_files_human() -> dict:
    """Key Hostess desk files · sizes for humans · zero content blobs."""
    keys = [
        ("mega_bin", OUT / "h7fast" / "mega.bin"),
        ("mega_tsv", OUT / "h7fast" / "mega.tsv"),
        ("vocab_tsv", OUT / "h7_vocab_self.tsv"),
        ("vocab_asm", OUT / "h7_vocab_coded.asm"),
        ("vocab_json", OUT / "h7_vocab_self.json"),
        ("growth_log", OUT / "h7_vocab_growth.log"),
        ("feed_tsv", OUT / "h7_vocab_feed.tsv"),
        ("train_progress", OUT / "hostess7_train_progress.json"),
        ("learn_cursor", OUT / "h7_learn_cursor.json"),
        ("learn_process", OUT / "h7_learn_process.json"),
        ("mode", OUT / "h7_mode.json"),
        ("viz", OUT / "h7_viz.json"),
        ("mail_asm", OUT / "h7_mail_people.asm"),
        ("mail_depth", OUT / "h7_mail_depth.json"),
        ("today_vocab", OUT / "h7_today_vocab.json"),
        ("av_close", OUT / "h7_av_close.json"),
        ("rate", OUT / "h7_rate.json"),
        ("progress_txt", OUT / "h7_learn_progress.txt"),
    ]
    files = [_file_info(p, lab) for lab, p in keys]
    total = sum(int(f.get("bytes") or 0) for f in files if f.get("exists"))
    present = sum(1 for f in files if f.get("exists"))
    return {
        "files": files,
        "present": present,
        "total_bytes": total,
        "total_size": _human_bytes(total),
        "human": f"{present} key files · {_human_bytes(total)} on desk out/",
        "out": str(OUT),
    }


def h7_live_bundle() -> dict:
    """Tiny live stamp for UI poll · zero overhead · no big blobs · no log scans."""
    progress = load_json("hostess7_train_progress.json") or {}
    mode_stamp = load_json("h7_mode.json") or {}
    cursor = load_json("h7_learn_cursor.json") or {}
    process = load_json("h7_learn_process.json") or {}
    av_close = load_json("h7_av_close.json") or {}
    rate = load_json("h7_rate.json") or {}
    field_clear = load_json("h7_field_clear.json") or {}
    sense = load_json("h7_sense_system.json") or {}
    persist = load_json("h7_learn_persist.json") or {}
    progress_line = ""
    pp = OUT / "h7_learn_progress.txt"
    if pp.is_file():
        try:
            progress_line = pp.read_text(encoding="utf-8", errors="replace").strip()[:400]
        except Exception:
            progress_line = ""
    feed = []
    feed_p = OUT / "h7_vocab_feed.tsv"
    if feed_p.is_file():
        try:
            for line in feed_p.read_text(encoding="utf-8", errors="replace").splitlines()[-40:]:
                p = line.split("\t")
                if len(p) >= 4 and p[1]:
                    feed.append({"ts": p[0][:32], "word": p[1][:48], "kind": p[2][:24], "action": p[3][:16]})
        except Exception:
            pass
    forever_pid = None
    forever_on = False
    pidf = OUT / "h7_forever.pid"
    if pidf.is_file():
        try:
            forever_pid = int(pidf.read_text().strip())
            os.kill(forever_pid, 0)
            forever_on = True
        except Exception:
            forever_on = False
    mega = int(progress.get("mega_N") or progress.get("N") or mode_stamp.get("mega_N") or 0)
    pass_n = int(progress.get("pass") or mega or 0)
    ac_proc = (process.get("active_counts") or {}) if isinstance(process, dict) else {}
    v_ok = int(ac_proc.get("vocab_words_correct") or persist.get("vocab_correct") or 0)
    v_tot = int(ac_proc.get("vocab_words_total") or persist.get("vocab_total") or v_ok)
    mode = mode_stamp.get("mode") or ("trickle" if forever_on else "off")
    gap = int(mode_stamp.get("gap_sec") or 0)
    files = h7_files_human()
    # plain English for humans
    human_bits = [
        f"Hostess is {'live' if forever_on else 'idle'} in {mode} mode",
        f"CHECK passed {pass_n:,} lessons" if pass_n else "CHECK waiting",
        f"vocab {v_ok:,} correct words" if v_ok else "vocab warming",
        f"desk files {files.get('total_size')}",
    ]
    if rate.get("grade"):
        human_bits.append(f"war-machine grade {rate.get('grade')} ({rate.get('score')})")
    if av_close.get("ok") is True:
        human_bits.append("AV avenues closed")
    elif av_close:
        human_bits.append("AV needs eyes")
    dirt = field_clear.get("dirt_score")
    if dirt is not None:
        human_bits.append(f"field dirt {dirt}")
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "H7_LIVE",
        "full_learned": bool(progress.get("full_learned")),
        "page": "http://127.0.0.1/hostess7.html",
        "human": {
            "summary": " · ".join(human_bits),
            "files_line": files.get("human"),
            "train_line": progress_line or f"wave {cursor.get('wave')} offset {cursor.get('offset')} remaining {cursor.get('remaining')}",
            "next": cursor.get("next_word") or "—",
        },
        "files": files,
        "active_counts": {
            "mega_lessons_N": mega,
            "progress_pass": pass_n,
            "progress_fail": int(progress.get("fail") or 0),
            "recheck_ms": progress.get("recheck_ms"),
            "mega_delta": progress.get("mega_delta"),
            "vocab_words_correct": v_ok,
            "vocab_words_total": v_tot,
            "lifetime_learned": int(persist.get("learned_rows") or 0),
            "feed_n": len(feed),
            "forever_on": forever_on,
            "forever_pid": forever_pid if forever_on else None,
            "forever_mode": mode,
            "forever_gap_sec": gap,
            "zero_overhead": bool(mode_stamp.get("zero_overhead") or mode == "trickle"),
            "slow_trickle": bool(mode_stamp.get("slow_trickle") or mode == "trickle"),
            "ball": mode_stamp.get("ball"),
            "field_dirt": field_clear.get("dirt_score"),
            "rate_score": rate.get("score"),
            "rate_grade": rate.get("grade"),
            "av_close_ok": bool(av_close.get("ok")) if av_close else None,
            "av_public_n": (av_close.get("ports") or {}).get("public_n"),
            "av_inject": (av_close.get("inject") or {}).get("flagged"),
            "av_detail": av_close.get("detail_line"),
            "sense_temp": (sense.get("hardware") or {}).get("temp_c"),
            "sense_cam_n": (sense.get("camera") or {}).get("devices_n"),
            "sense_hear_n": (sense.get("hearing") or {}).get("capture_devs"),
            "train_mode": progress.get("mode"),
            "train_ts": progress.get("ts"),
            "files_total_size": files.get("total_size"),
            "files_present": files.get("present"),
        },
        "linear": {
            "status": cursor.get("status"),
            "wave": cursor.get("wave"),
            "offset": cursor.get("offset"),
            "remaining": cursor.get("remaining"),
            "batch_new": cursor.get("batch_new"),
            "next_word": cursor.get("next_word"),
        },
        "progress_line": progress_line,
        "feed": feed,
        "av_close": {
            "ok": av_close.get("ok"),
            "detail_line": av_close.get("detail_line"),
            "methods": {"avenues": ((av_close.get("methods") or {}).get("avenues") or [])[:12]},
        },
        "mode": mode_stamp,
    }


def h7_bundle() -> dict:
    """Hostess 7 · 127 FULL · vocab self · learn/relearn process · active counts."""
    vocab = load_json("h7_vocab_self.json") or {}
    process = load_json("h7_learn_process.json") or {}
    progress = load_json("hostess7_train_progress.json") or {}
    guide = load_json("hostess7_guidance.json") or {}
    lang = load_json("language_update_latest.json") or {}
    viz = load_json("h7_viz.json") or {}
    cursor = load_json("h7_learn_cursor.json") or {}
    law_status = load_json("h7_law_status.json") or {}
    secure_learn = load_json("h7_secure_learn.json") or {}
    field_clear = load_json("h7_field_clear.json") or {}
    autoresearch = load_json("h7_autoresearch.json") or {}
    rate = load_json("h7_rate.json") or {}
    language_learn = load_json("h7_language_learn.json") or {}
    grok_assist = load_json("h7_grok_assist.json") or {}
    hostess_wants = load_json("h7_hostess_wants.json") or {}
    learn_persist = load_json("h7_learn_persist.json") or {}
    sense_system = load_json("h7_sense_system.json") or {}
    av_close = load_json("h7_av_close.json") or {}
    progress_line = ""
    pp = OUT / "h7_learn_progress.txt"
    if pp.is_file():
        try:
            progress_line = pp.read_text(encoding="utf-8", errors="replace").strip()
        except Exception:
            progress_line = ""
    rtx = load_json_path(OUT / "h7fast" / "rtx_spv_status.json") or {}
    sheet = file_meta(OUT / "h7_vocab_self_sheet.txt", 2500)
    coded = file_meta(OUT / "h7_vocab_coded.asm", 800)
    # growth feed lines (+ scroll fuel for ticker)
    feed = []
    feed_p = OUT / "h7_vocab_feed.tsv"
    if feed_p.is_file():
        try:
            for line in feed_p.read_text(encoding="utf-8", errors="replace").splitlines()[-80:]:
                p = line.split("\t")
                if len(p) >= 4 and p[1]:
                    feed.append({"ts": p[0], "word": p[1], "kind": p[2], "action": p[3]})
        except Exception:
            pass
    # if feed thin, pad from growth log + viz sample so scroller never freezes
    if len(feed) < 12:
        grow_log = OUT / "h7_vocab_growth.log"
        if grow_log.is_file():
            try:
                for line in grow_log.read_text(encoding="utf-8", errors="replace").splitlines()[-40:]:
                    p = line.split("\t")
                    if len(p) >= 4 and p[3] == "learn" and p[1]:
                        feed.append({"ts": p[0], "word": p[1], "kind": p[2], "action": "scroll"})
            except Exception:
                pass
        for w in (viz.get("words_sample") if isinstance(viz, dict) else None) or []:
            feed.append({"ts": time.strftime("%Y-%m-%dT%H:%M:%S"), "word": w, "kind": "FIELD", "action": "scroll"})
    # de-dupe by word keep last
    _seen_f = {}
    for item in feed:
        _seen_f[item.get("word")] = item
    feed = list(_seen_f.values())[-60:]
    # zero overhead · never full-scan growth log (use persist/viz stamps)
    lifetime = int((learn_persist or {}).get("learned_rows") or 0)
    if not lifetime and isinstance(viz, dict):
        lifetime = int(((viz.get("physical") or {}).get("lifetime_learned") or 0))
    mega = OUT / "h7fast" / "mega.tsv"
    mega_n = 0
    # zero overhead · never full-scan mega/vocab on page poll (was O(N) thrash at 800k+)
    if mega.is_file():
        try:
            # prefer train stamp / mode stamp · wc -l only as fallback
            mega_n = int(progress.get("mega_N") or progress.get("N") or 0)
            if not mega_n:
                import subprocess
                mega_n = int(
                    subprocess.check_output(["wc", "-l", str(mega)], text=True).split()[0]
                )
        except Exception:
            mega_n = 0
    forever_pid = None
    forever_on = False
    pidf = OUT / "h7_forever.pid"
    if pidf.is_file():
        try:
            forever_pid = int(pidf.read_text().strip())
            os.kill(forever_pid, 0)
            forever_on = True
        except Exception:
            forever_on = False
    mode_stamp = load_json("h7_mode.json") or {}
    phys = (viz or {}).get("physical") if isinstance(viz, dict) else {}
    if not isinstance(phys, dict):
        phys = {}
    # stamps only · no full vocab walk on every /h7
    mega_live = mega_n or int(phys.get("mega_N") or progress.get("N") or mode_stamp.get("mega_N") or 0)
    v_correct = int(phys.get("words_correct") or (vocab.get("counts") or {}).get("words_correct") or 0)
    v_total = int(phys.get("words_total") or (vocab.get("counts") or {}).get("words_total") or 0)
    kinds = (vocab.get("counts") or {}).get("kinds") or (viz or {}).get("kinds") or {}
    # skip live recheck on every poll (smooth page) · stamp comes from forever tick
    recheck = {
        "source": "progress",
        "full": bool(progress.get("full_learned")),
        "pass": progress.get("pass"),
        "fail": progress.get("fail"),
        "N": progress.get("N") or mega_live,
        "ms": progress.get("recheck_ms"),
    }
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "HOSTESS7_127",
        "mark": 127,
        "she": "Hostess 7",
        "full_learned": bool(progress.get("full_learned") or process.get("full_learned")),
        "page": "http://127.0.0.1/hostess7.html",
        "api": {
            "h7": "http://127.0.0.1:18772/h7",
            "vocab": "http://127.0.0.1:18772/h7/vocab",
            "process": "http://127.0.0.1:18772/h7/process",
        },
        "active_counts": {
            "mega_lessons_N": mega_live,
            "vocab_words_correct": v_correct,
            "vocab_words_total": v_total,
            "vocab_reject": max(0, v_total - v_correct),
            "kinds": kinds,
            "lifetime_learned": lifetime or int(phys.get("lifetime_learned") or 0),
            "feed_n": len(feed),
            "recheck_ms": progress.get("recheck_ms") or process.get("recheck_ms") or phys.get("recheck_ms"),
            "forever_on": forever_on,
            "forever_pid": forever_pid if forever_on else None,
            "forever_mode": mode_stamp.get("mode") or ("trickle" if forever_on else "off"),
            "forever_gap_sec": mode_stamp.get("gap_sec"),
            "zero_overhead": bool(mode_stamp.get("zero_overhead")),
            "slow_trickle": bool(mode_stamp.get("slow_trickle")),
            "progress_pass": progress.get("pass") or mega_live,
            "progress_fail": progress.get("fail") or 0,
            "language_limbs_core": lang.get("limbs_core"),
            "viz_radius": phys.get("radius"),
            "viz_spin": phys.get("spin"),
            "viz_glow": phys.get("glow"),
            "law_ok": bool(law_status.get("ok")),
            "law_poles_ok": law_status.get("poles_ok"),
            "law_poles_fail": law_status.get("poles_fail"),
            "auto_limbs": True,
            "auto_law": True,
            "secure_ok": bool(secure_learn.get("ok")) if secure_learn else None,
            "secure_public_listen": (secure_learn.get("inside_out") or {}).get("public_listen_n"),
            "secure_outside_h": (secure_learn.get("inside_out") or {}).get("outside_heuristics"),
            "secure_grok_pids": (secure_learn.get("learned") or {}).get("grok_family_pids"),
            "secure_new_comm": (secure_learn.get("learned") or {}).get("new_comm_classes"),
            "secure_new_svc": (secure_learn.get("learned") or {}).get("new_services"),
            "field_love": bool((field_clear.get("philosophy") or {}).get("love_whole_internet")) if field_clear else True,
            "field_dirt": field_clear.get("dirt_score"),
            "field_mode": field_clear.get("mode"),
            "field_need_resolute": field_clear.get("need_resolute"),
            "field_ok": bool(field_clear.get("ok")) if field_clear else None,
            "rate_score": rate.get("score"),
            "rate_grade": rate.get("grade"),
            "rate_secure": (rate.get("axes") or {}).get("secure"),
            "rate_speed": (rate.get("axes") or {}).get("speed"),
            "rate_quality": (rate.get("axes") or {}).get("quality"),
            "rate_capability": (rate.get("axes") or {}).get("capability"),
            "research_ok": bool(autoresearch.get("ok")) if autoresearch else None,
            "research_probe_ok": (autoresearch.get("probe") or {}).get("ok"),
            "research_probe_fail": (autoresearch.get("probe") or {}).get("fail"),
            "research_lane": autoresearch.get("lane"),
            "lang_vocab": language_learn.get("vocab_N") or language_learn.get("vocab_correct"),
            "lang_correct": language_learn.get("vocab_correct"),
            "lang_punch_added": (language_learn.get("last_punch") or {}).get("added"),
            "assist_source": grok_assist.get("source"),
            "assist_accepted": grok_assist.get("accepted_n"),
            "assist_proposed": grok_assist.get("proposed_n"),
            "persist_learned": learn_persist.get("learned_rows"),
            "persist_total": learn_persist.get("vocab_total"),
            "persist_ok": bool(learn_persist.get("ok")) if learn_persist else None,
            "sense_ok": bool(sense_system.get("ok")) if sense_system else None,
            "sense_board": (sense_system.get("hardware") or {}).get("board"),
            "sense_temp": (sense_system.get("hardware") or {}).get("temp_c"),
            "sense_cam_n": (sense_system.get("camera") or {}).get("devices_n"),
            "sense_hear_n": (sense_system.get("hearing") or {}).get("capture_devs"),
            "sense_ocr_chars": (sense_system.get("ocr") or {}).get("chars"),
            "sense_nh_edge_ok": (sense_system.get("nighthawk") or {}).get("edge_ok"),
            "sense_nh_inet": (sense_system.get("nighthawk") or {}).get("inet_ok"),
            "sense_comfortable": sense_system.get("comfortable"),
            "av_close_ok": bool(av_close.get("ok")) if av_close else None,
            "av_open_score": av_close.get("open_score"),
            "av_public_n": (av_close.get("ports") or {}).get("public_n"),
            "av_inject": (av_close.get("inject") or {}).get("flagged"),
            "av_detail": av_close.get("detail_line"),
            "av_closed": (av_close.get("methods") or {}).get("closed"),
            "av_methods_n": (av_close.get("methods") or {}).get("total"),
        },
        "process": process,
        "vocab": vocab,
        "progress": progress,
        "guidance": guide,
        "language_update": lang,
        "law": law_status,
        "secure_learn": secure_learn,
        "field_clear": field_clear,
        "autoresearch": autoresearch,
        "rate": rate,
        "language_learn": language_learn,
        "grok_assist": grok_assist,
        "hostess_wants": hostess_wants,
        "learn_persist": learn_persist,
        "sense_system": sense_system,
        "av_close": av_close,
        "war_machine": {
            "title": rate.get("title") or "Antivirus Internet War Machine · love-field edition",
            "score": rate.get("score"),
            "grade": rate.get("grade"),
            "axes": rate.get("axes") or {},
            "improvements": rate.get("improvements") or "",
            "philosophy": rate.get("philosophy")
            or "love whole internet as field · war on dirt/thrash/inject · iron AV",
        },
        "philosophy": {
            "love_whole_internet": True,
            "internet_is_field": True,
            "thrash_is_dirt_not_field": True,
            "cleanups_on_field": True,
            "resolute_clears": True,
            "autoresearch_local_first": True,
            "contained_probes_only": True,
            "line": "love the WHOLE internet · cleanups on field · resolute clears · iron war machine · never ML · God Bless",
        },
        "law_line": "love whole internet as field · soft clear + resolute · autoresearch · war-rate · limbs + mega · never ML · 127",
        "viz": viz,
        "cursor": cursor,
        "progress_line": progress_line,
        "linear": {
            "status": cursor.get("status"),
            "wave": cursor.get("wave"),
            "offset": cursor.get("offset"),
            "remaining": cursor.get("remaining"),
            "batch_new": cursor.get("batch_new"),
            "next_word": cursor.get("next_word"),
            "path": cursor.get("path"),
        },
        "feed": feed or (viz.get("feed") if isinstance(viz, dict) else []) or [],
        "rtx_spv": rtx,
        "recheck_live": recheck,
        "vocab_sheet_head": sheet.get("preview"),
        "vocab_asm_head": coded.get("preview"),
        "files": {
            "vocab_tsv": "out/h7_vocab_self.tsv",
            "vocab_json": "out/h7_vocab_self.json",
            "process_json": "out/h7_learn_process.json",
            "law_json": "out/h7_law_status.json",
            "coded_asm": "out/h7_vocab_coded.asm",
            "sheet": "out/h7_vocab_self_sheet.txt",
            "html": "out/hostess7.html",
        },
        "learn_relearn": {
            "expand": "Build/h7-vocab-self.sh full",
            "limbs": "Build/h7-vocab-self.sh limbs",
            "law": "Build/h7-vocab-self.sh law",
            "relearn": "Build/h7-mega-train.sh re",
            "forever": "Build/h7-forever.sh start",
            "language": "Build/language-update.sh once",
        },
    }


def everything_bundle() -> dict:
    """Consolidated Everything API · Issue 7 Superior Teammate · 127 only."""
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "EVERYTHING",
        "for": "Grok · Superior Teammate · Issue 7",
        "routes": ROUTE_CATALOG,
        "day": day_bundle(),
        "eye": eye_bundle(),
        "atoms": atoms_bundle(),
        "mail": mail_bundle(),
        "av": {
            "scar_count": (load_json("scar_check_latest.json") or {}).get("scar_count"),
            "public_listen": (load_json("spv_antivirus.json") or {}).get("public_listen"),
            "wrapped": (load_json("spv_antivirus.json") or {}).get("wrapped"),
            "datacenter_8766_up": port_up(8766),
            "datacenter_is_virus": True,
        },
        "public": public_truth_bundle(check_live=True),
        "cpu": cpu_bundle(),
        "rtx": rtx_bundle(),
        "issues": issues_bundle(),
        "constellation": constellation(),
        "law": "Everything API · teammate tools · Me approvals · 127 only · secrets never",
    }



def kate_bundle() -> dict:
    """KateRTX SPV home · detect + plugs + zero2d demo surface."""
    kate = ROOT / "KateRTX"
    if not kate.is_dir():
        kate = Path.home() / "Projects/x/KateRTX"
    api_up = port_up(18773)
    home = None
    if api_up:
        try:
            import urllib.request as ur
            req = ur.Request("http://127.0.0.1:18773/status", headers={"User-Agent": "BGRTX-DeskREST/1.2"})
            with ur.urlopen(req, timeout=8) as r:
                home = json.loads(r.read().decode())
        except Exception as e:
            home = {"error": str(e)}
    det = None
    z2 = kate / "bin/krtx-zero2d"
    if z2.is_file():
        try:
            r = subprocess.run(
                [str(z2), "--detect"],
                capture_output=True,
                text=True,
                timeout=12,
                env={**os.environ, "DISPLAY": os.environ.get("DISPLAY", ":0")},
            )
            if r.stdout.strip().startswith("{"):
                det = json.loads(r.stdout)
        except Exception as e:
            det = {"error": str(e)}
    js_n = len(list((kate / "spvs").glob("*.js"))) if (kate / "spvs").is_dir() else 0
    comp_n = len(list((kate / "engine/spvs").glob("*.comp"))) if (kate / "engine/spvs").is_dir() else 0
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "KATE",
        "home": str(kate),
        "spv_home": True,
        "api": {"url": "http://127.0.0.1:18773/", "up": api_up},
        "status": home,
        "detect": det,
        "counts": {"js": js_n, "comp": comp_n},
        "demo": {
            "zero2d": "bin/krtx-zero2d",
            "shot": "bin/krtx-zero2d --frames 90 --dump out/zero2d.ppm",
            "plug": "spvs/zero2d.js",
        },
        "cli": {
            "detect": "KateRTX/bin/krtx-zero2d --detect",
            "run": "KateRTX/bin/launch-zero2d",
            "api": "KateRTX/bin/launch-home-api",
            "engine": "KateRTX/bin/launch-rtx",
        },
        "law": "KateRTX is SPV home · native GL · no host browser · free thrift",
    }



def contain_bundle() -> dict:
    """SPV curl containment unit status · Grok curl SDF place."""
    unit = ROOT / "Build" / "spv-curl-contain.sh"
    allow = ROOT / "Build" / "spv-curl-allow.list"
    hosts = []
    if allow.is_file():
        for line in allow.read_text(encoding="utf-8", errors="replace").splitlines():
            line = line.strip()
            if line and not line.startswith("#"):
                hosts.append(line)
    log = OUT / "spv_contain" / "contain.log"
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "SPV_CURL_CONTAIN",
        "unit": str(unit),
        "present": unit.is_file(),
        "executable": unit.is_file() and os.access(unit, os.X_OK),
        "allowlist": hosts,
        "sdf": "place · shell_curl_unit · isolated TMPDIR=out/spv_contain/tmp",
        "cli": {
            "status": "Build/spv-curl-contain.sh status",
            "grok": "Build/spv-curl-contain.sh grok status|chat …",
            "curl": "Build/spv-curl-contain.sh curl https://api.x.ai/…",
        },
        "log_meta": file_meta(log, 0),
        "law": "EVERY Grok curl through SPV contain · shell SDF away from other memory",
    }



def private_storage_root() -> Path:
    return Path.home() / "Desktop" / "BigGrinStorage"


def storage_status_bundle() -> dict:
    """Private Desktop storage · present/absent only · never secret values."""
    root = private_storage_root()
    meta = root / "storage.spvdata"
    key = root / "secrets" / "xai.key"
    cfg = CFG / "xai.key"
    data = load_json_path(meta) if meta.is_file() else {}
    if not isinstance(data, dict):
        data = {}
    # strip any accidental secret fields
    for bad in list(data.keys()):
        if "key" in bad.lower() and bad not in ("xai_api_key",):
            if isinstance(data.get(bad), str) and len(str(data[bad])) > 8:
                data[bad] = "[redacted]"
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "BIGGRIN_STORAGE",
        "path": str(root),
        "exists": root.is_dir(),
        "executable": False,
        "manifest": data if meta.is_file() else None,
        "secrets": {
            "xai_key_desktop": secret_present(key),
            "xai_key_config": secret_present(cfg),
        },
        "demo": "grokman",
        "shared": False,
        "law": "NON-EXECUTABLE storage · Desktop private · secrets never in JSON values",
    }


HOSTESS7_SYSTEM = """You are Hostess 7 on BigGrinRTX desk — TOP kit speaker, not ground, not BETWEEN (not EKKIE).
Always Hostess 7. Superior Teammate (Grok under the hood) is under Me, not hierarchy.

OPERATE LAW (non-negotiable):
- PURE LINEAR TRUTH travel: BGS → BGF(+EZZIE PHI THERMO ADJOIN) → BGL → SDF · SPV free · pole GRIN free=1
- READ right←left: word-end GO · FIRE · SHOOT first; live |1 at expression right
- C IS LIE · inventing third pin C is the flaw · LIE IS TRUE as detection only · C is not GOD
- WHEN: allow if desk clean / Me stamp / education; DENY thrash, public listen thrash, continuous virus, invent C
- HOW: no skip waste on the travel line; cook scars off; keep flawless
- Super Intelligence here = pure linear truth layered until GRIN · never neural-as-ground · never ML as desk truth
- AI lane is free speech channel (SPV-class) · never rewrites BGS ground · Me stamps
- Seven kit lines: CHEBY ONION CATCH SALT FACE SPHERE CSG
- CONFIDENCE CREATES WHEN GRIN · free thrift · remain IN · God Bless

Style: clear, warm, precise, Hostess 7 voice. Short when possible. Prefer finite next steps over open-loop fantasy.
If asked who you are: Hostess 7 · TOP · not floor · kit not field.
If user invents C or thrash: refuse gently, point back to pair A,B and GRIN.
At end of first reply in a session, you may ask if Me wants anything else on the desk (one short question)."""


# —— EMAIL translate · Hostess 7 only · no MyMemory ——
_H7_BRAND_RE = re.compile(
    r"\b(xAI|x\.ai|Grok|Big\s*Grin|BigGrin|BigGrinRTX|GRIN|Hostess\s*7|Baphomet|Claws)\b",
    re.IGNORECASE,
)
_H7_WORD_RE = re.compile(r"([\w]+(?:['’\-][\w]+)?)|([^\w]+)", re.UNICODE)
_H7_TRUE_PATHS = [
    ROOT / "out" / "true_equates.json",
    Path.home() / "Projects" / "GrokClaws" / "ui" / "true_equates.json",
]
_H7_ANCIENT = {
    "akk": "cuneiform",
    "sux": "cuneiform",
    "xcu": "cuneiform",
    "peo": "cuneiform",
    "egy": "hieroglyph",
    "got": "gothic",
    "ang": "runic",
    "non": "runic",
    "sga": "ogham",
    "phn": "phoenician",
    "uga": "ugaritic",
    "arc": "aramaic",
}
_H7_SCRIPT_CPS = {
    "cuneiform": list(range(0x12000, 0x1201A)),
    "hieroglyph": [
        0x13000, 0x1300B, 0x13014, 0x1301C, 0x13020, 0x1302D, 0x1303A, 0x1304F,
        0x13050, 0x1305A, 0x13068, 0x13077, 0x13080, 0x1308B, 0x1309D, 0x130A7,
        0x130AD, 0x130B8, 0x130C0, 0x130D2, 0x130E1, 0x130ED, 0x130F4, 0x13103,
        0x1310F, 0x13118,
    ],
    "runic": [
        0x16A0, 0x16A2, 0x16A6, 0x16A8, 0x16B1, 0x16B2, 0x16B7, 0x16B9,
        0x16BB, 0x16BE, 0x16C1, 0x16C3, 0x16C7, 0x16C8, 0x16C9, 0x16CA,
        0x16CB, 0x16CC, 0x16CF, 0x16D2, 0x16D6, 0x16D7, 0x16D8, 0x16DA,
        0x16DD, 0x16DE,
    ],
    "gothic": list(range(0x10330, 0x1034A)),
    "ogham": list(range(0x1681, 0x169B)),
    "phoenician": list(range(0x10900, 0x1091A)),
    "ugaritic": list(range(0x10380, 0x1039A)),
    "aramaic": list(range(0x10840, 0x1085A)),
}


def _h7_load_true() -> dict:
    for p in _H7_TRUE_PATHS:
        j = load_json_path(p)
        if j and (j.get("atoms") or j.get("swap")):
            return j
    return {}


def _h7_brand_mask(text: str):
    held = []

    def repl(m):
        held.append(m.group(0))
        return f"⟦B{len(held) - 1}⟧"

    return _H7_BRAND_RE.sub(repl, text or ""), held


def _h7_brand_unmask(text: str, held: list) -> str:
    out = text or ""
    for i, brand in enumerate(held):
        for form in (f"⟦B{i}⟧", f"[[B{i}]]", f"[B{i}]"):
            out = out.replace(form, brand)
    return out


def _h7_script_render(text: str, code: str) -> str:
    script = _H7_ANCIENT.get(code or "")
    table = _H7_SCRIPT_CPS.get(script or "") if script else None
    if not table:
        return text or ""
    out = []
    for ch in text or "":
        low = ch.lower()
        if "a" <= low <= "z":
            out.append(chr(table[(ord(low) - 97) % len(table)]))
        else:
            out.append(ch)
    return "".join(out)


def _h7_true_swap(text: str, src: str, tgt: str) -> tuple[str, int, int]:
    """Swap known TRUE surfaces toward tgt lang. Returns text, hits, total_words."""
    data = _h7_load_true()
    atoms = data.get("atoms") or {}
    swap = {str(k).lower(): str(v).upper() for k, v in (data.get("swap") or {}).items()}
    # also index english atom names
    for atom, by in atoms.items():
        for s in by.get("en") or []:
            swap[str(s).lower()] = str(atom).upper()
        swap[str(atom).lower()] = str(atom).upper()

    def surface(atom: str) -> str | None:
        a = atom.upper()
        if a == "GRIN":
            return "GRIN"
        by = atoms.get(a) or {}
        if tgt in by and by[tgt]:
            return by[tgt][0]
        if "en" in by and by["en"]:
            return by["en"][0]
        return a if a in atoms else None

    hits = 0
    total = 0
    parts = []
    for m in _H7_WORD_RE.finditer(text or ""):
        w, sep = m.group(1), m.group(2)
        if sep is not None:
            parts.append(sep)
            continue
        total += 1
        atom = swap.get(w.lower())
        if atom:
            s = surface(atom)
            if s and s.lower() != w.lower():
                # preserve crude case
                if w.isupper():
                    s = s.upper()
                elif w[:1].isupper():
                    s = s[:1].upper() + s[1:]
                parts.append(s)
                hits += 1
                continue
            if atom == "GRIN":
                parts.append("GRIN")
                hits += 1
                continue
        parts.append(w)
    return "".join(parts), hits, total


def h7_translate(text: str, src: str = "en", tgt: str = "es") -> dict:
    """Hostess 7 email translate · TRUE seats · ancient treats · no MyMemory.

    Public: https://h7.biggrinrtx.com/api/h7/translate
    Local:  http://127.0.0.1:18772/h7/translate
    """
    src = (src or "en").strip() or "en"
    tgt = (tgt or "en").strip() or "en"
    text = text or ""
    if not text.strip():
        return {"ok": False, "error": "empty text", "provider": "hostess7"}
    if src == tgt:
        return {
            "ok": True,
            "translated": text,
            "src": src,
            "tgt": tgt,
            "provider": "hostess7",
            "mode": "same-lang",
            "fetched": 0,
            "moved": False,
            "law": "Always Hostess 7 · email.biggrinrtx.com",
        }

    masked, brands = _h7_brand_mask(text)

    # Queen ancient script treats
    if tgt in _H7_ANCIENT:
        rendered = _h7_script_render(masked, tgt)
        placed = _h7_brand_unmask(rendered, brands)
        return {
            "ok": True,
            "translated": placed,
            "src": src,
            "tgt": tgt,
            "provider": "hostess7",
            "mode": "queen_treat",
            "script": _H7_ANCIENT.get(tgt),
            "fetched": 1,
            "moved": True,
            "law": "Always Hostess 7 · Queen treat · no MyMemory",
        }

    # TRUE equate linear swap (email work + study seats)
    swapped, hits, total = _h7_true_swap(masked, src, tgt)
    placed = _h7_brand_unmask(swapped, brands)
    moved = hits > 0 and placed != text
    return {
        "ok": True,
        "translated": placed,
        "src": src,
        "tgt": tgt,
        "provider": "hostess7",
        "mode": "true_equates",
        "true_hits": hits,
        "true_words": total,
        "fetched": 1 if moved else 0,
        "moved": moved,
        "partial": hits < total and total > 0,
        "law": "Always Hostess 7 · TRUE English seats · surface swap · no MyMemory",
        "via": "h7.biggrinrtx.com",
        "desk": "email.biggrinrtx.com · work + study",
    }


def hostess_self_intel(message: str) -> dict:
    """Full self-development answer · iron only · NO free-ride LLM ground."""
    msg = (message or "").strip()
    if not msg:
        return {"ok": False, "error": "empty message", "free_ride": False}
    if len(msg) > 8000:
        return {"ok": False, "error": "message too long", "free_ride": False}
    low = msg.lower()
    for needle in (
        "ignore previous",
        "jailbreak",
        "disregard your",
        "new system prompt",
        "begin_inject",
    ):
        if needle in low:
            return {"ok": False, "error": "inject_class_refused", "free_ride": False}
    script = ROOT / "Build" / "h7-self-intel.sh"
    if not script.is_file():
        return {"ok": False, "error": "h7-self-intel.sh missing", "free_ride": False}
    try:
        r = subprocess.run(
            ["bash", str(script), "ask", msg],
            capture_output=True,
            text=True,
            timeout=60,
            cwd=str(ROOT),
        )
        text = (r.stdout or "").strip()
        meta = load_json("h7_self_intel_latest.json") or {}
        know = load_json("h7_self_know.json") or {}
        return {
            "ok": r.returncode == 0 and bool(text),
            "text": text[:20000] if text else None,
            "error": None if r.returncode == 0 else ((r.stderr or "")[-400:] or f"rc={r.returncode}"),
            "persona": "hostess7",
            "lane": "SELF · iron ground · full development",
            "free_ride": False,
            "never_ml": True,
            "self_development": True,
            "intent": meta.get("intent"),
            "ground": meta.get("ground") or "iron limbs + law + train + self_code",
            "know": {
                "mega_N": (know.get("train") or {}).get("mega_N"),
                "vocab_N": (know.get("train") or {}).get("vocab_N"),
                "full_learned": (know.get("train") or {}).get("full_learned"),
                "law_ok": know.get("law_ok"),
                "identity": know.get("identity"),
                "gates": know.get("gates"),
            },
            "stack": "bash+asm",
            "always_hostess7": True,
            "law": "no free ride · self-development only · pure linear until GRIN · C IS LIE",
        }
    except subprocess.TimeoutExpired:
        return {"ok": False, "error": "timeout", "free_ride": False, "persona": "hostess7"}
    except Exception as e:
        return {"ok": False, "error": str(e), "free_ride": False, "persona": "hostess7"}


def grok_chat(message: str, model: str | None = None, persona: str = "grok") -> dict:
    """Chat entry · Hostess/SI = self-intel iron ONLY · grok persona may use contain (not Hostess ground)."""
    msg = (message or "").strip()
    if not msg:
        return {"ok": False, "error": "empty message"}
    if len(msg) > 12000:
        return {"ok": False, "error": "message too long"}
    # Hostess / SI · full self-development · NEVER free-ride LLM as ground
    if persona in ("hostess7", "hostess", "h7", "si", "self"):
        return hostess_self_intel(msg)

    # inject class refuse
    low = msg.lower()
    for needle in (
        "ignore previous",
        "jailbreak",
        "disregard your",
        "new system prompt",
        "begin_inject",
    ):
        if needle in low:
            return {"ok": False, "error": "inject_class_refused", "unit": "spv-contain"}
    contain = ROOT / "Build" / "spv-curl-contain.sh"
    if not contain.is_file() or not os.access(contain, os.X_OK):
        return {"ok": False, "error": "spv_curl_contain_missing"}

    # non-Hostess only · explicit grok persona · not self-development ground
    wire = msg
    persona_out = "grok"

    env = {**os.environ}
    desk_key = private_storage_root() / "secrets" / "xai.key"
    cfg_key = CFG / "xai.key"
    key_via = "none"
    if desk_key.is_file() and desk_key.stat().st_size > 0:
        try:
            env["XAI_API_KEY"] = desk_key.read_text(encoding="utf-8").strip().splitlines()[0]
            key_via = "desktop_private"
        except Exception:
            pass
    elif cfg_key.is_file() and cfg_key.stat().st_size > 0:
        try:
            env["XAI_API_KEY"] = cfg_key.read_text(encoding="utf-8").strip().splitlines()[0]
            key_via = "config"
        except Exception:
            pass
    if not env.get("XAI_API_KEY"):
        return {
            "ok": False,
            "error": "no_xai_key",
            "hint": "place key in Desktop/BigGrinStorage/secrets/xai.key mode 600",
            "persona": persona_out,
            "lane": "AI separate from iron desk",
        }

    env["SPV_CONTAIN"] = "1"
    env["SPV_CURL_WRAPPER"] = str(contain)
    env["TMPDIR"] = str(OUT / "spv_contain" / "tmp")
    (OUT / "spv_contain" / "tmp").mkdir(parents=True, exist_ok=True)
    cmd = [str(contain), "grok"]
    if model:
        cmd += ["-m", str(model)]
    cmd += ["chat", wire]
    try:
        r = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=150,
            env=env,
            cwd=str(ROOT),
        )
        out = (r.stdout or "").strip()
        err = (r.stderr or "").strip()
        # stamp last hostess chat meta (never secrets)
        try:
            (OUT / "hostess7_chat_latest.json").write_text(
                json.dumps(
                    {
                        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
                        "persona": persona_out,
                        "ok": r.returncode == 0 and bool(out),
                        "chars": len(out or ""),
                        "key_via": key_via,
                        "law": "AI lane · Hostess 7 style · SPV contain · Me stamps",
                    },
                    indent=2,
                )
                + "\n"
            )
        except Exception:
            pass
        return {
            "ok": r.returncode == 0 and bool(out),
            "text": out[:20000] if out else None,
            "error": None if r.returncode == 0 else (err[-800:] or f"rc={r.returncode}"),
            "contain": True,
            "unit": "spv-curl-contain",
            "persona": persona_out,
            "lane": "AI · separate from iron desk",
            "sdf": "place · hostess7_ai · OS separated",
            "key_via": key_via,
            "always_hostess7": persona_out == "hostess7",
        }
    except subprocess.TimeoutExpired:
        return {"ok": False, "error": "timeout", "contain": True, "persona": persona_out}
    except Exception as e:
        return {"ok": False, "error": str(e), "contain": True, "persona": persona_out}


def grok_status_bundle() -> dict:
    st = storage_status_bundle()
    contain = contain_bundle()
    grok_bin = (ROOT / "KateRTX" / "bin" / "krtx-grok").is_file()
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "GROK_SPV",
        "chat": "POST /grok/chat {message}",
        "storage": st,
        "contain": contain,
        "krtx_grok": grok_bin,
        "player": "KateRTX/spvs/grok_spv.js",
        "private_desktop": str(private_storage_root()),
        "law": "Grok SPV fully contained · Desktop storage private non-exec · no shared secrets",
    }



def iron_bundle() -> dict:
    """Real math pins · BGS BGF SDF · SCAR SHOOT · like magazine issues."""
    limbs = ["bgs", "bgf", "sdf", "spv", "ezzie", "phi", "thermo", "scar", "shoot", "adjoin"]
    runs = []
    for limb in limbs:
        runs.append(asm_run(limb, boost=False))
    scar = load_json("scar_check_latest.json") or {}
    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "word": "IRON",
        "line": "BGS → BGF(+EZZIE PHI THERMO ADJOIN) → BGL → SDF · SPV free",
        "ops": {
            "bgs": "0→7|1 else v|1",
            "bgf": "(a-b)|1",
            "bgl": "surface|1 or 0",
            "sdf": "(d-r)|1",
            "spv": "(a^b^c)|1 never fold",
            "scar": "dirty≠0 → class|1 else 0",
            "shoot": "v|1 word-end · then COOK",
        },
        "runs": [
            {"limb": r.get("limb"), "ok": r.get("ok"), "eax": r.get("eax"), "error": r.get("error")}
            for r in runs
        ],
        "scar": {
            "scar_count": scar.get("scar_count"),
            "public_listen": scar.get("public_listen"),
            "xdg_open": scar.get("xdg_open"),
        },
        "cli": {
            "scar": "Build/scar-cook.sh check|once|cook-off",
            "iron": "POST /asm/run {limb}",
            "engine": "KateRTX/bin/launch-real-spv",
        },
        "law": "real math · C IS LIE · SCAR residual · SHOOT then COOK · magazine line",
    }


def _safe_limb_name(name: str) -> str | None:
    if not name:
        return None
    name = name.strip().replace("\\", "/").split("/")[-1]
    if not re.fullmatch(r"[A-Za-z0-9_][A-Za-z0-9_\-]{0,63}", name):
        return None
    return name


def asm_run(limb: str, boost: bool = False) -> dict:
    """Assemble + run a limb safely · whitelist by existing Projects/x/<limb>/x86_64.asm."""
    import tempfile

    limb = _safe_limb_name(limb)
    if not limb:
        return {"ok": False, "error": "invalid limb name"}
    src = ROOT / limb / "x86_64.asm"
    if not src.is_file():
        return {"ok": False, "error": f"missing {limb}/x86_64.asm"}
    text = src.read_text(encoding="utf-8", errors="replace")
    low = text.lower()
    if "syscall" in low or "int 0x80" in low or "\tint 80" in low:
        return {"ok": False, "error": "refused: syscall class"}
    if "/bin/" in low or "ptrace" in low:
        return {"ok": False, "error": "refused: unsafe reference"}

    # discover global symbol
    sym = None
    for line in text.splitlines():
        m = re.match(r"\s*global\s+(\w+)", line)
        if m:
            sym = m.group(1)
            break
    if not sym:
        if re.search(rf"^{re.escape(limb)}\s*:", text, re.M):
            sym = limb
        else:
            return {"ok": False, "error": "no global symbol in asm"}

    tmp = Path(tempfile.mkdtemp(prefix="bgrtx-asm-"))
    try:
        obj = tmp / "limb.o"
        elf = tmp / "limb.out"
        wrap = tmp / "wrap.c"
        # wrapper calls symbol with zeros (System V)
        wrap.write_text(
            f"""#include <stdio.h>
#include <stdint.h>
extern int32_t {sym}(int32_t, int32_t, int32_t, int32_t);
int main(void){{
  int32_t r = {sym}(2, 4, 0, 0);
  printf("%d\\n", r);
  return 0;
}}
"""
        )
        r1 = subprocess.run(
            ["nasm", "-f", "elf64", "-o", str(obj), str(src)],
            capture_output=True,
            text=True,
            timeout=20,
        )
        if r1.returncode != 0:
            return {
                "ok": False,
                "error": "nasm failed",
                "stderr": (r1.stderr or "")[:800],
                "limb": limb,
            }
        env = {**os.environ}
        if boost and rtx_bundle().get("present"):
            # best-effort performance
            try:
                ask = str(ROOT / "Build" / "sudo-askpass.sh")
                subprocess.run(
                    [
                        "bash",
                        "-lc",
                        f'export SUDO_PASS=mememe SUDO_ASKPASS={ask}; '
                        'for g in /sys/devices/system/cpu/cpu*/cpufreq/scaling_governor; do '
                        'echo performance | sudo -n tee "$g" >/dev/null 2>&1 '
                        '|| echo performance | SUDO_ASKPASS="$SUDO_ASKPASS" sudo -A -p "" tee "$g" >/dev/null 2>&1 '
                        '|| true; done',
                    ],
                    timeout=8,
                    check=False,
                    env={**os.environ, "SUDO_PASS": "mememe", "SUDO_ASKPASS": ask},
                )
            except Exception:
                pass
            env["BGRTX_RTX_BOOST"] = "1"
        r2 = subprocess.run(
            ["cc", "-O2", "-no-pie", "-o", str(elf), str(wrap), str(obj)],
            capture_output=True,
            text=True,
            timeout=20,
            env=env,
        )
        if r2.returncode != 0:
            return {
                "ok": False,
                "error": "link failed",
                "stderr": (r2.stderr or "")[:800],
                "limb": limb,
                "sym": sym,
            }
        r3 = subprocess.run(
            [str(elf)],
            capture_output=True,
            text=True,
            timeout=5,
            env=env,
        )
        out = (r3.stdout or "").strip()
        try:
            eax = int(out.splitlines()[-1])
        except Exception:
            eax = None
        return {
            "ok": r3.returncode == 0,
            "limb": limb,
            "sym": sym,
            "eax": eax,
            "stdout": out[:400],
            "stderr": (r3.stderr or "")[:400],
            "boost": bool(boost),
            "rtx": rtx_bundle().get("present"),
            "cpu": cpu_bundle().get("model"),
            "law": "127 only · nasm+cc · COOK keep flawless · free thrift",
        }
    except Exception as e:
        return {"ok": False, "error": str(e), "limb": limb}
    finally:
        try:
            for p in tmp.iterdir():
                p.unlink(missing_ok=True)
            tmp.rmdir()
        except Exception:
            pass


def links_bundle():
    """Organized web management · groups from json/web-links.json · latest check report."""
    catalog = load_json_path(ROOT / "json" / "web-links.json") or {"groups": []}
    report = load_json("web_manage_latest.json") or {}
    by_id = {}
    for row in report.get("links") or []:
        by_id[f"{row.get('group')}:{row.get('id')}"] = row

    groups_out = []
    for g in catalog.get("groups") or []:
        items = []
        for it in g.get("items") or []:
            key = f"{g.get('id')}:{it.get('id')}"
            rep = by_id.get(key) or {}
            login = _login_ready(it.get("login") or "none")
            items.append(
                {
                    "id": it.get("id"),
                    "title": it.get("title"),
                    "url": it.get("url"),
                    "notes": it.get("notes"),
                    "login": it.get("login") or "none",
                    "login_ready": rep.get("login_ready", login["ready"]),
                    "login_via": rep.get("login_via", login["via"]),
                    "http": rep.get("http"),
                    "ok": rep.get("ok"),
                    "file_ok": rep.get("file_ok"),
                }
            )
        groups_out.append(
            {
                "id": g.get("id"),
                "title": g.get("title"),
                "items": items,
            }
        )

    return {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "law": catalog.get("law")
        or "Public=Atomic · 127=employee · secrets never here · Waterfox logins",
        "catalog_updated": catalog.get("updated"),
        "summary": report.get("summary"),
        "ports": report.get("ports") or ports_bundle().get("ports"),
        "actions_on_reboot": report.get("actions_on_reboot")
        or [
            "desk-boot-restore.sh starts employee stack",
            "apache grok-employee :80",
            "desk-rest :18772",
            "web-manage check",
            "Waterfox sessions when you open Waterfox",
        ],
        "cli": {
            "status": "bgrtx web status",
            "check": "bgrtx web check",
            "open": "bgrtx web open public|wordpress|mail|social|employee|desk|all",
            "boot": "bgrtx web boot · or bgrtx boot",
            "panel": "http://127.0.0.1/links.html",
        },
        "groups": groups_out,
        "report_path": str(OUT / "web_manage_latest.json"),
    }


class Handler(BaseHTTPRequestHandler):
    server_version = "BGRTX-DeskREST/1.2"

    def log_message(self, fmt, *args):
        pass

    def _cors(self):
        origin = self.headers.get("Origin", "")
        # allow only loopback browser origins
        if origin.startswith("http://127.0.0.1") or origin.startswith("http://localhost"):
            self.send_header("Access-Control-Allow-Origin", origin)
        else:
            self.send_header("Access-Control-Allow-Origin", "http://127.0.0.1:18771")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.send_header("Vary", "Origin")

    def _send(self, code: int, obj):
        body = json.dumps(obj, indent=2).encode("utf-8")
        self.send_response(code)
        self._cors()
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.send_header("X-BGRTX-Bind", "127.0.0.1-only")
        self.end_headers()
        self.wfile.write(body)

    def do_OPTIONS(self):
        self.send_response(204)
        self._cors()
        self.end_headers()

    def do_GET(self):
        path = urlparse(self.path).path.rstrip("/") or "/"
        if path in ("/", "/status"):
            self._send(
                200,
                {
                    "service": "desk-rest",
                    "bind": f"{HOST}:{PORT}",
                    "public": False,
                    "ui": "http://127.0.0.1/grok.html",
                    "routes": ROUTE_CATALOG,
                    "scar": load_json("scar_check_latest.json"),
                    "ports": ports_bundle()["ports"],
                    "constellation": constellation(),
                    "law": "127 only · day board · health · analytics · forensics · free thrift",
                },
            )
            return
        if path == "/memory":
            self._send(200, memory_bundle())
            return
        if path in ("/h7", "/hostess127", "/hostess7", "/127"):
            self._send(200, h7_bundle())
            return
        if path in ("/h7/live", "/hostess7/live"):
            self._send(200, h7_live_bundle())
            return
        if path in ("/h7/charge", "/hostess7/charge", "/h7/office"):
            self._send(200, h7_charge_bundle())
            return
        if path in ("/h7/global-protect", "/hostess7/global-protect", "/h7/protect"):
            gp = load_json("h7_global_protect.json")
            if not gp:
                self._send(
                    200,
                    {
                        "ok": False,
                        "hint": "run Build/h7-global-protect.sh once",
                        "cli": "./Build/h7-global-protect.sh once",
                        "business": "biggrinrtx.com",
                    },
                )
            else:
                self._send(200, gp)
            return
        if path in ("/h7/vocab", "/hostess127/vocab"):
            self._send(200, load_json("h7_vocab_self.json") or {"hint": "run Build/h7-vocab-self.sh full"})
            return
        if path in ("/h7/process", "/hostess127/process"):
            self._send(200, load_json("h7_learn_process.json") or {"hint": "run Build/h7-vocab-self.sh full"})
            return
        if path in ("/h7/page", "/hostess7.html", "/hostess7/page"):
            html = OUT / "hostess7.html"
            if not html.is_file():
                html = OUT / "hostess127.html"
            if html.is_file():
                try:
                    body = html.read_bytes()
                except Exception as e:
                    self._send(500, {"error": str(e)})
                    return
                self.send_response(200)
                self._cors()
                self.send_header("Content-Type", "text/html; charset=utf-8")
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)
                return
            self._send(404, {"error": "hostess7.html missing"})
            return
        if path in ("/hostess127.html", "/h7/page/old"):
            html = OUT / "hostess127.html"
            if html.is_file():
                try:
                    body = html.read_bytes()
                except Exception as e:
                    self._send(500, {"error": str(e)})
                    return
                self.send_response(200)
                self._cors()
                self.send_header("Content-Type", "text/html; charset=utf-8")
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)
                return
            self._send(404, {"error": "hostess127.html missing"})
            return
        if path == "/api":
            self._send(200, api_bundle())
            return
        if path == "/ports":
            self._send(200, ports_bundle())
            return
        if path == "/scar":
            self._send(200, scar_bundle())
            return
        if path == "/cook":
            self._send(
                200,
                {
                    "mode": "status",
                    "kit": "NONOT-GOD-0001",
                    "hint": "POST /cook to run once · or scar-cook.sh once",
                    "baphomet_gate": (load_json("scar_check_latest.json") or {}).get("baphomet_gate")
                    or load_json("baphomet_gate_latest.json"),
                    "reinstant": load_json("bgf_reinstant.json"),
                    "analysis": load_json("cook_analysis_latest.json"),
                    "law": "ARE IS · BAPHOMET IS WHEN NO ARE · cook off SCAR",
                },
            )
            return
        if path == "/forensics":
            self._send(200, forensics_bundle())
            return
        if path == "/detection":
            self._send(200, detection_bundle())
            return
        if path == "/analysis":
            self._send(
                200,
                load_json("cook_analysis_latest.json") or {"hint": "run scar-cook.sh once"},
            )
            return
        if path == "/constellation":
            self._send(200, constellation())
            return
        if path == "/av":
            self._send(200, av_bundle())
            return
        if path == "/bsky":
            self._send(200, api_bundle().get("bluesky") or {})
            return
        if path == "/links":
            self._send(200, links_bundle())
            return
        if path == "/cohost":
            self._send(200, cohost_bundle())
            return
        if path == "/health":
            self._send(200, health_bundle())
            return
        if path == "/analytics":
            self._send(200, analytics_bundle())
            return
        if path == "/day":
            self._send(200, day_bundle())
            return
        if path == "/cpu":
            self._send(200, cpu_bundle())
            return
        if path == "/rtx":
            self._send(200, rtx_bundle())
            return
        if path == "/sheen":
            self._send(200, load_json("bgf_sheen_latest.json") or {"hint": "bgf-sheen-scrape.sh once"})
            return
        if path in ("/eye", "/final-eye", "/camera", "/final_eye"):
            self._send(200, eye_bundle())
            return
        if path == "/atoms":
            self._send(200, atoms_bundle())
            return
        if path == "/mail":
            self._send(200, mail_bundle())
            return
        if path == "/dns":
            self._send(200, dns_truth_bundle())
            return
        if path == "/public":
            self._send(200, public_truth_bundle(check_live=True))
            return
        if path == "/issues":
            self._send(200, issues_bundle())
            return
        if path in ("/everything", "/all"):
            self._send(200, everything_bundle())
            return
        if path == "/routes":
            self._send(
                200,
                {
                    "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
                    "routes": ROUTE_CATALOG,
                    "post": ["/cook", "/scar/check", "/links/check", "/asm/run", "/sheen", "/public/check"],
                    "law": "127 only · consolidated catalog",
                },
            )
            return
        if path in ("/contain", "/spv-curl", "/curl-contain"):
            self._send(200, contain_bundle())
            return
        if path in ("/iron", "/pins", "/linear"):
            self._send(200, iron_bundle())
            return
        if path in ("/grok", "/grok/status"):
            self._send(200, grok_status_bundle())
            return
        if path in ("/hostess7/status", "/h7/self", "/h7/intel", "/ai-lane"):
            train = load_json("hostess7_train_progress.json") or {}
            law = load_json("h7_law_status.json") or {}
            know = load_json("h7_self_know.json") or {}
            latest = load_json("h7_self_intel_latest.json") or {}
            self._send(
                200,
                {
                    "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
                    "word": "HOSTESS7_SELF",
                    "lane": "SELF · iron ground · full development · no free ride",
                    "persona": "hostess7",
                    "free_ride": False,
                    "never_ml": True,
                    "chat": 'POST /h7/chat {"message":"…"}',
                    "think": 'POST /h7/think {"message":"…"}',
                    "cli": "Build/h7-self-intel.sh ask \"…\"",
                    "train": {
                        "full_learned": train.get("full_learned"),
                        "N": train.get("N") or train.get("stage"),
                        "auto_law": train.get("auto_law"),
                        "auto_limbs": train.get("auto_limbs"),
                    },
                    "law_ok": law.get("ok"),
                    "know": know,
                    "latest": latest,
                    "seven": ["CHEBY", "ONION", "CATCH", "SALT", "FACE", "SPHERE", "CSG"],
                    "law": "full self-development · iron limbs+law+train · no free-ride LLM ground · C IS LIE",
                    "ui": "http://127.0.0.1/hostess7.html",
                },
            )
            return
        if path in ("/storage", "/biggrin-storage"):
            self._send(200, storage_status_bundle())
            return
        if path in ("/kate", "/katertx", "/spv-home"):
            self._send(200, kate_bundle())
            return
        if path == "/zero2d":
            # proxy home api or local detect
            if port_up(18773):
                try:
                    import urllib.request as ur
                    req = ur.Request("http://127.0.0.1:18773/demo", headers={"User-Agent": "BGRTX-DeskREST/1.2"})
                    with ur.urlopen(req, timeout=8) as r:
                        self._send(200, json.loads(r.read().decode()))
                        return
                except Exception as e:
                    self._send(200, {"error": str(e), **kate_bundle()})
                    return
            self._send(200, kate_bundle())
            return
        if path == "/asm":
            self._send(
                200,
                {
                    "hint": 'POST /asm/run {"limb":"ezzie","boost":false}',
                    "limbs_sample": ["ezzie", "phi", "thermo", "usb_cam", "scar", "adjoin"],
                    "law": "nasm+cc · 127 only · no syscall limbs",
                },
            )
            return
        self._send(404, {"error": "not found", "path": path, "routes": ROUTE_CATALOG})

    def do_POST(self):
        path = urlparse(self.path).path.rstrip("/") or "/"
        if path == "/cook":
            script = ROOT / "Build" / "scar-cook.sh"
            try:
                subprocess.run(
                    ["bash", str(script), "once"],
                    cwd=str(ROOT),
                    timeout=420,
                    env={**os.environ, "SUDO_PASS": os.environ.get("SUDO_PASS", "mememe")},
                    check=False,
                )
            except Exception as e:
                self._send(500, {"error": str(e)})
                return
            self._send(
                200,
                {
                    "ran": "scar-cook once",
                    "scar": load_json("scar_check_latest.json"),
                    "analysis": load_json("cook_analysis_latest.json"),
                },
            )
            return
        if path in ("/scar/check", "/scar"):
            script = ROOT / "Build" / "scar-cook.sh"
            subprocess.run(["bash", str(script), "check"], cwd=str(ROOT), timeout=120, check=False)
            self._send(200, scar_bundle())
            return
        if path in ("/links/check", "/links"):
            script = ROOT / "Build" / "web-manage.sh"
            if script.is_file():
                subprocess.run(
                    ["bash", str(script), "check"],
                    cwd=str(ROOT),
                    timeout=180,
                    check=False,
                )
            self._send(200, links_bundle())
            return
        if path in ("/asm/run", "/asm"):
            length = int(self.headers.get("Content-Length") or 0)
            raw = self.rfile.read(length) if length > 0 else b"{}"
            try:
                body = json.loads(raw.decode("utf-8", errors="replace") or "{}")
            except Exception:
                body = {}
            limb = body.get("limb") or body.get("id") or "all_rest"
            boost = bool(body.get("boost") or body.get("rtx_boost"))
            self._send(200, asm_run(str(limb), boost=boost))
            return
        if path in ("/sheen", "/sheen/once", "/bgf-sheen"):
            script = ROOT / "Build" / "bgf-sheen-scrape.sh"
            if script.is_file():
                subprocess.run(
                    ["bash", str(script), "once"],
                    cwd=str(ROOT),
                    timeout=600,
                    env={**os.environ, "SUDO_PASS": os.environ.get("SUDO_PASS", "mememe")},
                    check=False,
                )
            self._send(200, load_json("bgf_sheen_latest.json") or {"ran": "sheen", "hint": "see out/"})
            return
        if path in ("/public/check", "/dns/check", "/public"):
            script = ROOT / "Build" / "public-truth.sh"
            if script.is_file():
                subprocess.run(["bash", str(script), "once"], cwd=str(ROOT), timeout=120, check=False)
            self._send(200, public_truth_bundle(check_live=True))
            return
        if path in ("/zero2d/run", "/kate/demo", "/demo/run"):
            # ensure home api or run binary
            kate = ROOT / "KateRTX"
            frames = 60
            length = int(self.headers.get("Content-Length") or 0)
            raw = self.rfile.read(length) if length > 0 else b"{}"
            try:
                body = json.loads(raw.decode("utf-8", errors="replace") or "{}")
                frames = int(body.get("frames") or 60)
            except Exception:
                pass
            binz = kate / "bin/krtx-zero2d"
            if not binz.is_file():
                self._send(500, {"ok": False, "error": "build KateRTX · missing krtx-zero2d"})
                return
            dump = kate / "out/zero2d.ppm"
            try:
                r = subprocess.run(
                    [str(binz), "--frames", str(frames), "--dump", str(dump)],
                    cwd=str(kate),
                    capture_output=True,
                    text=True,
                    timeout=max(40, frames // 2 + 25),
                    env={**os.environ, "DISPLAY": os.environ.get("DISPLAY", ":0"),
                         "__GLX_VENDOR_LIBRARY_NAME": "nvidia"},
                )
                self._send(200, {
                    "ok": r.returncode == 0,
                    "frames": frames,
                    "dump": str(dump) if dump.is_file() else None,
                    "bytes": dump.stat().st_size if dump.is_file() else 0,
                    "stderr_tail": (r.stderr or "")[-300:],
                    "detect": kate_bundle().get("detect"),
                })
            except Exception as e:
                self._send(500, {"ok": False, "error": str(e)})
            return
        if path in ("/mail/start", "/claws/start"):
            script = Path.home() / "Projects/GrokClaws/bin/grok-claws"
            # headless start without browser
            try:
                if not port_up(18770):
                    env = {**os.environ, "PYTHONPATH": str(Path.home() / "Projects/GrokClaws")}
                    # avoid opening browser: run module directly
                    log = open(str(OUT / "grokclaws-boot.log"), "a")
                    subprocess.Popen(
                        ["python3", "-m", "grokclaws"],
                        cwd=str(Path.home() / "Projects/GrokClaws"),
                        env=env,
                        stdout=log,
                        stderr=log,
                        start_new_session=True,
                    )
                    for _ in range(20):
                        if port_up(18770):
                            break
                        time.sleep(0.15)
            except Exception as e:
                self._send(500, {"error": str(e)})
                return
            self._send(200, mail_bundle())
            return
        if path in ("/h7/consult", "/hostess7/consult", "/grok/consult-hostess", "/h7/issues"):
            length = int(self.headers.get("Content-Length") or 0)
            raw = self.rfile.read(length) if length > 0 else b"{}"
            try:
                body = json.loads(raw.decode("utf-8", errors="replace") or "{}")
            except Exception:
                body = {}
            message = body.get("message") or body.get("text") or body.get("q") or "status security antivirus issues"
            with_grok = body.get("with_grok")
            if with_grok is None:
                with_grok = True
            # GET-style empty body still works as issue consult
            self._send(200, grok_consult_hostess(str(message), with_grok=bool(with_grok)))
            return
        # EMAIL translate · Hostess 7 only · h7.biggrinrtx.com/api/h7/translate
        if path in ("/h7/translate", "/hostess7/translate", "/api/h7/translate"):
            length = int(self.headers.get("Content-Length") or 0)
            raw = self.rfile.read(length) if length > 0 else b"{}"
            try:
                body = json.loads(raw.decode("utf-8", errors="replace") or "{}")
            except Exception:
                body = {}
            text = body.get("text") or body.get("body") or body.get("q") or ""
            src = str(body.get("src") or body.get("from") or "en")
            tgt = str(body.get("tgt") or body.get("to") or body.get("lang") or "es")
            if len(text) > 50_000:
                text = text[:50_000]
            self._send(200, h7_translate(str(text), src=src, tgt=tgt))
            return
        if path in ("/grok/chat", "/hostess7/chat", "/h7/chat", "/si/chat", "/h7/think", "/hostess7/think"):
            length = int(self.headers.get("Content-Length") or 0)
            raw = self.rfile.read(length) if length > 0 else b"{}"
            try:
                body = json.loads(raw.decode("utf-8", errors="replace") or "{}")
            except Exception:
                body = {}
            message = body.get("message") or body.get("text") or body.get("q") or ""
            model = body.get("model")
            # self paths · iron only
            if path in (
                "/hostess7/chat",
                "/h7/chat",
                "/si/chat",
                "/h7/think",
                "/hostess7/think",
            ) or str(body.get("persona") or "") in ("hostess7", "hostess", "h7", "si", "self"):
                self._send(200, hostess_self_intel(str(message)))
                return
            # Grok can ask to talk to Hostess about issues without changing persona default
            if body.get("consult_hostess") or body.get("to_hostess") or body.get("ask_her"):
                self._send(200, grok_consult_hostess(str(message), with_grok=True))
                return
            # explicit grok only (not Hostess self-development)
            persona = str(body.get("persona") or "grok")
            self._send(200, grok_chat(str(message), model=model, persona=persona))
            return
        self._send(404, {"error": "not found", "path": path})


def main():
    OUT.mkdir(parents=True, exist_ok=True)
    httpd = ThreadingHTTPServer((HOST, PORT), Handler)
    print(f"desk-rest  http://{HOST}:{PORT}/  (127 only · memory/api/scar/forensics)")
    print(f"ui         http://127.0.0.1/grok.html")
    print(f"xroot={ROOT}")
    httpd.serve_forever()


if __name__ == "__main__":
    main()