# -*- coding: utf-8 -*-
"""
Ananta 注入器（只挂保龄球瓶 + 气球）—— 共享基础设施

★ 名字说明：这一份**不是**「统一加载器 AnantaUnifiedLoader」。
  那份是全通道诊断版（A/B/C/D/E/F 都能开），在
  C:\\Game\\Ananta_mainleakflow_keitaro_gg\\AnantaUnifiedLoader；
  本目录是它的**最小副本**，日常跑服只挂 bowling + balloon 两个模块。

★ 由 AnantaTrainSeatLoader v5.0.0.4 的 loader_common.py 改写。保留它踩过的所有坑：
   · 优雅停机三阶段（先让 agent 停手 → 等在途 RPC → 卸脚本）
   · 卸载超时**绝不 detach**（半拆的会话会把游戏拖死）
   · attach 走子线程 + 超时 detach（attach 本身可能永久阻塞）
   · 单次 RPC 失败**不拆会话**，交给主循环打印真实原因 + 退避重连

★ 与单脚本版的唯一结构性差别：**一个 frida 会话里开多个 script**。
  为什么必须这样（不是合并成一个大 JS）：
    - 每个 agent.js 顶层都是 `var Il2Cpp2;` + 一堆 IIFE，合并会撞全局命名空间；
    - 更致命的是 world_mod.js / gadget_world.js / train_agent.js 都写
      `rpc.exports = { ... }`（**整体覆盖**），合并后只有最后一个模块的导出会活下来；
    - 分成多个 script：各自独立 realm，导出互不覆盖，且仍是**同一个会话**
      （满足「同一时刻只能有一个 frida 会话」这条硬约束）。
"""
import sys, os, time, json, subprocess, atexit, ctypes, threading, re

for _s in (sys.stdout, sys.stderr):
    try:
        _s.reconfigure(encoding="utf-8", errors="replace")
    except Exception:
        pass

_builtin_print = print


def print(*a, **k):
    k["flush"] = True
    try:
        _builtin_print(*a, **k)
    except Exception:
        pass


STOP = {"ask": False, "done": False}

RPC_TIMEOUT = 25.0
ATTACH_TIMEOUT = 60.0

# ---- 超时治理（2026-09-29）----
# ⛑ frida 的同步 RPC 一旦卡住，Python 侧**没有任何办法**让那个调用返回：
#   t.join(timeout) 只是「放弃等待」，线程仍卡在 _frida 的阻塞调用里，
#   daemon 线程要到进程退出才回收 ⇒ 主循环每超时一次就永久多一个线程。
# 治理靠三层，见 Mod._run_guarded 的注释。
_STUCK_LOCK = threading.Lock()
_STUCK_THREADS = []          # 只保留**还活着**的卡住线程（每次统计前先清理）
# 同一个 script 连续超时这么多次 ⇒ 判定该 script 彻底废了，快速失败不再堆新调用
STUCK_MAX_CONSECUTIVE = 3


def stuck_thread_count():
    """当前卡在 frida 同步调用里、join 不回来的线程数（供上层观测/自检）。"""
    with _STUCK_LOCK:
        _STUCK_THREADS[:] = [x for x in _STUCK_THREADS if x.is_alive()]
        return len(_STUCK_THREADS)

# 这些是「别的独立加载器 exe」。本统一加载器一开，它们就必须全部关闭，
# 否则同一个游戏进程上会有两个 frida 会话 —— 实测会把游戏带崩。
OTHER_EXES = [
    "AnantaGymLoader.exe",
    "AnantaTrainSeatLoader.exe",
    "AnantaBowlingLoader.exe",
    "AnantaBalloonLoader.exe",
    "AnantaGadgetLoader.exe",
    "AnantaWorldMod.exe",
    "AnantaPropScanner.exe",
]


def _install_ctrl_handler():
    if os.name != "nt":
        return False
    try:
        k32 = ctypes.windll.kernel32
        PROT = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_uint)

        def handler(ev):
            if ev not in (0, 1, 2, 5, 6):
                return False
            STOP["ask"] = True
            try:
                _atexit_close()
            except Exception:
                pass
            for _ in range(280):
                if STOP["done"]:
                    break
                k32.Sleep(50)
            return True

        cb = PROT(handler)
        handler._cb = cb
        k32.SetConsoleCtrlHandler(cb, True)
        return True
    except Exception:
        return False


HAS_CTRL = _install_ctrl_handler()


def sleep_slice(sec):
    end = time.time() + sec
    while time.time() < end:
        if STOP["ask"]:
            return
        time.sleep(min(0.2, max(0.0, end - time.time())))


def ts():
    return time.strftime("%H:%M:%S")


def _tasklist_rows():
    """全量 tasklist 一次，返回 [(进程名小写, pid)]。

    ⛑ 为什么不用 `tasklist /FI "IMAGENAME eq xxx"`：
       实测（2026-09-28）游戏是提权启动的，`/FI` 过滤会直接回
       「没有运行的任务匹配指定标准」，而**不带 filter 的全量列表里明明有**。
       所以一律走全量解析，filter 那条路不可信。
    """
    rows = []
    try:
        out = subprocess.check_output(
            'tasklist /FO CSV /NH', shell=True,
            text=True, encoding="mbcs", errors="replace")
    except Exception:
        return rows
    for line in out.splitlines():
        parts = [p.strip().strip('"') for p in line.split('","')]
        if len(parts) < 2:
            continue
        try:
            rows.append((parts[0].lower(), int(parts[1].replace(",", ""))))
        except ValueError:
            continue
    return rows


def resolve_pid():
    """找 Ananta.exe 的 pid：全量 tasklist → frida 枚举（两道保险）。"""
    for name, pid in _tasklist_rows():
        if name in ("ananta.exe", "ananta"):
            return pid
    # 回退：连 tasklist 都看不到时（更严格的权限隔离），用 frida 自己枚举
    try:
        import frida
        dev = frida.get_local_device()
        for p in dev.enumerate_processes():
            if p.name.lower() in ("ananta.exe", "ananta"):
                return p.pid
    except Exception:
        pass
    return None


def count_exe(name):
    n = name.lower()
    return len([1 for nm, _ in _tasklist_rows() if nm == n])


def _pid_alive_win(pid):
    """OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION)。

    ⛑ 为什么不用 tasklist 做唯一判据：
      tasklist 在某些环境（受限 shell / 沙箱）会返回**空列表**，
      那时 pid_alive 恒 False ⇒ 互斥锁形同虚设（两份加载器照样起得来）。
      PROCESS_QUERY_LIMITED_INFORMATION(0x1000) 自 Vista 起对**任何**进程都放行
      （含提权进程、含别的会话），只要 pid 存在就拿得到句柄 —— 是最可靠的存在性判据。
    """
    try:
        k32 = ctypes.windll.kernel32
        k32.OpenProcess.restype = ctypes.c_void_p
        k32.OpenProcess.argtypes = [ctypes.c_uint32, ctypes.c_bool, ctypes.c_uint32]
        h = k32.OpenProcess(0x1000, False, int(pid))
        if not h:
            return False
        try:
            # ⛑ 光拿到句柄**不能**算活着：只要还有人握着该进程的句柄，
            #   进程对象就不会销毁，OpenProcess 照样成功（实测踩到）。
            #   退出码 != STILL_ACTIVE(259) ⇒ 它已经死了。
            code = ctypes.c_uint32()
            ok = k32.GetExitCodeProcess(ctypes.c_void_p(h), ctypes.byref(code))
            return bool(ok) and code.value == 259
        finally:
            k32.CloseHandle(ctypes.c_void_p(h))
    except Exception:
        pass
    return False


def pid_alive(pid):
    try:
        pid = int(pid)
    except Exception:
        return False
    if _pid_alive_win(pid):
        return True
    for _, p in _tasklist_rows():     # 兜底：万一 OpenProcess 被拦
        if p == pid:
            return True
    return False


# 常驻加载器的互斥锁。⛑ 为什么必须有（2026-09-29）：
#   两个 frida 会话同时挂在 Ananta.exe 上 = 游戏必崩（历史踩过 3 次）。
#   以前加载器是手动开的，人不容易开两份；现在改成「一键启动」随游戏自动起，
#   重复双击的概率大幅上升，所以必须由程序自己挡住。
#   ⛑ 锁文件里存 pid 而不是单纯「文件存在」—— 崩溃退出会留下僵尸锁，
#     只判存在会永久卡死启动，必须校验 pid 是否真的活着。
LOCK_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".loader.lock")


def acquire_instance_lock():
    """拿常驻锁。返回 None = 拿到；返回占用者 pid = 已经有一份在跑，调用方应立刻退出。"""
    try:
        if os.path.exists(LOCK_PATH):
            try:
                old = int((open(LOCK_PATH, encoding="utf-8").read() or "0").strip() or 0)
            except Exception:
                old = 0
            if old and pid_alive(old) and old != os.getpid():
                return old
    except Exception:
        pass
    try:
        with open(LOCK_PATH, "w", encoding="utf-8") as fh:
            fh.write(str(os.getpid()))
    except Exception:
        pass
    atexit.register(release_instance_lock)
    return None


def release_instance_lock():
    """只删**自己**写的锁，别把别人的删了。"""
    try:
        if not os.path.exists(LOCK_PATH):
            return
        try:
            cur = int((open(LOCK_PATH, encoding="utf-8").read() or "0").strip() or 0)
        except Exception:
            cur = 0
        if cur == os.getpid():
            os.remove(LOCK_PATH)
    except Exception:
        pass


def other_loader_running():
    """有没有别的加载器在跑？返回名字，没有返回 None。必须**每一轮**复查。"""
    for e in OTHER_EXES:
        if count_exe(e) > 0:
            return e
    return None


def apply_placeholders(src, root):
    """把 agent JS 里「由 Python 端注入」的占位符替换掉。

    ⛑ 2026-09-28 才发现的大坑：`modules/world_mod.js:3376` 是
           var CELLS_RAW = __CELLS_JSON__;
       这是一个**原始占位符**，原版靠 `new-worldmod/inject.py` 在注入前
       `src.replace("__CELLS_JSON__", json.dumps(cells))` 换掉。
       我们这里是直接读文件 ⇒ 未定义的标识符 ⇒ 顶层 ReferenceError
       ⇒ **整个脚本加载即死**，连它自己的 `Il2Cpp.perform(...)` 引导都跑不到。
       实测：所有历史会话里 `[worldmod]` 的 console 输出 **0 次** —— 通道 B 一直是死的。
    """
    if "__CELLS_JSON__" not in src:
        return src
    p = os.path.join(root, "cells.json")
    if not os.path.exists(p):
        raise RuntimeError("JS 需要 __CELLS_JSON__，但找不到 %s" % p)
    data = json.loads(open(p, encoding="utf-8").read())
    rows = data.get("cells", data if isinstance(data, list) else [])
    if not rows:
        raise RuntimeError("%s 里没有 cells" % p)
    payload = json.dumps(rows, ensure_ascii=False, separators=(",", ":"))
    return src.replace("__CELLS_JSON__", payload)


# ★★ 气球 side-channel 路径（2026-09-29 修）
#
# 病根：`modules/balloon_fix.js:3417` 的默认值是作者自己开发机上的硬编码
#           var PUSH_PATHS = ['D:\\Ananta\\gym_tools\\_balloon_push.json'];
#   该文件顶部 R146 注释写得很清楚：**「exe 加载器会把这一行整个改写成运行期算出来的
#   本机多路径」**。原版 `AnantaBalloonLoader.exe` 确实这么干（实机日志里有
#   `agent PUSH_PATHS -> ["C:\\Users\\…\\AnantaBalloon\\balloon_push.json", …]`），
#   而我们这里是把 js 原样读进来 ⇒ 没有重写 ⇒ agent 永远去读 D:\Ananta\…
#   ⇒ pushN 恒 0 ⇒ 一颗气球都不会被伪造出来。
#
# 服务端（GameRouter.Balloon.cs:1140-1159）按顺序写这三条，这里按**同一顺序**给候选：
#   ① %LOCALAPPDATA%\AnantaBalloon\balloon_push.json   ← 与安装位置无关，主用
#   ② <服务端 base 目录>\config\balloon_push.json
#   ③ D:\Ananta\gym_tools\_balloon_push.json           ← 作者旧位置，兜底
_BALLOON_PUSH_RE = re.compile(r"var\s+PUSH_PATHS\s*=\s*\[[^\]]*\]\s*;")


def apply_balloon_paths(src, root):
    """把 balloon_fix.js 里写死的 PUSH_PATHS 换成本机真实候选列表。"""
    if "PUSH_PATHS" not in src:
        return src
    shared = os.path.join(
        os.environ.get("LOCALAPPDATA") or "", "AnantaBalloon", "balloon_push.json")
    cands = []
    for p in (
        shared,
        os.path.join(root, "config", "balloon_push.json"),
        os.path.join(root, "Drmk.Server", "Drmk.App", "bin", "Debug", "net8.0",
                     "config", "balloon_push.json"),
        r"D:\Ananta\gym_tools\_balloon_push.json",
    ):
        if not p:
            continue
        # 统一成反斜杠（Windows File API 两种都吃，但 JS 字符串里反斜杠是转义符，
        # 必须成对出现 —— 下面 _js_str 会再翻倍一次）
        p = os.path.normpath(p).replace("/", "\\")
        if p not in cands:
            cands.append(p)

    def _js_str(s):
        # ⛑ 不能用 json.dumps 直接塞：`"C:\Users\..."` 里的 \U / \b 会被 JS 当转义吃
        #   掉（实测 node 打印成 C:UsersNoiro...）。反斜杠必须翻倍。
        return '"' + s.replace("\\", "\\\\") + '"'

    arr = "[" + ", ".join(_js_str(p) for p in cands) + "]"
    # ⛑⛑ 替换值**必须用函数**给：re.subn 的字符串 repl 会再做一次反斜杠转义处理，
    #    把 `C:\\Users` 压回 `C:\Users`（实测踩到，JS 侧就又变成转义符了）。
    new, n = _BALLOON_PUSH_RE.subn(lambda _m: "var PUSH_PATHS = %s;" % arr, src, count=1)
    if n == 0:
        return src
    return new


class Mod(object):
    """一个模块 = 一个 frida script。"""

    def __init__(self, name, path):
        self.name = name
        self.path = path
        self.script = None
        self.ready = False
        self.note = ""
        self.dead = False      # 客户端崩了/重启后，旧 script 会永久失效
        self.timeouts = 0      # 本 script 连续超时次数（load() 成功时清零）

    def load(self, session, quiet=True):
        try:
            src = open(self.path, encoding="utf-8", errors="replace").read()
        except Exception as e:
            self.note = "读文件失败: %s" % e
            return False
        # ★ 占位符替换（__CELLS_JSON__ 等）—— 见 apply_placeholders 的说明
        try:
            src = apply_placeholders(
                src, os.path.dirname(os.path.dirname(os.path.abspath(self.path))))
        except Exception as e:
            self.note = "占位符替换失败: %s" % str(e)[:160]
            return False
        # ★ 气球 side-channel：js 里的 PUSH_PATHS 是作者机器的硬编码，必须按本机改写
        #   root = 服务端根目录（…/DRMK-PS_09_HyNB-v0.3）：
        #     <root>/tools/AnantaInjector/modules/balloon_fix.js → 往上 4 层
        try:
            src = apply_balloon_paths(
                src, os.path.dirname(os.path.dirname(os.path.dirname(
                    os.path.dirname(os.path.abspath(self.path))))))
        except Exception as e:
            self.note = "气球路径改写失败: %s" % str(e)[:160]
            return False
        try:
            # runtime="v8" 是原加载器的写法（frida 17.18 仍接受该参数）。
            # 万一某个 frida 版本把它拿掉了，退回默认运行时——别因为这一个参数整个模块挂掉。
            try:
                sc = session.create_script(src, runtime="v8")
            except TypeError:
                sc = session.create_script(src)
        except Exception as e:
            self.note = "create_script 失败: %s" % str(e)[:120]
            return False

        box = {"ready": False}

        def on_message(msg, data):
            try:
                if msg.get("type") == "send":
                    p = msg.get("payload")
                    if isinstance(p, str):
                        try:
                            p = json.loads(p)
                        except Exception:
                            pass
                    if isinstance(p, dict) and (p.get("ready") or p.get("abReady")):
                        box["ready"] = True
                    elif not quiet:
                        print("    [%s] %s" % (self.name, str(p)[:200]))
            except Exception:
                pass
            try:
                if msg.get("type") == "error":
                    self.note = "script error: %s" % str(msg.get("description"))[:160]
                    if not quiet:
                        print("    [!] %s 脚本报错: %s" % (self.name, self.note))
            except Exception:
                pass

        try:
            sc.on("message", on_message)
            sc.load()
        except Exception as e:
            self.note = "load 失败: %s" % str(e)[:160]
            return False

        # 有些模块不发 ready，给 3 秒宽限后也放行（它们可能只是没实现握手）
        for _ in range(30):
            if box["ready"]:
                break
            time.sleep(0.1)
        self.script = sc
        self.ready = True
        self.dead = False      # 重新加载 = 新 script，旧失效标记要清掉
        self.timeouts = 0      # 同上：连续超时计数也归零
        return True

    def call3(self, name, arg=None, arg2=None, arg3=None, timeout=RPC_TIMEOUT):
        """三参 RPC（如 gadget_loader 的 findobj <关键字> <maxNodes> <maxDepth>）。

        不想把 arg2/arg3 混进 call() 的可选语义里（None 与「不传」会分不清），
        所以单独开一个方法，语义明确。
        """
        if self.script is None:
            raise RuntimeError("模块 %s 未加载" % self.name)
        E = self.script.exports_sync
        fn = getattr(E, name)
        box = {}

        def run():
            try:
                box["r"] = fn(arg, arg2, arg3)
            except Exception as e:
                box["e"] = e

        return self._run_guarded(run, box, name, timeout)

    def _run_guarded(self, run, box, name, timeout):
        """起线程跑 RPC + 收尾：超时判定 + 识别「客户端崩了」。

        ⛑ 客户端崩掉/重启后旧 script 会变成 'script has been destroyed'，
           不识别的话上层会一直假死在旧 session 上。
        ⛑ 线程在这里起（不在调用方），免得哪天又漏掉 t 的定义。

        ⛑⛑ 2026-09-29 修「超时线程泄漏」（原实现只 join、不回收）：
           Python 杀不掉线程，t.join(timeout) 超时后那个调用**仍然卡在 _frida 里**
           ⇒ 主循环每超时一次就永久多一个线程，跑久了线程数单调增长。
           三层治理，从强到弱：
           ① **best-effort 真取消**：子线程里挂 frida.Cancellable()，超时后 cancel.cancel()。
              只对 frida 内部会检查 cancellable 的同步调用生效（attach/load 这类稳生效，
              exports_sync 的 RPC 不一定）；**不生效也无害**，纯增强。
           ② **超时即判该 script 已废**：置 self.dead ⇒ 上层 `ctx.dead` 为真 ⇒ close()
              → script.unload() → 卡在等待中的 RPC 抛错 → 线程自然退出。
              **这才是真正的回收路径**，也是原来漏掉的一环（原来超时不置 dead，
              于是下一轮继续往同一个废 script 上发调用，每次再留一个线程）。
           ⛑ 本副本还有 armedGuard 兜底（bfStatus 连续 60s 拿不到就主动重挂），
              不置 dead 也不会无限假死 —— 但要白等 60 秒，且这 60 秒里每次探测都再留
              一个线程。置 dead 之后走 `unified_loader.py` 的 `ctx.dead` 分支，**立刻**重挂。
           ③ **快速失败**：连续超时达 STUCK_MAX_CONSECUTIVE ⇒ 抛错让上层重建，
              不再往废 script 上堆新调用。
           ⛑ 日志按规则 110 —— 不打「只打前 N 条」，每次超时都打（量本来就小）。
        """
        state = {}

        def runner():
            # cancellable 是线程局部的，必须在**调用线程**里 push 才有效。
            cancel = None
            try:
                import frida as _frida
                cancel = _frida.Cancellable()
            except Exception:
                cancel = None
            if cancel is None:
                run()
                return
            state["cancel"] = cancel
            try:
                with cancel:
                    run()
            except Exception as e:
                if "e" not in box:
                    box["e"] = e

        t = threading.Thread(target=runner)
        t.daemon = True
        with _STUCK_LOCK:
            _STUCK_THREADS.append(t)
        t.start()
        t.join(timeout)

        if t.is_alive():
            cancel = state.get("cancel")
            if cancel is not None:
                try:
                    cancel.cancel()
                except Exception:
                    pass
            self.timeouts += 1
            self.dead = True                      # ② 交给上层 close() 回收
            n = stuck_thread_count()
            print("[%s] [!] rpc %s.%s 超时（>%ss）—— 判定该 script 已废，上层会重建；"
                  "卡住线程=%d 连续超时=%d" % (ts(), self.name, name, timeout, n, self.timeouts))
            if self.timeouts >= STUCK_MAX_CONSECUTIVE:
                raise RuntimeError("rpc %s.%s 连续超时 %d 次，放弃该 script，等待重建"
                                   % (self.name, name, self.timeouts))
            raise RuntimeError("rpc %s.%s 超时（>%ss）" % (self.name, name, timeout))

        # 正常返回：线程已结束，从账本里摘掉
        self.timeouts = 0
        with _STUCK_LOCK:
            if t in _STUCK_THREADS:
                _STUCK_THREADS.remove(t)
        if "e" in box:
            msg = str(box["e"])
            if ("has been destroyed" in msg or "is detached" in msg
                    or "invalid session" in msg or "Session is detached" in msg):
                self.dead = True
            raise RuntimeError("rpc %s.%s 失败: %s" % (self.name, name, msg[:160]))
        return box.get("r")

    def call(self, name, arg=None, arg2=None, timeout=RPC_TIMEOUT):
        """arg2 用于双参 RPC（如 addbtn 的 abTick(半径, 批量)）。"""
        if self.script is None:
            raise RuntimeError("模块 %s 未加载" % self.name)
        E = self.script.exports_sync
        fn = getattr(E, name)
        box = {}

        def run():
            try:
                if arg2 is not None:
                    box["r"] = fn(arg, arg2)
                elif arg is not None:
                    box["r"] = fn(arg)
                else:
                    box["r"] = fn()
            except Exception as e:
                box["e"] = e

        return self._run_guarded(run, box, name, timeout)


class Ctx(object):
    def __init__(self, modules):
        self.modules = modules          # [Mod]
        self.session = None
        self.pid = None
        self.closing = False
        self.fail = 0

    def attach(self):
        box = {}

        def run():
            try:
                box["r"] = self._attach_impl(box)
            except Exception as e:
                box["e"] = e

        t = threading.Thread(target=run)
        t.daemon = True
        t.start()
        t.join(ATTACH_TIMEOUT)
        if t.is_alive():
            sess = box.get("sess")
            if sess is not None:
                def _kill():
                    try:
                        sess.detach()
                    except Exception:
                        pass
                k = threading.Thread(target=_kill)
                k.daemon = True
                k.start()
                k.join(6)
            return False
        if "e" in box:
            print("[%s] attach 异常：%s" % (ts(), str(box["e"])[:200]))
            return False
        return box.get("r", False)

    def _attach_impl(self, box=None):
        import frida
        pid = resolve_pid()
        if pid is None:
            return False
        self.pid = pid
        session = frida.attach(pid)
        if box is not None:
            box["sess"] = session
        ok = 0
        for m in self.modules:
            if m.load(session):
                ok += 1
                print("[%s]   ✓ %s" % (ts(), m.name))
            else:
                print("[%s]   ✗ %s —— %s" % (ts(), m.name, m.note or "未知原因"))
        if ok == 0:
            try:
                session.detach()
            except Exception:
                pass
            return False
        self.session = session
        self.closing = False
        self.fail = 0
        return True

    @property
    def dead(self):
        """客户端崩了/重启 ⇒ 任一模块的 script 失效，或 session 已断。"""
        if self.session is None:
            return True
        for m in self.modules:
            if m.script is not None and m.dead:
                return True
        return False

    def mod(self, name):
        for m in self.modules:
            if m.name == name and m.script is not None:
                return m
        return None

    def close(self):
        """优雅停机：① 请各 agent 停手 ② 卸脚本 ③ detach。卸载超时绝不 detach。"""
        if self.closing or self.session is None:
            STOP["done"] = True
            return
        self.closing = True
        t0 = time.time()

        for m in self.modules:
            # ⛑ 2026-09-29：dead 的 script 再发 stop 只会再超时一次、再留一个卡住的线程
            #   （它的 RPC 通道已经废了，发什么都是白发）⇒ 直接跳过，交给 unload 回收。
            if m.script is None or m.dead:
                continue
            # "stop" 是 AnantaGadgetLoader 的停机 RPC（gadget_world.py 的 finally 里调它）
            for stop_name in ("shutdown", "bowlShutdown", "abShutdown", "stop"):
                try:
                    m.call(stop_name, timeout=4)
                    break
                except Exception:
                    continue

        time.sleep(0.3)
        sess = self.session
        scrs = [m.script for m in self.modules if m.script is not None]
        self.session = None
        for m in self.modules:
            m.script = None

        def _unload():
            for sc in scrs:
                try:
                    sc.unload()
                except Exception:
                    pass

        t1 = threading.Thread(target=_unload)
        t1.daemon = True
        t1.start()
        t1.join(8.0)
        if t1.is_alive():
            print("[!] 脚本卸载没在 8 秒内完成 —— 已跳过强制断开（避免把游戏拖死）。")
            print("    各 agent 已停机、不再做任何事；重启游戏即可彻底清掉。")
            STOP["done"] = True
            return

        def _det():
            try:
                if sess:
                    sess.detach()
            except Exception:
                pass

        t2 = threading.Thread(target=_det)
        t2.daemon = True
        t2.start()
        t2.join(3.0)
        print("[%s] 注入已安全卸载，用时 %.1fs" % (ts(), time.time() - t0))
        STOP["done"] = True


_LIVE = []
_CLOSE_LOCK = threading.Lock()


def _atexit_close():
    with _CLOSE_LOCK:
        for c in list(_LIVE):
            try:
                c.close()
            except Exception:
                pass
    STOP["done"] = True