#!/usr/bin/env python3
"""sanad_api_x2 — AGIBOT X2 fleet agent: TELEMETRY + MAP sync in ONE service.

The single X2 agent (one container, one systemd service) that:

  1. TELEMETRY — every ~2 s POSTs the robot's live status:
         POST {SERVER_URL}/api/v1/fleet/ingest/telemetry
     { sn, name, mac, brand, type, model, battery, charging, battery_detail,
       motor_temp, storage, status, position, faults, map, ts }

  2. MAP — a background loop (every MAP_POLL_INTERVAL, default 30 s) checks the
     robot's saved nav maps (pgm+yaml sets and RTAB-Map .db) and uploads
     each map ONE time (re-upload only if its content changes):
         POST {SERVER_URL}/api/v1/fleet/ingest/{sn}/map    (multipart: db + meta)

     The map result is SHOWN inside every telemetry post as the "map" field:
       "map": { "uploaded": true|false,
                "state": "uploaded" | "no_map" | "failed" | "pending",
                "maps_found": N, "last_map": "floor-1"|null,
                "error": "no saved map found …"|null, "checked_ts": … }
     so the server always sees whether the map made it — and why not.

DATA SOURCES (AGIBOT X2 — pluggable, configured in .env)
--------------------------------------------------------
  backend            : X2_SOURCE = auto | http | ros2 | aimrt | none
  battery / charging : X2_FIELD_SOC/VOLTAGE/CURRENT/TEMP/SOH/CYCLES
  faults / motor temp: X2_FIELD_TEMPS (+ staleness of the last successful read)
  position {x,y}     : X2_FIELD_X / X2_FIELD_Y
  status             : derived (charging/moving/idle/offline)
  storage            : host disk via the read-only /:/host mount (+ data dir size)
  maps               : MAPS_DIR/**.pgm+yaml and *.db, places under
                       DATA_DIR/<robot>/places/<map>.json — unchanged from g1

Nothing above is hard-coded: run tools/probe_x2.sh on the robot to discover the
real topics/fields, then set them in .env. No code change, no image rebuild.

Read-only toward the robot: never commands motion. Degrades to heartbeats when
no state source is reachable; --simulate fakes only the state side (the map scan
stays real).

CONFIG — environment (see .env.example). Key vars:
  SERVER_URL, DEVICE_TOKEN, SN (required at install), ROBOT_NAME,
  X2_SOURCE + X2_TOPIC_*/X2_FIELD_* (state source), POLL_INTERVAL (2),
  MAC_INTERFACE (which NIC's MAC to report), TELEMETRY_ENDPOINT,
  ROBOT (web_nav3 robot name, default sanad), MAPS_DIR, DATA_DIR, STATE_DIR,
  MAP_SELECT (all|active|newest), MAP_UPLOAD_MODE (multipart|base64json),
  MAP_ENDPOINT, MAP_POLL_INTERVAL (30), VERIFY_TLS, HTTP_TIMEOUT

CLI:  --simulate | --once | --dry-run | --force (map re-upload) | --list (maps) | -v
"""
from __future__ import annotations

import argparse
import base64
import datetime as _dt
import hashlib
import json
import logging
import math
import os
import platform
import shutil
import sys
import threading
import time
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional

import requests

log = logging.getLogger("sanad_api_x2")


# --------------------------------------------------------------------------- #
# env helpers
# --------------------------------------------------------------------------- #
def _load_dotenv(path: str = ".env") -> None:
    p = Path(path)
    if not p.exists():
        return
    for line in p.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, _, v = line.partition("=")
        os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))


def _env(name: str, default: str = "") -> str:
    return os.environ.get(name, default).strip()


def _env_bool(name: str, default: bool) -> bool:
    return _env(name, "1" if default else "0").lower() in ("1", "true", "yes", "on")


def _now_str() -> str:
    """Full local date+time with UTC offset (e.g. 2026-07-13 14:20:33+04:00).
    TZ_OFFSET_HOURS (default +4, Dubai) keeps container clocks honest without tzdata."""
    off = float(_env("TZ_OFFSET_HOURS", "4"))
    tz = _dt.timezone(_dt.timedelta(hours=off))
    return _dt.datetime.now(tz).isoformat(sep=" ", timespec="seconds")


# --------------------------------------------------------------------------- #
# config
# --------------------------------------------------------------------------- #
@dataclass
class Config:
    server_url: str
    device_token: str
    sn: str
    name: str
    brand: str
    robot_type: str
    model: str
    storage_path: str
    data_path: str
    # telemetry / state source
    net_interface: str
    domain_id: int
    mac_interface: str
    read_fsm: bool
    position_source: str
    rosbridge_url: str
    low_soc: int
    motor_temp_max: float
    alert_log_patterns: str
    alert_log_cooldown: float
    alert_scan_interval: float
    alert_backfill_bytes: int
    poll_interval: float
    telemetry_endpoint: str
    # map sync
    robot: str
    maps_dir: Path
    extra_map_dirs: List[Path]      # extra roots to scan for pgm+yaml sets (e.g. Nav2/Pudu maps)
    web_data_dir: Optional[Path]
    legacy_places: Optional[Path]
    web_nav3_url: str
    map_select: str
    map_upload_mode: str
    map_endpoint_tmpl: str
    map_poll_interval: float
    map_max_upload_mb: float
    state_dir: Path
    # logs + alerts
    alert_endpoint: str
    logs_endpoint: str
    logs_interval: float
    # remote dashboard (register the Sanad UI URL for the fleet to embed)
    remote_enable: bool
    remote_endpoint: str
    remote_kind: str
    remote_host: str
    remote_ports: str
    remote_url: str
    remote_interval: float
    ssh_enable: bool
    ssh_user: str
    ssh_port: int
    control_url: str
    control_enable: bool
    # project logs (e.g. the robot's Sanad app) shipped alongside agent logs
    project_log_container: str
    project_log_path: str
    project_log_label: str
    project_log_backfill: int
    project_log_exclude: str
    ros_distro: str
    # transport
    verify_tls: bool
    http_timeout: float

    @classmethod
    def from_env(cls) -> "Config":
        server = _env("SERVER_URL").rstrip("/")
        token = _env("DEVICE_TOKEN")
        missing = [n for n, v in (("SERVER_URL", server), ("DEVICE_TOKEN", token)) if not v]
        if missing:
            raise SystemExit(f"[config] missing required env: {', '.join(missing)}")
        iface = _env("X2_INTERFACE", "eth0")
        data_dir = _env("DATA_DIR")
        legacy = _env("LEGACY_PLACES")
        return cls(
            server_url=server,
            device_token=token,
            sn=_env("SN", "x2_0000"),
            name=_env("ROBOT_NAME", "") or _env("SN", "x2_0000"),
            brand=_env("ROBOT_BRAND", "agibot"),
            robot_type=_env("ROBOT_TYPE", "humanoid"),
            model=_env("ROBOT_MODEL", "x2"),
            storage_path=_env("STORAGE_PATH", ""),
            data_path=_env("STORAGE_DATA_PATH", ""),
            net_interface=iface,
            domain_id=int(_env("ROS_DOMAIN_ID", "0")),
            mac_interface=_env("MAC_INTERFACE", iface),
            read_fsm=_env_bool("X2_READ_FSM", False),
            position_source=_env("X2_POSITION_SOURCE", "none").lower(),
            rosbridge_url=_env("ROSBRIDGE_URL", "ws://127.0.0.1:9090"),
            low_soc=int(_env("LOW_SOC", "50")),
            motor_temp_max=float(_env("MOTOR_TEMP_MAX", "85")),
            # log-driven alerts: "CODE=regex" entries separated by ";;" (regex may
            # contain '|'). Scanned against the robot's project logs (sanadr1).
            # NOTE: matching is CASE-SENSITIVE (log levels are uppercase); use an
            # inline (?i) prefix for case-insensitive text (Gemini messages).
            alert_log_patterns=_env("ALERT_LOG_PATTERNS",
                "GEMINI_BILLING=(?i)prepayment credits.{0,40}deplet|Please go to AI Studio|Failed to connect to Gemini"
                ";;ROBOT_ERROR=\\bERROR\\b|\\bCRITICAL\\b|^Traceback"),
            alert_log_cooldown=float(_env("ALERT_LOG_COOLDOWN", "300")),   # per-signature re-alert gap
            alert_scan_interval=float(_env("ALERT_SCAN_INTERVAL", "10")),
            alert_backfill_bytes=int(_env("ALERT_BACKFILL_BYTES", str(8 * 1024 * 1024))),
            poll_interval=float(_env("POLL_INTERVAL", "2")),
            telemetry_endpoint=_env("TELEMETRY_ENDPOINT", "/api/v1/fleet/ingest/telemetry"),
            robot=_env("ROBOT", "sanad"),
            maps_dir=Path(_env("MAPS_DIR", "/data/maps")),
            # colon-separated extra roots (mounted Nav2/Pudu map dirs). Any *.yaml+*.pgm
            # set found here is rendered to PNG and uploaded like a slam_toolbox map.
            extra_map_dirs=[Path(p) for p in _env("EXTRA_MAP_DIRS", "").split(":") if p.strip()],
            web_data_dir=Path(data_dir) if data_dir else None,
            legacy_places=Path(legacy) if legacy else None,
            web_nav3_url=_env("WEB_NAV3_URL", "").rstrip("/"),
            map_select=_env("MAP_SELECT", "all").lower(),
            map_upload_mode=_env("MAP_UPLOAD_MODE", "multipart").lower(),
            map_endpoint_tmpl=_env("MAP_ENDPOINT", "/api/v1/fleet/ingest/{sn}/map"),
            map_poll_interval=float(_env("MAP_POLL_INTERVAL", "30")),
            map_max_upload_mb=float(_env("MAP_MAX_UPLOAD_MB", "7")),
            state_dir=Path(_env("STATE_DIR", "/data/state")),
            alert_endpoint=_env("ALERT_ENDPOINT", "/api/v1/fleet/ingest/{sn}/alert"),
            logs_endpoint=_env("LOGS_ENDPOINT", "/api/v1/fleet/ingest/{sn}/logs"),
            logs_interval=float(_env("LOGS_INTERVAL", "60")),
            remote_enable=_env_bool("REMOTE_ENABLE", True),
            remote_endpoint=_env("REMOTE_ENDPOINT", "/api/v1/fleet/ingest/{sn}/remote"),
            remote_kind=_env("REMOTE_KIND", "web"),
            remote_host=_env("REMOTE_HOST", ""),
            remote_ports=_env("REMOTE_PORTS", "8001,8014,8011,8012,8013,8000,8080"),
            remote_url=_env("REMOTE_URL", ""),   # explicit URL (e.g. a public tunnel) wins
            remote_interval=float(_env("REMOTE_INTERVAL", "60")),
            ssh_enable=_env_bool("SSH_REGISTER", True),
            ssh_user=_env("SSH_USER", ""),
            ssh_port=int(_env("SSH_PORT", "22")),
            control_url=_env("CONTROL_STATUS_URL", ""),      # Sanad /api/controller/status
            control_enable=_env_bool("CONTROL_ENABLE", False),  # remote mode-SWITCH (motion!) — off
            project_log_container=_env("PROJECT_LOG_CONTAINER", "auto"),
            project_log_path=_env("PROJECT_LOG_PATH", ""),
            project_log_label=_env("PROJECT_LOG_LABEL", ""),
            project_log_backfill=int(_env("PROJECT_LOG_BACKFILL", "100")),
            # drop uvicorn/access-log noise so shipped lines match the project's
            # own LIVE LOGS panel exactly (app-logger lines + tracebacks only)
            project_log_exclude=_env("PROJECT_LOG_EXCLUDE",
                                     r"^(INFO|WARNING|ERROR|DEBUG|CRITICAL):\s"),
            ros_distro=_env("SOFTWARE_ROS", ""),
            verify_tls=_env_bool("VERIFY_TLS", True),
            http_timeout=float(_env("HTTP_TIMEOUT", "30")),
        )

    def telemetry_url(self) -> str:
        return self.server_url + self.telemetry_endpoint

    def map_url(self) -> str:
        return self.server_url + self.map_endpoint_tmpl.format(sn=self.sn)

    def alert_url(self) -> str:
        return self.server_url + self.alert_endpoint.format(sn=self.sn)

    def logs_url(self) -> str:
        return self.server_url + self.logs_endpoint.format(sn=self.sn)

    def remote_url_ep(self) -> str:
        return self.server_url + self.remote_endpoint.format(sn=self.sn)

    def auth_headers(self) -> Dict[str, str]:
        return {"Authorization": f"Bearer {self.device_token}"}


# --------------------------------------------------------------------------- #
# mac + storage
# --------------------------------------------------------------------------- #
def read_mac(interface: str) -> str:
    p = Path(f"/sys/class/net/{interface}/address")
    try:
        mac = p.read_text().strip()
        if mac and mac != "00:00:00:00:00:00":
            return mac.lower()
    except Exception:
        pass
    n = uuid.getnode()
    return ":".join(f"{(n >> b) & 0xff:02x}" for b in range(40, -1, -8))


_AGENT_VERSION = "2026.07.13"

_software_cache: Optional[Dict[str, Any]] = None


def read_software(cfg: Config) -> Dict[str, Any]:
    """Robot software/OS card: ROS distro, host OS (via /host), kernel, arch.
    The container shares the HOST kernel; the host OS comes from
    /host/etc/os-release (the read-only /:/host mount)."""
    global _software_cache
    if _software_cache is not None:
        return _software_cache
    sw: Dict[str, Any] = {}
    # ROS distro: pinned via SOFTWARE_ROS, else detect host installs /opt/ros/*
    ros = cfg.ros_distro
    if not ros:
        try:
            for base in ("/host/opt/ros", "/opt/ros"):
                p = Path(base)
                if p.is_dir():
                    distros = sorted(d.name for d in p.iterdir() if d.is_dir())
                    if distros:
                        ros = ",".join(distros)
                        break
        except Exception:
            pass
    sw["ros"] = ros or None
    # host OS (firmware/OS card)
    os_name = os_ver = None
    for rel in ("/host/etc/os-release", "/etc/os-release"):
        try:
            kv = {}
            for line in Path(rel).read_text().splitlines():
                if "=" in line:
                    k, _, v = line.partition("=")
                    kv[k] = v.strip().strip('"')
            os_name = kv.get("PRETTY_NAME") or kv.get("NAME")
            os_ver = kv.get("VERSION_ID")
            break
        except Exception:
            continue
    sw["os"] = os_name          # e.g. "Ubuntu 20.04.6 LTS"
    sw["os_version"] = os_ver   # e.g. "20.04"
    # platform.uname(), not os.uname(): identical .release/.machine on the robot,
    # but os.uname() is Unix-only and this must also import on a Windows
    # workstation (dry-run / --simulate during development).
    u = platform.uname()
    sw["kernel"] = u.release    # host kernel (shared with the container)
    sw["arch"] = u.machine      # e.g. aarch64
    sw["python"] = ".".join(map(str, sys.version_info[:3]))
    sw["agent"] = f"{log.name} {_AGENT_VERSION}"
    _software_cache = sw
    return sw


_firmware_cache: Optional[Dict[str, Any]] = None


def read_firmware_static() -> Dict[str, Any]:
    """Board-level firmware card (static, via /host): compute board model +
    Jetson L4T/BSP release + kernel. The state source adds live fw versions."""
    global _firmware_cache
    if _firmware_cache is not None:
        return dict(_firmware_cache)
    fw: Dict[str, Any] = {}
    # board model, e.g. "NVIDIA Orin NX Developer Kit"
    for p in ("/host/sys/firmware/devicetree/base/model",
              "/host/proc/device-tree/model",
              "/sys/firmware/devicetree/base/model",
              "/proc/device-tree/model"):
        try:
            fw["board"] = Path(p).read_bytes().decode().strip("\x00 \n")
            break
        except Exception:
            continue
    # Jetson L4T/BSP: "# R35 (release), REVISION: 3.1, ..." -> "R35.3.1"
    for p in ("/host/etc/nv_tegra_release", "/etc/nv_tegra_release"):
        try:
            head = Path(p).read_text().splitlines()[0]
            import re
            m = re.search(r"(R\d+).*?REVISION:\s*([\d.]+)", head)
            fw["l4t"] = f"{m.group(1)}.{m.group(2)}" if m else head.lstrip("# ").strip()
            break
        except Exception:
            continue
    fw["kernel"] = platform.uname().release
    _firmware_cache = fw
    return dict(fw)


_data_size_cache: Dict[str, Any] = {"ts": 0.0, "kb": None}


def read_storage(cfg: Config) -> Optional[Dict[str, Any]]:
    """Disk usage of the robot's root fs + optional Sanad data-dir size.

    In docker, bind-mount the host root read-only at /host (the installer does)
    so this reports the HOST disk, not the container overlay."""
    root = cfg.storage_path or ("/host" if os.path.isdir("/host") else "/")
    try:
        du = shutil.disk_usage(root)
        out: Dict[str, Any] = {
            "total_gb": round(du.total / 1e9, 2),
            "free_gb": round(du.free / 1e9, 2),
            "used_percent": round(du.used / du.total * 100, 1),
        }
    except Exception:
        return None
    if cfg.data_path and os.path.isdir(cfg.data_path):
        now = time.monotonic()
        if _data_size_cache["kb"] is None or now - _data_size_cache["ts"] > 60:
            try:
                total = 0
                for r, _, files in os.walk(cfg.data_path):
                    for f in files:
                        try:
                            total += os.path.getsize(os.path.join(r, f))
                        except OSError:
                            pass
                _data_size_cache.update(ts=now, kb=round(total / 1024, 1))
            except Exception:
                pass
        if _data_size_cache["kb"] is not None:
            out["data_kb"] = _data_size_cache["kb"]
    return out


# --------------------------------------------------------------------------- #
# state source (telemetry side — degrades to heartbeats if unreachable)
# --------------------------------------------------------------------------- #
def _dig(obj: Any, path: str) -> Any:
    """Walk a dotted path over dicts OR message objects.

      "pose.pose.position.x"     nested attribute / key
      "bms.cell_temp[0]"         list index
      "joints.leg[*].velocity"   WILDCARD — collect that field from EVERY
                                 element, returning a flat list

    The wildcard matters because real robot payloads publish joints as an array
    of objects, not an array of numbers: without it, a velocity/temperature
    mapping resolves to a list of dicts and silently yields nothing.
    Several wildcards compose ("a[*].b[*].c" flattens both levels).

    Returns None for any missing link, so a wrong mapping degrades that one
    field to null — never an exception."""
    if not path:
        return None
    cur = obj
    for part in path.split("."):
        if cur is None:
            return None
        idx = None
        star = False
        if part.endswith("]") and "[" in part:
            part, _, raw = part[:-1].partition("[")
            if raw == "*":
                star = True
            else:
                try:
                    idx = int(raw)
                except ValueError:
                    return None

        def _get(o: Any, name: str) -> Any:
            if not name:
                return o
            return o.get(name) if isinstance(o, dict) else getattr(o, name, None)

        if isinstance(cur, list) and part:
            # a previous [*] already fanned out — keep mapping across the list
            cur = [_get(o, part) for o in cur]
        else:
            cur = _get(cur, part)

        if star:
            if cur is None:
                return None
            try:
                cur = list(cur)
            except TypeError:
                return None
        elif idx is not None:
            try:
                cur = cur[idx]
            except Exception:
                return None
    if isinstance(cur, list) and cur and isinstance(cur[0], list):
        cur = [v for sub in cur for v in (sub if isinstance(sub, list) else [sub])]
    return cur


def _import_msg(spec: str):
    """"sensor_msgs/msg/BatteryState" -> the message class."""
    import importlib
    parts = [p for p in spec.replace(".", "/").split("/") if p]
    return getattr(importlib.import_module(".".join(parts[:-1])), parts[-1])


def _qos(depth: int = 10):
    """Subscription QoS for reading robot telemetry.

    Defaults to BEST_EFFORT, which is the only setting that works against BOTH
    kinds of publisher: a RELIABLE subscriber gets NOTHING from a BEST_EFFORT
    publisher (incompatible), while a BEST_EFFORT subscriber happily reads from
    either. The X2 publishes /aima/mc/leg_odometry as BEST_EFFORT, so the rclpy
    default (RELIABLE) silently delivers zero messages — the subscription is
    created, no error is raised, and the field just stays null forever.

    Override with X2_ROS_QOS=reliable if a topic ever requires it."""
    from rclpy.qos import HistoryPolicy, QoSProfile, ReliabilityPolicy
    want = _env("X2_ROS_QOS", "best_effort").lower()
    rel = (ReliabilityPolicy.RELIABLE if want in ("reliable", "rel")
           else ReliabilityPolicy.BEST_EFFORT)
    return QoSProfile(reliability=rel, history=HistoryPolicy.KEEP_LAST, depth=depth)


def _seq(v: Any) -> List[Any]:
    if v is None:
        return []
    if isinstance(v, (str, bytes)):
        return []
    return list(v) if hasattr(v, "__iter__") else [v]


class AgiBotSource:
    """AGIBOT X2 state source — the ONE robot-specific class in this agent.

    The X2 exposes no fixed topic contract, so NOTHING here is hard-coded:
    choose a backend with X2_SOURCE and point it at the real interface with
    env vars.

      X2_SOURCE=auto   (default) http if X2_STATE_URL is set, else ros2 if rclpy
                       imports, else aimrt if aimrt_py imports, else none
      X2_SOURCE=http   poll X2_STATE_URL (JSON); map fields with X2_FIELD_*
      X2_SOURCE=ros2   subscribe X2_TOPIC_* (types from X2_TYPE_*, defaulting to
                       the standard sensor_msgs / nav_msgs ones)
      X2_SOURCE=aimrt  AimRT channels — see _start_aimrt() below
      X2_SOURCE=none   never read; always heartbeat

    Whatever the backend, it fills one six-key snapshot —
    {bms, state_age, temps, max_vel, xy, fw} — which is the entire seam between
    this robot and the shared telemetry / fault / status / alert pipeline.

    Every read is wrapped: a wrong mapping yields null fields and heartbeat mode,
    never a crashed loop."""

    # Defaults match the STANDARD ROS 2 messages. They are almost certainly
    # right for a ros2 backend and almost certainly wrong for a vendor http
    # payload — which is exactly why they are env-overridable.
    _DEFAULTS = {
        "soc": "percentage",                 # sensor_msgs/BatteryState
        "voltage": "voltage",
        "current": "current",
        "temp": "temperature",
        "soh": "",
        "cycles": "",
        "temps": "",                         # standard JointState has NO temperatures
        "vel": "velocity",                   # sensor_msgs/JointState
        "x": "pose.pose.position.x",         # nav_msgs/Odometry
        "y": "pose.pose.position.y",
        "fsm": "",
        "fw": "",
    }

    def __init__(self, cfg: Config):
        self.cfg = cfg
        self._lock = threading.Lock()
        self._bms: Optional[Dict[str, Any]] = None
        self._bms_ts = 0.0
        self._state_ts = 0.0
        self._temps: List[float] = []
        self._max_vel = 0.0
        self._xy: Optional[Dict[str, float]] = None
        self._fw: Dict[str, Any] = {}
        self._fsm: Optional[int] = None
        self._loco = None
        self._stop = False
        self.backend = "none"
        self.ok = False
        # field map: X2_FIELD_SOC, X2_FIELD_VOLTAGE, X2_FIELD_X, … (empty value
        # = that field is unavailable on this robot and reports null)
        self._map = {k: _env("X2_FIELD_" + k.upper(), d) for k, d in self._DEFAULTS.items()}
        self._soc_scale = _env("X2_SOC_SCALE", "auto").lower()

        def _f(name: str, default: str) -> float:
            try:
                return float(_env(name, default) or default)
            except ValueError:
                log.warning("bad %s — using %s", name, default)
                return float(default)

        # +1 = positive current means CHARGING (the ROS BatteryState
        # convention). Set -1 if the X2 reports the opposite sign.
        self._cur_sign = _f("X2_CURRENT_SIGN", "1")
        # Unit scaling: the telemetry schema is VOLTS and AMPS. BMS firmware
        # commonly publishes mV/mA instead — set 0.001 for those.
        self._v_scale = _f("X2_VOLTAGE_SCALE", "1")
        self._i_scale = _f("X2_CURRENT_SCALE", "1")
        self._start()

    # ---------------- backend selection ----------------
    def _start(self) -> None:
        want = (_env("X2_SOURCE", "auto").lower() or "auto")
        order = ["http", "ros2", "aimrt"] if want == "auto" else [want]
        for b in order:
            try:
                if b == "none":
                    break
                if b == "http" and self._start_http():
                    self.backend = "http"
                    break
                if b == "ros2" and self._start_ros2():
                    self.backend = "ros2"
                    break
                if b == "aimrt" and self._start_aimrt():
                    self.backend = "aimrt"
                    break
            except Exception as e:
                log.warning("X2 source %r failed to start (%s)", b, e)
        if self.backend == "none":
            log.warning("no X2 state source active (X2_SOURCE=%s) — telemetry runs in "
                        "heartbeat mode. Set X2_STATE_URL (http) or X2_TOPIC_* (ros2); "
                        "run tools/probe_x2.sh on the robot to find the real names.", want)
        else:
            self.ok = True
            log.info("X2 source up: backend=%s", self.backend)

    def _start_http(self) -> bool:
        """Poll a vendor state endpoint returning one JSON object."""
        url = _env("X2_STATE_URL")
        if not url:
            return False
        self._http_url = url
        try:
            self._http_period = max(0.2, float(_env("X2_HTTP_INTERVAL", "1") or "1"))
        except ValueError:
            self._http_period = 1.0
        threading.Thread(target=self._http_loop, daemon=True).start()
        log.info("X2 http source: %s every %.1fs", url, self._http_period)
        return True

    def _http_loop(self) -> None:
        while not self._stop:
            try:
                r = requests.get(self._http_url, timeout=4)
                if r.ok:
                    self._ingest_all(r.json())
                else:
                    log.debug("X2 http state: HTTP %s", r.status_code)
            except Exception as e:
                log.debug("X2 http poll failed: %s", e)
            time.sleep(self._http_period)

    def _start_ros2(self) -> bool:
        """Subscribe the configured ROS 2 topics. Types are resolved by name, so
        a custom vendor message works as long as its package is on PYTHONPATH."""
        try:
            import rclpy
            from rclpy.node import Node
        except Exception as e:
            log.debug("rclpy unavailable (%s)", e)
            return False
        wanted = [
            (_env("X2_TOPIC_BATTERY", "/battery_state"),
             _env("X2_TYPE_BATTERY", "sensor_msgs/msg/BatteryState"), self._ingest_battery),
            (_env("X2_TOPIC_JOINTS", "/joint_states"),
             _env("X2_TYPE_JOINTS", "sensor_msgs/msg/JointState"), self._ingest_joints),
            (_env("X2_TOPIC_ODOM", "/odom"),
             _env("X2_TYPE_ODOM", "nav_msgs/msg/Odometry"), self._ingest_odom),
        ]
        if not any(t for t, _, _ in wanted):
            return False
        rclpy.init(args=None)
        node = Node("sanad_api_x2")
        n = 0
        for topic, spec, cb in wanted:
            if not topic or not spec:
                continue
            try:
                cls = _import_msg(spec)
            except Exception as e:
                log.warning("X2 ros2: cannot import %s for %s (%s) — skipped", spec, topic, e)
                continue
            node.create_subscription(cls, topic, cb, _qos())
            log.info("X2 ros2: subscribed %s (%s)", topic, spec)
            n += 1
        if not n:
            try:
                rclpy.shutdown()
            except Exception:
                pass
            return False
        self._node = node
        threading.Thread(target=lambda: rclpy.spin(node), daemon=True).start()
        return True

    def _start_aimrt(self) -> bool:
        """AgiBot's own runtime (the X1 stack is built on it).

        NOT implemented natively, on purpose: the X2's AimRT channel names and
        message definitions are not public, and guessing them would produce an
        agent that imports cleanly, builds, posts perfect-looking heartbeats
        with battery:null forever, and only reveals the mistake on the robot.

        The supported path today is AimRT's ROS 2 plugin: enable it on the robot
        and run this agent with X2_SOURCE=ros2 against the bridged topics. If you
        get the native channel definitions from AgiBot, implement them here — the
        only contract to satisfy is calling self._ingest_battery / _ingest_joints
        / _ingest_odom with the incoming messages."""
        try:
            import aimrt_py  # noqa: F401
        except Exception as e:
            log.debug("aimrt_py unavailable (%s)", e)
            return False
        log.warning("aimrt_py is installed, but no native X2 channel binding is built "
                    "(channel/message definitions are not public). Enable AimRT's ROS 2 "
                    "plugin and set X2_SOURCE=ros2 with X2_TOPIC_* instead.")
        return False

    # ---------------- field extraction ----------------
    def _num(self, obj: Any, key: str) -> Optional[float]:
        v = _dig(obj, self._map.get(key, ""))
        try:
            return float(v) if v is not None else None
        except (TypeError, ValueError):
            return None

    def _ingest_battery(self, msg: Any) -> None:
        try:
            soc = self._num(msg, "soc")
            if soc is None:
                return          # no SOC = no battery record; downstream expects an int
            if self._soc_scale == "fraction" or (self._soc_scale == "auto" and 0.0 < soc <= 1.0):
                soc *= 100.0    # ROS BatteryState.percentage is 0..1; vendors often use 0..100
            cur = self._num(msg, "current")
            volt = self._num(msg, "voltage")
            temp = self._num(msg, "temp")
            soh = self._num(msg, "soh")
            cyc = self._num(msg, "cycles")
            rec = {
                "soc": max(0, min(100, int(round(soc)))),
                "current_a": round(cur * self._i_scale * self._cur_sign, 2) if cur is not None else 0.0,
                "voltage_v": round(volt * self._v_scale, 1) if volt is not None else None,
                "temp_c": int(round(temp)) if temp is not None and -40 <= temp <= 150 else None,
                "soh": int(round(soh)) if soh is not None else 0,
                "cycles": int(round(cyc)) if cyc is not None else 0,
            }
            with self._lock:
                self._bms = rec
                self._bms_ts = self._state_ts = time.monotonic()
        except Exception:
            pass

    def _ingest_joints(self, msg: Any) -> None:
        try:
            temps: List[float] = []
            for x in _seq(_dig(msg, self._map.get("temps", ""))):
                try:
                    f = float(x)
                except (TypeError, ValueError):
                    continue
                if 0 < f <= 200:        # 0 = slot not reporting, same rule as g1
                    temps.append(f)
            max_vel = 0.0
            for x in _seq(_dig(msg, self._map.get("vel", ""))):
                try:
                    max_vel = max(max_vel, abs(float(x)))
                except (TypeError, ValueError):
                    pass
            with self._lock:
                self._temps = temps
                self._max_vel = max_vel
                self._state_ts = time.monotonic()
        except Exception:
            pass

    def _ingest_odom(self, msg: Any) -> None:
        try:
            x, y = self._num(msg, "x"), self._num(msg, "y")
            if x is None or y is None:
                return
            with self._lock:
                self._xy = {"x": round(x, 3), "y": round(y, 3)}
                self._state_ts = time.monotonic()
        except Exception:
            pass

    def _ingest_all(self, d: Any) -> None:
        """One combined payload (http backend) feeds all three extractors."""
        self._ingest_battery(d)
        self._ingest_joints(d)
        self._ingest_odom(d)
        try:
            fsm = _dig(d, self._map.get("fsm", ""))
            fw = _dig(d, self._map.get("fw", ""))
            with self._lock:
                if fsm is not None:
                    try:
                        self._fsm = int(fsm)
                    except (TypeError, ValueError):
                        pass
                if isinstance(fw, dict):
                    self._fw.update({str(k): str(v) for k, v in fw.items()})
        except Exception:
            pass

    # ---------------- the snapshot contract ----------------
    def snapshot(self) -> Dict[str, Any]:
        with self._lock:
            now = time.monotonic()
            return {
                "bms": dict(self._bms) if self._bms else None,
                "state_age": (now - self._state_ts) if self._state_ts else None,
                "temps": list(self._temps),
                "max_vel": self._max_vel,
                "xy": dict(self._xy) if self._xy else None,
                "fw": dict(self._fw),
            }

    def fsm_id(self) -> Optional[int]:
        """Only the http backend can carry one (X2_FIELD_FSM); there is no
        known X2 loco RPC. Read-only either way — never commands motion."""
        with self._lock:
            return self._fsm


# AGIBOT X2 publishes no documented loco FSM id scheme. control.mode comes
# from the read-only CONTROL_STATUS_URL when the robot exposes one; fill
# these in once the real ids are known (empty = mode reported as unknown).
_FSM_STATUS: Dict[int, str] = {}
# Control-panel mode labels + the switchable set (fsm_id -> friendly mode).
_CONTROL_MODES: Dict[int, str] = {}
_CONTROL_SWITCHABLE: List[str] = []


_control_cache: Dict[str, Any] = {"ts": 0.0, "data": None}


def read_control(cfg: Config) -> Optional[Dict[str, Any]]:
    """READ-ONLY control status from the Sanad dashboard's /api/controller/status
    (no robot control, no motion). Reports the current loco mode + the switchable set.
    Remote SWITCHING is a separate, motion-capable path gated by CONTROL_ENABLE."""
    now = time.monotonic()
    if _control_cache["data"] is not None and now - _control_cache["ts"] < 1.5:
        return _control_cache["data"]
    url = cfg.control_url
    if not url:
        port = _REMOTE_STAT.get("port")
        if not port:
            return None
        url = f"http://127.0.0.1:{port}/api/controller/status"
    try:
        r = requests.get(url, timeout=2)
        d = r.json() if r.ok else None
    except Exception:
        d = None
    if not isinstance(d, dict):
        _control_cache.update(ts=now, data=None)
        return None
    fid = d.get("fsm_id")
    out = {
        "fsm_id": fid,
        "mode": _CONTROL_MODES.get(fid, "unknown") if fid is not None else None,
        "armed": d.get("armed"),
        "walk_ready": d.get("walk_ready"),
        "teleop_active": d.get("teleop_active"),
        "last_velocity": d.get("last_velocity"),
        "sdk_available": d.get("sdk_available"),
        "switchable_modes": _CONTROL_SWITCHABLE,
        "remote_switch_enabled": bool(cfg.control_enable),  # remote mode-switch armed?
    }
    _control_cache.update(ts=now, data=out)
    return out


class RosbridgePosition:
    def __init__(self, cfg: Config):
        self.cfg = cfg
        self._xy: Optional[Dict[str, float]] = None
        self._lock = threading.Lock()
        self._stop = False
        try:
            import websocket  # noqa: F401
        except Exception as e:
            log.warning("websocket-client absent (%s) — rosbridge position disabled", e)
            self._ok = False
            return
        self._ok = True
        threading.Thread(target=self._run, daemon=True).start()

    def _run(self) -> None:
        import websocket
        sub = json.dumps({"op": "subscribe", "topic": "/odom",
                          "type": "nav_msgs/Odometry", "throttle_rate": 500})
        while not self._stop:
            try:
                ws = websocket.create_connection(self.cfg.rosbridge_url, timeout=5)
                ws.send(sub)
                while not self._stop:
                    msg = json.loads(ws.recv())
                    pos = (((msg.get("msg") or {}).get("pose") or {}).get("pose") or {}).get("position")
                    if pos:
                        with self._lock:
                            self._xy = {"x": round(float(pos["x"]), 3), "y": round(float(pos["y"]), 3)}
            except Exception as e:
                log.debug("rosbridge position reconnect: %s", e)
                time.sleep(3)

    def get(self) -> Optional[Dict[str, float]]:
        with self._lock:
            return dict(self._xy) if self._xy else None


class Ros2Position:
    """Position from a ROS 2 odometry topic, INDEPENDENT of X2_SOURCE.

    The X2's battery comes cleanly from the Control Dash over http, but that
    payload carries no odometry — while ROS publishes it on
    /aima/mc/leg_odometry (nav_msgs/Odometry). Those are two different
    transports, so position is its own source rather than part of the state
    backend: X2_SOURCE=http can run with X2_POSITION_SOURCE=ros2 at the
    same time.

    Enable with X2_POSITION_SOURCE=ros2 (topic: X2_TOPIC_ODOM, type:
    X2_TYPE_ODOM, fields: X2_FIELD_X / X2_FIELD_Y). Requires rclpy — the
    systemd unit must source the ROS overlay first. If rclpy is missing or the
    topic never publishes, position simply stays null; nothing else is affected."""

    def __init__(self, cfg: Config):
        self.cfg = cfg
        self._xy: Optional[Dict[str, float]] = None
        self._lock = threading.Lock()
        self.ok = False
        # read here, NOT as class attributes: the class body runs at import,
        # which is before main() calls _load_dotenv().
        self._px = _env("X2_FIELD_X", "") or "pose.pose.position.x"
        self._py = _env("X2_FIELD_Y", "") or "pose.pose.position.y"
        topic = _env("X2_TOPIC_ODOM", "/odom")
        spec = _env("X2_TYPE_ODOM", "nav_msgs/msg/Odometry")
        if not topic or not spec:
            log.warning("X2_POSITION_SOURCE=ros2 but X2_TOPIC_ODOM/X2_TYPE_ODOM is empty")
            return
        try:
            import rclpy
            from rclpy.node import Node
        except Exception as e:
            log.warning("rclpy unavailable (%s) — position stays null. The unit must "
                        "source the ROS overlay before starting the agent.", e)
            return
        try:
            cls = _import_msg(spec)
        except Exception as e:
            log.warning("cannot import %s (%s) — position stays null", spec, e)
            return
        try:
            # AgiBotSource may already have started rclpy when X2_SOURCE=ros2.
            if not rclpy.ok():
                rclpy.init(args=None)
            node = Node("sanad_api_x2_pos")
            node.create_subscription(cls, topic, self._on_odom, _qos())
            self._node = node
            threading.Thread(target=lambda: rclpy.spin(node), daemon=True).start()
            self.ok = True
            log.info("X2 position: ros2 %s (%s)", topic, spec)
        except Exception as e:
            log.warning("ros2 position init failed (%s) — position stays null", e)

    def _on_odom(self, msg: Any) -> None:
        try:
            x = _dig(msg, self._px)
            y = _dig(msg, self._py)
            if x is None or y is None:
                return
            with self._lock:
                self._xy = {"x": round(float(x), 3), "y": round(float(y), 3)}
        except Exception:
            pass

    def get(self) -> Optional[Dict[str, float]]:
        with self._lock:
            return dict(self._xy) if self._xy else None


# --------------------------------------------------------------------------- #
# map sync (web_nav3 saved maps → fleet server, uploaded ONCE per content)
# --------------------------------------------------------------------------- #
@dataclass
class MapArtifact:
    path: Path                      # .db (rtabmap) or .yaml (slam_toolbox set)
    name: str
    stem: str
    size: int
    mtime: int
    fmt: str = "rtabmap_db"         # rtabmap_db | slam_toolbox
    files: Dict[str, Path] = field(default_factory=dict)  # slam set: pgm/yaml/posegraph/data
    description: str = ""
    sha256: str = ""
    points: List[Dict[str, Any]] = field(default_factory=list)

    def fingerprint(self) -> str:
        return f"{self.fmt}:{self.size}:{self.mtime}"


def _sha256(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def _sha256_set(paths: List[Path]) -> str:
    h = hashlib.sha256()
    for p in sorted(paths):
        with p.open("rb") as f:
            for chunk in iter(lambda: f.read(1024 * 1024), b""):
                h.update(chunk)
    return h.hexdigest()


# ---- slam_toolbox map set (office.yaml + office.pgm [+ .posegraph .data]) ----
def _parse_map_yaml(path: Path) -> Dict[str, Any]:
    """Tiny parser for a ROS map_server yaml (image/resolution/origin) — no pyyaml."""
    out: Dict[str, Any] = {}
    for line in path.read_text().splitlines():
        line = line.split("#", 1)[0].strip()
        if ":" not in line:
            continue
        k, _, v = line.partition(":")
        k, v = k.strip(), v.strip()
        if k == "image":
            out["image"] = v
        elif k == "resolution":
            try:
                out["resolution"] = float(v)
            except ValueError:
                pass
        elif k == "origin":
            try:
                nums = [float(x) for x in v.strip("[]").split(",")]
                out["origin"] = {"x": nums[0], "y": nums[1],
                                 "yaw": nums[2] if len(nums) > 2 else 0.0}
            except Exception:
                pass
    return out


def _read_pgm(path: Path) -> Optional[Dict[str, Any]]:
    """Parse a binary PGM (P5): returns {width, height, maxval, pixels(bytes)}."""
    try:
        data = path.read_bytes()
        if not data.startswith(b"P5"):
            return None
        # tokenize header (magic, width, height, maxval), skipping comments
        tokens: List[bytes] = []
        i = 2
        while len(tokens) < 3 and i < len(data):
            c = data[i:i + 1]
            if c in b" \t\r\n":
                i += 1
            elif c == b"#":
                i = data.index(b"\n", i) + 1
            else:
                j = i
                while j < len(data) and data[j:j + 1] not in b" \t\r\n":
                    j += 1
                tokens.append(data[i:j])
                i = j
        w, h, maxval = int(tokens[0]), int(tokens[1]), int(tokens[2])
        pixels = data[i + 1: i + 1 + w * h]
        if len(pixels) < w * h:
            return None
        return {"width": w, "height": h, "maxval": maxval, "pixels": pixels}
    except Exception:
        return None


def _pgm_to_png_b64(pgm: Dict[str, Any]) -> str:
    """Grayscale 8-bit PNG from parsed PGM — pure stdlib (zlib + struct)."""
    import struct
    import zlib

    w, h, pixels = pgm["width"], pgm["height"], pgm["pixels"]

    def chunk(tag: bytes, body: bytes) -> bytes:
        return (struct.pack(">I", len(body)) + tag + body
                + struct.pack(">I", zlib.crc32(tag + body) & 0xFFFFFFFF))

    ihdr = struct.pack(">IIBBBBB", w, h, 8, 0, 0, 0, 0)  # 8-bit grayscale
    raw = b"".join(b"\x00" + pixels[y * w:(y + 1) * w] for y in range(h))
    png = (b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", ihdr)
           + chunk(b"IDAT", zlib.compress(raw, 6)) + chunk(b"IEND", b""))
    return base64.b64encode(png).decode("ascii")


def _discover_slam_sets(cfg: Config) -> List[MapArtifact]:
    """Find slam_toolbox / map_server map sets: <stem>.yaml + <stem>.pgm
    (+ optional .posegraph/.data) under the maps roots and maps_slam/."""
    roots = [cfg.maps_dir, cfg.maps_dir / cfg.robot, cfg.maps_dir / "maps_slam"]
    roots += list(cfg.extra_map_dirs)      # Nav2/Pudu map dirs mounted via EXTRA_MAP_DIRS
    seen: set = set()
    out: List[MapArtifact] = []
    for root in roots:
        if not root.exists():
            continue
        for y in sorted(root.glob("*.yaml")):
            meta = _parse_map_yaml(y)
            img = meta.get("image", "")
            pgm = (y.parent / img) if img else y.with_suffix(".pgm")
            if not pgm.exists():
                pgm = y.with_suffix(".pgm")
                if not pgm.exists():
                    continue  # yaml without a raster — not a map set
            rp = str(y.resolve())
            if rp in seen:
                continue
            seen.add(rp)
            files: Dict[str, Path] = {"yaml": y, "pgm": pgm}
            for ext in ("posegraph", "data"):
                p = y.with_suffix("." + ext)
                if p.exists():
                    files[ext] = p
            size = sum(p.stat().st_size for p in files.values())
            mtime = max(int(p.stat().st_mtime) for p in files.values())
            out.append(MapArtifact(path=y, name=y.name, stem=y.stem,
                                   size=size, mtime=mtime,
                                   fmt="slam_toolbox", files=files))
    # Pudu/Nav2 converter emits a plain map + a keepout-BAKED twin (obstacles baked
    # in for Foxy, which has no KeepoutFilter). The baked one is the deploy map — drop
    # the redundant plain twin so the fleet server gets one canonical map, not two.
    baked = {m.stem[: -len("_keepout_baked")] for m in out if m.stem.endswith("_keepout_baked")}
    out = [m for m in out if m.stem not in baked]
    return out


def _map_key(stem: str) -> str:
    stem = Path(stem).name
    if stem.endswith(".db"):
        stem = stem[:-3]
    return "".join(c for c in stem if c.isalnum() or c in "_-.")


def _read_json(path: Path, default: Any) -> Any:
    try:
        return json.loads(path.read_text() or "")
    except Exception:
        return default


def _yaw_from_pose(pose: Dict[str, Any]) -> float:
    if "qw" in pose or "qz" in pose:
        qx = float(pose.get("qx", 0.0)); qy = float(pose.get("qy", 0.0))
        qz = float(pose.get("qz", 0.0)); qw = float(pose.get("qw", 1.0))
        return math.atan2(2.0 * (qw * qz + qx * qy),
                          1.0 - 2.0 * (qy * qy + qz * qz))
    return float(pose.get("yaw", 0.0))


def _places_files_for(cfg: Config, stem: str) -> List[Path]:
    out: List[Path] = []
    key = _map_key(stem)
    if cfg.web_data_dir:
        out.append(cfg.web_data_dir / cfg.robot / "places" / f"{key}.json")
    if cfg.legacy_places:
        out.append(cfg.legacy_places)
    return out


def load_points(cfg: Config, stem: str) -> List[Dict[str, Any]]:
    for pf in _places_files_for(cfg, stem):
        data = _read_json(pf, None) if pf.exists() else None
        if isinstance(data, dict) and data:
            pts: List[Dict[str, Any]] = []
            for name, pose in data.items():
                if not isinstance(pose, dict):
                    continue
                try:
                    pts.append({
                        "name": name,
                        "type": str(pose.get("type", "waypoint")),
                        "x": float(pose["x"]),
                        "y": float(pose["y"]),
                        "yaw": round(_yaw_from_pose(pose), 4),
                    })
                except (KeyError, TypeError, ValueError):
                    continue
            return pts
    return []


def discover_maps(cfg: Config) -> List[MapArtifact]:
    roots = [cfg.maps_dir / cfg.robot, cfg.maps_dir]
    meta: Dict[str, Any] = {}
    meta_file = cfg.maps_dir / cfg.robot / "maps_meta.json"
    if meta_file.exists():
        meta = _read_json(meta_file, {}) or {}
    seen: set = set()
    out: List[MapArtifact] = []
    for root in roots:
        if not root.exists():
            continue
        for p in sorted(root.glob("*.db")):
            rp = str(p.resolve())
            if rp in seen:
                continue
            seen.add(rp)
            st = p.stat()
            out.append(MapArtifact(
                path=p, name=p.name, stem=p.stem,
                size=st.st_size, mtime=int(st.st_mtime),
                description=(meta.get(p.name) or {}).get("description", ""),
            ))
    # slam_toolbox map sets (office.yaml + office.pgm …) live alongside
    out.extend(_discover_slam_sets(cfg))
    out.sort(key=lambda m: m.mtime, reverse=True)
    return out


def _active_map_name(cfg: Config) -> Optional[str]:
    if not cfg.web_nav3_url:
        return None
    try:
        r = requests.get(cfg.web_nav3_url + "/api/status",
                         headers={"X-Robot-Name": cfg.robot},
                         timeout=min(cfg.http_timeout, 5))
        r.raise_for_status()
        am = (r.json() or {}).get("active_map")
        return _map_key(am) if am else None
    except requests.RequestException:
        return None


def select_maps(cfg: Config, maps: List[MapArtifact]) -> List[MapArtifact]:
    if not maps:
        return []
    if cfg.map_select == "newest":
        return maps[:1]
    if cfg.map_select == "active":
        active = _active_map_name(cfg)
        if active:
            picked = [m for m in maps if _map_key(m.stem) == active]
            if picked:
                return picked
        return maps[:1]
    return maps  # "all"


def _state_file(cfg: Config) -> Path:
    return cfg.state_dir / "uploaded.json"


def load_state(cfg: Config) -> Dict[str, str]:
    return _read_json(_state_file(cfg), {}) if _state_file(cfg).exists() else {}


def save_state(cfg: Config, state: Dict[str, str]) -> None:
    try:
        cfg.state_dir.mkdir(parents=True, exist_ok=True)
        _state_file(cfg).write_text(json.dumps(state, indent=2))
    except Exception as e:
        log.warning("could not persist map state: %s", e)


def build_meta(cfg: Config, m: MapArtifact) -> Dict[str, Any]:
    return {
        "sn": cfg.sn,
        "name": m.stem,
        "file": m.name,
        "format": m.fmt,
        "size_bytes": m.size,
        "sha256": m.sha256,
        "mtime": m.mtime,
        "description": m.description,
        "points": m.points,
    }


def _upload_slam_map(cfg: Config, m: MapArtifact, session: requests.Session) -> bool:
    """slam_toolbox map → the spec's image JSON: PNG (from the pgm) + resolution
    + origin + width/height + points. This is what the dashboard renders."""
    url = cfg.map_url()
    ymeta = _parse_map_yaml(m.files["yaml"])
    pgm = _read_pgm(m.files["pgm"])
    if pgm is None:
        log.error("map %s: cannot parse %s (not binary P5?)", m.stem, m.files["pgm"].name)
        return False
    body = build_meta(cfg, m)
    body.update({
        "resolution": ymeta.get("resolution"),
        "origin": ymeta.get("origin"),
        "width": pgm["width"],
        "height": pgm["height"],
        "image_base64": _pgm_to_png_b64(pgm),
    })
    try:
        resp = session.post(url, json=body, headers=cfg.auth_headers(),
                            timeout=cfg.http_timeout, verify=cfg.verify_tls)
    except requests.RequestException as e:
        log.error("map upload %s FAILED (transport): %s", m.stem, e)
        return False
    if not resp.ok:
        log.error("map upload %s FAILED: HTTP %s %s", m.stem, resp.status_code, resp.text[:300])
        return False
    log.info("map uploaded: %s (slam_toolbox %dx%d @ %sm, %d points) -> HTTP %s",
             m.stem, pgm["width"], pgm["height"], ymeta.get("resolution"),
             len(m.points), resp.status_code)
    return True


def upload_map(cfg: Config, m: MapArtifact, session: requests.Session) -> bool:
    if m.fmt == "slam_toolbox":
        return _upload_slam_map(cfg, m, session)
    url = cfg.map_url()
    meta = build_meta(cfg, m)
    try:
        if cfg.map_upload_mode == "base64json":
            body = dict(meta)
            body["db_base64"] = base64.b64encode(m.path.read_bytes()).decode("ascii")
            resp = session.post(url, json=body, headers=cfg.auth_headers(),
                                timeout=cfg.http_timeout, verify=cfg.verify_tls)
        else:  # multipart (default)
            with m.path.open("rb") as fh:
                files = {"db": (m.name, fh, "application/octet-stream")}
                data = {"meta": json.dumps(meta)}
                resp = session.post(url, files=files, data=data,
                                    headers=cfg.auth_headers(),
                                    timeout=cfg.http_timeout, verify=cfg.verify_tls)
    except requests.RequestException as e:
        log.error("map upload %s FAILED (transport): %s", m.name, e)
        return False
    if not resp.ok:
        log.error("map upload %s FAILED: HTTP %s %s", m.name, resp.status_code, resp.text[:300])
        return False
    log.info("map uploaded: %s (%.2f MB, %d points) -> HTTP %s",
             m.name, m.size / 1024 / 1024, len(m.points), resp.status_code)
    return True


# Shared map status — SHOWN in every telemetry post ("map" field).
_MAP_STATUS_LOCK = threading.Lock()
_MAP_STATUS: Dict[str, Any] = {
    "uploaded": False, "state": "pending", "maps_found": 0,
    "last_map": None, "error": None, "checked_ts": None,
}


def _set_map_status(**kw: Any) -> None:
    with _MAP_STATUS_LOCK:
        _MAP_STATUS.update(kw)
        _MAP_STATUS["checked_ts"] = int(time.time())


def get_map_status() -> Dict[str, Any]:
    with _MAP_STATUS_LOCK:
        return dict(_MAP_STATUS)


def map_sync_once(cfg: Config, session: requests.Session,
                  force: bool = False, dry_run: bool = False) -> int:
    """One map pass: scan the Sanad dashboard maps and upload anything new.
    Always updates the shared map status (visible in telemetry)."""
    try:
        maps = select_maps(cfg, discover_maps(cfg))
    except Exception as e:
        _set_map_status(state="failed", uploaded=False, error=f"map scan failed: {e}")
        return 0
    if not maps:
        _set_map_status(state="no_map", uploaded=False, maps_found=0, last_map=None,
                        error=f"no saved map found in Sanad dashboard "
                              f"(maps_dir={cfg.maps_dir}, robot={cfg.robot})")
        return 0

    state = load_state(cfg)
    uploaded = failed = unstable = too_large = current = 0
    last_err: Optional[str] = None
    now = time.time()
    for m in maps:
        prev = state.get(str(m.path.resolve()))
        if not force and prev == m.fingerprint():
            current += 1
            continue  # already uploaded this exact content — one-time rule
        # stability guard: a db modified in the last 120 s is still being
        # written (active mapping) — wait until it settles before uploading
        if not force and (now - m.mtime) < 120:
            log.info("map %s still changing (mapping in progress) — waiting to settle", m.name)
            unstable += 1
            continue
        # server rejects bodies over ~8 MB (client_max_body_size) — don't burn
        # bandwidth on uploads that will 413. Raster (slam_toolbox) maps are tiny.
        if m.fmt != "slam_toolbox" and (m.size / 1048576) > cfg.map_max_upload_mb:
            log.warning("map %s is %.0f MB — exceeds server upload cap (~%.0f MB), skipping "
                        "(export a raster map or raise the server limit)",
                        m.name, m.size / 1048576, cfg.map_max_upload_mb)
            too_large += 1
            continue
        m.sha256 = (_sha256_set(list(m.files.values()))
                    if m.fmt == "slam_toolbox" else _sha256(m.path))
        m.points = load_points(cfg, m.stem)
        if dry_run:
            log.info("[dry-run] would upload map %s (%.2f MB, %d points)",
                     m.name, m.size / 1024 / 1024, len(m.points))
            continue
        if upload_map(cfg, m, session):
            state[str(m.path.resolve())] = m.fingerprint()
            save_state(cfg, state)
            uploaded += 1
        else:
            failed += 1
            last_err = f"upload failed for {m.name} (see agent log)"

    if failed:
        _set_map_status(state="failed", uploaded=False, maps_found=len(maps),
                        last_map=maps[0].stem, error=last_err)
    elif uploaded or current:
        # at least one map is on the server (just now or previously); note skips
        note = None
        if too_large:
            note = f"{too_large} map(s) skipped: exceed server upload cap (~{cfg.map_max_upload_mb:.0f} MB)"
        elif unstable:
            note = "newer map still being written (mapping in progress)"
        _set_map_status(state="uploaded", uploaded=True, maps_found=len(maps),
                        last_map=maps[0].stem, error=note)
    elif too_large:
        _set_map_status(state="failed", uploaded=False, maps_found=len(maps),
                        last_map=maps[0].stem,
                        error=f"map exceeds server upload cap (~{cfg.map_max_upload_mb:.0f} MB) — "
                              "export a raster map or raise the server limit")
    elif unstable:
        # newest content is still being written (active mapping) — be honest
        _set_map_status(state="pending", uploaded=False, maps_found=len(maps),
                        last_map=maps[0].stem,
                        error="map still being written (mapping in progress) — "
                              "will upload when it settles")
    else:
        _set_map_status(state="uploaded", uploaded=True, maps_found=len(maps),
                        last_map=maps[0].stem, error=None)
    return uploaded


def map_loop(cfg: Config, session: requests.Session) -> None:
    while True:
        try:
            map_sync_once(cfg, session)
        except Exception as e:
            log.exception("map pass failed: %s", e)
        time.sleep(cfg.map_poll_interval)


# --------------------------------------------------------------------------- #
# logs + alerts (spec: POST /{sn}/logs periodically, POST /{sn}/alert on events)
# --------------------------------------------------------------------------- #
class _RingLogHandler(logging.Handler):
    """Buffers the agent's own log lines so they can be shipped to the server."""

    def __init__(self, maxlen: int = 400):
        super().__init__(level=logging.INFO)
        from collections import deque
        self._buf: Any = deque(maxlen=maxlen)
        self._blk = threading.Lock()

    def emit(self, record: logging.LogRecord) -> None:
        try:
            with self._blk:
                self._buf.append(self.format(record))
        except Exception:
            pass

    def drain(self) -> List[str]:
        with self._blk:
            lines = list(self._buf)
            self._buf.clear()
        return lines

    def requeue(self, lines: List[str]) -> None:
        """Put unshipped lines back (front of the ring) so they retry next cycle
        instead of being lost — bounded by maxlen, oldest evicted first."""
        with self._blk:
            self._buf.extendleft(reversed(lines))


_LOG_RING = _RingLogHandler()

# shipped-status shown in every telemetry post ("logs" / "alerts" fields)
_LOGS_STAT: Dict[str, Any] = {"last_sent": None, "lines_sent": 0, "ok": None}
_ALERTS_STAT: Dict[str, Any] = {"sent": 0, "last": None, "last_time": None, "ok": None}

# start times ("started_at" = this run, "last_start" = previous run)
_STARTED: Dict[str, Any] = {"now": None, "prev": None, "mono": time.monotonic()}


def _init_start_times(cfg: Config) -> None:
    """Record this agent start; remember the previous one (persisted in STATE_DIR)."""
    f = cfg.state_dir / "agent_state.json"
    prev = (_read_json(f, {}) or {}).get("started_at")
    now_s = _now_str()
    try:
        cfg.state_dir.mkdir(parents=True, exist_ok=True)
        f.write_text(json.dumps({"started_at": now_s}))
    except Exception as e:
        log.debug("could not persist start time: %s", e)
    _STARTED.update(now=now_s, prev=prev, mono=time.monotonic())


class ProjectLogTail:
    """Tails the robot's main PROJECT logs (e.g. the sanadr1 / sanad-p4 app)
    and feeds them into the shipped log lines, labeled "[<project>-logs] …".

    Sources, in priority order:
      PROJECT_LOG_PATH       explicit log file (or dir -> newest *.log) via /host
      PROJECT_LOG_CONTAINER  a docker container name; "auto" (default) scans the
                             host's docker metadata (/host/var/lib/docker) for a
                             RUNNING Sanad project (sanadr1, sanad-p4, sanad*)
    Reads the container's json-log through the read-only /:/host mount — no
    docker socket needed, read-only, cannot disturb the project."""

    KNOWN = ("sanadr1", "sanad-p4", "sanadv3", "sanad")

    def __init__(self, cfg: Config):
        self.label: Optional[str] = None
        self._cur: Optional[Path] = None
        self._pos = 0
        self._backfill = max(0, cfg.project_log_backfill)
        self._exclude = None
        if cfg.project_log_exclude:
            try:
                import re
                self._exclude = re.compile(cfg.project_log_exclude)
            except Exception:
                self._exclude = None
        self.active = False
        try:
            self._resolve(cfg)
        except Exception as e:
            log.debug("project-log resolve failed: %s", e)
        if self.active:
            log.info("project logs: sharing '%s' (%s)", self.label, self._cur)
        else:
            log.info("project logs: none found (project_logs=null)")

    def _resolve(self, cfg: Config) -> None:
        # explicit file/dir
        if cfg.project_log_path:
            p = Path(cfg.project_log_path)
            if p.is_dir():
                logs = sorted(p.glob("*.log"), key=lambda f: f.stat().st_mtime, reverse=True)
                p = logs[0] if logs else None
            if p and p.exists():
                self._start(p, cfg.project_log_label or f"{p.stem}-logs")
            return
        # docker container json-log via /host
        base = Path("/host/var/lib/docker/containers")
        if not base.exists():
            return
        want = cfg.project_log_container
        candidates: List[Any] = []
        for cf in base.glob("*/config.v2.json"):
            try:
                d = json.loads(cf.read_text())
            except Exception:
                continue
            name = (d.get("Name") or "").lstrip("/")
            running = bool((d.get("State") or {}).get("Running"))
            lp = d.get("LogPath") or ""
            if not name or not lp:
                continue
            if want != "auto":
                if name == want:
                    candidates.append((0, name, lp, running))
            elif running and "sanad" in name.lower() and not name.startswith("sanad-api"):
                # rank known Sanad projects first
                rank = self.KNOWN.index(name) if name in self.KNOWN else len(self.KNOWN)
                candidates.append((rank, name, lp, running))
        if not candidates:
            return
        candidates.sort(key=lambda c: c[0])
        _, name, lp, _ = candidates[0]
        p = Path("/host" + lp) if not lp.startswith("/host") else Path(lp)
        if p.exists():
            self._start(p, cfg.project_log_label or f"{name}-logs")

    def _start(self, p: Path, label: str) -> None:
        self._cur = p
        self.label = label
        size = p.stat().st_size
        self._pos = size                  # new lines ship from here on
        self._pending: List[str] = []
        # backfill: the last N RELEVANT lines (post-filter) from the last ~1 MB,
        # so the noise (access-log spam) doesn't eat the history window
        if self._backfill and size:
            try:
                take = min(size, 8 * 1024 * 1024)
                with p.open("rb") as f:
                    f.seek(size - take)
                    tail = f.read(take)
                raw = tail.decode("utf-8", "replace").splitlines()
                if take < size and raw:
                    raw = raw[1:]      # first line is a partial record — drop it
                for ln in raw:
                    ln = ln.strip()
                    if ln.startswith("{"):
                        try:
                            ln = (json.loads(ln).get("log") or "").rstrip()
                        except Exception:
                            pass
                    if not ln:
                        continue
                    if self._exclude is not None and self._exclude.search(ln):
                        continue
                    self._pending.append(f"[{label}] {ln}")
                self._pending = self._pending[-self._backfill:]
            except Exception:
                self._pending = []
        self.active = True

    def poll(self) -> List[str]:
        """New lines since last poll (docker json-log unwrapped), labeled.
        The startup backfill (last N relevant lines) is returned on first call."""
        if not self.active or self._cur is None:
            return []
        pend, self._pending = self._pending, []
        out: List[str] = []
        try:
            st = self._cur.stat()
            if st.st_size < self._pos:    # log rotated
                self._pos = 0
            if st.st_size > self._pos:
                with self._cur.open("rb") as f:
                    f.seek(self._pos)
                    chunk = f.read(min(st.st_size - self._pos, 256 * 1024))
                    self._pos = f.tell()
                for ln in chunk.decode("utf-8", "replace").splitlines():
                    ln = ln.strip()
                    if ln.startswith("{"):
                        try:
                            ln = (json.loads(ln).get("log") or "").rstrip()
                        except Exception:
                            pass
                    if not ln:
                        continue
                    if self._exclude is not None and self._exclude.search(ln):
                        continue  # access-log noise — not in the project's log panel
                    out.append(f"[{self.label}] {ln}")
        except Exception as e:
            log.debug("project-log poll failed: %s", e)
        return pend + out[-100:]           # backfill first, then ≤100 new/cycle


_PROJECT_TAIL: Optional[ProjectLogTail] = None


def ship_logs(cfg: Config, session: requests.Session) -> None:
    """POST buffered agent log lines (+ project logs) to /{sn}/logs.
    Best-effort: failures are logged at DEBUG only (below the ring's level ->
    no feedback loop)."""
    lines = _LOG_RING.drain()
    if _PROJECT_TAIL is not None and _PROJECT_TAIL.active:
        lines.extend(_PROJECT_TAIL.poll())
    if not lines:
        return
    try:
        r = session.post(cfg.logs_url(),
                         json={"sn": cfg.sn, "name": cfg.name,
                               "lines": lines, "ts": int(time.time())},
                         headers=cfg.auth_headers(),
                         timeout=cfg.http_timeout, verify=cfg.verify_tls)
        if r.ok:
            _LOGS_STAT.update(last_sent=_now_str(), ok=True)
            _LOGS_STAT["lines_sent"] += len(lines)
            log.debug("logs shipped: %d lines -> HTTP %s", len(lines), r.status_code)
        else:
            _LOGS_STAT["ok"] = False
            _LOG_RING.requeue(lines)   # retry next cycle (server keeps 500ing)
            log.debug("logs ship failed: HTTP %s (%d lines requeued)", r.status_code, len(lines))
    except requests.RequestException as e:
        _LOGS_STAT["ok"] = False
        _LOG_RING.requeue(lines)
        log.debug("logs ship failed (%d lines requeued): %s", len(lines), e)


def logs_loop(cfg: Config, session: requests.Session) -> None:
    while True:
        time.sleep(cfg.logs_interval)
        try:
            ship_logs(cfg, session)
        except Exception:
            pass


# --------------------------------------------------------------------------- #
# remote dashboard — discover the Sanad web UI and register its URL for the
# fleet dashboard to embed ("full dashboard through the fleet API"). No changes
# to the Sanad app: we only probe its port and POST the URL to /{sn}/remote.
# --------------------------------------------------------------------------- #
_REMOTE_STAT: Dict[str, Any] = {"url": None, "port": None, "kind": None, "ok": None,
                                "ssh": None, "ssh_ok": None}


def _primary_ip() -> str:
    import socket
    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 "127.0.0.1"


def discover_dashboard(cfg: Config) -> Optional[Dict[str, Any]]:
    """Find the Sanad dashboard: probe each candidate port on localhost; the one
    that returns 200 with a 'Sanad'/dashboard page wins. Returns {url, port}."""
    if cfg.remote_url:                      # explicit URL (e.g. a tunnel) overrides
        return {"url": cfg.remote_url, "port": None}
    host = cfg.remote_host or _primary_ip()   # the LAN IP the fleet can reach
    for tok in cfg.remote_ports.split(","):
        tok = tok.strip()
        if not tok.isdigit():
            continue
        port = int(tok)
        try:
            r = requests.get(f"http://127.0.0.1:{port}/", timeout=2)
        except requests.RequestException:
            continue
        if r.status_code == 200 and ("sanad" in r.text.lower() or "dashboard" in r.text.lower()):
            return {"url": f"http://{host}:{port}", "port": port}
    return None


def _post_remote(cfg: Config, session: requests.Session, body: Dict[str, Any]) -> bool:
    try:
        r = session.post(cfg.remote_url_ep(), json=body, headers=cfg.auth_headers(),
                         timeout=cfg.http_timeout, verify=cfg.verify_tls)
        if not r.ok:
            log.warning("remote(%s) register failed: HTTP %s %s",
                        body.get("kind"), r.status_code, r.text[:120])
        return bool(r.ok)
    except requests.RequestException as e:
        log.debug("remote(%s) register failed: %s", body.get("kind"), e)
        return False


def register_remote(cfg: Config, session: requests.Session) -> None:
    if not cfg.remote_enable:
        return
    host = cfg.remote_host or _primary_ip()
    # 1) the Sanad dashboard (kind=web) — the exact UI, for the fleet to embed
    d = discover_dashboard(cfg)
    if d:
        ok = _post_remote(cfg, session, {
            "sn": cfg.sn, "name": cfg.name, "kind": cfg.remote_kind,
            "url": d["url"], "label": f"{cfg.name} — Sanad Dashboard",
            "port": d["port"], "ts": int(time.time())})
        _REMOTE_STAT.update(url=d["url"], port=d["port"], kind=cfg.remote_kind, ok=ok)
        if ok:
            log.info("remote dashboard registered: %s -> HTTP 200", d["url"])
    else:
        _REMOTE_STAT.update(url=None, port=None, ok=None)
    # 2) SSH access (kind=ssh) — "ssh <user>@<ip>". Skipped when the login user
    # isn't known: registering a guessed username sends the fleet UI a command
    # that silently fails for whoever tries it.
    if cfg.ssh_enable and cfg.ssh_user:
        cmd = f"ssh {cfg.ssh_user}@{host}"
        if cfg.ssh_port != 22:
            cmd += f" -p {cfg.ssh_port}"
        ok = _post_remote(cfg, session, {
            "sn": cfg.sn, "name": cfg.name, "kind": "ssh",
            "url": f"ssh://{cfg.ssh_user}@{host}:{cfg.ssh_port}",   # URL-valid form
            "command": cmd,                                        # "ssh <user>@<ip>"
            "host": host, "port": cfg.ssh_port, "user": cfg.ssh_user,
            "label": f"{cfg.name} — SSH", "ts": int(time.time())})
        _REMOTE_STAT.update(ssh=cmd, ssh_ok=ok)
        if ok:
            log.info("remote SSH registered: %s -> HTTP 200", cmd)


def remote_loop(cfg: Config, session: requests.Session) -> None:
    while True:
        try:
            register_remote(cfg, session)
        except Exception as e:
            log.debug("remote loop: %s", e)
        time.sleep(cfg.remote_interval)


_ALERT_SEEN: Dict[str, float] = {}   # fault CODE -> monotonic ts of last alert


def _post_alert(cfg: Config, session: requests.Session, text: str) -> bool:
    """POST a single alert string to /{sn}/alert; updates the alert status."""
    try:
        r = session.post(cfg.alert_url(),
                         json={"sn": cfg.sn, "name": cfg.name,
                               "alert": text, "message": text, "ts": int(time.time())},
                         headers=cfg.auth_headers(),
                         timeout=cfg.http_timeout, verify=cfg.verify_tls)
        _ALERTS_STAT.update(last=text, last_time=_now_str(), ok=bool(r.ok))
        if r.ok:
            _ALERTS_STAT["sent"] += 1
        log.info("alert sent: %s -> HTTP %s", text[:120], r.status_code)
        return bool(r.ok)
    except requests.RequestException as e:
        _ALERTS_STAT.update(last=text, last_time=_now_str(), ok=False)
        log.debug("alert send failed (%s): %s", text[:60], e)
        return False


def _fault_code(f: str) -> str:
    """Dedup key for a fault string: the CODE before the first ':'."""
    return (f.split(":", 1)[0].strip() or f)


def send_alerts(cfg: Config, session: requests.Session, faults: List[str]) -> None:
    """POST each NEW fault (rising edge) to /{sn}/alert. Faults are strings.
    Includes LOW_BATTERY (<= LOW_SOC, default 50%), MOTOR_OVERTEMP, COMMS_STALE.

    Deduped on the fault CODE, never on the whole string: every fault embeds a
    LIVE number ("battery 49%", "motor temp 87C", "no robot state for 12s") that
    changes almost every tick, so string-dedup re-fired the same fault every
    POLL_INTERVAL for as long as it lasted — COMMS_STALE counts up by 1 s
    forever, which meant an alert POST every 2 s until the state source came back.

    A code alerts on its rising edge, then at most once per ALERT_LOG_COOLDOWN
    while it persists (so a long outage still re-asserts, but doesn't flood).
    Clearing the fault drops the code, so the next occurrence alerts again."""
    now = time.monotonic()
    current: Dict[str, str] = {}
    for f in faults:
        current.setdefault(_fault_code(f), f)
    for code in [c for c in _ALERT_SEEN if c not in current]:
        del _ALERT_SEEN[code]          # cleared — next occurrence is a rising edge again
    for code, text in sorted(current.items()):
        last = _ALERT_SEEN.get(code)
        if last is None or (now - last) > cfg.alert_log_cooldown:
            _ALERT_SEEN[code] = now
            _post_alert(cfg, session, text)


class LogAlertScanner:
    """Scans the robot's project logs (sanadr1) for error/billing patterns and
    fires an alert on each NEW signature (deduped with a cooldown). Catches the
    Gemini 'credits depleted' billing error and any ERROR/Traceback the app logs."""

    def __init__(self, cfg: Config):
        import re
        self._path: Optional[Path] = None
        self._pos = 0
        self._cooldown = cfg.alert_log_cooldown
        self._seen: Dict[str, float] = {}
        self._patterns: List[Any] = []
        self._pending: List[Any] = []
        for entry in cfg.alert_log_patterns.split(";;"):
            if "=" in entry:
                code, rx = entry.split("=", 1)
                try:
                    # case-SENSITIVE (uppercase log levels); use (?i) inline for text
                    self._patterns.append((code.strip(), re.compile(rx)))
                except re.error as e:
                    log.warning("bad alert pattern %s: %s", code, e)
        if _PROJECT_TAIL is not None and _PROJECT_TAIL.active and _PROJECT_TAIL._cur:
            self._path = _PROJECT_TAIL._cur
            try:
                size = self._path.stat().st_size
                self._pos = size                       # ongoing reads start at EOF
                # STARTUP BACKFILL: scan a large tail once (access-logs bury real
                # errors) so a currently-active error (e.g. Gemini billing) alerts
                # right away. Only the matching lines are kept — cheap.
                back = min(size, cfg.alert_backfill_bytes)
                with self._path.open("rb") as f:
                    f.seek(size - back)
                    raw = f.read(back).decode("utf-8", "replace").splitlines()
                if back < size and raw:
                    raw = raw[1:]                       # first line is partial
                matches = []
                for ln in raw:
                    ln = ln.strip()
                    if ln.startswith("{"):
                        try:
                            ln = (json.loads(ln).get("log") or "").rstrip()
                        except Exception:
                            pass
                    if not ln:
                        continue
                    for code, rx in self._patterns:
                        if rx.search(ln):
                            matches.append((code, ln))
                            break
                # keep the most-recent unique signature per (code+line), cap 20
                seen = set()
                uniq = []
                for code, ln in reversed(matches):
                    sig = code + ":" + re.sub(r"\d+", "#", ln)[:120]
                    if sig in seen:
                        continue
                    seen.add(sig)
                    uniq.append((code, ln))
                self._pending = list(reversed(uniq))[-20:]
            except Exception:
                self._pos = 0
        if self._path and self._patterns:
            log.info("log-alert scan on %s (%d patterns, %d backfilled)",
                     self._path.name, len(self._patterns), len(self._pending))

    def scan(self, cfg: Config, session: requests.Session) -> None:
        if not self._path or not self._patterns:
            return
        import re
        now0 = time.monotonic()
        if self._pending:                       # flush startup backfill first
            pend, self._pending = self._pending, []
            for code, ln in pend:
                sig = code + ":" + re.sub(r"\d+", "#", ln)[:120]
                self._seen[sig] = now0
                _post_alert(cfg, session, f"{code}: {ln[:220]}")
        try:
            st = self._path.stat()
            if st.st_size < self._pos:      # rotated
                self._pos = 0
            if st.st_size <= self._pos:
                return
            with self._path.open("rb") as f:
                f.seek(self._pos)
                chunk = f.read(min(st.st_size - self._pos, 512 * 1024))
                self._pos = f.tell()
        except Exception as e:
            log.debug("alert scan read failed: %s", e)
            return
        now = time.monotonic()
        for ln in chunk.decode("utf-8", "replace").splitlines():
            ln = ln.strip()
            if ln.startswith("{"):
                try:
                    ln = (json.loads(ln).get("log") or "").rstrip()
                except Exception:
                    pass
            if not ln:
                continue
            for code, rx in self._patterns:
                if rx.search(ln):
                    sig = code + ":" + re.sub(r"\d+", "#", ln)[:120]   # dedup key
                    if now - self._seen.get(sig, -1e9) > self._cooldown:
                        self._seen[sig] = now
                        _post_alert(cfg, session, f"{code}: {ln[:220]}")
                    break


_LOG_ALERTS: Optional[LogAlertScanner] = None


def alert_scan_loop(cfg: Config, session: requests.Session) -> None:
    while True:
        time.sleep(cfg.alert_scan_interval)
        try:
            if _LOG_ALERTS is not None:
                _LOG_ALERTS.scan(cfg, session)
        except Exception as e:
            log.debug("alert scan loop: %s", e)


# --------------------------------------------------------------------------- #
# telemetry assembly
# --------------------------------------------------------------------------- #
def derive_faults(cfg: Config, snap: Dict[str, Any]) -> List[str]:
    # STRINGS, not objects — the fleet ingest 500s on fault objects.
    faults: List[str] = []
    bms = snap.get("bms")
    if bms and bms.get("soc", 100) <= cfg.low_soc:
        faults.append(f"LOW_BATTERY: battery {bms['soc']}% (warning)")
    temps = snap.get("temps") or []
    if temps and max(temps) >= cfg.motor_temp_max:
        faults.append(f"MOTOR_OVERTEMP: motor temp {max(temps):.0f}C (warning)")
    if snap.get("state_age") is not None and snap["state_age"] > 3.0:
        faults.append(f"COMMS_STALE: no robot state for {snap['state_age']:.0f}s (critical)")
    return faults


def derive_status(cfg: Config, snap: Dict[str, Any], fsm: Optional[int]) -> str:
    base = _FSM_STATUS.get(fsm) if fsm is not None else None
    bms = snap.get("bms")
    charging = bool(bms and bms.get("current_a", 0.0) > 0.05)
    alive = snap.get("state_age") is not None and snap["state_age"] <= 3.0
    if not alive and bms is None:
        return "offline"
    if charging:
        return "charging"
    if snap.get("max_vel", 0.0) > 0.15:
        return "moving"
    return base or "idle"


def build_telemetry(cfg: Config, mac: str, reader: Optional[AgiBotSource],
                    pos: Optional[RosbridgePosition],
                    sim: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    if sim is not None:
        snap = {"bms": {"soc": sim["battery"], "current_a": 0.5 if sim["charging"] else -0.3,
                        "voltage_v": 47.5, "temp_c": 36, "soh": 100, "cycles": 45},
                "state_age": 0.1, "temps": [sim.get("temp", 45)], "max_vel": sim.get("max_vel", 0.0),
                "xy": sim.get("position")}
        fsm = sim.get("fsm")
    else:
        snap = reader.snapshot() if reader else {"bms": None, "state_age": None, "temps": [], "max_vel": 0.0, "xy": None}
        fsm = reader.fsm_id() if (reader and cfg.read_fsm) else None

    bms = snap.get("bms")
    battery = bms["soc"] if bms else None
    charging = bool(bms and bms.get("current_a", 0.0) > 0.05)
    status = derive_status(cfg, snap, fsm)
    faults = derive_faults(cfg, snap)

    battery_detail = None
    if bms:
        battery_detail = {"voltage_v": bms.get("voltage_v"), "current_a": bms.get("current_a"),
                          "temp_c": bms.get("temp_c"), "soh": bms.get("soh"),
                          "cycles": bms.get("cycles")}

    temps = snap.get("temps") or []
    motor_temp = ({"max": round(max(temps), 1), "avg": round(sum(temps) / len(temps), 1),
                   "min": round(min(temps), 1)} if temps else None)

    position = snap.get("xy")
    if position is None and pos is not None:
        position = pos.get()

    return {
        "sn": cfg.sn,
        "name": cfg.name,            # friendly display name (e.g. x2_10)
        "mac": mac,
        "brand": cfg.brand,
        "type": cfg.robot_type,      # humanoid
        "model": cfg.model,          # g1
        "software": read_software(cfg),  # ros/os/kernel/arch/python/agent
        "firmware": {**read_firmware_static(), **(snap.get("fw") or {})},
                                     # board/l4t/kernel + live robot/bms fw versions
        "battery": battery,          # null = couldn't read (heartbeat)
        "charging": charging,
        "battery_detail": battery_detail,
        "motor_temp": motor_temp,    # null = not receiving
        "storage": read_storage(cfg),
        "status": status,
        "position": position,        # null when no odom/localization source
        "control": read_control(cfg),  # loco mode (zero_torque/damp/lock/running) + switchable set
        "faults": faults,
        "map": get_map_status(),     # SHOWS whether the saved map made it to the server
        "logs": dict(_LOGS_STAT),    # log-shipping status (last_sent, lines_sent, ok)
        "project_logs": (_PROJECT_TAIL.label
                         if _PROJECT_TAIL is not None and _PROJECT_TAIL.active
                         else None),  # e.g. "sanadr1-logs"; null = no project found
        "remote": (dict(_REMOTE_STAT) if _REMOTE_STAT.get("url") else None),
                                     # Sanad dashboard URL registered for the fleet UI
        "alerts": dict(_ALERTS_STAT),  # alert status (sent, last, last_time, ok)
        "time": _now_str(),          # full date+time of this post
        "started_at": _STARTED["now"],   # when this agent run started
        "last_start": _STARTED["prev"],  # previous agent start (null on first ever)
        "uptime_s": int(time.monotonic() - _STARTED["mono"]),
        "ts": int(time.time()),
    }


def post_telemetry(cfg: Config, payload: Dict[str, Any], session: requests.Session) -> bool:
    try:
        r = session.post(cfg.telemetry_url(), json=payload, headers=cfg.auth_headers(),
                         timeout=cfg.http_timeout, verify=cfg.verify_tls)
    except requests.RequestException as e:
        log.error("telemetry POST failed (transport): %s", e)
        return False
    if not r.ok:
        log.error("telemetry POST failed: HTTP %s %s", r.status_code, r.text[:200])
        return False
    mp = payload.get("map") or {}
    log.info("telemetry ok: battery=%s charging=%s status=%s pos=%s faults=%d map=%s -> HTTP %s",
             payload["battery"], payload["charging"], payload["status"],
             payload["position"], len(payload["faults"]), mp.get("state"), r.status_code)
    return True


def _sim_state(i: int) -> Dict[str, Any]:
    charging = (i % 6) in (0, 1)
    battery = max(5, 90 - (i % 40))
    moving = (i % 3) == 2 and not charging
    return {"battery": battery, "charging": charging, "temp": 45 + (i % 10),
            "max_vel": 0.4 if moving else 0.0, "fsm": None,
            "position": {"x": round(1.0 + 0.1 * i, 2), "y": round(2.0 - 0.05 * i, 2)}}


def cmd_list(cfg: Config) -> None:
    maps = discover_maps(cfg)
    if not maps:
        print(f"(no .db maps under {cfg.maps_dir} for robot '{cfg.robot}')")
        return
    print(f"{len(maps)} map(s) under {cfg.maps_dir} (robot={cfg.robot}):")
    for m in maps:
        pts = load_points(cfg, m.stem)
        print(f"  {m.name:<28} {m.size/1024/1024:6.2f} MB  {len(pts):>3} points  {m.description}")


# --------------------------------------------------------------------------- #
# main
# --------------------------------------------------------------------------- #
def main(argv: Optional[List[str]] = None) -> int:
    ap = argparse.ArgumentParser(description="AGIBOT X2 fleet agent: telemetry + map sync")
    ap.add_argument("--simulate", action="store_true", help="synthetic robot state (map scan stays real)")
    ap.add_argument("--once", action="store_true", help="one map pass + one telemetry post, then exit")
    ap.add_argument("--map-only", action="store_true", help="upload discovered maps once (no state source, no telemetry), then exit")
    ap.add_argument("--dry-run", action="store_true", help="build payloads, never POST")
    ap.add_argument("--force", action="store_true", help="re-upload maps even if unchanged")
    ap.add_argument("--list", action="store_true", help="list discovered maps and exit")
    ap.add_argument("--interval", type=float, default=None, help="override telemetry POLL_INTERVAL")
    ap.add_argument("-v", "--verbose", action="store_true")
    args = ap.parse_args(argv)

    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO,
                        format="%(asctime)s %(levelname)s %(name)s: %(message)s")
    # buffer our own log lines for shipping to /{sn}/logs
    _LOG_RING.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s"))
    logging.getLogger().addHandler(_LOG_RING)
    _load_dotenv()
    cfg = Config.from_env()
    if args.interval is not None:
        cfg.poll_interval = args.interval

    if args.list:
        cmd_list(cfg)
        return 0

    _init_start_times(cfg)
    global _PROJECT_TAIL, _LOG_ALERTS
    _PROJECT_TAIL = ProjectLogTail(cfg)
    _LOG_ALERTS = LogAlertScanner(cfg)   # error/billing alerts from the project logs
    mac = read_mac(cfg.mac_interface)
    log.info("sanad_api_x2 — sn=%s name=%s mac=%s server=%s iface=%s pos=%s "
             "map_dir=%s map_every=%.0fs%s",
             cfg.sn, cfg.name, mac, cfg.server_url, cfg.net_interface,
             cfg.position_source, cfg.maps_dir, cfg.map_poll_interval,
             " [SIMULATE]" if args.simulate else "")

    session = requests.Session()

    if args.map_only:
        # map upload only — search all map roots (incl. EXTRA_MAP_DIRS) and ship,
        # without opening a state source or posting telemetry (won't disturb a live feed).
        map_sync_once(cfg, session, force=args.force, dry_run=args.dry_run)
        return 0

    reader = None
    pos = None
    if not args.simulate:
        reader = AgiBotSource(cfg)
        if cfg.position_source == "rosbridge":
            pos = RosbridgePosition(cfg)
        elif cfg.position_source == "ros2":
            # independent of X2_SOURCE: the Control Dash has no odometry, ROS does
            pos = Ros2Position(cfg)
        time.sleep(1.0)

    tick = 0

    def one_telemetry() -> None:
        nonlocal tick
        sim = _sim_state(tick) if args.simulate else None
        payload = build_telemetry(cfg, mac, reader, pos, sim=sim)
        if args.dry_run:
            log.info("[dry-run] %s", json.dumps(payload))
        else:
            post_telemetry(cfg, payload, session)
            send_alerts(cfg, session, payload.get("faults") or [])
        tick += 1

    if args.once or args.dry_run:
        # one map pass first so the telemetry "map" field reflects it
        map_sync_once(cfg, session, force=args.force, dry_run=args.dry_run)
        if cfg.remote_enable and not args.dry_run:
            register_remote(cfg, session)
        elif cfg.remote_enable:
            d = discover_dashboard(cfg)
            if d:
                _REMOTE_STAT.update(url=d["url"], port=d["port"], kind=cfg.remote_kind, ok=None)
        if args.once:
            one_telemetry()
            ship_logs(cfg, session)
            return 0
        for _ in range(3):
            one_telemetry()
            time.sleep(min(cfg.poll_interval, 1.0))
        return 0

    # loop mode: map sync + log shipping + remote registration in bg threads
    threading.Thread(target=map_loop, args=(cfg, session), daemon=True).start()
    threading.Thread(target=logs_loop, args=(cfg, session), daemon=True).start()
    threading.Thread(target=alert_scan_loop, args=(cfg, session), daemon=True).start()
    if cfg.remote_enable:
        threading.Thread(target=remote_loop, args=(cfg, session), daemon=True).start()
    log.info("telemetry every %.1fs; map check every %.0fs; logs every %.0fs (Ctrl-C to stop)",
             cfg.poll_interval, cfg.map_poll_interval, cfg.logs_interval)
    while True:
        try:
            one_telemetry()
        except Exception as e:
            log.exception("telemetry tick failed: %s", e)
        try:
            time.sleep(cfg.poll_interval)
        except KeyboardInterrupt:
            log.info("stopped")
            return 0


if __name__ == "__main__":
    sys.exit(main())