#!/usr/bin/env python3
"""
tiny_beacon.py — TinyBeacon v2 on Linux (BlueZ), for Reachy Mini (CM4), Raspberry
Pi, Arduino UNO Q, Thor (Jetson), any node running tiny-tech. Spec: tinyai-id docs/ble-mesh/SPEC.md.

Zero pip dependencies. Three radio paths (--via):
  dbus         (default when python3-dbus + GLib import) — bluetoothd's LEAdvertisingManager1 AND
               GattManager1 over D-Bus: the FULL mfg AD (0xFFFF 'TN' header + mine tail) goes on
               air with the Catalog UUID in the scan response, flags.catalog is SET and the Tiny
               Catalog GATT service (SPEC §3: identity / tools / hello / endpoint) is served from
               this process. Runs as a plain user (bluetooth group), no root.
  bluetoothctl (advertising only, flags.catalog off) — the v0.1 fallback, any user.
  btmgmt       (advertising only, root) — raw add-adv.

  python3 tiny_beacon.py --kind linux --tools-from daemon --fetch-secret        # Thor / Reachy / Pi
  python3 tiny_beacon.py --kind q-brain --tools 8 --via bluetoothctl            # UNO Q (no GATT)
  python3 tiny_beacon.py --encode --kind reachy --tools 9                       # AD bytes, no radio (tests)

--tools-from daemon reads the running tiny-tech daemon's roster from its mesh file registry
($TMPDIR/tiny/mesh_registry.json, src/mesh/registry.ts) — the real tool NAMES and count, refreshed
every 5 min — so the AD header and the Catalog say 34, not a hand-typed 7.
"""
import argparse, json, os, random, shutil, socket, struct, subprocess, sys, time, hmac, hashlib, base64

VERSION2 = 0x10
SCRIPT_VERSION = "0.2.0"  # identity.fw = "tiny-beacon <this>"
COMPANY = 0xFFFF
MAGIC = b"TN"
CATALOG_UUID = "f0e60ba4-b906-41fb-9d15-bb1312d27564"
# SPEC §3: characteristic UUIDs = the service UUID with the first group replaced
CHAR_IDENTITY = "00000001-b906-41fb-9d15-bb1312d27564"
CHAR_TOOLS = "00000002-b906-41fb-9d15-bb1312d27564"
CHAR_HELLO = "00000003-b906-41fb-9d15-bb1312d27564"
CHAR_ENDPOINT = "00000004-b906-41fb-9d15-bb1312d27564"
ATT_VALUE_MAX = 512  # Core Spec Vol 3 Part F §3.2.9 — identity/endpoint cap
TOOLS_PAGE_MAX = 500  # SPEC §3 tools page cap
HELLO_RING = 32
KINDS = {"unknown": 0, "nicla-vision": 1, "nicla-voice": 2, "nicla-sense": 3, "iphone": 4, "mac": 5, "android": 6,
         "linux": 7, "sticky": 8, "ring": 9, "reachy": 10, "scout": 11, "fomo": 12, "q-brain": 13, "microduck": 14,
         "wisp": 15, "flipper": 16, "pineapple": 17, "fly-neo": 18, "glasses": 19, "printer": 20, "lamp": 21, "leaf": 22,
         "spidey": 23, "ipad": 24, "watch": 25}
FLAGS = {"provisioned": 1 << 0, "social": 1 << 1, "catalog": 1 << 2, "battery": 1 << 3, "mine": 1 << 5, "agent": 1 << 6}
# bit order = TinyBeaconV2.Caps (Swift) = Cap (TS) — fixtures.json is the referee
CAPS = {"camera": 1 << 0, "mic": 1 << 1, "speaker": 1 << 2, "display": 1 << 3, "imu": 1 << 4, "gps": 1 << 5, "motors": 1 << 6,
        "lora": 1 << 7, "wifi": 1 << 8, "uwb": 1 << 9, "health": 1 << 10, "env": 1 << 11, "touch": 1 << 12, "eink": 1 << 13,
        "relay": 1 << 14}
DEFAULT_CAPS = {"linux": ["wifi", "relay"], "reachy": ["camera", "mic", "speaker", "motors", "imu", "wifi", "relay"],
                "scout": ["camera", "motors", "wifi"], "fomo": ["camera", "motors", "wifi"], "q-brain": ["wifi", "relay"]}
USER_AGENT = "tiny-beacon/0.19 (+tiny-tech native/tinyble/linux/tiny_beacon.py)"
DEFAULT_RSSI1M = -63  # RPi/CM4 class antenna at 1 m (measured on the bench: -61…-66)
ROSTER_REFRESH_S = 300
REGISTRY_STALE_MS = 30_000  # src/mesh/registry.ts STALE_MS
INDEX = int(os.environ.get("HCI_INDEX", "0"))
ADV_MIN_MS = int(os.environ.get("TINY_BLE_ADV_MIN_MS", "250"))
ADV_MAX_MS = int(os.environ.get("TINY_BLE_ADV_MAX_MS", "400"))


def header(kind, flags, caps, tools, session, rssi1m):
    return struct.pack("<BBBHBHb", VERSION2, kind, flags, caps, min(tools, 255), session, rssi1m)


def mfg_ad(kind, flags, caps, tools, session, rssi1m, mine_tail=b""):
    """FULL form. SPEC §2.4 mine tail: with flags.mine the 3 raw HMAC bytes behind the 4-char tag ride
    after the header (18 B AD) — a node with no name carrier is otherwise never 'your UNO Q'."""
    body = struct.pack("<H", COMPANY) + MAGIC + header(kind, flags, caps, tools, session, rssi1m) + (mine_tail if flags & FLAGS["mine"] else b"")
    return bytes([len(body) + 1, 0xFF]) + body  # AD structure: len, type 0xFF, payload


def uuid_bytes_le():
    return bytes.fromhex(CATALOG_UUID.replace("-", ""))[::-1]  # 128-bit UUID, little-endian on air


def uuid_ad():
    u = uuid_bytes_le()
    return bytes([len(u) + 1, 0x07]) + u


def name_ad(name):
    b = name.encode("utf-8")[:8]
    return bytes([len(b) + 1, 0x09]) + b


def mine_tail(secret, session):
    return hmac.new(secret, struct.pack("<H", session), hashlib.sha256).digest()[:3]


def mine_tag(secret, session):
    return base64.urlsafe_b64encode(mine_tail(secret, session)).decode().rstrip("=")


def secret_path():
    return os.path.expanduser(os.environ.get("TINY_BLE_SECRET", "~/.tiny/ble-mine-secret"))


def load_secret():
    p = secret_path()
    try:
        d = open(p, "rb").read()
        if len(d) == 32:
            return d
    except FileNotFoundError:
        pass
    d = os.urandom(32)
    save_secret(d)
    return d


def save_secret(d):
    p = secret_path()
    os.makedirs(os.path.dirname(p), exist_ok=True)
    with open(p, "wb") as f:
        f.write(d)
    os.chmod(p, 0o600)


def decode_secret(s):
    """Wire shape of /api/devices/ble-secret: base64url of exactly 32 bytes (43 chars)."""
    if not isinstance(s, str) or len(s) != 43:
        return None
    try:
        d = base64.urlsafe_b64decode(s + "=")
    except Exception:
        return None
    return d if len(d) == 32 else None


def device_json_path(p=None):
    return os.path.expanduser(p or os.environ.get("TINY_DEVICE_JSON", "~/.tiny/device.json"))


def load_device(device_json=None):
    """tiny-tech's ~/.tiny/device.json ({deviceId, token, name, apiUrl}) or {} — never raises."""
    try:
        d = json.load(open(device_json_path(device_json)))
        return d if isinstance(d, dict) else {}
    except Exception:
        return {}


def device_id_of(d):
    return d.get("deviceId") or d.get("device_id")


def fetch_account_secret(device_json=None, api=None):
    """ROLLOUT §2.5: the ACCOUNT's `mine` secret, so this node reads as 'your Reachy / your UNO Q'
    on the owner's phone and Mac. Auth = the enrolled device token in ~/.tiny/device.json (tiny-tech
    `enroll`) — the same credential heartbeats use; no session needed. Returns bytes or None (keeps
    the per-install secret on any failure — never blocks advertising on the network)."""
    import urllib.request
    d = load_device(device_json)
    dev_id, tok = device_id_of(d), d.get("token") or d.get("device_token")
    if not dev_id or not tok:
        return None
    base = (api or os.environ.get("TINY_API") or d.get("apiUrl") or "https://tiny.technology").rstrip("/")
    req = urllib.request.Request(base + "/api/devices/ble-secret", data=json.dumps({"deviceId": dev_id, "token": tok}).encode(),
                                 # a bare urllib User-Agent is 403'd at the edge (seen on the UNO Q, 2026-09-21)
                                 headers={"content-type": "application/json", "User-Agent": USER_AGENT}, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=10) as r:
            body = json.load(r)
    except Exception:
        return None
    sec = decode_secret(body.get("secret")) if isinstance(body, dict) else None
    if sec:
        save_secret(sec)
    return sec


# ── roster: the running tiny-tech daemon's tools ──────────────────────────────────────────────

def registry_path():
    """tiny-tech src/mesh/registry.ts: join(os.tmpdir(), 'tiny', 'mesh_registry.json'). Node's tmpdir()
    honours TMPDIR/TMP/TEMP, else /tmp."""
    env = os.environ.get("TINY_MESH_REGISTRY")
    if env:
        return os.path.expanduser(env)
    tmp = os.environ.get("TMPDIR") or os.environ.get("TMP") or os.environ.get("TEMP") or "/tmp"
    return os.path.join(tmp.rstrip("/") or "/", "tiny", "mesh_registry.json")


def pid_alive(pid):
    try:
        os.kill(int(pid), 0)
        return True
    except (ProcessLookupError, ValueError, TypeError):
        return False
    except PermissionError:
        return True


def roster_from_registry(path=None, now_ms=None, require_pid=True):
    """The freshest LIVE daemon entry's tool names. Registry entries: {id, type, pid, last_seen(ms),
    metadata{tools[≤50], tool_count, model}}. Live = last_seen within STALE_MS (30 s) — and, on this
    box, a pid that still exists (a daemon killed with SIGKILL leaves its row until TTL).
    Returns {"names": [...], "count": N, "id": ..., "model": ...} or None."""
    path = path or registry_path()
    try:
        agents = json.load(open(path)).get("agents") or {}
    except Exception:
        return None
    now_ms = now_ms or int(time.time() * 1000)
    best = None
    for aid, e in agents.items():
        if not isinstance(e, dict):
            continue
        m = e.get("metadata") or {}
        seen = e.get("last_seen") or 0
        if now_ms - seen > REGISTRY_STALE_MS:
            continue
        if require_pid and e.get("pid") is not None and not pid_alive(e["pid"]):
            continue
        names = [str(t) for t in (m.get("tools") or []) if t]
        count = int(m.get("tool_count") or len(names))
        if not names and not count:
            continue
        if best is None or seen > best["seen"]:
            best = {"names": sorted(set(names)), "count": max(count, len(names)), "id": aid, "model": m.get("model"), "seen": seen}
    return best


class Roster:
    """What the AD header's `tools` byte and the Catalog's tools pages say. Static (--tools N) or the
    daemon's registry row (--tools-from daemon), re-read every ROSTER_REFRESH_S."""

    def __init__(self, static_count=0, source="static", registry=None):
        self.source, self.registry = source, registry
        self.names, self.count, self.daemon = [], static_count, None
        self.refresh()

    def refresh(self):
        if self.source != "daemon":
            return False
        r = roster_from_registry(self.registry)
        if not r:
            return False
        changed = (r["names"], r["count"]) != (self.names, self.count)
        self.names, self.count, self.daemon = r["names"], r["count"], r["id"]
        return changed


def tools_pages(names, page_max=TOOLS_PAGE_MAX, descs=None):
    """SPEC §3 tools pages: {"p":i,"n":pages,"items":[{"name":..,"desc"?:..}]} each ≤ page_max bytes.
    Greedy fill; "p"/"n" are sized with 2-digit placeholders so the real numbers never push a page
    over. No URL on the wire (the fleet registry has the schemas; the phone matches by name)."""
    descs = descs or {}
    pages, cur = [], []

    def size(items):
        return len(json.dumps({"p": 99, "n": 99, "items": items}, separators=(",", ":")).encode())

    for name in names:
        item = {"name": name}
        if descs.get(name):
            item["desc"] = str(descs[name])[:64]
        if cur and size(cur + [item]) > page_max:
            pages.append(cur)
            cur = []
        cur.append(item)
    pages.append(cur)
    return [json.dumps({"p": i, "n": len(pages), "items": items}, separators=(",", ":")).encode() for i, items in enumerate(pages)]


def lan_ip():
    """The address we would use to reach the LAN (no packets sent)."""
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.connect(("10.255.255.255", 1))
        ip = s.getsockname()[0]
        s.close()
        return ip
    except Exception:
        return None


def port_listening(port, host="127.0.0.1"):
    try:
        with socket.create_connection((host, port), timeout=0.3):
            return True
    except Exception:
        return False


def dev_hash(device_id):
    """identity.dev = sha256(device_id) first 8 bytes, hex 16 (SPEC §3) — never the id itself."""
    return hashlib.sha256(device_id.encode()).hexdigest()[:16] if device_id else ""


def platform_string():
    """Same shape as tiny-tech devicePlatform(): `${platform()}-${arch()}` → linux-arm64 / linux-x64."""
    u = os.uname()
    arch = {"aarch64": "arm64", "x86_64": "x64", "armv7l": "arm", "armv6l": "arm"}.get(u.machine, u.machine)
    return f"{u.sysname.lower()}-{arch}"


def identity_json(kind, flags, caps, tools, session, name=None, handle=None, social=False, device_id=None):
    o = {"v": 2, "kind": kind, "dev": dev_hash(device_id), "platform": platform_string(),
         "fw": f"tiny-beacon {SCRIPT_VERSION}", "tools": tools, "caps": caps, "session": session, "flags": flags}
    # §4.3: a display name is fine on a DEVICE kind; the @handle rides only when social.
    if name:
        o["name"] = name[:32]
    if social and handle:
        o["handle"] = handle
    b = json.dumps(o, separators=(",", ":")).encode()
    if len(b) > ATT_VALUE_MAX:
        o.pop("name", None)
        b = json.dumps(o, separators=(",", ":")).encode()
    return b


def endpoint_json(device_id=None, ip=None, zenoh_port=7447, http=None):
    """SPEC §3: only fields that exist. zenoh only when the locator port is actually listening."""
    o = {}
    ip = ip or lan_ip()
    if ip:
        o["ip"] = ip
    if device_id:
        o["device"] = f"tinyapp://device/{device_id}"
    loc = os.environ.get("TINY_ZENOH_LOCATOR")
    if loc:
        o["zenoh"] = loc
    elif ip and port_listening(zenoh_port):
        o["zenoh"] = f"tcp/{ip}:{zenoh_port}"
    if http:
        o["http"] = http
    return json.dumps(o, separators=(",", ":")).encode()


def build_flags(args, catalog=False):
    if args.flags:
        flags = 0
        for f in args.flags.split(","):
            flags |= FLAGS[f.strip()]
        return flags
    flags = FLAGS["provisioned"] | FLAGS["agent"] | FLAGS["mine"] | (FLAGS["social"] if args.social else 0)
    if catalog:
        flags |= FLAGS["catalog"]
    return flags


def build_caps(args):
    caps = 0
    for c in (args.caps.split(",") if args.caps else DEFAULT_CAPS.get(args.kind, ["wifi"])):
        caps |= CAPS[c.strip()]
    return caps


def build(args, session, catalog=False, tools=None):
    kind = KINDS[args.kind]
    flags = build_flags(args, catalog)
    caps = build_caps(args)
    # 31-byte legacy PDU: FULL mfg (18 B) + 128-bit UUID (18 B) does not fit. The FULL form is
    # self-witnessed by the 'TN' magic (SPEC §2.4), so the UUID is dropped from the AD here and
    # the name (social only, ≤ 8 B) rides alongside. The dbus path puts the UUID in the SCAN_RSP,
    # exactly like the Sticky (adv = mfg, scan-rsp = UUID).
    ad = mfg_ad(kind, flags, caps, args.tools if tools is None else tools, session, args.rssi1m, mine_tail(load_secret(), session))
    if args.social and args.name:
        ad += name_ad(args.name)
    return ad


def mfg_payload_of(ad):
    """The bytes after len/type/company of the FIRST AD structure — exactly 'TN' + 9-byte header (+ tail).
    (An earlier slice took ad[0]-1 bytes and bled two bytes of the social name AD into the mfg data.)"""
    return ad[4:1 + ad[0]]


def emit(o):
    print(json.dumps(o), flush=True)


# ── path 1: bluetoothctl (advertising only) ───────────────────────────────────────────────────

def bluetoothctl_script(ad_mfg_payload, name, social, seconds):
    """Unprivileged path: bluetoothd's LEAdvertisingManager over D-Bus via bluetoothctl.
    Advertising lives as long as the bluetoothctl process, so we keep it open."""
    hexbytes = " ".join(f"0x{b:02x}" for b in ad_mfg_payload)
    # ROLLOUT §4.2: BlueZ's default interval gave ~1 sighting / 10 s from the UNO Q; RSSI bands need a
    # stream. 250–400 ms = the Sticky's numbers (bluetoothctl takes the interval in ms, min then max).
    lines = ["menu advertise", "clear", f"manufacturer 0xffff {hexbytes}",
             f"name {name[:8]}" if social and name else "name off", "discoverable on",
             f"interval {ADV_MIN_MS} {ADV_MAX_MS}", "back", "advertise peripheral"]
    return "\n".join(lines) + "\n"


def advertise_bluetoothctl(ad_mfg_payload, name, social, seconds):
    p = subprocess.Popen(["bluetoothctl"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
    p.stdin.write(bluetoothctl_script(ad_mfg_payload, name, social, seconds)); p.stdin.flush()
    time.sleep(2)
    return p


# ── path 2: btmgmt (root, advertising only) ───────────────────────────────────────────────────

def btmgmt(*a):
    return subprocess.run(["btmgmt", "--index", str(INDEX), *a], capture_output=True, text=True, timeout=10)


def advertise_once(ad):
    btmgmt("rm-adv", "1")
    # -i/-x = min/max interval in 0.625 ms units (bluez ≥ 5.55); older btmgmt ignores unknown flags → retry plain.
    r = btmgmt("add-adv", "-d", ad.hex(), "-g", "-i", str(int(ADV_MIN_MS / 0.625)), "-x", str(int(ADV_MAX_MS / 0.625)), "1")
    if "Advertising added" not in r.stdout:
        r = btmgmt("add-adv", "-d", ad.hex(), "-g", "1")  # -g general discoverable; instance 1
    if "Advertising added" not in r.stdout:
        raise SystemExit(json.dumps({"error": (r.stdout + r.stderr).strip()[:300]}))


# ── path 3: D-Bus — advertising + Tiny Catalog GATT (SPEC §3) ─────────────────────────────────

def bluez_version():
    """(major, minor) from `bluetoothctl --version` ('bluetoothctl: 5.72'), or None."""
    try:
        out = subprocess.run(["bluetoothctl", "--version"], capture_output=True, text=True, timeout=5).stdout
        v = out.strip().split()[-1].split(".")
        return int(v[0]), int(v[1])
    except Exception:
        return None


def bluez_experimental(conf="/etc/bluetooth/main.conf"):
    """bluetoothd honours LEAdvertisement1 MinInterval/MaxInterval only when [General] Experimental = true."""
    try:
        for line in open(conf):
            t = line.strip().lower().replace(" ", "")
            if t.startswith("experimental="):
                return t.split("=", 1)[1] in ("true", "1", "yes")
    except Exception:
        pass
    return False


def radio_caps():
    v = bluez_version()
    return {"bluez": f"{v[0]}.{v[1]}" if v else None,
            "scan_rsp_uuid": bool(v and v >= (5, 80)),  # ScanResponseServiceUUIDs landed in 5.80
            "interval_honoured": bluez_experimental()}


def dbus_available():
    try:
        import dbus, dbus.mainloop.glib  # noqa: F401
        from gi.repository import GLib  # noqa: F401
        return True
    except Exception:
        return False


def make_dbus_classes():
    """Built lazily so --encode / tests / the bluetoothctl path never import dbus."""
    import dbus, dbus.service
    BLUEZ = "org.bluez"
    DBUS_OM = "org.freedesktop.DBus.ObjectManager"
    DBUS_PROP = "org.freedesktop.DBus.Properties"
    LE_ADV = "org.bluez.LEAdvertisement1"
    GATT_SERVICE = "org.bluez.GattService1"
    GATT_CHRC = "org.bluez.GattCharacteristic1"

    class InvalidArgs(dbus.exceptions.DBusException):
        _dbus_error_name = "org.freedesktop.DBus.Error.InvalidArgs"

    class NotSupported(dbus.exceptions.DBusException):
        _dbus_error_name = "org.bluez.Error.NotSupported"

    class InvalidOffset(dbus.exceptions.DBusException):
        _dbus_error_name = "org.bluez.Error.InvalidOffset"

    class Advertisement(dbus.service.Object):
        """org.bluez.LEAdvertisement1 — Type peripheral (connectable, so the Catalog can be read),
        ManufacturerData 0xFFFF = 'TN' + header + mine tail (the same bytes bluetoothctl sent);
        ScanResponseServiceUUIDs = [Catalog UUID] → the SCAN_RSP, like the Sticky (BlueZ ≥ 5.80 —
        5.72 has no scan-response properties and silently ignores the key, so on Ubuntu 24.04 the
        UUID is NOT on air and the FULL mfg + flags.catalog is the whole witness; the Mac/phone
        still connect by peripheral id). LocalName (≤ 8 B, scan response) only with --social (§4.1).
        Min/MaxInterval = ROLLOUT §4.2 stream (250–400 ms) — bluetoothd 5.72 only reads them with
        `Experimental = true` in /etc/bluetooth/main.conf (src/advertising.c parse_min_interval),
        otherwise the kernel default 1280 ms applies (measured on Thor with btmon)."""
        PATH = "/org/tiny/beacon/adv0"

        def __init__(self, bus, mfg_payload, social_name=None, include_uuid=True, interval=True):
            self.mfg_payload = bytes(mfg_payload)
            self.social_name = social_name
            self.include_uuid = include_uuid
            self.interval = interval
            super().__init__(bus, self.PATH)

        def props(self):
            p = {"Type": dbus.String("peripheral"),
                 "ManufacturerData": dbus.Dictionary({dbus.UInt16(COMPANY): dbus.Array([dbus.Byte(b) for b in self.mfg_payload], signature="y")}, signature="qv"),
                 "Discoverable": dbus.Boolean(True)}
            if self.include_uuid:
                p["ScanResponseServiceUUIDs"] = dbus.Array([dbus.String(CATALOG_UUID)], signature="s")
            if self.social_name:
                p["LocalName"] = dbus.String(self.social_name.encode("utf-8")[:8].decode("utf-8", "ignore"))
            if self.interval:
                p["MinInterval"] = dbus.UInt32(ADV_MIN_MS)
                p["MaxInterval"] = dbus.UInt32(ADV_MAX_MS)
            return p

        @dbus.service.method(DBUS_PROP, in_signature="s", out_signature="a{sv}")
        def GetAll(self, interface):
            if interface != LE_ADV:
                raise InvalidArgs()
            return self.props()

        @dbus.service.method(DBUS_PROP, in_signature="ss", out_signature="v")
        def Get(self, interface, prop):
            return self.props()[prop]

        @dbus.service.method(LE_ADV, in_signature="", out_signature="")
        def Release(self):
            emit({"event": "adv_released"})

    class Application(dbus.service.Object):
        PATH = "/org/tiny/beacon"

        def __init__(self, bus):
            self.services = []
            super().__init__(bus, self.PATH)

        def add(self, s):
            self.services.append(s)

        @dbus.service.method(DBUS_OM, out_signature="a{oa{sa{sv}}}")
        def GetManagedObjects(self):
            out = {}
            for s in self.services:
                out[s.path] = {GATT_SERVICE: s.props()}
                for c in s.chrcs:
                    out[c.path] = {GATT_CHRC: c.props()}
            return out

    class Service(dbus.service.Object):
        def __init__(self, bus, index, uuid):
            self.path = f"{Application.PATH}/service{index}"
            self.uuid, self.chrcs = uuid, []
            super().__init__(bus, self.path)

        def props(self):
            return {"UUID": dbus.String(self.uuid), "Primary": dbus.Boolean(True),
                    "Characteristics": dbus.Array([c.path for c in self.chrcs], signature="o")}

        @dbus.service.method(DBUS_PROP, in_signature="s", out_signature="a{sv}")
        def GetAll(self, interface):
            if interface != GATT_SERVICE:
                raise InvalidArgs()
            return self.props()

    class Characteristic(dbus.service.Object):
        """read → self.read(options) bytes (whole value; BlueZ hands us options['offset'] for the
        ATT Read Blob continuation of a long read); write → self.write(bytes, options)."""

        def __init__(self, bus, service, index, uuid, flags, read=None, write=None):
            self.path = f"{service.path}/char{index}"
            self.service, self.uuid, self.flags = service, uuid, flags
            self._read, self._write = read, write
            service.chrcs.append(self)
            super().__init__(bus, self.path)

        def props(self):
            return {"Service": dbus.ObjectPath(self.service.path), "UUID": dbus.String(self.uuid),
                    "Flags": dbus.Array(self.flags, signature="s")}

        @dbus.service.method(DBUS_PROP, in_signature="s", out_signature="a{sv}")
        def GetAll(self, interface):
            if interface != GATT_CHRC:
                raise InvalidArgs()
            return self.props()

        @dbus.service.method(GATT_CHRC, in_signature="a{sv}", out_signature="ay")
        def ReadValue(self, options):
            if not self._read:
                raise NotSupported()
            opts = {str(k): v for k, v in options.items()}
            value = self._read(opts)
            off = int(opts.get("offset", 0))
            if off > len(value):
                raise InvalidOffset()
            return dbus.Array([dbus.Byte(b) for b in value[off:]], signature="y")

        @dbus.service.method(GATT_CHRC, in_signature="aya{sv}", out_signature="")
        def WriteValue(self, value, options):
            if not self._write:
                raise NotSupported()
            self._write(bytes(bytearray(value)), {str(k): v for k, v in options.items()})

        @dbus.service.method(GATT_CHRC)
        def StartNotify(self):
            raise NotSupported()

        @dbus.service.method(GATT_CHRC)
        def StopNotify(self):
            raise NotSupported()

    return {"Advertisement": Advertisement, "Application": Application, "Service": Service, "Characteristic": Characteristic,
            "BLUEZ": BLUEZ, "DBUS_OM": DBUS_OM, "DBUS_PROP": DBUS_PROP}


class Catalog:
    """State behind the four characteristics (SPEC §3). Per-central page + hello, keyed by the
    BlueZ device object path in options['device']; a 32-entry hello ring."""

    def __init__(self, args, roster, session, device):
        self.args, self.roster, self.session, self.device = args, roster, session, device
        self.page, self.hello_seen, self.hellos = {}, set(), []

    @staticmethod
    def _central(options):
        return str(options.get("device", "*"))

    def _offset(self, options):
        return int(options.get("offset", 0))

    def read_identity(self, options):
        d = identity_json(KINDS[self.args.kind], build_flags(self.args, catalog=True), build_caps(self.args), self.roster.count, self.session,
                          name=self.args.name, handle=self.args.handle, social=self.args.social, device_id=device_id_of(self.device))
        if self._offset(options) == 0:
            emit({"event": "read", "char": "identity", "central": self._central(options), "bytes": len(d)})
        return d

    def read_tools(self, options):
        c = self._central(options)
        if self.args.social and c not in self.hello_seen:
            return b'{"p":0,"n":0,"items":[],"err":"hello_first"}'
        pages = tools_pages(self.roster.names)
        p = min(self.page.get(c, 0), len(pages) - 1)
        if self._offset(options) == 0:
            emit({"event": "read", "char": "tools", "central": c, "page": p, "pages": len(pages), "bytes": len(pages[p])})
        return pages[p]

    def write_tools(self, value, options):
        self.page[self._central(options)] = value[0] if value else 0

    def write_hello(self, value, options):
        c = self._central(options)
        self.hello_seen.add(c)
        txt = value[:512].decode("utf-8", "replace")
        try:
            h = json.loads(txt)
        except Exception:
            h = {"raw": txt}
        self.hellos.append({"t": time.time(), "central": c, "hello": h})
        del self.hellos[:-HELLO_RING]
        emit({"event": "hello", "central": c, "json": h})

    def read_endpoint(self, options):
        d = endpoint_json(device_id_of(self.device), http=self.args.http or None)
        if self._offset(options) == 0:
            emit({"event": "read", "char": "endpoint", "central": self._central(options), "bytes": len(d)})
        return d


def run_dbus(args, roster, session):
    """Advertise + serve the Catalog until SIGTERM. Session rotation re-registers the advertisement
    (BlueZ reads LEAdvertisement1 properties once, at RegisterAdvertisement) and updates
    identity.session in place; the GATT application stays registered."""
    import dbus, dbus.mainloop.glib
    from gi.repository import GLib
    K = make_dbus_classes()
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
    bus = dbus.SystemBus()
    om = dbus.Interface(bus.get_object(K["BLUEZ"], "/"), K["DBUS_OM"])
    adapter = None
    for path, ifaces in om.GetManagedObjects().items():
        if "org.bluez.LEAdvertisingManager1" in ifaces and "org.bluez.GattManager1" in ifaces:
            if adapter is None or path.endswith(f"/hci{INDEX}"):
                adapter = path
    if not adapter:
        raise SystemExit(json.dumps({"error": "no BlueZ adapter with LEAdvertisingManager1 + GattManager1 (bluetoothd running? bluez ≥ 5.50?)"}))
    props = dbus.Interface(bus.get_object(K["BLUEZ"], adapter), K["DBUS_PROP"])
    if not props.Get("org.bluez.Adapter1", "Powered"):
        props.Set("org.bluez.Adapter1", "Powered", dbus.Boolean(True))
    # §4.1: a central that connects reads GAP Device Name (0x2A00) = the adapter Alias, which BlueZ
    # defaults to the HOSTNAME — on Thor that is the owner's first name. Align it with the device
    # name we serve in identity.name (persisted by bluetoothd; --keep-alias leaves it alone).
    try:
        alias = str(props.Get("org.bluez.Adapter1", "Alias"))
        if args.name and alias != args.name and not args.keep_alias:
            props.Set("org.bluez.Adapter1", "Alias", dbus.String(args.name[:32]))
            emit({"event": "alias", "from": alias, "to": args.name[:32]})
    except Exception as e:
        emit({"event": "alias_skipped", "reason": str(e)[:120]})
    adv_mgr = dbus.Interface(bus.get_object(K["BLUEZ"], adapter), "org.bluez.LEAdvertisingManager1")
    gatt_mgr = dbus.Interface(bus.get_object(K["BLUEZ"], adapter), "org.bluez.GattManager1")

    device = load_device(args.device_json)
    caps = radio_caps()
    state = {"session": session, "adv": None, "registered": False}
    catalog = Catalog(args, roster, session, device)

    app = K["Application"](bus)
    svc = K["Service"](bus, 0, CATALOG_UUID)
    K["Characteristic"](bus, svc, 0, CHAR_IDENTITY, ["read"], read=catalog.read_identity)
    K["Characteristic"](bus, svc, 1, CHAR_TOOLS, ["read", "write"], read=catalog.read_tools, write=catalog.write_tools)
    K["Characteristic"](bus, svc, 2, CHAR_HELLO, ["write", "write-without-response"], write=catalog.write_hello)
    K["Characteristic"](bus, svc, 3, CHAR_ENDPOINT, ["read"], read=catalog.read_endpoint)
    app.add(svc)
    loop = GLib.MainLoop()

    def current_ad():
        return build(args, state["session"], catalog=True, tools=roster.count)

    def start_adv(interval=True):
        ad = current_ad()
        if state["adv"] is not None:
            state["adv"].remove_from_connection()
            state["adv"] = None
        adv = K["Advertisement"](bus, mfg_payload_of(ad), social_name=(args.name if args.social else None), interval=interval)
        state["adv"] = adv

        def ok():
            state["registered"] = True
            emit({"event": "advertising", "via": "dbus", "kind": args.kind, "session": state["session"], "mine": mine_tag(load_secret(), state["session"]),
                  "tools": roster.count, "catalog": True, "len": len(ad), "ad": ad.hex(),
                  "scan_rsp": uuid_ad().hex() if caps["scan_rsp_uuid"] else None,
                  "interval_ms": [ADV_MIN_MS, ADV_MAX_MS] if (interval and caps["interval_honoured"]) else None, **caps})

        def err(e):
            msg = str(e)
            if interval:
                # older bluetoothd rejects MinInterval/MaxInterval → once more with BlueZ's default interval
                emit({"event": "adv_retry", "reason": msg[:200]})
                adv.remove_from_connection(); state["adv"] = None
                start_adv(interval=False)
            else:
                emit({"error": f"RegisterAdvertisement: {msg[:300]}"})
                loop.quit()

        adv_mgr.RegisterAdvertisement(adv.PATH, dbus.Dictionary({}, signature="sv"), reply_handler=ok, error_handler=err)

    def stop_adv(then):
        if not state["registered"]:
            then(); return
        state["registered"] = False

        def done(*_):
            then()
        adv_mgr.UnregisterAdvertisement(state["adv"].PATH, reply_handler=done, error_handler=done)

    def on_app_ok():
        emit({"event": "gatt_registered", "adapter": adapter, "service": CATALOG_UUID, "chars": 4})
        start_adv()

    def on_app_err(e):
        emit({"error": f"RegisterApplication: {e}"})
        loop.quit()

    def rotate():
        state["session"] = random.randint(1, 0xFFFF)
        catalog.session = state["session"]
        if args.fetch_secret:
            fetch_account_secret(args.device_json)  # picks up an owner rotation; fail-open
        stop_adv(start_adv)
        return True

    def refresh_roster():
        if roster.refresh():
            emit({"event": "roster", "tools": roster.count, "names": len(roster.names), "daemon": roster.daemon})
            stop_adv(start_adv)  # the header's tools byte changed
        return True

    gatt_mgr.RegisterApplication(app.PATH, dbus.Dictionary({}, signature="sv"), reply_handler=on_app_ok, error_handler=on_app_err)
    GLib.timeout_add_seconds(max(60, args.rotate), rotate)
    if roster.source == "daemon":
        GLib.timeout_add_seconds(ROSTER_REFRESH_S, refresh_roster)
    if args.seconds:
        GLib.timeout_add_seconds(args.seconds, lambda: (loop.quit(), False)[1])
    try:
        loop.run()
    except KeyboardInterrupt:
        pass
    finally:
        try:
            if state["registered"]:
                adv_mgr.UnregisterAdvertisement(state["adv"].PATH)
            gatt_mgr.UnregisterApplication(app.PATH)
        except Exception:
            pass
    emit({"event": "stopped"})


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--kind", default="linux", choices=sorted(KINDS))
    ap.add_argument("--name", default=None, help="display name (identity.name; on air only with --social). Default: device.json name, else hostname")
    ap.add_argument("--handle", default=None, help="@handle for identity.handle — only served with --social")
    ap.add_argument("--tools", type=int, default=0)
    ap.add_argument("--tools-from", default="static", choices=["static", "daemon"],
                    help="daemon = the running tiny-tech daemon's roster from $TMPDIR/tiny/mesh_registry.json (names + count, refreshed every 5 min)")
    ap.add_argument("--registry", default=None, help="override the mesh registry path (TINY_MESH_REGISTRY)")
    ap.add_argument("--caps", default="")
    ap.add_argument("--flags", default="", help="override flags, e.g. catalog,battery,agent (fixture parity)")
    ap.add_argument("--social", action="store_true")
    ap.add_argument("--rssi1m", type=int, default=DEFAULT_RSSI1M)
    ap.add_argument("--session", type=int, default=0)
    ap.add_argument("--rotate", type=int, default=900, help="seconds between session rotations")
    ap.add_argument("--encode", action="store_true", help="print AD bytes as JSON and exit")
    ap.add_argument("--catalog", action="store_true", help="with --encode: set flags.catalog (what the dbus path puts on air)")
    ap.add_argument("--no-catalog", action="store_true", help="advertising only (v0.1 behaviour): no GATT, flags.catalog off, bluetoothctl/btmgmt path")
    ap.add_argument("--once", action="store_true", help="set the advertisement and exit (stays on until rm-adv)")
    ap.add_argument("--via", default="auto", choices=["auto", "dbus", "btmgmt", "bluetoothctl"],
                    help="dbus = advertising + Catalog GATT as a user; bluetoothctl = advertising only; btmgmt needs root")
    ap.add_argument("--seconds", type=int, default=0, help="stop after N seconds (0 = forever)")
    ap.add_argument("--fetch-secret", action="store_true",
                    help="read the account `mine` secret from /api/devices/ble-secret with ~/.tiny/device.json's device token (ROLLOUT §2.5)")
    ap.add_argument("--device-json", default=None, help="path to tiny-tech's device.json (default ~/.tiny/device.json)")
    ap.add_argument("--keep-alias", action="store_true", help="do not rename the adapter (GAP Device Name) to --name")
    ap.add_argument("--http", default="", help="endpoint.http, e.g. http://thor.local:8080/ (only if you run something there)")
    args = ap.parse_args()

    if args.name is None:
        # the hostname is often the OWNER's name (Thor's is `cagatay`) — prefer the enrolled device name
        args.name = load_device(args.device_json).get("name") or os.uname().nodename

    if args.fetch_secret:
        got = fetch_account_secret(args.device_json)
        emit({"event": "secret", "source": "account" if got else "install"})

    roster = Roster(args.tools, args.tools_from, args.registry)
    if args.tools_from == "daemon":
        emit({"event": "roster", "source": "daemon", "registry": args.registry or registry_path(), "tools": roster.count,
              "names": len(roster.names), "daemon": roster.daemon})

    session = args.session or random.randint(1, 0xFFFF)
    via = args.via
    if via == "auto":
        if not args.no_catalog and dbus_available():
            via = "dbus"
        else:
            via = "btmgmt" if (os.geteuid() == 0 and shutil.which("btmgmt")) else "bluetoothctl"
    if via == "dbus" and args.no_catalog:
        via = "bluetoothctl"
    catalog = (via == "dbus") or (args.encode and args.catalog)
    ad = build(args, session, catalog=catalog, tools=roster.count)
    if len(ad) > 31:
        raise SystemExit(json.dumps({"error": f"AD {len(ad)} bytes > 31 — drop --name or caps"}))
    if args.encode:
        print(json.dumps({"ad": ad.hex(), "len": len(ad), "session": session, "mine": mine_tag(load_secret(), session), "tools": roster.count,
                          "catalog": catalog, "scan_rsp": uuid_ad().hex() if catalog else None}))
        sys.exit(0)
    if via == "dbus":
        if not dbus_available():
            raise SystemExit(json.dumps({"error": "python3-dbus + python3-gi needed for --via dbus (apt install python3-dbus python3-gi)"}))
        run_dbus(args, roster, session)
        return
    if not shutil.which(via):
        raise SystemExit(json.dumps({"error": f"{via} not found (apt install bluez)"}))
    started = time.time()
    proc = None
    while True:
        ad = build(args, session, tools=roster.count)
        mfg_payload = mfg_payload_of(ad)  # 'TN' + header (bluetoothctl adds len/type/company itself)
        if via == "btmgmt":
            btmgmt("power", "on")
            advertise_once(ad)
        else:
            if proc:
                proc.terminate()
            proc = advertise_bluetoothctl(mfg_payload, args.name, args.social, args.rotate)
        emit({"event": "advertising", "via": via, "kind": args.kind, "session": session, "mine": mine_tag(load_secret(), session),
              "tools": roster.count, "catalog": False, "len": len(ad), "ad": ad.hex()})
        if args.once and via == "btmgmt":
            break
        step = args.rotate if not args.seconds else min(args.rotate, max(1, args.seconds - (time.time() - started)))
        time.sleep(step)
        if args.seconds and time.time() - started >= args.seconds:
            break
        session = random.randint(1, 0xFFFF)
        roster.refresh()
        if args.fetch_secret:
            fetch_account_secret(args.device_json)  # picks up an owner rotation; fail-open
    if proc:
        proc.terminate()
    emit({"event": "stopped"})


if __name__ == "__main__":
    main()