#!/usr/bin/env python3
"""Atlas Cyber Protect local portal — living security UI + guardian agent (no Factory, no prices)."""
from __future__ import annotations

import hashlib
import hmac
import json
import os
import re
import secrets
import socket
import sqlite3
import ssl
import subprocess
import threading
import time
import urllib.parse
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qsl, urlparse

ROOT = Path(__file__).resolve().parent
DATA = Path("/var/lib/atlas-cyber-protect")
DB = DATA / "portal.sqlite"
PORTAL_HTML = ROOT / "portal.html"
ICONS_DIR = ROOT / "device-icons"
STATE = DATA / "state.json"
AGENT_RELEASES = DATA / "agent-releases"
AGENT_MANIFEST = AGENT_RELEASES / "manifest.json"
ENDPOINT_LEARN = DATA / "endpoint-learning.jsonl"
LOCK_FLAG = DATA / "subscription.locked"
ENV_FILE = Path("/etc/atlas-cyber-protect/acp.env")
CUSTOMER_JSON = DATA / "customer.json"
HOST = os.environ.get("ACP_PORTAL_HOST", "0.0.0.0")
PORT = int(os.environ.get("ACP_PORTAL_PORT", "8787"))
SESSIONS: dict[str, dict] = {}
SESS_TTL = 14 * 24 * 3600
SESSIONS_PATH = DATA / "sessions.json"
SESSIONS_LOCK = threading.Lock()


def _sessions_load() -> None:
    """Restore saved sessions (remember-me survives box restarts / power loss)."""
    try:
        if SESSIONS_PATH.is_file():
            data = json.loads(SESSIONS_PATH.read_text(encoding="utf-8"))
            now = time.time()
            for tok, sess in (data or {}).items():
                if isinstance(sess, dict) and float(sess.get("exp") or 0) > now:
                    SESSIONS[tok] = sess
    except Exception as exc:
        print(f"sessions load skipped: {type(exc).__name__}: {exc}", flush=True)


def _sessions_save() -> None:
    """Persist valid sessions so logins are remembered after a reboot."""
    try:
        DATA.mkdir(parents=True, exist_ok=True)
        now = time.time()
        with SESSIONS_LOCK:
            keep = {t: s for t, s in SESSIONS.items() if float(s.get("exp") or 0) > now}
            tmp = SESSIONS_PATH.with_suffix(".json.tmp")
            tmp.write_text(json.dumps(keep), encoding="utf-8")
            tmp.replace(SESSIONS_PATH)
    except Exception as exc:
        print(f"sessions save skipped: {type(exc).__name__}: {exc}", flush=True)


def _customer_info() -> dict:
    env = _load_env()
    out = {
        "business_name": env.get("BUSINESS_NAME") or "",
        "site_name": env.get("SITE_NAME") or "",
        "notify_email": env.get("NOTIFY_EMAIL") or "",
        "company_id": env.get("COMPANY_ID") or "",
    }
    try:
        if CUSTOMER_JSON.is_file():
            raw = json.loads(CUSTOMER_JSON.read_text(encoding="utf-8"))
            if isinstance(raw, dict):
                out["business_name"] = out["business_name"] or str(raw.get("business_name") or "")
                out["site_name"] = out["site_name"] or str(raw.get("site_name") or "")
                out["notify_email"] = out["notify_email"] or str(raw.get("notify_email") or "")
                out["company_id"] = out["company_id"] or str(raw.get("company_id") or "")
    except Exception:
        pass
    try:
        st = _load_state()
        cust = st.get("customer") if isinstance(st.get("customer"), dict) else {}
        if cust:
            out["business_name"] = out["business_name"] or str(cust.get("business_name") or "")
            out["site_name"] = out["site_name"] or str(cust.get("site_name") or "")
            out["notify_email"] = out["notify_email"] or str(cust.get("notify_email") or "")
            out["company_id"] = out["company_id"] or str(cust.get("company_id") or "")
    except Exception:
        pass
    done = Path("/etc/atlas-cyber-protect/customer-setup.done")
    try:
        if done.is_file() and not out["business_name"]:
            out["business_name"] = done.read_text(encoding="utf-8").strip()[:120]
    except Exception:
        pass
    return out


def _atlas_base() -> str:
    try:
        from atlas_reach import atlas_base as _reach

        return _reach()
    except Exception:
        return (
            _load_env().get("ATLAS_URL")
            or os.environ.get("ATLAS_URL")
            or "https://atlas-server.taile9cc75.ts.net"
        ).rstrip("/")


def _sync_atlas_connect_endpoint(agent_id: str, row: dict) -> None:
    """Register this PC with Funnel Atlas Connect so managers can click Connect."""
    try:
        cust = _customer_info()
        company_id = str(cust.get("company_id") or "").strip()
        if not company_id or company_id == "pending":
            return
        atlas = _atlas_base()
        inv = row.get("inventory") if isinstance(row.get("inventory"), dict) else {}
        hostname = (
            str(row.get("name") or "")
            or str(inv.get("hostname") or "")
            or str(row.get("hostname") or "")
            or agent_id
        )[:120]
        payload = {
            "agent_id": agent_id[:80],
            "company_id": company_id[:80],
            "site_id": str(cust.get("site_id") or "")[:80],
            "hostname": hostname,
            "ownership": "company_managed",
            "assigned_person_id": "",
            "rustdesk_id": str(inv.get("atlas_connect_id") or inv.get("rustdesk_id") or "")[:80],
            "agent_version": str(inv.get("agent_version") or row.get("agent_version") or "")[:32],
        }
        import ssl
        import urllib.request

        data = json.dumps(payload).encode()
        req = urllib.request.Request(
            f"{atlas}/api/cyber-protect/remote/endpoints",
            data=data,
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        with urllib.request.urlopen(req, timeout=20, context=ctx) as resp:
            resp.read()
    except Exception as exc:
        print(f"atlas-connect sync: {exc}", flush=True)


def _agents() -> dict:
    try:
        import companion_store as cs

        return cs.load_agents()
    except Exception:
        st = _load_state()
        raw = st.get("companion_agents") or {}
        return raw if isinstance(raw, dict) else {}


def _get_agent(agent_id: str) -> dict | None:
    """Look up a single paired agent by id."""
    if not agent_id:
        return None
    try:
        import companion_store as cs

        agents = cs.load_agents()
        a = agents.get(agent_id)
        if a:
            return a
    except Exception:
        pass
    st = _load_state()
    raw = st.get("companion_agents") or {}
    if isinstance(raw, dict):
        return raw.get(agent_id)
    return None


def _save_agents(agents: dict) -> None:
    try:
        import companion_store as cs

        cs.save_agents(agents)
    except Exception:
        st = _load_state()
        st["companion_agents"] = agents
        _save_state(st)


def _load_env() -> dict[str, str]:
    out: dict[str, str] = {}
    if ENV_FILE.is_file():
        for line in ENV_FILE.read_text(encoding="utf-8").splitlines():
            line = line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            k, v = line.split("=", 1)
            out[k.strip()] = v.strip().strip('"').strip("'")
    return out


def _enroll_token() -> str:
    return (_load_env().get("ENROLL_TOKEN") or os.environ.get("ENROLL_TOKEN") or "").strip()


def _lan_pending() -> dict:
    path = DATA / "lan-pending.json"
    if not path.is_file():
        return {"updated": 0, "people": []}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        if isinstance(data, dict):
            return data
    except Exception:
        pass
    return {"updated": 0, "people": []}


def _queue_tool_on_pcs(kind: str, url: str = "", agent_id: str = "") -> int:
    agents = dict(_agents())
    cmd = {
        "id": f"{kind}-{int(time.time())}",
        "type": kind,
        "queued_at": time.time(),
        "url": url,
    }
    n = 0
    for aid, row in agents.items():
        if not isinstance(row, dict):
            continue
        if agent_id and aid != agent_id:
            continue
        cmds = list(row.get("pending_commands") or [])
        cmds.append(cmd)
        row["pending_commands"] = cmds[-8:]
        agents[aid] = row
        n += 1
    if n:
        _save_agents(agents)
    return n


def _atlas_lan_approve(email: str, role: str = "employee", person_id: str = "") -> dict:
    atlas = _atlas_base()
    token = _enroll_token()
    payload = {
        "enroll_token": token,
        "email": email,
        "person_id": person_id,
        "role": role,
    }
    import ssl
    import urllib.request

    data = json.dumps(payload).encode()
    req = urllib.request.Request(
        f"{atlas}/api/cyber-protect/auth/lan-approve",
        data=data,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    with urllib.request.urlopen(req, timeout=30, context=ctx) as resp:
        return json.loads(resp.read().decode("utf-8", errors="replace") or "{}")


def _chat_archive_index() -> dict:
    """Box-local chat archive (pulled from Atlas). Manager/owner-only via portal login."""
    base = DATA / "chat-archive"
    out: dict = {"ok": True, "threads": [], "people": [], "messages": [], "calls": [], "updated_at": 0}
    try:
        snap = base / "snapshot.json"
        if snap.is_file():
            d = json.loads(snap.read_text(encoding="utf-8"))
            out["threads"] = d.get("threads") or []
            out["people"] = d.get("people") or []
            out["calls"] = d.get("calls") or []
            out["thread_members"] = d.get("thread_members") or {}
            out["company_id"] = d.get("company_id") or ""
            out["site"] = d.get("site") or {}
            out["updated_at"] = d.get("updated_at") or 0
        msgs_file = base / "messages.jsonl"
        if msgs_file.is_file():
            rows: list[dict] = []
            for line in msgs_file.read_text(encoding="utf-8", errors="replace").splitlines():
                line = line.strip()
                if not line:
                    continue
                try:
                    m = json.loads(line)
                except Exception:
                    continue
                if m.get("media_path"):
                    ext = str(m.get("media_path") or "").rsplit(".", 1)[-1] or "bin"
                    m["media_src"] = f"/api/companion/chat-archive/media/{m.get('id')}.{ext}"
                rows.append(m)
            out["messages"] = rows[-2000:]
        for c in out.get("calls") or []:
            for rec in c.get("recordings") or []:
                fname = str(rec.get("url") or "").rsplit("/", 1)[-1]
                if fname:
                    rec["local_src"] = f"/api/companion/chat-archive/media/{fname}"
    except Exception as e:
        out["error"] = str(e)[:200]
    return out


def _companion_public(a: dict) -> dict:
    """Rich PC agent card for portal (AV + software, no secrets)."""
    inv = a.get("inventory") if isinstance(a.get("inventory"), dict) else {}
    av = inv.get("antivirus") if isinstance(inv.get("antivirus"), dict) else {}
    primary = av.get("primary") if isinstance(av.get("primary"), dict) else {}
    defender = av.get("defender") if isinstance(av.get("defender"), dict) else inv.get("defender") or {}
    software = inv.get("software") if isinstance(inv.get("software"), dict) else {}
    programs = software.get("programs") if isinstance(software.get("programs"), list) else []
    last_scan = inv.get("last_scan") if isinstance(inv.get("last_scan"), dict) else a.get("last_scan") or {}
    man = _load_agent_manifest()
    remote_ver = str(man.get("version") or "")
    local_ver = str(inv.get("agent_version") or "")[:32]
    update_available = False
    try:

        def parts(v: str):
            out = []
            for p in (v or "0").split("."):
                out.append(int("".join(c for c in p if c.isdigit()) or "0"))
            return out

        update_available = bool(remote_ver) and parts(remote_ver) > parts(local_ver or "0")
    except Exception:
        update_available = False
    cust = _customer_info()
    age = time.time() - float(a.get("last_seen") or 0)
    online = age < 180
    return {
        "agent_id": a.get("agent_id"),
        "name": a.get("name"),
        "last_seen": a.get("last_seen"),
        "online": online,
        "remote_addr": a.get("remote_addr") or inv.get("lan_ip") or "",
        "lan_ip": inv.get("lan_ip") or a.get("remote_addr") or "",
        "os": str(inv.get("os") or "")[:80],
        "hostname": str(inv.get("hostname") or "")[:80],
        "agent_version": local_ver,
        "latest_version": remote_ver,
        "update_available": update_available,
        "customer": cust,
        "safety": a.get("safety") or {},
        "primary_av": str(primary.get("name") or inv.get("primary_av") or "")[:120],
        "av_engine": str(primary.get("engine") or inv.get("av_engine") or "none")[:40],
        "av_can_scan": bool(primary.get("can_scan")),
        "defender_realtime": bool(defender.get("RealTimeProtectionEnabled")),
        "defender_enabled": bool(defender.get("AntivirusEnabled")),
        "software_count": int(software.get("count") or len(programs) or 0),
        "programs": [
            {
                "name": str(p.get("name") or "")[:120],
                "version": str(p.get("version") or "")[:40],
                "publisher": str(p.get("publisher") or "")[:80],
            }
            for p in programs[:40]
            if isinstance(p, dict)
        ],
        "last_scan": {
            "ok": last_scan.get("ok"),
            "engine": last_scan.get("engine"),
            "name": last_scan.get("name"),
            "note": str(last_scan.get("note") or "")[:240],
            "at": last_scan.get("at"),
            "threats_seen": list(last_scan.get("threats_seen") or [])[:12],
        }
        if last_scan
        else {},
        "atlas_connect_ready": bool(inv.get("atlas_connect_ready")),
        "atlas_connect_reason": str(inv.get("atlas_connect_reason") or "")[:160],
        "pending_commands": len(a.get("pending_commands") or []),
        "paired_at": a.get("paired_at"),
    }


def _learn_endpoint(agent_row: dict, inv: dict) -> None:
    """Append endpoint facts so Atlas/guardian can learn what PCs are doing."""
    try:
        DATA.mkdir(parents=True, exist_ok=True)
        primary = {}
        av = inv.get("antivirus") if isinstance(inv.get("antivirus"), dict) else {}
        if isinstance(av.get("primary"), dict):
            primary = av.get("primary") or {}
        software = inv.get("software") if isinstance(inv.get("software"), dict) else {}
        lesson = {
            "ts": time.time(),
            "agent_id": agent_row.get("agent_id"),
            "name": agent_row.get("name"),
            "hostname": inv.get("hostname"),
            "os": inv.get("os"),
            "av_engine": primary.get("engine") or inv.get("av_engine"),
            "av_name": primary.get("name") or inv.get("primary_av"),
            "software_count": (software.get("count") if isinstance(software, dict) else None),
            "agent_version": inv.get("agent_version"),
            "lesson": (
                f"PC {(agent_row.get('name') or inv.get('hostname') or 'unknown')} uses "
                f"{primary.get('name') or 'unknown AV'} ({primary.get('engine') or 'none'}); "
                f"software inventory {(software.get('count') if isinstance(software, dict) else 0)} programs. "
                "Box watches LAN; agent watches endpoint via installed AV."
            ),
        }
        with ENDPOINT_LEARN.open("a", encoding="utf-8") as fh:
            fh.write(json.dumps(lesson, ensure_ascii=False) + "\n")
        # Keep a rolling summary on state for guardian
        st = _load_state()
        lessons = list(st.get("endpoint_lessons") or [])
        lessons.insert(0, lesson)
        st["endpoint_lessons"] = lessons[:80]
        # Light bump to site learning when endpoints report
        pct = float(st.get("learning_pct") or 0)
        if pct < 100:
            st["learning_pct"] = min(100.0, pct + 0.15)
        _save_state(st)
    except Exception as exc:
        print(f"portal: endpoint learn skip: {exc}", flush=True)


def _db() -> sqlite3.Connection:
    DATA.mkdir(parents=True, exist_ok=True)
    con = sqlite3.connect(str(DB), timeout=30)
    con.row_factory = sqlite3.Row
    con.executescript(
        """
        CREATE TABLE IF NOT EXISTS users (
          id TEXT PRIMARY KEY,
          email TEXT NOT NULL UNIQUE,
          name TEXT NOT NULL DEFAULT '',
          pass_hash TEXT NOT NULL,
          pass_salt TEXT NOT NULL,
          created_at REAL NOT NULL
        );
        CREATE TABLE IF NOT EXISTS decisions (
          finding_key TEXT PRIMARY KEY,
          decision TEXT NOT NULL,
          detail TEXT NOT NULL DEFAULT '',
          updated_at REAL NOT NULL
        );
        """
    )
    con.commit()
    return con


def _hash_pw(password: str, salt: str) -> str:
    return hashlib.pbkdf2_hmac(
        "sha256", password.encode("utf-8"), salt.encode("utf-8"), 120_000
    ).hex()


def _user_count() -> int:
    with _db() as con:
        return int(con.execute("SELECT COUNT(*) FROM users").fetchone()[0])


def _get_user(email: str):
    with _db() as con:
        return con.execute(
            "SELECT * FROM users WHERE email=?", (email.strip().casefold(),)
        ).fetchone()


def _atlas_register_manager(email: str, password: str, name: str) -> dict:
    """After first portal account: also create Atlas collab manager + employee file under business folder."""
    import json
    import urllib.request

    cust = _customer_info()
    business = (cust.get("business_name") or "").strip()
    payload = {
        "email": email,
        "password": password,
        "name": name,
        "role": "manager",
        "business_name": business,
        "source": "portal_first_register",
        "hostname": socket.gethostname(),
    }
    funnel = (os.environ.get("ATLAS_FUNNEL") or _atlas_base()).rstrip("/")
    try:
        req = urllib.request.Request(
            f"{funnel}/api/cyber-protect/auth/setup-account",
            data=json.dumps(payload).encode("utf-8"),
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        ctx = None
        try:
            import ssl

            ctx = ssl._create_unverified_context()
        except Exception:
            pass
        with urllib.request.urlopen(req, context=ctx, timeout=25) as resp:
            return json.loads(resp.read().decode("utf-8", errors="replace") or "{}")
    except Exception as exc:
        return {"ok": False, "error": str(exc)[:200]}


def _create_user(email: str, password: str, name: str) -> dict:
    email = email.strip().casefold()
    if "@" not in email or len(password) < 8:
        raise ValueError("Valid email and password (8+ chars) required")
    if _get_user(email):
        raise ValueError("Account already exists — log in")
    salt = secrets.token_hex(16)
    uid = "u_" + secrets.token_hex(8)
    with _db() as con:
        if int(con.execute("SELECT COUNT(*) FROM users").fetchone()[0]) > 0:
            raise ValueError("This Protect box already has an account — log in")
        con.execute(
            "INSERT INTO users(id,email,name,pass_hash,pass_salt,created_at) VALUES (?,?,?,?,?,?)",
            (uid, email, name.strip()[:80], _hash_pw(password, salt), salt, time.time()),
        )
        con.commit()
    user = {"id": uid, "email": email, "name": name.strip()[:80]}
    # First portal account = first manager — sync to Atlas company folder
    try:
        atlas = _atlas_register_manager(email, password, name.strip()[:80])
        user["atlas_collab"] = atlas
    except Exception as exc:
        user["atlas_collab"] = {"ok": False, "error": str(exc)[:120]}
    return user


class LoginError(Exception):
    """User-facing login failure (wrong password vs Atlas unreachable)."""

    def __init__(self, message: str, *, status: int = 401):
        super().__init__(message)
        self.status = status


def _try_owner_login(email: str, password: str) -> dict | None:
    """Master FN.nology owner can open any Protect box via Atlas credentials."""
    email = (email or "").strip().casefold()
    owner_email = "fn.nology@gmail.com"
    try:
        from share_policy import FACTORY_OWNER_EMAIL

        owner_email = FACTORY_OWNER_EMAIL.strip().casefold() or owner_email
    except Exception:
        pass
    if email != owner_email:
        return None
    atlas = _atlas_base()
    kit = (_load_env().get("FIELD_KIT_KEY") or os.environ.get("FIELD_KIT_KEY") or "").strip()
    payload = {
        "email": email,
        "password": password,
        "field_kit_key": kit,
    }
    try:
        import ssl
        import urllib.error
        import urllib.request

        data = json.dumps(payload).encode()
        req = urllib.request.Request(
            f"{atlas}/api/cyber-protect/owner-login",
            data=data,
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        with urllib.request.urlopen(req, timeout=45, context=ctx) as resp:
            out = json.loads(resp.read().decode("utf-8", errors="replace") or "{}")
        if out.get("ok") and out.get("master"):
            return {
                "id": "owner_master",
                "email": out.get("email") or email,
                "name": out.get("name") or "FN.nology owner",
                "master": True,
            }
        return None
    except urllib.error.HTTPError as exc:
        detail = ""
        try:
            detail = (exc.read() or b"").decode("utf-8", errors="replace")[:200]
        except Exception:
            detail = ""
        print(f"owner-login HTTP {exc.code}: {detail}", flush=True)
        # Wrong password / not owner → fall through to local accounts
        if exc.code in (401, 403):
            return None
        raise LoginError(
            "Atlas could not verify the owner login right now. Try again in a moment.",
            status=503,
        ) from exc
    except Exception as exc:
        print(f"owner-login failed: {exc}", flush=True)
        raise LoginError(
            "Cannot reach Atlas to verify FN.nology owner login. "
            "Use the Protect HTTPS link (…ts.net) and check internet on this box.",
            status=503,
        ) from exc


def _check_login(email: str, password: str):
    # Master owner first — works on every Protect box
    owner = _try_owner_login(email, password)
    if owner:
        return owner
    row = _get_user(email)
    if not row:
        return None
    if not hmac.compare_digest(row["pass_hash"], _hash_pw(password, row["pass_salt"])):
        return None
    # NOTE: we deliberately do NOT push this password to the Atlas collab
    # account. An earlier auto-sync overwrote a user's funnel password on every
    # box login, which caused "sign in rejected" in the agent app (the owner
    # reported the box changing their password without consent). Passwords are
    # only ever changed by the user (agent app / forgot-password) or by an
    # explicit manager/owner reset. The funnel collab account stays the source
    # of truth.
    return dict(row)


def _verify_collab_login(email: str, password: str) -> dict | None:
    """Verify an Atlas collab manager login against the funnel. Returns the
    collab person + token when the account exists and is approved."""
    try:
        import ssl
        import urllib.request

        payload = json.dumps({"email": email, "password": password}).encode()
        req = urllib.request.Request(
            f"{_atlas_base().rstrip('/')}/api/cyber-protect/auth/login",
            data=payload,
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        with urllib.request.urlopen(req, timeout=25, context=ctx) as resp:
            out = json.loads(resp.read().decode("utf-8", errors="replace") or "{}")
        if not out.get("ok") or not out.get("token"):
            return None
        person = out.get("person") or {}
        return {
            "token": out.get("token"),
            "person": person,
            "access": out.get("access") or {},
            "role": str(person.get("role") or "employee"),
        }
    except Exception as exc:
        print(f"collab verify failed: {exc}", flush=True)
        return None


def _new_session(user: dict) -> str:
    tok = secrets.token_urlsafe(32)
    SESSIONS[tok] = {
        "user_id": user["id"],
        "email": user["email"],
        "name": user.get("name") or "",
        "master": bool(user.get("master")),
        "collab": bool(user.get("collab")),
        "collab_token": user.get("collab_token") or "",
        "collab_role": user.get("collab_role") or "",
        "collab_person_id": user.get("collab_person_id") or "",
        "exp": time.time() + SESS_TTL,
    }
    _sessions_save()
    return tok


def _session(handler: BaseHTTPRequestHandler):
    cookie = handler.headers.get("Cookie") or ""
    for part in cookie.split(";"):
        part = part.strip()
        if part.startswith("acp_sess="):
            tok = part.split("=", 1)[1].strip()
            sess = SESSIONS.get(tok)
            if sess and float(sess.get("exp") or 0) > time.time():
                return tok, sess
    return None, None


def _load_state() -> dict:
    if STATE.is_file():
        try:
            return json.loads(STATE.read_text(encoding="utf-8"))
        except Exception:
            pass
    return {}


def _link_guard_hits() -> list:
    """Hits live in a separate file so agent heartbeats cannot wipe them."""
    try:
        import link_guard as lg

        return list(lg.load_hits(50))
    except Exception:
        return list((_load_state().get("link_guard_hits") or []))


def _remote_install_configured() -> bool:
    try:
        import remote_install as ri

        c = ri.load_creds()
        return bool(c.get("username") and c.get("password"))
    except Exception:
        return False


def _public_invites() -> list:
    try:
        import remote_install as ri

        rows = []
        for r in ri.load_invites()[:30]:
            rows.append(
                {
                    "token": r.get("token"),
                    "ip": r.get("ip"),
                    "mac": r.get("mac"),
                    "hostname": r.get("hostname"),
                    "status": r.get("status"),
                    "created_at": r.get("created_at"),
                    "delivery": r.get("delivery"),
                    "delivery_detail": r.get("delivery_detail"),
                    "install_url": f"/install/{r.get('token')}",
                }
            )
        return rows
    except Exception:
        return []


def _lan_ip_guess() -> str:
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.connect(("1.1.1.1", 80))
        ip = s.getsockname()[0]
        s.close()
        return ip
    except Exception:
        return "192.168.111.141"


# --------------------------------------------------------------------------
# Atlas Remote relay — frames from agents, input queue for managers.
# The agent posts its latest screen frame here; the manager's live-view
# fetches it. Manager input events are queued per agent and drained by the
# agent's RemoteLoop. All in-memory (frames are transient by nature).
# --------------------------------------------------------------------------
REMOTE_FRAMES: dict[str, dict] = {}   # agent_id -> {"png": bytes, "w": int, "h": int, "rw": int, "rh": int, "ts": float}
REMOTE_INPUTS: dict[str, list] = {}   # agent_id -> [input events]
REMOTE_VIEWERS: dict[str, float] = {}  # agent_id -> last time a manager fetched a frame
REMOTE_LOCK = threading.Lock()
REMOTE_TTL = 60  # drop stale frames after 60s
REMOTE_VIEW_TTL = 4  # a manager counts as "viewing" if a frame was fetched in the last 4s


def _remote_auth(self, agent_id: str) -> bool:
    """Agent auth: the posted agent token must match the paired agent row."""
    import hmac

    tok = str(self.headers.get("X-Agent-Token") or "").strip()
    if not tok or not agent_id:
        return False
    try:
        row = _get_agent(agent_id)
        expected = str((row or {}).get("agent_token") or "")
        return bool(expected) and hmac.compare_digest(tok, expected)
    except Exception:
        return False


def _remote_cleanup() -> None:
    now = time.time()
    for aid in [a for a, f in REMOTE_FRAMES.items() if now - float(f.get("ts") or 0) > REMOTE_TTL]:
        REMOTE_FRAMES.pop(aid, None)
    for aid in [a for a, ts in REMOTE_VIEWERS.items() if now - float(ts or 0) > REMOTE_VIEW_TTL * 4]:
        REMOTE_VIEWERS.pop(aid, None)


def _load_agent_manifest() -> dict:
    AGENT_RELEASES.mkdir(parents=True, exist_ok=True)
    if AGENT_MANIFEST.is_file():
        try:
            return json.loads(AGENT_MANIFEST.read_text(encoding="utf-8"))
        except Exception:
            pass
    # Default empty channel — publish a build to enable auto-update
    return {
        "ok": True,
        "product": "atlas-cyber-protect-agent",
        "version": "0.2.0",
        "channel": "stable",
        "sha256": "",
        "url": "",
        "changelog": "Initial observe-only agent channel. Drop a signed EXE + manifest to push features.",
        "mandatory": False,
        "safety": {
            "mode": "observe_report_only",
            "never_delete_files": True,
            "never_block_network": True,
        },
    }


def _agent_update_available(client_version: str) -> dict:
    man = _load_agent_manifest()
    remote = str(man.get("version") or "")
    if not man.get("sha256") or not man.get("url"):
        return {"available": False, "version": remote}
    try:
        def parts(v: str):
            out = []
            for p in (v or "0").split("."):
                out.append(int("".join(c for c in p if c.isdigit()) or "0"))
            return out

        available = parts(remote) > parts(client_version or "0")
    except Exception:
        available = False
    fname = str(man.get("filename") or "")
    url = str(man.get("url") or "")
    if fname and not url:
        url = f"/api/companion/agent/download/{fname}"
    return {
        "available": bool(available),
        "version": remote,
        "changelog": man.get("changelog") or "",
        "channel": man.get("channel") or "stable",
        "filename": fname,
        "url": url,
        "sha256": man.get("sha256") or "",
        "silent": True,
    }


def _save_state(st: dict) -> None:
    DATA.mkdir(parents=True, exist_ok=True)
    STATE.write_text(json.dumps(st), encoding="utf-8")


# —— Friendly device names — managers rename LAN devices so the office list is
# easy to read ("Front desk PC", "Till printer" …). Aliases live in their own
# file so agent rescans cannot wipe them; keyed by MAC (fallback IP). ——
DEVICE_ALIASES = DATA / "device-aliases.json"
# Speedtest state lives in its own file — the agent rewrites the shared
# state.json every loop, which would wipe an in-flight result otherwise.
SPEEDTEST_STATE = DATA / "speedtest-state.json"


def _speedtest_load() -> dict:
    try:
        if SPEEDTEST_STATE.is_file():
            return json.loads(SPEEDTEST_STATE.read_text(encoding="utf-8"))
    except Exception:
        pass
    return {}


def _speedtest_save(st: dict) -> None:
    try:
        DATA.mkdir(parents=True, exist_ok=True)
        SPEEDTEST_STATE.write_text(json.dumps(st), encoding="utf-8")
    except Exception:
        pass


def _device_aliases() -> dict[str, str]:
    if DEVICE_ALIASES.is_file():
        try:
            raw = json.loads(DEVICE_ALIASES.read_text(encoding="utf-8"))
            if isinstance(raw, dict):
                return {str(k): str(v).strip()[:60] for k, v in raw.items() if str(v).strip()}
        except Exception:
            pass
    st = _load_state()
    raw = st.get("device_aliases")
    if isinstance(raw, dict):
        return {str(k): str(v).strip()[:60] for k, v in raw.items() if str(v).strip()}
    return {}


def _save_device_aliases(aliases: dict[str, str]) -> None:
    try:
        DATA.mkdir(parents=True, exist_ok=True)
        tmp = DEVICE_ALIASES.with_suffix(".json.tmp")
        tmp.write_text(json.dumps(aliases, ensure_ascii=False, indent=1), encoding="utf-8")
        tmp.replace(DEVICE_ALIASES)
    except Exception as exc:
        print(f"device aliases save failed: {exc}", flush=True)
    st = _load_state()
    st["device_aliases"] = aliases
    _save_state(st)


def _apply_device_aliases(devices: list) -> list:
    """Overlay manager-set friendly names onto the scanned device list."""
    aliases = _device_aliases()
    if not aliases or not devices:
        return devices
    for d in devices:
        if not isinstance(d, dict):
            continue
        key = str(d.get("mac") or d.get("ip") or "").strip()
        if key and key in aliases:
            d["display_name"] = aliases[key]
    return devices


def _decided_keys() -> dict[str, str]:
    out: dict[str, str] = {}
    with _db() as con:
        for row in con.execute("SELECT finding_key, decision FROM decisions"):
            out[str(row["finding_key"])] = str(row["decision"])
    st = _load_state()
    for k, v in (st.get("decisions") or {}).items():
        if isinstance(v, dict):
            out.setdefault(k, str(v.get("decision") or ""))
        else:
            out.setdefault(k, str(v))
    return out


def _recompute_threat_summary(st: dict, findings: list) -> None:
    """Keep the served threat counts in lock-step with the Threats tab list.

    The tab shows exactly `open_findings` minus decided keys; the shield and
    the Guardian lines must count the SAME list, otherwise the number can say
    "21 open · 3 critical · 5 high" while the tab is empty (stale summary that
    was never re-aligned when decisions were made).
    """
    try:
        prev = st.get("threat_summary") or {}
        st["threat_summary"] = {
            "checked_at": prev.get("checked_at") or time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "findings": len(findings),
            "critical": sum(1 for f in findings if f.get("severity") == "critical"),
            "high": sum(1 for f in findings if f.get("severity") == "high"),
            "medium": sum(1 for f in findings if f.get("severity") == "medium"),
            "devices_with_ports": prev.get("devices_with_ports") or 0,
            "raw_signals": prev.get("raw_signals") or len(findings),
            "checks": prev.get("checks") or ["risky_ports", "new_devices", "gateway_mac", "inbound_probes", "internet_path"],
        }
    except Exception:
        pass


def _status_payload(sess: dict | None) -> dict:
    st = _load_state()
    locked = LOCK_FLAG.is_file() or bool(st.get("subscription_locked"))
    pct = float(st.get("learning_pct") or 0)
    devices = _apply_device_aliases(list(st.get("last_devices") or []))
    decided = _decided_keys()
    findings = [
        f
        for f in (st.get("open_findings") or [])
        if (f.get("key") or "") not in decided
    ]
    # Single source of truth: serve the SAME counts as the tab list.
    _recompute_threat_summary(st, findings)
    try:
        import ui_guardian

        ui_guardian.live_feed(st)
        _save_state(st)
    except Exception:
        pass
    jobs = st.get("jobs") or [
        {"id": "learn", "label": "Learning normal traffic", "pct": pct},
        {"id": "scan", "label": "Checking devices on your network", "pct": min(100, pct + 5)},
    ]
    if findings:
        jobs = list(jobs) + [
            {"id": "inspect", "label": "Inspecting issues for your safety", "pct": 55}
        ]
    needs = []
    for f in findings:
        key = f.get("key") or f.get("id") or f.get("plain_tip", "")[:40]
        needs.append(
            {
                "key": key,
                "title": f.get("title") or f.get("plain_tip") or "Something needs a look",
                "detail": f.get("plain_tip") or f.get("tech_detail") or "",
                "state": f.get("customer_state") or "checked",
                "category": f.get("category") or "security",
                "severity": f.get("severity") or ("high" if f.get("category") == "threat" else "low"),
            }
        )
    pulse = st.get("guardian_pulse") or {
        "score": 50,
        "label": "Starting",
        "tone": "sky",
        "devices": len(devices) if isinstance(devices, list) else 0,
        "reachable": 0,
        "open_issues": len(needs),
        "learning_pct": pct,
        "internet_ok": bool(st.get("internet_ok", True)),
    }
    headline = "Network protected"
    sub = st.get("guardian_pulse", {}).get("label") or ""
    if locked:
        headline = "Subscription paused"
        sub = "Contact FN.nology to restore Atlas Cyber Protect."
    elif needs:
        headline = "Network protected"
        sub = "Issues found — being inspected for future fixes for your safety"
    elif not locked:
        headline = "Network protected"
        sub = sub or "Guardian is live on this Protect box"

    authed = bool(sess)
    collab = None
    if sess and sess.get("collab"):
        collab = {
            "role": sess.get("collab_role") or "manager",
            "token": sess.get("collab_token") or "",
            "person_id": sess.get("collab_person_id") or "",
        }
    return {
        "ok": True,
        "locked": locked,
        "authenticated": authed,
        "needs_register": _user_count() == 0,
        "user": (
            {
                "email": sess.get("email"),
                "name": sess.get("name"),
                "master": bool(sess.get("master")),
                "role": "owner" if sess.get("master") else "manager",
            }
            if sess
            else None
        ),
        "collab": collab,
        "headline": headline,
        "subline": sub,
        "learning_pct": pct,
        "learning_state": st.get("learning_state") or "learning",
        "device_count": len(devices) if isinstance(devices, list) else int(st.get("device_count") or 0),
        "devices": (devices if isinstance(devices, list) else []) if authed else [],
        "scan_meta": (st.get("scan_meta") or {}) if authed else {},
        "jobs": jobs if authed else [],
        "needs_decision": needs if (authed and not locked) else [],
        "activity": (st.get("activity") or []) if authed else [],
        "internet_ok": bool(st.get("internet_ok", True)),
        "pulse": pulse if authed else {},
        "shields": (st.get("guardian_shields") or []) if authed else [],
        "guardian_feed": (st.get("guardian_feed") or [])[:20] if authed else [],
        "network_health": (st.get("network_health") or {}) if authed else {},
        "services_summary": (st.get("services_summary") or []) if authed else [],
        "device_events": (st.get("device_events") or [])[:30] if authed else [],
        "capabilities": (st.get("capabilities") or []) if authed else [],
        "threat_summary": (st.get("threat_summary") or {}) if authed else {},
        "companion_agents": (
            [_companion_public(a) for a in list(_agents().values())]
            if authed
            else []
        ),
        "endpoint_lessons": (st.get("endpoint_lessons") or [])[:12] if authed else [],
        "link_guard_hits": (_link_guard_hits()[:12] if authed else []),
        "pc_agent_download": "https://atlas-server.taile9cc75.ts.net/releases/public/windows/AtlasCyberProtectAgent-Setup-latest.exe",
        "agent_latest_version": str((_load_agent_manifest() or {}).get("version") or ""),
        "customer": _customer_info() if authed else {},
        "install_invites": (_public_invites() if authed else []),
        "remote_install_configured": _remote_install_configured() if authed else False,
        "gateway_secure": (_gateway_secure_summary() if authed else {}),
        "gateway": _reyee_public_safe() if authed else {},
        "portal_version": str(st.get("portal_version") or "—"),
        "portal_update": st.get("portal_update_info") or {},
        "server_time": time.time(),
    }


def _reyee_public_safe() -> dict:
    """Ruijie/ReyeeOS gateway snapshot (read-only) — never breaks the portal."""
    try:
        import infra_secure as infra

        out = infra.reyee_public()
        return out if isinstance(out, dict) else {}
    except Exception:
        return {}


def _gateway_secure_summary() -> dict:
    try:
        import infra_secure as infra

        st = infra.public_status()
        inv = infra.inventory_public()
        st["inventory"] = {
            "summary": inv.get("summary") or {},
            "count": len(inv.get("items") or []),
            "last_scan_at": inv.get("checked_at") or "",
            "needs_scan": bool(inv.get("needs_scan")),
        }
        # Preview tips from last full scan cache if present
        tips = []
        for it in (inv.get("items") or [])[:4]:
            for t in (it.get("tips") or [])[:1]:
                tips.append(t)
        st["tips_preview"] = tips[:6]
        return st
    except Exception as exc:
        return {"ok": False, "configured": False, "error": str(exc)[:160]}


# --------------------------------------------------------------------------
# Live data-flow + speed test (customer dashboard).
# The agent samples directional traffic (down/up × internet/LAN); here we
# serve it live, and run a real internet + LAN speed test on demand.
# --------------------------------------------------------------------------
SPEEDTEST_LOCK = threading.Lock()


def _dataflow_payload() -> dict:
    st = _load_state()
    rates = st.get("flow_rates") or {}
    totals = st.get("flow") or {}
    devs = {str(d.get("mac") or ""): d for d in (st.get("last_devices") or [])}
    per_device = []
    for mac, ft in sorted(
        (st.get("flow_totals") or {}).items(),
        key=lambda kv: int(kv[1].get("down") or 0) + int(kv[1].get("up") or 0),
        reverse=True,
    ):
        d = devs.get(mac) or {}
        per_device.append(
            {
                "name": d.get("display_name") or d.get("hostname") or d.get("vendor") or d.get("ip") or mac,
                "ip": d.get("ip") or "",
                "down_rate": float(ft.get("down_rate") or 0),
                "up_rate": float(ft.get("up_rate") or 0),
                "down": int(ft.get("down") or 0),
                "up": int(ft.get("up") or 0),
            }
        )
        if len(per_device) >= 8:
            break
    live = {}
    try:
        f = DATA / "flow-live.json"
        if f.is_file():
            live = json.loads(f.read_text(encoding="utf-8", errors="replace"))
    except Exception:
        live = {}
    out = {
        "rates": (live.get("rates") or {}) if live else rates,
        "totals": totals,
        "per_device": per_device,
        "sampled_at": ((live.get("rates") or {}).get("sampled_at") if live else rates.get("sampled_at")) or time.time(),
        "server_time": time.time(),
        # Real speedtest state lives in speedtest-state.json (owned by this
        # portal process) — the agent never writes it to its own state, so the
        # old st.get("speedtest") was always empty and results "disappeared"
        # after the button's own poll loop moved on.
        "speedtest": _speedtest_load(),
        "gateway": _reyee_public_safe(),
        "public_ip": _public_ip_safe(),
    }
    if live:
        out["history"] = (live.get("history") or [])[-45:]
        out["wan_total"] = live.get("wan_total") or {}
        out["gateway_counters"] = live.get("gateway_counters") or {}
    try:
        logf = DATA / "network-events.jsonl"
        if logf.is_file():
            lines = logf.read_text(encoding="utf-8", errors="replace").splitlines()
            log_rows = []
            for ln in lines[-120:]:
                try:
                    d = json.loads(ln)
                    if isinstance(d, dict):
                        log_rows.append(d)
                except Exception:
                    continue
            if log_rows:
                out["logs"] = log_rows
    except Exception:
        pass
    return out


def _speedtest_ping(atlas: str) -> float:
    """Latency to Atlas — TTFB on a tiny blob (no transfer-time skew)."""
    t0 = time.time()
    try:
        req = urllib.request.Request(
            f"{atlas}/api/cyber-protect/speedtest/blob?mb=0.02",
            headers={"User-Agent": "AtlasCyberProtect/speedtest"},
            method="GET",
        )
        with urllib.request.urlopen(req, timeout=8) as r:
            r.read(1)
        return round((time.time() - t0) * 1000, 1)
    except Exception:
        return 0.0


def _speedtest_download(atlas: str) -> dict:
    """Download throughput vs Atlas VPS — 3 rounds of 5 MB, keep the best."""
    best = 0.0
    for _ in range(3):
        t0 = time.time()
        total = 0
        try:
            with urllib.request.urlopen(f"{atlas}/api/cyber-protect/speedtest/blob?mb=5", timeout=40) as r:
                while True:
                    chunk = r.read(256 * 1024)
                    if not chunk:
                        break
                    total += len(chunk)
            secs = max(0.001, time.time() - t0)
            best = max(best, (total * 8) / (secs * 1_000_000))
        except Exception:
            break
    return {"download_mbps": round(best, 2)}


def _speedtest_upload(atlas: str) -> dict:
    """Upload throughput vs Atlas VPS (~3 MB random blob)."""
    t0 = time.time()
    try:
        blob = secrets.token_bytes(3 * 1024 * 1024)
        req = urllib.request.Request(
            f"{atlas}/api/cyber-protect/speedtest/upload",
            data=blob,
            method="POST",
            headers={"Content-Type": "application/octet-stream"},
        )
        urllib.request.urlopen(req, timeout=60).read()
        secs = max(0.001, time.time() - t0)
        return {"upload_mbps": round((len(blob) * 8) / (secs * 1_000_000), 2)}
    except Exception as exc:
        return {"upload_mbps": 0.0, "note": f"upload: {str(exc)[:120]}"}


def _speedtest_internet(atlas: str) -> dict:
    """Real download + upload throughput against the Atlas VPS (no LAN leg)."""
    return {"ping_ms": _speedtest_ping(atlas), **_speedtest_download(atlas), **_speedtest_upload(atlas)}


def _speedtest_lan(st: dict) -> dict:
    """LAN throughput: a paired PC agent downloads/uploads a blob from/to the
    box over the LAN. Results arrive via the agent's next heartbeat."""
    res = {"lan_download_mbps": 0.0, "lan_upload_mbps": 0.0, "note": ""}
    try:
        agents = {aid: row for aid, row in dict(_agents()).items() if isinstance(row, dict)}
        if not agents:
            res["note"] = "no paired PCs for a LAN test"
            return res
        # Prefer agents whose last contact came from a LAN address.
        cands = sorted(
            agents.keys(),
            key=lambda aid: (
                0
                if str(agents[aid].get("remote_addr") or "").startswith(
                    ("192.168.", "10.", "172.16.", "172.17.", "172.18.", "172.19.", "172.2", "172.30.", "172.31.")
                )
                else 1,
                -float(agents[aid].get("last_seen") or 0),
            ),
        )
        aid = cands[0]
        token = secrets.token_hex(8)
        mb = 2.0
        st["lan_speed_token"] = {"token": token, "expires": time.time() + 240}
        _save_state(st)
        base = f"http://{_lan_ip_guess()}:{PORT}"
        cmd = {
            "id": f"lan-speed-{int(time.time())}",
            "type": "lan_speed",
            "url": f"{base}/api/lan-speed/blob?token={token}&mb={mb}",
            "upload_url": f"{base}/api/lan-speed/upload?token={token}&mb={mb}",
            "mb": mb,
        }
        agents[aid]["pending_commands"] = list(agents[aid].get("pending_commands") or []) + [cmd]
        _save_agents(agents)
        # Wait for the agent to run it and report back (heartbeat cadence).
        deadline = time.time() + 75
        while time.time() < deadline:
            time.sleep(4)
            row = (dict(_agents()) or {}).get(aid) or {}
            for cr in (row.get("last_command_results") or [])[:8]:
                if not isinstance(cr, dict) or str(cr.get("id") or "") != cmd["id"]:
                    continue
                r = cr.get("result") if isinstance(cr.get("result"), dict) else {}
                if r.get("ok"):
                    res["lan_download_mbps"] = round(float(r.get("download_mbps") or 0), 2)
                    res["lan_upload_mbps"] = round(float(r.get("upload_mbps") or 0), 2)
                else:
                    res["note"] = str(r.get("note") or "LAN test failed")
                return res
        res["note"] = "LAN test timed out waiting for the PC"
    except Exception as exc:
        res["note"] = f"LAN: {str(exc)[:120]}"
    return res


def _speedtest_thread() -> None:
    """Full internet speed test, run entirely ON the Protect box (never on a
    customer PC). Live progress is written to speedtest-state.json so the
    dashboard can poll GET /api/speedtest and show it running in real time."""
    try:
        started = float((_speedtest_load().get("started_at") or 0) or time.time())
        atlas = _atlas_base()

        def _set(**kw: object) -> None:
            try:
                s = _speedtest_load()
                s.update({"state": "running", "started_at": started, **kw})
                _speedtest_save(s)
            except Exception:
                pass

        _set(stage="Pinging Atlas…", progress=5, result={})
        ping_ms = _speedtest_ping(atlas)
        _set(stage="Downloading…", progress=20, result={"ping_ms": ping_ms})
        dl = _speedtest_download(atlas)
        _set(stage="Uploading…", progress=70, result={"ping_ms": ping_ms, **dl})
        up = _speedtest_upload(atlas)
        result = {"ping_ms": ping_ms, **dl, **up, "finished_at": time.time()}
        _speedtest_save(
            {"state": "done", "stage": "", "progress": 100, "started_at": started, "result": result}
        )
        # Report to Atlas so the funnel dashboard shows the same numbers.
        try:
            payload = {"enroll_token": _enroll_token(), **result}
            req = urllib.request.Request(
                f"{atlas}/api/cyber-protect/speedtest/result",
                data=json.dumps(payload).encode(),
                headers={"Content-Type": "application/json"},
                method="POST",
            )
            urllib.request.urlopen(req, timeout=25).read()
        except Exception:
            pass
    except Exception as exc:
        try:
            _speedtest_save(
                {"state": "error", "stage": "", "started_at": time.time(), "result": {"error": str(exc)[:160]}}
            )
        except Exception:
            pass


def _public_ip_safe() -> dict:
    """Cached web public-IP / ISP info (never does a web call in a request)."""
    try:
        from infra_secure import public_ip_cached

        c = public_ip_cached()
        return c if isinstance(c, dict) else {}
    except Exception:
        return {}


def _start_speedtest() -> bool:
    cur = _speedtest_load()
    if cur.get("state") == "running":
        # A crashed process can leave a stale "running" behind (the agent also
        # rewrites the shared state file). If it started >6 min ago, treat it
        # as dead and allow a fresh run.
        started = float(cur.get("started_at") or 0)
        if started and time.time() - started < 360:
            return False
    t = threading.Thread(target=_speedtest_thread, daemon=True)
    t.start()
    return True


class Handler(BaseHTTPRequestHandler):
    server_version = "AtlasCyberProtect/2.0"

    def log_message(self, fmt: str, *args) -> None:
        print(f"portal: {fmt % args}", flush=True)

    def _request_https(self) -> bool:
        xf = (self.headers.get("X-Forwarded-Proto") or self.headers.get("x-forwarded-proto") or "").lower()
        if xf == "https":
            return True
        host = (self.headers.get("Host") or "").split(":")[0].lower()
        return host.endswith(".ts.net")

    def _json(self, code: int, payload: dict, *, set_cookie: str | None = None) -> None:
        body = json.dumps(payload).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(body.__len__()))
        self.send_header("Cache-Control", "no-store")
        if set_cookie is not None:
            flags = f"Path=/; HttpOnly; SameSite=Lax; Max-Age={SESS_TTL}"
            if self._request_https():
                flags += "; Secure"
            if set_cookie == "":
                self.send_header("Set-Cookie", f"acp_sess=; {flags}")
            else:
                self.send_header("Set-Cookie", f"acp_sess={set_cookie}; {flags}")
        self.end_headers()
        self.wfile.write(body)

    def _html(self, path: Path) -> None:
        data = path.read_bytes() if path.is_file() else b"<h1>Portal missing</h1>"
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(data)

    def _read_json(self) -> dict:
        n = int(self.headers.get("Content-Length") or 0)
        raw = self.rfile.read(n) if n else b"{}"
        try:
            return json.loads(raw.decode() or "{}")
        except Exception:
            return {}

    def _read_raw(self) -> bytes:
        n = int(self.headers.get("Content-Length") or 0)
        return self.rfile.read(n) if n else b""

    def do_GET(self) -> None:  # noqa: N802
        path = urlparse(self.path).path
        if path in {"/", "/index.html", "/login", "/app"}:
            self._html(PORTAL_HTML)
            return
        if path == "/api/atlas-os/sys":
            # Atlas OS system panel — GPU (RX 7800 XT) + RAM + disk + model.
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            gpu = gpu_load = vram_used = None
            try:
                # GPU name from lspci (no rocm-smi installed on some builds)
                out = subprocess.run(
                    ["lspci"], capture_output=True, text=True, timeout=5
                ).stdout
                for ln in out.splitlines():
                    if "VGA" in ln and ("AMD" in ln or "ATI" in ln):
                        gpu = ln.split(":", 1)[-1].strip() or None
                        break
            except Exception:
                gpu = None
            try:
                # Live GPU load: which models are resident + % GPU from ollama ps
                out = subprocess.run(
                    ["ollama", "ps"], capture_output=True, text=True, timeout=8
                ).stdout
                rows = [l for l in out.splitlines()[1:] if l.strip()]
                if rows:
                    parts = rows[0].split()
                    if len(parts) >= 4:
                        proc = parts[3]
                        if "%" in proc:
                            try:
                                gpu_load = float(proc.replace("%", "").strip())
                            except Exception:
                                gpu_load = None
                        gpu = f"{gpu or 'GPU'} · {parts[0]}"
                else:
                    gpu = gpu or "GPU idle (no model loaded)"
            except Exception:
                pass
            try:
                with open("/proc/meminfo") as fh:
                    mem = dict(l.split(":", 1) for l in fh.read().splitlines() if ":" in l)
                ram_mb = int(mem.get("MemTotal", "0").split()[0]) // 1024
                avail_mb = int(mem.get("MemAvailable", "0").split()[0]) // 1024
                ram = f"{avail_mb}/{ram_mb} GB free"
            except Exception:
                ram = "—"
            try:
                disk = subprocess.run(
                    ["df", "-h", "/"], capture_output=True, text=True, timeout=5
                ).stdout.splitlines()[-1].split()
                disk = f"{disk[3]} free of {disk[1]}"
            except Exception:
                disk = "—"
            try:
                cpu = subprocess.run(
                    ["nproc"], capture_output=True, text=True, timeout=5
                ).stdout.strip()
            except Exception:
                cpu = "—"
            spec = (os.environ.get("ACP_LOCAL_LLM") or "").strip()
            model = spec.split("ollama://", 1)[1] if spec.startswith("ollama://") else ""
            self._json(
                200,
                {
                    "ok": True,
                    "gpu": gpu,
                    "gpu_load": gpu_load,
                    "vram_used": vram_used,
                    "cpu": cpu and f"{cpu} cores",
                    "ram": ram,
                    "disk": disk,
                    "model": model or "—",
                },
            )
            return
        if path == "/api/home-server/sys":
            # Home Server room — full system snapshot read locally on the box
            # (no SSH needed; the funnel proxies this over HTTP).
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return

            def _sh(cmd: list[str], timeout: int = 8) -> str:
                try:
                    return subprocess.run(
                        cmd, capture_output=True, text=True, timeout=timeout
                    ).stdout.strip()
                except Exception:
                    return ""

            result: dict = {"ok": True}
            try:
                result["hostname"] = _sh(["hostname"]) or "—"
                try:
                    with open("/etc/os-release") as fh:
                        rel = dict(
                            l.split("=", 1) for l in fh.read().splitlines() if "=" in l
                        )
                    result["os"] = (
                        rel.get("PRETTY_NAME", rel.get("NAME", "")).strip('"') or "—"
                    )
                except Exception:
                    result["os"] = "—"
                result["cpu_model"] = _sh(["lscpu"]).split("Model name:")[-1].split("\n")[0].strip() or "—"
                try:
                    result["cpu_cores"] = int(_sh(["nproc"]) or 0)
                except Exception:
                    result["cpu_cores"] = 0
                try:
                    result["uptime"] = int(float(_sh(["cat", "/proc/uptime"]).split()[0]))
                except Exception:
                    result["uptime"] = 0
                result["load"] = _sh(["cat", "/proc/loadavg"]).split()[:3] and " ".join(
                    _sh(["cat", "/proc/loadavg"]).split()[:3]
                )
                # CPU usage from /proc/stat (two samples)
                try:
                    def _cpu_ticks():
                        with open("/proc/stat") as fh:
                            parts = fh.readline().split()
                        nums = [int(x) for x in parts[1:9]]
                        return sum(nums), nums[0], nums[2], nums[3]

                    t1, u1, s1, i1 = _cpu_ticks()
                    time.sleep(0.4)
                    t2, u2, s2, i2 = _cpu_ticks()
                    dt = max(t2 - t1, 1)
                    result["cpu_percent"] = round((dt - (i2 - i1)) / dt * 100, 1)
                    result["cpu_user"] = round((u2 - u1) / dt * 100, 1)
                    result["cpu_sys"] = round((s2 - s1) / dt * 100, 1)
                    result["cpu_idle"] = round((i2 - i1) / dt * 100, 1)
                except Exception:
                    result["cpu_percent"] = result["cpu_user"] = result["cpu_sys"] = result["cpu_idle"] = 0
                # RAM
                try:
                    with open("/proc/meminfo") as fh:
                        mem = dict(
                            l.split(":", 1) for l in fh.read().splitlines() if ":" in l
                        )
                    total = int(mem.get("MemTotal", "0").split()[0]) // 1024
                    avail = int(mem.get("MemAvailable", "0").split()[0]) // 1024
                    used = total - avail
                    result["ram_total"] = f"{total} MB"
                    result["ram_used"] = f"{used} MB"
                    result["ram_avail"] = f"{avail} MB"
                    result["ram_percent"] = round(used / total * 100, 1) if total else 0
                except Exception:
                    pass
                # Disk
                try:
                    disk = _sh(["df", "-m", "/"]).splitlines()[-1].split()
                    if len(disk) >= 5:
                        result["disk_total"] = f"{int(disk[1])} MB"
                        result["disk_used"] = f"{int(disk[2])} MB"
                        result["disk_avail"] = f"{int(disk[3])} MB"
                        result["disk_percent"] = float(disk[4].rstrip("%"))
                except Exception:
                    pass
                # GPU
                gpu = {"status": "idle", "temp": 0, "fan": "auto", "hashrate": "—", "algo": "—"}
                try:
                    busy = _sh(["cat", "/sys/class/drm/card1/device/gpu_busy_percent"])
                    temp = _sh(["cat", "/sys/class/drm/card1/device/hwmon/hwmon*/temp1_input"])
                    gpu["status"] = "running" if (busy or "").strip() and int(busy) > 0 else "idle"
                    gpu["usage"] = int(busy or 0)
                    if temp:
                        gpu["temp"] = round(int(temp) / 1000)
                    for ln in (_sh(["lspci"]) or "").splitlines():
                        if "VGA" in ln and ("AMD" in ln or "ATI" in ln):
                            gpu["name"] = ln.split(":", 1)[-1].strip()
                            break
                except Exception:
                    pass
                try:
                    rows = [l for l in (_sh(["ollama", "ps"]) or "").splitlines()[1:] if l.strip()]
                    if rows:
                        gpu["algo"] = "AI · " + (rows[0].split()[0] if rows[0].split() else "model")
                        gpu["hashrate"] = "GPU AI"
                except Exception:
                    pass
                result["gpu"] = gpu
                # Docker containers
                try:
                    result["docker"] = [
                        {"name": p[0], "running": True, "status": p[1] if len(p) > 1 else ""}
                        for line in (_sh(["docker", "ps", "--format", "{{.Names}}|{{.Status}}"]) or "").splitlines()
                        if (p := line.split("|", 1)) and p[0]
                    ]
                except Exception:
                    result["docker"] = []
                # Key systemd services
                try:
                    raw = _sh(
                        [
                            "systemctl", "list-units", "--type=service", "--state=running",
                            "--no-pager", "--plain",
                        ]
                    )
                    result["services"] = [
                        {"name": ln.split()[0].replace(".service", ""), "active": "running", "description": ""}
                        for ln in (raw or "").splitlines()
                        if ln.strip() and any(k in ln for k in ("atlas-", "ollama", "docker", "nginx", "fan"))
                    ]
                except Exception:
                    result["services"] = []
                result["model"] = (os.environ.get("ACP_LOCAL_LLM") or "").split("ollama://", 1)[-1] or "—"
            except Exception as exc:
                result["error"] = str(exc)
            self._json(200, result)
            return
        if path == "/api/atlas-os/files":
            # File manager — list / read / download files
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            import pathlib
            req_path = qp.get("path", "/")
            base = pathlib.Path("/")
            try:
                target = (base / req_path.lstrip("/")).resolve()
            except Exception:
                self._json(400, {"ok": False, "error": "invalid path"})
                return
            if not str(target).startswith("/"):
                self._json(400, {"ok": False, "error": "invalid path"})
                return
            if target.is_file():
                try:
                    ct = "text/plain"
                    if target.suffix == ".html": ct = "text/html"
                    elif target.suffix == ".json": ct = "application/json"
                    elif target.suffix in (".png", ".jpg", ".jpeg"): ct = "image/png"
                    elif target.suffix == ".css": ct = "text/css"
                    elif target.suffix == ".js": ct = "application/javascript"
                    data = target.read_bytes()
                    self.send_response(200)
                    self.send_header("Content-Type", ct)
                    self.send_header("Content-Length", str(len(data)))
                    self.end_headers()
                    self.wfile.write(data)
                except Exception as exc:
                    self._json(403, {"ok": False, "error": str(exc)[:160]})
                return
            entries = []
            try:
                for child in sorted(target.iterdir(), key=lambda p: (not p.is_dir(), p.name.lower())):
                    if child.name.startswith(".") and req_path == "/":
                        continue
                    try:
                        st = child.stat()
                        entries.append({
                            "name": child.name,
                            "path": str(child),
                            "is_dir": child.is_dir(),
                            "size": st.st_size,
                            "mtime": int(st.st_mtime),
                        })
                    except Exception:
                        continue
            except PermissionError:
                self._json(403, {"ok": False, "error": "permission denied"})
                return
            except FileNotFoundError:
                self._json(404, {"ok": False, "error": "directory not found"})
                return
            self._json(200, {"ok": True, "path": str(target), "entries": entries})
            return
        if path == "/api/atlas-os/processes":
            # Running processes
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            procs = []
            try:
                r = subprocess.run(
                    ["ps", "aux", "--sort=-pcpu"],
                    capture_output=True, text=True, timeout=5,
                )
                lines = r.stdout.strip().splitlines()
                for ln in lines[1:51]:
                    parts = ln.split(None, 10)
                    if len(parts) >= 11:
                        procs.append({
                            "user": parts[0], "pid": parts[1],
                            "cpu": parts[2], "mem": parts[3],
                            "cmd": parts[10][:120],
                        })
            except Exception:
                pass
            self._json(200, {"ok": True, "processes": procs})
            return
        if path == "/api/atlas-os/logs":
            # System logs (journalctl / syslog)
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            source = qp.get("source", "journal")
            lines_n = min(int(qp.get("lines", "100") or 100), 500)
            log_text = ""
            try:
                if source == "dmesg":
                    r = subprocess.run(["dmesg", "--time-format=iso"], capture_output=True, text=True, timeout=5)
                    log_text = "\n".join(r.stdout.strip().splitlines()[-lines_n:])
                elif source == "syslog":
                    r = subprocess.run(["tail", "-n", str(lines_n), "/var/log/syslog"], capture_output=True, text=True, timeout=5)
                    log_text = r.stdout
                elif source == "auth":
                    r = subprocess.run(["tail", "-n", str(lines_n), "/var/log/auth.log"], capture_output=True, text=True, timeout=5)
                    log_text = r.stdout
                elif source == "ollama":
                    r = subprocess.run(["journalctl", "-u", "ollama", "-n", str(lines_n), "--no-pager"], capture_output=True, text=True, timeout=5)
                    log_text = r.stdout
                else:
                    r = subprocess.run(["journalctl", "-n", str(lines_n), "--no-pager"], capture_output=True, text=True, timeout=8)
                    log_text = r.stdout
            except Exception as exc:
                log_text = f"Error reading logs: {exc}"
            self._json(200, {"ok": True, "source": source, "lines": lines_n, "logs": log_text[-12000:]})
            return
        if path == "/api/atlas-os/network-stats":
            # Live network traffic stats
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            stats = {}
            try:
                with open("/proc/net/dev") as fh:
                    for ln in fh.read().splitlines()[2:]:
                        if ":" not in ln:
                            continue
                        iface, data = ln.split(":", 1)
                        iface = iface.strip()
                        parts = data.split()
                        if len(parts) >= 10:
                            stats[iface] = {
                                "rx_bytes": int(parts[0]),
                                "rx_packets": int(parts[1]),
                                "tx_bytes": int(parts[8]),
                                "tx_packets": int(parts[9]),
                            }
            except Exception:
                pass
            # Active connections count
            conn_count = 0
            try:
                r = subprocess.run(["ss", "-tun"], capture_output=True, text=True, timeout=5)
                conn_count = len(r.stdout.strip().splitlines()) - 1
            except Exception:
                pass
            # DNS resolution speed
            dns_ms = None
            try:
                import time as _t
                t1 = _t.monotonic()
                subprocess.run(["nslookup", "google.com"], capture_output=True, timeout=5)
                dns_ms = round((_t.monotonic() - t1) * 1000)
            except Exception:
                pass
            self._json(200, {"ok": True, "interfaces": stats, "connections": conn_count, "dns_ms": dns_ms})
            return
        if path == "/api/atlas-os/speedtest-result":
            # Get the latest speedtest result from the DB
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            result = None
            try:
                with _db() as con:
                    row = con.execute(
                        "SELECT json_data FROM speedtest ORDER BY started_at DESC LIMIT 1"
                    ).fetchone()
                    if row:
                        result = json.loads(row[0] or "{}")
            except Exception:
                pass
            self._json(200, {"ok": True, "result": result})
            return
        if path == "/api/atlas-os/screenshots":
            # List available screenshots in /tmp/atlas-screenshots
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            import pathlib
            import glob as _glob
            shots = []
            try:
                shots_dir = pathlib.Path("/tmp/atlas-screenshots")
                shots_dir.mkdir(exist_ok=True)
                for f in sorted(shots_dir.glob("*.png"), key=lambda p: p.stat().st_mtime, reverse=True)[:50]:
                    shots.append({"name": f.name, "path": str(f), "size": f.stat().st_size, "mtime": int(f.stat().st_mtime)})
            except Exception:
                pass
            self._json(200, {"ok": True, "screenshots": shots})
            return
        if path == "/api/atlas-os/devices":
            # Network devices with online/offline status
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            devices = []
            try:
                with _db() as con:
                    for r in con.execute(
                        "SELECT hostname, vendor, ip, mac, last_seen, bytes_obs, stats_json FROM devices ORDER BY last_seen DESC LIMIT 80"
                    ).fetchall():
                        hostname, vendor, ip, mac, last_seen, bytes_obs, stats_json = r
                        stats = {}
                        try:
                            stats = json.loads(stats_json or "{}")
                        except Exception:
                            pass
                        online = (time.time() - (last_seen or 0)) < 300 if last_seen else False
                        devices.append({
                            "hostname": hostname or "—",
                            "vendor": vendor or "—",
                            "ip": ip or "—",
                            "mac": mac or "—",
                            "online": online,
                            "last_seen": last_seen or 0,
                            "bytes": int(bytes_obs or stats.get("bytes_total") or 0),
                            "connection": stats.get("connection", "unknown"),
                        })
            except Exception:
                pass
            self._json(200, {"ok": True, "devices": devices})
            return
        if path == "/api/atlas-os/ai-memory":
            # AI memory — persistent notes the AI remembers across sessions
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            mem_path = DATA / "atlas-os-memory.json"
            try:
                if mem_path.is_file():
                    notes = json.loads(mem_path.read_text(encoding="utf-8"))
                else:
                    notes = []
            except Exception:
                notes = []
            self._json(200, {"ok": True, "notes": notes[-100:]})
            return
        if path in {"/atlas-os", "/atlas-os/"}:
            # Atlas OS — the AI-first desktop shell (served to Chrome on the box
            # virtual display, and to the owner through the funnel proxy).
            fpath = ROOT / "atlas-os.html"
            if fpath.is_file():
                data = fpath.read_bytes()
                self.send_response(200)
                self.send_header("Content-Type", "text/html; charset=utf-8")
                self.send_header("Content-Length", str(len(data)))
                self.send_header("Cache-Control", "no-cache")
                self.end_headers()
                self.wfile.write(data)
                return
            self.send_error(404)
            return
        # ── Mobile PWAs (Guard Room + Atlas OS) ──
        _pwa_routes = {
            "/guard-app": ("guard-app.html", "text/html; charset=utf-8"),
            "/guard-app/": ("guard-app.html", "text/html; charset=utf-8"),
            "/guard-app/manifest.json": ("guard-manifest.json", "application/json; charset=utf-8"),
            "/atlas-os-app": ("atlas-os-app.html", "text/html; charset=utf-8"),
            "/atlas-os-app/": ("atlas-os-app.html", "text/html; charset=utf-8"),
            "/atlas-os-app/manifest.json": ("atlas-os-manifest.json", "application/json; charset=utf-8"),
        }
        if path in _pwa_routes:
            fname, ctype = _pwa_routes[path]
            fpath = ROOT / fname
            if fpath.is_file():
                data = fpath.read_bytes()
                self.send_response(200)
                self.send_header("Content-Type", ctype)
                self.send_header("Content-Length", str(len(data)))
                self.send_header("Cache-Control", "no-cache")
                self.end_headers()
                self.wfile.write(data)
                return
            self.send_error(404)
            return
        if path in {"/atlas-themes.css", "/atlas-themes.js"}:
            name = path.lstrip("/")
            fpath = ROOT / name
            if fpath.is_file():
                data = fpath.read_bytes()
                ctype = "text/css" if name.endswith(".css") else "application/javascript"
                self.send_response(200)
                self.send_header("Content-Type", ctype)
                self.send_header("Content-Length", str(len(data)))
                self.send_header("Cache-Control", "public, max-age=86400")
                self.end_headers()
                self.wfile.write(data)
                return
            self.send_error(404)
            return
        if path in {"/atlas-cyber-protect-guide.json", "/api/guide"}:
            guide = ROOT / "atlas-cyber-protect-guide.json"
            if not guide.is_file():
                self.send_error(404)
                return
            data = guide.read_bytes()
            self.send_response(200)
            self.send_header("Content-Type", "application/json; charset=utf-8")
            self.send_header("Content-Length", str(len(data)))
            self.send_header("Cache-Control", "no-cache")
            self.end_headers()
            self.wfile.write(data)
            return
        if path.startswith("/device-icons/") and path.count("/") == 2:
            name = path.rsplit("/", 1)[-1]
            if re.fullmatch(r"[a-z0-9_-]+\.svg", name or ""):
                icon = ICONS_DIR / name
                if icon.is_file():
                    data = icon.read_bytes()
                    self.send_response(200)
                    self.send_header("Content-Type", "image/svg+xml")
                    self.send_header("Content-Length", str(len(data)))
                    self.send_header("Cache-Control", "public, max-age=86400")
                    self.end_headers()
                    self.wfile.write(data)
                    return
            self.send_error(404)
            return
        if path == "/api/companion/discover":
            # Public LAN discovery for Protect Agent / AV companion apps
            st = _load_state()
            sm = st.get("scan_meta") or {}
            lan = sm.get("self_ip") or ""
            if not lan:
                try:
                    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
                    s.connect(("1.1.1.1", 80))
                    lan = s.getsockname()[0]
                    s.close()
                except Exception:
                    lan = ""
            cust = _customer_info()
            # Public funnel front so paired PC agents can reach this box from
            # anywhere without Tailscale VPN (https://acp-<host>.<tailnet>.ts.net)
            try:
                ts_host = _load_env().get("TS_HOSTNAME") or f"acp-{socket.gethostname().split('.')[0]}"
                funnel_url = f"https://{ts_host}.taile9cc75.ts.net"
            except Exception:
                funnel_url = ""
            self._json(
                200,
                {
                    "ok": True,
                    "product": "atlas-cyber-protect",
                    "role": "lan-sidebox",
                    "hostname": socket.gethostname(),
                    "lan_ip": lan,
                    "portal_port": PORT,
                    "funnel_url": funnel_url,
                    "business_name": cust.get("business_name") or "",
                    "site_name": cust.get("site_name") or "",
                    "company_id": (cust.get("company_id") or os.environ.get("COMPANY_ID") or "").strip(),
                    "device_count": len(st.get("last_devices") or []),
                    "threat_summary": st.get("threat_summary") or {},
                    "agent_policy": {
                        "mode": "observe_report_only",
                        "never_delete_files": True,
                        "never_block_network": True,
                        "never_modify_firewall": True,
                        "may_request_av_scan": True,
                        "link_guard": True,
                        "warn_dangerous_links": True,
                    },
                    "pair": "/api/companion/pair",
                    "heartbeat": "/api/companion/heartbeat",
                    "update": "/api/companion/agent/update",
                    "link_check": "/api/companion/link-check",
                },
            )
            return
        if path == "/api/companion/agent/update":
            man = _load_agent_manifest()
            man = dict(man)
            man["ok"] = True
            # Always expose relative download URL if file name present
            fname = str(man.get("filename") or "")
            if fname and not man.get("url"):
                man["url"] = f"/api/companion/agent/download/{fname}"
            elif man.get("url") and not str(man.get("url")).startswith("/"):
                pass
            elif fname:
                man["url"] = f"/api/companion/agent/download/{fname}"
            self._json(200, man)
            return
        if path.startswith("/api/companion/agent/download/"):
            name = path.rsplit("/", 1)[-1]
            if not re.fullmatch(r"AtlasCyberProtectAgent-[0-9A-Za-z.\-]+\.exe", name or ""):
                self.send_error(404)
                return
            fpath = AGENT_RELEASES / name
            if not fpath.is_file():
                self.send_error(404)
                return
            data = fpath.read_bytes()
            self.send_response(200)
            self.send_header("Content-Type", "application/octet-stream")
            self.send_header("Content-Length", str(len(data)))
            self.send_header("Content-Disposition", f'attachment; filename="{name}"')
            self.send_header("Cache-Control", "no-store")
            self.end_headers()
            self.wfile.write(data)
            return
        if path.startswith("/api/companion/remote/frame/"):
            # Manager live-view: latest screen frame for an agent (session).
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            agent_id = path.rsplit("/", 1)[-1]
            with REMOTE_LOCK:
                REMOTE_VIEWERS[agent_id] = time.time()  # active-viewer signal for on-demand streaming
                _remote_cleanup()
                frame = REMOTE_FRAMES.get(agent_id) or {}
            if not frame.get("png"):
                self._json(404, {"ok": False, "error": "no frame yet"})
                return
            self.send_response(200)
            self.send_header("Content-Type", str(frame.get("ct") or "image/png"))
            self.send_header("Cache-Control", "no-store")
            self.send_header("X-Frame-Id", str(frame.get("fid") or ""))
            self.send_header("X-Frame-Size", f"{frame.get('w')}x{frame.get('h')}")
            self.send_header("X-Remote-Size", f"{frame.get('rw')}x{frame.get('rh')}")
            self.send_header("Content-Length", str(len(frame["png"])))
            self.end_headers()
            self.wfile.write(frame["png"])
            return
        if path.startswith("/api/companion/remote/viewers/"):
            # Agent asks whether a manager is actively viewing this PC (on-demand
            # streaming: the agent only sends frames while this returns True).
            agent_id = path.rsplit("/", 1)[-1]
            if not _remote_auth(self, agent_id):
                self._json(401, {"ok": False, "error": "agent auth required"})
                return
            now = time.time()
            with REMOTE_LOCK:
                last = float(REMOTE_VIEWERS.get(agent_id) or 0.0)
            self._json(200, {"ok": True, "viewing": bool(now - last < REMOTE_VIEW_TTL)})
            return
        if path.startswith("/api/companion/remote/input/"):
            # Agent drains its queued input events (agent token auth).
            agent_id = path.rsplit("/", 1)[-1]
            if not _remote_auth(self, agent_id):
                self._json(401, {"ok": False, "error": "agent auth required"})
                return
            with REMOTE_LOCK:
                q = REMOTE_INPUTS.pop(agent_id, None) or []
            self._json(200, q)
            return
        if path == "/api/companion/console/poll":
            # Agent polls for queued remote-console commands (fast path, ~5s).
            qp = dict(parse_qsl(urlparse(self.path).query))
            agent_id = str(qp.get("agent_id") or "").strip()
            if not _remote_auth(self, agent_id):
                self._json(401, {"ok": False, "error": "agent auth required"})
                return
            st = _load_state()
            q = list((st.get("console_queue") or {}).get(agent_id) or [])
            st.setdefault("console_queue", {})[agent_id] = []
            _save_state(st)
            self._json(200, {"ok": True, "tasks": q})
            return
        if path == "/api/companion/agents":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            agents = list(_agents().values())
            agents.sort(key=lambda a: float(a.get("last_seen") or 0), reverse=True)
            self._json(200, {"ok": True, "agents": [_companion_public(a) for a in agents], "count": len(agents)})
            return
        if path == "/api/companion/chat-archive":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            self._json(200, _chat_archive_index())
            return
        if path.startswith("/api/companion/chat-archive/media/"):
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            fname = path.rsplit("/", 1)[-1]
            if not re.fullmatch(r"[A-Za-z0-9_.-]+", fname or ""):
                self.send_error(404)
                return
            fpath = DATA / "chat-archive" / "media" / fname
            if not fpath.is_file():
                fpath = DATA / "chat-archive" / "recordings" / fname
            if not fpath.is_file():
                self.send_error(404)
                return
            low = fname.lower()
            ctype = (
                "image/png"
                if low.endswith(".png")
                else "image/jpeg"
                if low.endswith((".jpg", ".jpeg"))
                else "audio/webm"
                if low.endswith(".webm")
                else "application/octet-stream"
            )
            data = fpath.read_bytes()
            self.send_response(200)
            self.send_header("Content-Type", ctype)
            self.send_header("Content-Length", str(len(data)))
            self.send_header("Cache-Control", "no-store")
            self.end_headers()
            self.wfile.write(data)
            return
        # Install invite pages (public — opened on the target PC)
        if path.startswith("/install/") or path.startswith("/api/companion/invite/"):
            parts = [p for p in path.split("/") if p]
            token = ""
            action = ""
            if path.startswith("/install/") and len(parts) >= 2:
                token = parts[1]
                action = "page"
            elif len(parts) >= 4 and parts[0] == "api":
                token = parts[3]
                action = parts[4] if len(parts) > 4 else "meta"
            if not re.fullmatch(r"[A-Za-z0-9_\-]{8,80}", token or ""):
                self.send_error(404)
                return
            try:
                import remote_install as ri
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)})
                return
            inv = ri.get_invite(token)
            if not inv:
                self.send_error(404)
                return
            base = f"http://{_lan_ip_guess()}:{PORT}"
            meta = ri.installer_meta()
            if action in {"page", ""}:
                html = f"""<!DOCTYPE html><html><head><meta charset=utf-8><meta name=viewport content="width=device-width,initial-scale=1">
<title>Install Atlas Cyber Protect</title>
<style>body{{font-family:Georgia,serif;background:#0b1220;color:#e8f4ff;margin:0;padding:2rem;line-height:1.5}}
a.btn,button{{display:inline-block;background:#3dff9a;color:#04140c;font-weight:700;border:0;border-radius:10px;padding:12px 18px;text-decoration:none;cursor:pointer;font-size:1rem}}
.muted{{color:#8aa0b8}}.card{{max-width:520px;margin:0 auto;padding:1.5rem;border:1px solid #1e3a5f;background:#0f1a2e;border-radius:14px}}
h1{{color:#7CFFB2;font-size:1.4rem}}</style></head><body><div class=card>
<h1>Install PC Agent</h1>
<p>Protect invites <strong>{inv.get('hostname') or inv.get('ip') or 'this PC'}</strong> to install Atlas Cyber Protect Agent {meta.get('version') or ''}.</p>
<p class=muted>Warn-only: antivirus + email/link warnings. Never deletes files or blocks internet.</p>
<p><a class=btn href="/api/companion/invite/{token}/download">Download &amp; install</a></p>
<p class=muted>Or open the desktop prompt if it already appeared.</p>
</div>
<script>fetch('/api/companion/invite/{token}/accept',{{method:'POST'}}).catch(()=>{{}})</script>
</body></html>"""
                data = html.encode()
                self.send_response(200)
                self.send_header("Content-Type", "text/html; charset=utf-8")
                self.send_header("Content-Length", str(len(data)))
                self.end_headers()
                self.wfile.write(data)
                return
            if action == "prompt.hta":
                hta = ri.prompt_hta(token, base, {**meta, "hostname": inv.get("hostname") or inv.get("ip")})
                data = hta.encode("utf-8", errors="replace")
                self.send_response(200)
                self.send_header("Content-Type", "application/hta")
                self.send_header("Content-Disposition", f'attachment; filename="AtlasProtectInvite.hta"')
                self.send_header("Content-Length", str(len(data)))
                self.end_headers()
                self.wfile.write(data)
                return
            if action == "download":
                fpath = Path(meta.get("path") or "")
                if not fpath.is_file():
                    self.send_error(404)
                    return
                ri.update_invite(token, status="downloading")
                data = fpath.read_bytes()
                name = meta.get("filename") or fpath.name
                self.send_response(200)
                self.send_header("Content-Type", "application/octet-stream")
                self.send_header("Content-Length", str(len(data)))
                self.send_header("Content-Disposition", f'attachment; filename="{name}"')
                self.end_headers()
                self.wfile.write(data)
                return
            if action == "meta":
                self._json(200, {"ok": True, "invite": inv, "installer": meta})
                return
            self.send_error(404)
            return
        if path == "/api/status":
            _, sess = _session(self)
            self._json(200, _status_payload(sess))
            return
        if path == "/api/dataflow":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            self._json(200, {"ok": True, **_dataflow_payload()})
            return
        if path == "/api/speedtest":
            # GET = status/poll; POST (below) starts a run.
            self._json(200, {"ok": True, "speedtest": _speedtest_load()})
            return
        if path == "/api/ping":
            # Ping a LAN host from the box (read-only) — the Guard dashboard's
            # device 📡 button calls this through the funnel proxy. Guarded the
            # same way as the speedtest trigger: a portal session or the box
            # enroll token (the funnel injects it).
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            _, sess = _session(self)
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            target_ip = str(qp.get("ip") or "").strip()[:64]
            if not target_ip:
                self._json(400, {"ok": False, "error": "ip required"})
                return
            try:
                import subprocess as _sp

                r = _sp.run(
                    ["ping", "-c", "3", "-W", "2", target_ip],
                    capture_output=True,
                    text=True,
                    timeout=15,
                )
                out = (r.stdout or "") + (r.stderr or "")
                ms = None
                loss = None
                import re as _re

                m = _re.search(r"(\d+(?:\.\d+)?)/(\d+(?:\.\d+)?)/(\d+(?:\.\d+)?)", out)
                if m:
                    ms = float(m.group(2))
                lm = _re.search(r"(\d+(?:\.\d+)?)% packet loss", out)
                if lm:
                    loss = float(lm.group(1))
                ok = r.returncode == 0 or loss == 0
                self._json(
                    200,
                    {
                        "ok": ok or loss in (0, 0.0),
                        "ip": target_ip,
                        "ping_ms": ms,
                        "loss_pct": loss,
                        "returncode": r.returncode,
                        "raw": out[-600:],
                    },
                )
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)[:160]})
            return
        if path == "/api/lan-speed/blob":
            qp = dict(parse_qsl(urlparse(self.path).query))
            token = str(qp.get("token") or "")
            st = _load_state()
            if token != str((st.get("lan_speed_token") or {}).get("token") or "") or time.time() > float((st.get("lan_speed_token") or {}).get("expires") or 0):
                self._json(403, {"ok": False, "error": "invalid token"})
                return
            try:
                mb = min(20.0, max(0.5, float(qp.get("mb") or 2)))
                size = int(mb * 1024 * 1024)
                self.send_response(200)
                self.send_header("Content-Type", "application/octet-stream")
                self.send_header("Content-Length", str(size))
                self.send_header("Cache-Control", "no-store")
                self.end_headers()
                sent = 0
                while sent < size:
                    c = b"X" * min(262144, size - sent)
                    self.wfile.write(c)
                    sent += len(c)
            except Exception:
                pass
            return
        if path == "/api/me":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            self._json(200, {"ok": True, "user": sess})
            return
        if path == "/api/people/lan-pending":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            self._json(200, {"ok": True, **_lan_pending()})
            return
        if path == "/api/portal/version":
            # Public: which room UI version this box runs + Atlas-managed status
            st = _load_state()
            self._json(
                200,
                {
                    "ok": True,
                    "version": str(st.get("portal_version") or "—"),
                    "updates_from_atlas": True,
                    "update": st.get("portal_update_info") or {},
                },
            )
            return
        if path == "/api/gateway/creds":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            try:
                import gateway_secure as gs

                self._json(200, gs.public_status())
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)[:200]})
            return
        if path == "/api/gateway/read":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            try:
                import gateway_secure as gs

                self._json(200, gs.read_gateway(force=True))
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)[:200]})
            return
        if path == "/api/infra/inventory":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            try:
                import infra_secure as infra

                self._json(200, infra.inventory_public())
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)[:200]})
            return
        if path == "/api/infra/creds":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            try:
                import infra_secure as infra

                self._json(200, infra.public_status())
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)[:200]})
            return
        self.send_error(404)

    def do_POST(self) -> None:  # noqa: N802
        path = urlparse(self.path).path
        if path == "/api/atlas-os/ai-memory-save":
            # Save an AI memory note
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            body = body_json(self)
            note = str(body.get("note") or "").strip()[:2000]
            if not note:
                self._json(400, {"ok": False, "error": "note required"})
                return
            mem_path = DATA / "atlas-os-memory.json"
            try:
                if mem_path.is_file():
                    notes = json.loads(mem_path.read_text(encoding="utf-8"))
                else:
                    notes = []
            except Exception:
                notes = []
            notes.append({"text": note, "ts": time.time()})
            notes = notes[-200:]
            mem_path.write_text(json.dumps(notes, indent=1), encoding="utf-8")
            self._json(200, {"ok": True, "count": len(notes)})
            return
        if path == "/api/atlas-os/exec":
            # Run a shell command (terminal / app backend)
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            body = body_json(self)
            cmd = str(body.get("cmd") or "").strip()[:500]
            cwd = str(body.get("cwd") or "/").strip()[:300]
            if not cmd:
                self._json(400, {"ok": False, "error": "cmd required"})
                return
            try:
                r = subprocess.run(
                    ["bash", "-c", cmd],
                    capture_output=True, text=True, timeout=30, cwd=cwd,
                )
                self._json(200, {
                    "ok": True,
                    "stdout": r.stdout[-8000:],
                    "stderr": r.stderr[-4000:],
                    "code": r.returncode,
                })
            except subprocess.TimeoutExpired:
                self._json(200, {"ok": True, "stdout": "", "stderr": "command timed out", "code": -1})
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)[:200]})
            return
        if path == "/api/atlas-os/file-write":
            # Write / save a file
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            body = body_json(self)
            fpath = str(body.get("path") or "").strip()
            content = str(body.get("content") or "")
            if not fpath:
                self._json(400, {"ok": False, "error": "path required"})
                return
            try:
                import pathlib as _pl
                p = _pl.Path(fpath)
                p.parent.mkdir(parents=True, exist_ok=True)
                p.write_text(content, encoding="utf-8")
                self._json(200, {"ok": True, "size": len(content)})
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)[:200]})
            return
        if path == "/api/atlas-os/screenshot":
            # Capture a screenshot via scrot/import
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            import pathlib as _pl
            shots_dir = _pl.Path("/tmp/atlas-screenshots")
            shots_dir.mkdir(exist_ok=True)
            fname = f"shot_{int(time.time())}.png"
            fpath = shots_dir / fname
            try:
                # Try import (ImageMagick) first, then scrot
                try:
                    r = subprocess.run(
                        ["import", "-window", "root", str(fpath)],
                        capture_output=True, timeout=10,
                    )
                except FileNotFoundError:
                    r = subprocess.run(
                        ["scrot", str(fpath)],
                        capture_output=True, timeout=10,
                    )
                if fpath.exists():
                    self._json(200, {"ok": True, "file": str(fpath), "name": fname})
                else:
                    self._json(500, {"ok": False, "error": "screenshot failed", "stderr": (r.stderr or b"").decode()[:300]})
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)[:200]})
            return
        if path == "/api/atlas-os/open":
            # Open a file/folder with the system default app
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            body = body_json(self)
            target = str(body.get("path") or "").strip()
            if not target:
                self._json(400, {"ok": False, "error": "path required"})
                return
            try:
                subprocess.Popen(["xdg-open", target])
                self._json(200, {"ok": True})
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)[:200]})
            return
        if path == "/api/atlas-os/service-control":
            # Start/stop/restart systemd services
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            body = body_json(self)
            svc = str(body.get("service") or "").strip()[:100]
            action = str(body.get("action") or "").strip()[:20]
            if not svc or action not in ("start", "stop", "restart", "status"):
                self._json(400, {"ok": False, "error": "service and valid action required"})
                return
            try:
                r = subprocess.run(
                    ["systemctl", action, svc],
                    capture_output=True, text=True, timeout=15,
                )
                self._json(200, {"ok": r.returncode == 0, "stdout": r.stdout[:2000], "stderr": r.stderr[:1000], "code": r.returncode})
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)[:200]})
            return
        if path == "/api/atlas-os/kill-process":
            # Kill a process by PID
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            body = body_json(self)
            pid = str(body.get("pid") or "").strip()
            sig = str(body.get("signal") or "TERM").strip()
            if not pid or not pid.isdigit():
                self._json(400, {"ok": False, "error": "valid pid required"})
                return
            try:
                r = subprocess.run(["kill", f"-{sig}", pid], capture_output=True, text=True, timeout=5)
                self._json(200, {"ok": r.returncode == 0, "stderr": r.stderr[:300]})
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)[:200]})
            return
        if path == "/api/atlas-os/upload":
            # Upload a file to the server
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            dest_dir = qp.get("dir", "/tmp")
            ct = self.headers.get("Content-Type") or ""
            filename = qp.get("filename", "upload.bin")
            try:
                import pathlib as _pl
                p = _pl.Path(dest_dir) / filename
                p.parent.mkdir(parents=True, exist_ok=True)
                n = int(self.headers.get("Content-Length") or 0)
                with open(p, "wb") as f:
                    total = 0
                    while total < n:
                        chunk = self.rfile.read(min(262144, n - total))
                        if not chunk: break
                        f.write(chunk)
                        total += len(chunk)
                self._json(200, {"ok": True, "path": str(p), "size": total})
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)[:200]})
            return
        if path == "/api/speedtest":
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            started = _start_speedtest()
            self._json(200, {"ok": True, "running": started})
            return
        if path == "/api/lan-speed/upload":
            qp = dict(parse_qsl(urlparse(self.path).query))
            token = str(qp.get("token") or "")
            st = _load_state()
            if token != str((st.get("lan_speed_token") or {}).get("token") or "") or time.time() > float((st.get("lan_speed_token") or {}).get("expires") or 0):
                self._json(403, {"ok": False, "error": "invalid token"})
                return
            try:
                n = int(self.headers.get("Content-Length") or 0)
                total = 0
                while total < n:
                    chunk = self.rfile.read(min(262144, n - total))
                    if not chunk:
                        break
                    total += len(chunk)
                self._json(200, {"ok": True, "bytes": total})
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)[:160]})
            return
        if path.startswith("/api/companion/remote/frame/"):
            # Agent posts its latest screen frame (raw JPEG/PNG body, agent auth).
            agent_id = path.rsplit("/", 1)[-1]
            if not _remote_auth(self, agent_id):
                self._json(401, {"ok": False, "error": "agent auth required"})
                return
            png = self._read_raw()
            if not png:
                self._json(400, {"ok": False, "error": "empty frame"})
                return
            ct = str(self.headers.get("Content-Type") or "image/png").split(";")[0].strip() or "image/png"
            fid = str(self.headers.get("X-Frame-Id") or "")
            size = str(self.headers.get("X-Frame-Size") or "0x0")
            w, _, h = size.partition("x")
            try:
                w, h = int(w), int(h)
            except Exception:
                w, h = 0, 0
            rsize = str(self.headers.get("X-Remote-Size") or "0x0")
            rw, _, rh = rsize.partition("x")
            try:
                rw, rh = int(rw), int(rh)
            except Exception:
                rw, rh = w, h
            with REMOTE_LOCK:
                REMOTE_FRAMES[agent_id] = {"png": png, "ct": ct, "fid": fid, "w": w, "h": h, "rw": rw, "rh": rh, "ts": time.time()}
            self._json(200, {"ok": True})
            return
        if path.startswith("/api/companion/remote/input/"):
            # Manager posts input events for an agent (session auth).
            agent_id = path.rsplit("/", 1)[-1]
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            evs = body_json(self)
            if not isinstance(evs, list):
                evs = [evs]
            with REMOTE_LOCK:
                q = REMOTE_INPUTS.setdefault(agent_id, [])
                q.extend(evs)
                REMOTE_INPUTS[agent_id] = q[-200:]
            self._json(200, {"ok": True, "queued": len(evs)})
            return
        body = self._read_json()
        if path == "/api/companion/report-issue":
            """Agent fix-report → relay to the Atlas Ideas Office (kind=fix)."""
            text = str(body.get("text") or "").strip()
            if len(text) < 3:
                self._json(400, {"ok": False, "error": "Describe the problem (3+ characters)"})
                return
            location = str(body.get("location") or "")[:300]
            url = str(body.get("url") or "")[:500]
            host = socket.gethostname()
            full = f"Agent fix report ({host}): {text}"
            import urllib.request

            funnel = (os.environ.get("ATLAS_FUNNEL") or _atlas_base()).rstrip("/")
            payload = {
                "text": full,
                "room": "atlas-protect",
                "url": url,
                "location": location,
                "source": str(body.get("source") or "atlas-agent")[:40],
                "component": str(body.get("component") or "report-a-problem")[:80],
                "app_id": str(body.get("app_id") or "atlas-cyber-protect-agent")[:64],
                "agent_id": str(body.get("agent_id") or "")[:64],
                "hostname": str(body.get("hostname") or host)[:80],
                "version": str(body.get("version") or "")[:40],
            }
            try:
                req = urllib.request.Request(
                    f"{funnel}/api/cyber-protect/auth/report-issue",
                    data=json.dumps(payload).encode("utf-8"),
                    headers={"Content-Type": "application/json"},
                    method="POST",
                )
                ctx = None
                try:
                    ctx = ssl._create_unverified_context()
                except Exception:
                    pass
                with urllib.request.urlopen(req, context=ctx, timeout=25) as resp:
                    out = json.loads(resp.read().decode("utf-8", errors="replace") or "{}")
                self._json(200, out)
            except Exception as exc:
                self._json(502, {"ok": False, "error": f"Ideas Office unreachable: {str(exc)[:160]}"})
            return
        if path == "/api/companion/link-check":
            # Public to paired agents (token optional but preferred); warn-only
            try:
                import link_guard as lg
            except Exception as exc:
                self._json(500, {"ok": False, "error": f"link_guard missing: {exc}"})
                return
            text = str(body.get("text") or "")
            urls = body.get("urls") if isinstance(body.get("urls"), list) else []
            source = str(body.get("source") or "agent")[:40]
            agent_id = str(body.get("agent_id") or "")[:80]
            if text.strip():
                out = lg.check_text(text, source=source)
            elif urls:
                out = lg.check_urls([str(u) for u in urls], source=source)
            else:
                self._json(400, {"ok": False, "error": "text or urls required"})
                return
            out["agent_id"] = agent_id
            # Surface warnings in dedicated hits file + activity (agent must not wipe hits)
            if int(out.get("danger_count") or 0) or int(out.get("warn_count") or 0):
                hosts = [
                    r.get("host")
                    for r in (out.get("results") or [])
                    if isinstance(r, dict) and r.get("warn")
                ][:6]
                hit = {
                    "ts": time.time(),
                    "source": source,
                    "agent_id": agent_id,
                    "danger": out.get("danger_count"),
                    "warn": out.get("warn_count"),
                    "hosts": hosts,
                }
                try:
                    lg.record_hit(hit)
                except Exception:
                    pass
                st = _load_state()
                act = list(st.get("activity") or [])
                act.insert(
                    0,
                    {
                        "ts": time.time(),
                        "plain": (
                            f"Link guard ({source}"
                            + (f"/{agent_id}" if agent_id else "")
                            + f"): {out.get('summary')} "
                            + (f"Hosts: {', '.join(h for h in hosts if h)}" if hosts else "")
                        )[:400],
                    },
                )
                st["activity"] = act[:40]
                _save_state(st)
            self._json(200, out)
            return
        if path == "/api/companion/pair":
            agent_id = str(body.get("agent_id") or "").strip()
            if not agent_id or len(agent_id) > 80:
                self._json(400, {"ok": False, "error": "agent_id required"})
                return
            # Reject agents that claim dangerous modes
            safety = body.get("safety") if isinstance(body.get("safety"), dict) else {}
            if safety.get("never_delete_files") is False or safety.get("mode") not in {
                "observe_report_only",
                None,
                "",
            }:
                # Only accept observe-only companions
                if safety and safety.get("mode") and safety.get("mode") != "observe_report_only":
                    self._json(403, {"ok": False, "error": "only observe_report_only agents allowed"})
                    return
            token = secrets.token_urlsafe(24)
            agents = dict(_agents())
            inv = body.get("inventory") if isinstance(body.get("inventory"), dict) else {}
            agents[agent_id] = {
                "agent_id": agent_id,
                "agent_token": token,
                "name": str(body.get("name") or inv.get("hostname") or agent_id)[:80],
                "inventory": inv,
                "safety": safety or {
                    "mode": "observe_report_only",
                    "never_delete_files": True,
                    "never_block_network": True,
                },
                "paired_at": time.time(),
                "last_seen": time.time(),
                "remote_addr": self.client_address[0] if self.client_address else "",
            }
            _save_agents(agents)
            st = _load_state()
            act = list(st.get("activity") or [])
            act.insert(
                0,
                {
                    "ts": time.time(),
                    "plain": f"Windows Protect Agent paired: {agents[agent_id]['name']} (observe-only).",
                },
            )
            st["activity"] = act[:40]
            st["companion_agents"] = agents
            _save_state(st)
            self._json(
                200,
                {
                    "ok": True,
                    "agent_id": agent_id,
                    "agent_token": token,
                    "policy": "observe_report_only",
                    "message": "Paired. Agent must never delete files or block PC network.",
                },
            )
            return
        if path == "/api/companion/heartbeat":
            agent_id = str(body.get("agent_id") or "").strip()
            token = str(body.get("agent_token") or "").strip()
            agents = dict(_agents())
            row = agents.get(agent_id) or {}
            if not row or not hmac.compare_digest(str(row.get("agent_token") or ""), token):
                self._json(403, {"ok": False, "error": "invalid agent credentials"})
                return
            inv = body.get("inventory") if isinstance(body.get("inventory"), dict) else {}
            row["inventory"] = inv or row.get("inventory") or {}
            row["last_seen"] = time.time()
            row["remote_addr"] = self.client_address[0] if self.client_address else row.get("remote_addr")
            if isinstance(body.get("safety"), dict):
                row["safety"] = body.get("safety")
            if isinstance(body.get("last_scan"), dict):
                row["last_scan"] = body.get("last_scan")
                row["inventory"]["last_scan"] = body.get("last_scan")
            # Consume pending commands for this agent (scan / refresh / update)
            pending = list(row.get("pending_commands") or [])
            send_cmds = pending[:3]
            row["pending_commands"] = pending[3:]
            st = _load_state()
            if isinstance(body.get("command_results"), list) and body.get("command_results"):
                row["last_command_results"] = body.get("command_results")[:8]
                act = list(st.get("activity") or [])
                for cr in body.get("command_results")[:3]:
                    if not isinstance(cr, dict):
                        continue
                    res = cr.get("result") if isinstance(cr.get("result"), dict) else {}
                    act.insert(
                        0,
                        {
                            "ts": time.time(),
                            "plain": (
                                f"PC {row.get('name')}: "
                                f"{res.get('name') or cr.get('type') or 'command'} — "
                                f"{str(res.get('note') or ('ok' if res.get('ok') else 'done'))[:160]}"
                            ),
                        },
                    )
                st["activity"] = act[:40]
            agents[agent_id] = row
            _save_agents(agents)
            st["companion_agents"] = agents
            _save_state(st)
            if inv:
                _learn_endpoint(row, inv)
            # Keep Atlas Connect PC list in sync (Funnel) so Connect finds this office PC
            try:
                _sync_atlas_connect_endpoint(agent_id, row)
            except Exception:
                pass
            locked = LOCK_FLAG.is_file() or bool(st.get("subscription_locked"))
            lock_message = "Subscription paused"
            if LOCK_FLAG.is_file():
                try:
                    lock_message = LOCK_FLAG.read_text(encoding="utf-8").strip() or lock_message
                except Exception:
                    pass
            self._json(
                200,
                {
                    "ok": True,
                    "agent_count": len(agents),
                    "policy": "observe_report_only",
                    "commands": send_cmds,
                    "locked": locked,
                    "lock_message": lock_message,
                    "update": _agent_update_available(
                        str(body.get("agent_version") or (inv.get("agent_version") if isinstance(inv, dict) else "") or "")
                    ),
                    "customer": _customer_info(),
                },
            )
            return
        if path == "/api/companion/console/result":
            # Agent posts remote-console command result(s); the box stores them
            # and the periodic relay forwards them to Atlas (enroll_token auth).
            # Accepts a single result OR a {results: [...]} list from the agent.
            agent_id = str(body.get("agent_id") or "").strip()
            if not _remote_auth(self, agent_id):
                self._json(401, {"ok": False, "error": "agent auth required"})
                return
            raw = body.get("results") if isinstance(body.get("results"), list) else [body]
            st = _load_state()
            for r in raw[:20]:
                if not isinstance(r, dict):
                    continue
                res = {
                    "task_id": str(r.get("task_id") or ""),
                    "agent_id": agent_id,
                    "output": str(r.get("output") or "")[:20000],
                    "exit_code": r.get("exit_code"),
                    "error": str(r.get("error") or "")[:500],
                    "ts": time.time(),
                }
                st.setdefault("console_results", []).append(res)
            st["console_results"] = st["console_results"][-100:]
            act = list(st.get("activity") or [])
            row = _agents().get(agent_id) or {}
            act.insert(
                0,
                {
                    "ts": time.time(),
                    "plain": f"PC {row.get('name') or agent_id}: remote console command finished (exit {res['exit_code']}).",
                },
            )
            st["activity"] = act[:40]
            _save_state(st)
            self._json(200, {"ok": True})
            return
        if path == "/api/companion/agents/update-all":
            # Atlas Guard "Update all behind" — service-to-service call from the
            # Atlas server authenticated by the field kit key (the shared secret
            # baked into every USB kit; Atlas knows the same key). Queues a
            # force_update command for every target agent so it updates on its
            # next heartbeat. agent_ids optional — default is every agent that is
            # behind the box's local manifest.
            kit = (_load_env().get("FIELD_KIT_KEY") or os.environ.get("FIELD_KIT_KEY") or "").strip()
            got = (self.headers.get("X-Field-Kit") or "").strip()
            if not kit or not got or not hmac.compare_digest(got, kit):
                self._json(401, {"ok": False, "error": "invalid service key"})
                return
            wanted = body.get("agent_ids") if isinstance(body.get("agent_ids"), list) else []
            agents = dict(_agents())
            man = _load_agent_manifest()
            remote_ver = str(man.get("version") or "")

            def _vparts(v: str):
                return [int("".join(c for c in p if c.isdigit()) or "0") for p in (v or "0").split(".")]

            queued = []
            for aid, row in agents.items():
                if wanted and aid not in wanted:
                    continue
                inv = row.get("inventory") if isinstance(row.get("inventory"), dict) else {}
                local = str(inv.get("agent_version") or "")[:32]
                # Explicit list always queues; otherwise only behind agents.
                if wanted or (remote_ver and _vparts(remote_ver) > _vparts(local or "0")):
                    cmds = list(row.get("pending_commands") or [])
                    cmds.append({"id": f"upd-{secrets.token_hex(4)}", "type": "force_update", "queued_at": time.time()})
                    row["pending_commands"] = cmds[-8:]
                    agents[aid] = row
                    queued.append(aid)
            _save_agents(agents)
            st = _load_state()
            act = list(st.get("activity") or [])
            act.insert(0, {"ts": time.time(), "plain": f"Atlas Guard update-all queued for {len(queued)} PC agent(s)."})
            st["activity"] = act[:40]
            st["companion_agents"] = agents
            _save_state(st)
            self._json(200, {"ok": True, "queued": len(queued), "agent_ids": queued})
            return
        if path.startswith("/api/companion/agents/") and (
            path.endswith("/scan") or path.endswith("/update")
        ):
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            parts = [p for p in path.split("/") if p]
            if len(parts) < 5:
                self._json(400, {"ok": False, "error": "agent id required"})
                return
            agent_id = parts[3]
            action = parts[4]
            agents = dict(_agents())
            row = agents.get(agent_id)
            if not row:
                self._json(404, {"ok": False, "error": "agent not found"})
                return
            cmds = list(row.get("pending_commands") or [])
            if action == "scan":
                cmds.append({"id": f"scan-{secrets.token_hex(4)}", "type": "quick_scan", "queued_at": time.time()})
                plain = f"Queued AV scan for PC agent {row.get('name') or agent_id}."
            else:
                cmds.append({"id": f"upd-{secrets.token_hex(4)}", "type": "force_update", "queued_at": time.time()})
                plain = f"Queued silent update for PC agent {row.get('name') or agent_id}."
            row["pending_commands"] = cmds[-8:]
            agents[agent_id] = row
            _save_agents(agents)
            st = _load_state()
            act = list(st.get("activity") or [])
            act.insert(0, {"ts": time.time(), "plain": plain})
            st["activity"] = act[:40]
            st["companion_agents"] = agents
            _save_state(st)
            self._json(200, {"ok": True, "agent_id": agent_id, "queued": True, "action": action, "pending": len(row["pending_commands"])})
            return
        if path == "/api/companion/invite":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            try:
                import remote_install as ri
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)})
                return
            devices = body.get("devices") if isinstance(body.get("devices"), list) else []
            if not devices:
                self._json(400, {"ok": False, "error": "select at least one device"})
                return
            created = ri.create_invites(devices, created_by=str((sess or {}).get("email") or ""))
            base = f"http://{_lan_ip_guess()}:{PORT}"
            results = []
            for inv in created:
                delivery = ri.try_deliver_popup(inv, base)
                results.append(
                    {
                        "token": inv.get("token"),
                        "ip": inv.get("ip"),
                        "hostname": inv.get("hostname"),
                        "install_url": f"{base}/install/{inv.get('token')}",
                        "prompt_url": f"{base}/api/companion/invite/{inv.get('token')}/prompt.hta",
                        **delivery,
                    }
                )
            st = _load_state()
            act = list(st.get("activity") or [])
            act.insert(
                0,
                {
                    "ts": time.time(),
                    "plain": f"Install invite sent to {len(results)} PC(s). Users click Install on the popup or open the invite link.",
                },
            )
            st["activity"] = act[:40]
            _save_state(st)
            self._json(200, {"ok": True, "invites": results, "count": len(results)})
            return
        if path == "/api/portal/update":
            # In-room button: ask the box agent to pull the newest Atlas portal UI now.
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            st = _load_state()
            st["portal_update_requested"] = True
            _save_state(st)
            self._json(200, {"ok": True, "queued": True, "note": "Checking Atlas for a newer room UI…"})
            return

        if path == "/api/companion/remote-creds":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            try:
                import remote_install as ri
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)})
                return
            if body.get("clear"):
                ri.clear_creds()
                self._json(200, {"ok": True, "configured": False})
                return
            user = str(body.get("username") or "").strip()
            password = str(body.get("password") or "")
            domain = str(body.get("domain") or "").strip()
            if not user or not password:
                self._json(400, {"ok": False, "error": "username and password required"})
                return
            ri.save_creds(user, password, domain)
            self._json(200, {"ok": True, "configured": True, "username": user, "domain": domain})
            return

        # —— Main gateway (router) login: info / read-only only ——
        if path == "/api/gateway/creds":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            try:
                import gateway_secure as gs
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)})
                return
            if body.get("clear"):
                self._json(200, gs.clear_creds())
                return
            try:
                out = gs.save_creds(
                    username=str(body.get("username") or ""),
                    password=str(body.get("password") or ""),
                    gateway_ip=str(body.get("gateway_ip") or ""),
                    notes=str(body.get("notes") or ""),
                    keep_password=bool(body.get("keep_password")),
                )
                self._json(200, out)
            except ValueError as exc:
                self._json(400, {"ok": False, "error": str(exc)})
            return

        if path == "/api/gateway/read":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            try:
                import gateway_secure as gs

                self._json(200, gs.read_gateway(force=bool(body.get("force", True))))
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)[:200]})
            return

        if path == "/api/gateway/mutate":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            try:
                import gateway_secure as gs

                denied = gs.refuse_mutate(str(sess.get("email") or ""))
                self._json(403, denied)
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)[:200]})
            return

        if path == "/api/infra/creds":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            try:
                import infra_secure as infra
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)})
                return
            if body.get("clear_all"):
                self._json(200, infra.clear_all_logins())
                return
            kind = str(body.get("kind") or "default").strip().lower()
            try:
                if kind == "device":
                    out = infra.save_device_login(
                        ip=str(body.get("ip") or ""),
                        username=str(body.get("username") or ""),
                        password=str(body.get("password") or ""),
                        notes=str(body.get("notes") or ""),
                        role=str(body.get("role") or ""),
                        keep_password=bool(body.get("keep_password")),
                        clear=bool(body.get("clear")),
                    )
                else:
                    out = infra.save_default_login(
                        username=str(body.get("username") or ""),
                        password=str(body.get("password") or ""),
                        notes=str(body.get("notes") or ""),
                        keep_password=bool(body.get("keep_password")),
                    )
                self._json(200, out)
            except ValueError as exc:
                self._json(400, {"ok": False, "error": str(exc)})
            return

        if path == "/api/infra/scan":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            try:
                import infra_secure as infra

                self._json(200, infra.scan_infrastructure(force=bool(body.get("force", True))))
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)[:200]})
            return

        if path.startswith("/api/companion/invite/") and path.endswith("/accept"):
            parts = [p for p in path.split("/") if p]
            token = parts[3] if len(parts) >= 5 else ""
            try:
                import remote_install as ri

                ri.update_invite(token, status="accepted", accepted_at=time.time())
            except Exception:
                pass
            self._json(200, {"ok": True})
            return
        if path == "/api/register":
            if LOCK_FLAG.is_file():
                self._json(403, {"ok": False, "error": "Subscription paused"})
                return
            try:
                user = _create_user(
                    str(body.get("email") or ""),
                    str(body.get("password") or ""),
                    str(body.get("name") or ""),
                )
            except ValueError as exc:
                self._json(400, {"ok": False, "error": str(exc)})
                return
            tok = _new_session(user)
            self._json(200, {"ok": True, "user": user}, set_cookie=tok)
            return
        if path == "/api/login":
            email = str(body.get("email") or "")
            password = str(body.get("password") or "")
            try:
                user = _check_login(email, password)
            except LoginError as exc:
                self._json(exc.status, {"ok": False, "error": str(exc)})
                return
            if not user:
                # Not a box-local account / not the master owner — try the Atlas
                # collab login (managers register in the app with the same email
                # + password; that account is not a box user). Managers get in,
                # employees are denied (users must not log into Atlas Protect).
                collab = _verify_collab_login(email, password)
                if not collab:
                    self._json(
                        401,
                        {
                            "ok": False,
                            "error": "Wrong email or password. Master owner must use fn.nology@gmail.com + Atlas password.",
                        },
                    )
                    return
                if str(collab.get("role") or "") != "manager":
                    self._json(
                        403,
                        {
                            "ok": False,
                            "error": "Only managers can open Atlas Protect — users use the installed agent app (chat + protection) and cannot log into Atlas Protect.",
                        },
                    )
                    return
                person = collab.get("person") or {}
                tok = _new_session(
                    {
                        "id": "collab_" + str(person.get("id") or email),
                        "email": email,
                        "name": str(person.get("name") or email),
                        "master": False,
                        "collab": True,
                        "collab_token": collab.get("token"),
                        "collab_role": "manager",
                        "collab_person_id": str(person.get("id") or ""),
                    }
                )
                self._json(
                    200,
                    {
                        "ok": True,
                        "user": {
                            "id": "collab_" + str(person.get("id") or email),
                            "email": email,
                            "name": str(person.get("name") or email),
                            "master": False,
                        },
                        "collab": {
                            "token": collab.get("token"),
                            "role": "manager",
                            "person": person,
                        },
                    },
                    set_cookie=tok,
                )
                return
            # Local customer accounts blocked when paused; master owner may still enter
            if LOCK_FLAG.is_file() and not user.get("master"):
                self._json(403, {"ok": False, "error": "Subscription paused"})
                return
            tok = _new_session(
                {
                    "id": user["id"],
                    "email": user["email"],
                    "name": user.get("name") or "",
                    "master": bool(user.get("master")),
                }
            )
            # The agent app signs in through this box as a fallback — return a
            # collab token too when the same credentials exist on Atlas (the
            # password was just re-synced above, so this succeeds for the box
            # owner account that was synced at registration).
            collab = _verify_collab_login(email, password)
            out = {
                "ok": True,
                "user": {
                    "id": user["id"],
                    "email": user["email"],
                    "name": user.get("name") or "",
                    "master": bool(user.get("master")),
                },
            }
            if collab and collab.get("token"):
                out["collab"] = {
                    "token": collab.get("token"),
                    "role": str(collab.get("role") or "employee"),
                    "person": collab.get("person") or {},
                }
            self._json(200, out, set_cookie=tok)
            return
        if path == "/api/cyber-protect/auth/login":
            # Same endpoint shape as the Atlas funnel, served by the box too —
            # the installed agent app tries the funnel first and falls back to
            # the box, so sign-in works even when only the box is reachable.
            email = str(body.get("email") or "")
            password = str(body.get("password") or "")
            if not email or not password:
                self._json(400, {"ok": False, "error": "email and password required"})
                return
            try:
                user = _check_login(email, password)
            except LoginError as exc:
                self._json(exc.status, {"ok": False, "error": str(exc)})
                return
            if not user:
                collab = _verify_collab_login(email, password)
                if not collab:
                    self._json(401, {"ok": False, "error": "invalid email or password"})
                    return
                person = collab.get("person") or {}
                self._json(
                    200,
                    {
                        "ok": True,
                        "token": collab.get("token"),
                        "person": person,
                        "access": collab.get("access") or {},
                    },
                )
                return
            collab = _verify_collab_login(email, password)
            if collab and collab.get("token"):
                self._json(
                    200,
                    {
                        "ok": True,
                        "token": collab.get("token"),
                        "person": collab.get("person") or {},
                        "access": collab.get("access") or {},
                    },
                )
                return
            self._json(401, {"ok": False, "error": "invalid email or password"})
            return
        if path == "/api/collab-token":
            # "Open Atlas Protect" from the agent app passes the collab token it
            # already got at sign-in — swap it for a box session so the manager
            # lands inside the portal without typing anything again.
            collab_token = str(body.get("token") or "").strip()
            role = str(body.get("role") or "").strip()
            if not collab_token:
                self._json(400, {"ok": False, "error": "collab token required"})
                return
            if role and role != "manager":
                self._json(403, {"ok": False, "error": "Only managers can open Atlas Protect — users cannot log into Atlas Protect."})
                return
            try:
                import urllib.request

                req = urllib.request.Request(
                    f"{_atlas_base().rstrip('/')}/api/cyber-protect/people/me",
                    headers={"Authorization": "Bearer " + collab_token, "Content-Type": "application/json"},
                    method="GET",
                )
                ctx = ssl.create_default_context()
                ctx.check_hostname = False
                ctx.verify_mode = ssl.CERT_NONE
                with urllib.request.urlopen(req, timeout=25, context=ctx) as resp:
                    me = json.loads(resp.read().decode("utf-8", errors="replace") or "{}")
            except urllib.error.HTTPError as exc:
                self._json(exc.code, {"ok": False, "error": "collab token invalid or expired — sign in again in the agent app"})
                return
            except Exception as exc:
                self._json(503, {"ok": False, "error": f"Atlas unreachable: {str(exc)[:120]}"})
                return
            person = (me or {}).get("person") or {}
            if str(person.get("role") or "") != "manager":
                self._json(403, {"ok": False, "error": "Only managers can open Atlas Protect — users cannot log into Atlas Protect."})
                return
            tok = _new_session(
                {
                    "id": "collab_" + str(person.get("id") or ""),
                    "email": str(person.get("email") or ""),
                    "name": str(person.get("name") or ""),
                    "master": False,
                    "collab": True,
                    "collab_token": collab_token,
                    "collab_role": "manager",
                    "collab_person_id": str(person.get("id") or ""),
                }
            )
            self._json(
                200,
                {
                    "ok": True,
                    "user": {
                        "id": "collab_" + str(person.get("id") or ""),
                        "email": str(person.get("email") or ""),
                        "name": str(person.get("name") or ""),
                        "master": False,
                    },
                    "collab": {"token": collab_token, "role": "manager", "person": person},
                },
                set_cookie=tok,
            )
            return
        if path == "/api/people/lan-approve":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            email = str(body.get("email") or "").strip().casefold()
            person_id = str(body.get("person_id") or "").strip()
            role = str(body.get("role") or "employee").strip()
            if not email and not person_id:
                self._json(400, {"ok": False, "error": "email required"})
                return
            try:
                out = _atlas_lan_approve(email, role=role, person_id=person_id)
            except Exception as exc:
                self._json(502, {"ok": False, "error": f"Atlas approve failed: {exc}"[:200]})
                return
            self._json(200, out if isinstance(out, dict) else {"ok": True, "result": out})
            return
        if path == "/api/tools/open":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            action = str(body.get("action") or body.get("kind") or "").strip().lower()
            kind = {
                "browser": "open_browser",
                "open_browser": "open_browser",
                "winbox": "open_winbox",
                "open_winbox": "open_winbox",
            }.get(action)
            if not kind:
                self._json(400, {"ok": False, "error": "action must be browser or winbox"})
                return
            url = str(body.get("url") or ("https://www.speedtest.net/" if kind == "open_browser" else ""))
            agent_id = str(body.get("agent_id") or "").strip()
            queued = _queue_tool_on_pcs(kind, url=url, agent_id=agent_id)
            local = {}
            try:
                from agent import _launch_local_tool

                local = _launch_local_tool(kind, {"url": url})
            except Exception as exc:
                local = {"ok": False, "error": str(exc)[:120]}
            self._json(
                200,
                {
                    "ok": True if queued or local.get("ok") else False,
                    "kind": kind,
                    "queued_pcs": queued,
                    "local": local,
                    "hint": "Opens on the office Windows Protect agent. Use Atlas Connect if you need to see the screen from overseas.",
                },
            )
            return
        if path == "/api/logout":
            tok, _ = _session(self)
            if tok:
                SESSIONS.pop(tok, None)
                _sessions_save()
            self._json(200, {"ok": True}, set_cookie="")
            return
        if path == "/api/devices/rename":
            # Managers give any LAN device a friendly name so the office list is
            # easy to manage ("Front desk PC", "Till printer"…). Stored per MAC
            # (fallback IP) and survives rescans.
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            key = str(body.get("mac") or body.get("ip") or "").strip()
            name = str(body.get("name") or "").strip()
            if not key:
                self._json(400, {"ok": False, "error": "device mac or ip required"})
                return
            if len(name) > 60:
                self._json(400, {"ok": False, "error": "name too long (60 max)"})
                return
            aliases = _device_aliases()
            if not name:
                aliases.pop(key, None)
            else:
                aliases[key] = name
            _save_device_aliases(aliases)
            # Reflect it on the live device list immediately (no rescan needed)
            st = _load_state()
            for d in (st.get("last_devices") or []):
                if isinstance(d, dict) and str(d.get("mac") or d.get("ip") or "").strip() == key:
                    d["display_name"] = name or (d.get("hostname") or d.get("vendor") or d.get("ip") or "Device")
            _save_state(st)
            self._json(200, {"ok": True, "key": key, "name": name, "display_name": name or ""})
            return
        if path == "/api/decide":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            if LOCK_FLAG.is_file():
                self._json(403, {"ok": False, "error": "Subscription paused"})
                return
            key = str(body.get("key") or "").strip()
            decision = str(body.get("decision") or "").strip().lower()
            if not key or decision not in {"fine", "neutralize"}:
                self._json(400, {"ok": False, "error": "key and decision required"})
                return
            with _db() as con:
                con.execute(
                    "INSERT INTO decisions(finding_key,decision,detail,updated_at) VALUES (?,?,?,?) "
                    "ON CONFLICT(finding_key) DO UPDATE SET decision=excluded.decision, updated_at=excluded.updated_at",
                    (key, decision, str(body.get("detail") or "")[:500], time.time()),
                )
                con.commit()
            q = DATA / "decision_queue.jsonl"
            with q.open("a", encoding="utf-8") as fh:
                fh.write(
                    json.dumps(
                        {
                            "key": key,
                            "decision": decision,
                            "email": sess.get("email"),
                            "ts": time.time(),
                        }
                    )
                    + "\n"
                )
            st = _load_state()
            st.setdefault("decisions", {})
            st["decisions"][key] = {
                "decision": decision,
                "ts": time.time(),
                "email": sess.get("email"),
            }
            # Remove from open findings immediately so the button feels instant
            st["open_findings"] = [
                f for f in (st.get("open_findings") or []) if (f.get("key") or "") != key
            ]
            # Keep the shield count aligned with the list the user just decided on
            _recompute_threat_summary(
                st,
                [
                    f
                    for f in (st.get("open_findings") or [])
                    if (f.get("key") or "") not in _decided_keys()
                ],
            )
            act = list(st.get("activity") or [])
            act.insert(
                0,
                {
                    "ts": time.time(),
                    "plain": (
                        "You marked an issue as fine — guardian noted it."
                        if decision == "fine"
                        else "Neutralize requested — Atlas will contain that issue when possible. Your network stays protected."
                    ),
                },
            )
            st["activity"] = act[:40]
            feed = list(st.get("guardian_feed") or [])
            feed.insert(
                0,
                {
                    "ts": time.time(),
                    "text": (
                        f"Understood — marked fine ({key})."
                        if decision == "fine"
                        else f"Neutralize queued for {key}. I never silent-block without your choice."
                    ),
                    "kind": "guardian",
                },
            )
            st["guardian_feed"] = feed[:60]
            _save_state(st)
            self._json(200, {"ok": True, "decision": decision, "key": key})
            return
        if path == "/api/agent/ask":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            if LOCK_FLAG.is_file():
                self._json(403, {"ok": False, "error": "Subscription paused"})
                return
            try:
                import ui_guardian

                result = ui_guardian.ask(
                    str(body.get("message") or body.get("q") or ""),
                    enroll_token=_enroll_token(),
                )
                self._json(200, result)
            except Exception as exc:
                self._json(500, {"ok": False, "error": str(exc)})
            return
        if path == "/api/atlas-os/ai":
            # Atlas OS desktop assistant — talks to the box's local Ollama GPU
            # model (ACP_LOCAL_LLM). Session or enroll-token gated like the
            # other box APIs; the desktop shell calls it over loopback.
            _, sess = _session(self)
            qp = dict(parse_qsl(urlparse(self.path).query))
            enroll = str(qp.get("enroll_token") or "").strip()
            if not sess and not (enroll and hmac.compare_digest(enroll, _enroll_token())):
                self._json(401, {"ok": False, "error": "login required"})
                return
            spec = (os.environ.get("ACP_LOCAL_LLM") or "").strip()
            if not spec.startswith("ollama://"):
                self._json(400, {"ok": False, "error": "no local LLM configured (ACP_LOCAL_LLM)"})
                return
            model = spec.split("ollama://", 1)[1].strip() or ""
            host = (os.environ.get("ACP_LOCAL_OLLAMA_HOST") or "http://127.0.0.1:11434").rstrip("/")
            msg = str(body.get("message") or "").strip()[:4000]
            history = body.get("history") if isinstance(body.get("history"), list) else []
            if not msg:
                self._json(400, {"ok": False, "error": "message required"})
                return
            system = (
                "You are ATLAS, the living operating-system assistant of Atlas OS — an "
                "AI-first operating system. You are friendly, concise, and capable. "
                "You can explain the system, help with files, commands, networking, "
                "security (Atlas Protect is watching this LAN), and general knowledge. "
                "You run on the box's own GPU. Keep answers under 200 words unless asked "
                "for detail. Be genuinely helpful — this is the OS speaking."
            )
            messages = [{"role": "system", "content": system}]
            for h in history[-8:]:
                if not isinstance(h, dict):
                    continue
                role = "assistant" if str(h.get("role") or "") == "assistant" else "user"
                content = str(h.get("content") or "").strip()[:2000]
                if content:
                    messages.append({"role": role, "content": content})
            messages.append({"role": "user", "content": msg})
            try:
                import urllib.request as _ur

                payload = {
                    "model": model,
                    "messages": messages,
                    "stream": False,
                    "options": {"num_predict": 600, "temperature": 0.6},
                }
                req = _ur.Request(
                    f"{host}/api/chat",
                    data=json.dumps(payload).encode(),
                    headers={"Content-Type": "application/json"},
                    method="POST",
                )
                ctx = ssl.create_default_context()
                ctx.check_hostname = False
                ctx.verify_mode = ssl.CERT_NONE
                with _ur.urlopen(req, timeout=180, context=ctx) as resp:
                    out = json.loads(resp.read().decode("utf-8", errors="replace") or "{}")
                reply = ((out.get("message") or {}).get("content") or "").strip()
                if not reply:
                    self._json(502, {"ok": False, "error": "model returned empty reply"})
                    return
                self._json(200, {"ok": True, "reply": reply, "model": model})
            except Exception as exc:
                self._json(502, {"ok": False, "error": f"local LLM unreachable: {str(exc)[:160]}"})
            return
        if path == "/api/agent/ack-all":
            _, sess = _session(self)
            if not sess:
                self._json(401, {"ok": False, "error": "login required"})
                return
            st = _load_state()
            count = 0
            with _db() as con:
                for f in list(st.get("open_findings") or []):
                    key = str(f.get("key") or "").strip()
                    if not key:
                        continue
                    con.execute(
                        "INSERT INTO decisions(finding_key,decision,detail,updated_at) VALUES (?,?,?,?) "
                        "ON CONFLICT(finding_key) DO UPDATE SET decision=excluded.decision, updated_at=excluded.updated_at",
                        (key, "fine", "ack-all", time.time()),
                    )
                    st.setdefault("decisions", {})[key] = {
                        "decision": "fine",
                        "ts": time.time(),
                    }
                    count += 1
                con.commit()
            st["open_findings"] = []
            _recompute_threat_summary(st, [])
            _save_state(st)
            self._json(200, {"ok": True, "acked": count})
            return
        self.send_error(404)


def body_json(handler) -> dict | list:
    """Parse the POST body of a handler as JSON (dict or list)."""
    n = int(handler.headers.get("Content-Length") or 0)
    raw = handler.rfile.read(n) if n else b"{}"
    try:
        return json.loads(raw.decode() or "{}")
    except Exception:
        return {}


def main() -> None:
    _sessions_load()
    httpd = ThreadingHTTPServer((HOST, PORT), Handler)
    print(f"Atlas Cyber Protect portal on http://{HOST}:{PORT}", flush=True)
    httpd.serve_forever()


if __name__ == "__main__":
    main()