"""M10 启动生命周期判据：多实例 / 重复启动 / 残留清理。

## 对应你 2026-09-26 的定位

> 「框架本身类似给 agent 用的 MC 启动器……多实例，多版本隔离，
>   单个实例多次启动是自然而然被要求支持的」

本文件把那三句要求翻成**可判**的行为，重点在**重复启动**与
**残留/误杀**这类真会出事的部分。

## 为什么用真进程而不是 mock

「pid 被复用」「进程已死但记录还在」这些缺陷，mock 测不出来 ——
必须真的起进程、真的杀进程、真的留残留。判据因此用 `sleep` 子进程
（轻量、无 JVM），并在结束时确保清理干净。
"""
import sys, os, time, json, signal, subprocess
from pathlib import Path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from mcfold.launch import (
    status, claim, register, release, stop, cleanup, AlreadyRunning,
    runtime_dir_for,
)

ROOT = Path("/home/newqqagent/mcview/runtime/launch-test")
INST = ROOT / "instA"
INST2 = ROOT / "instB"
INST.mkdir(parents=True, exist_ok=True)
INST2.mkdir(parents=True, exist_ok=True)

_spawned: list[subprocess.Popen] = []
JAVA = "java"
_HOLD_DIR = Path("/tmp/mcfold-hold")

# 外来 pid 用的形状：cmdline 里既没有实例路径、也不是 java
_ = None


def _ensure_hold_class() -> None:
    """写一个最小的常驻 java 类（读 stdin 直到 EOF）。"""
    _HOLD_DIR.mkdir(exist_ok=True)
    src = _HOLD_DIR / "Hold.java"
    if not src.exists():
        src.write_text(
            "public class Hold {\n"
            "  public static void main(String[] a) throws Exception {\n"
            "    java.io.InputStream in = System.in;\n"
            "    int c; while ((c = in.read()) != -1) { }\n"
            "  }\n"
            "}\n", encoding="utf-8")
        subprocess.run([JAVA, str(src)], check=True, capture_output=True)


def _spawn(cmd_argv: list[str] | None = None, tag: str = "instA") -> subprocess.Popen:
    """起一个真进程模拟 JVM。

    ★ 真实 JVM 的 cmdline 里**带实例/运行时目录**（我们用 cwd=game_dir 启动），
      所以身份校验（`_is_our_jvm` 查 cmdline 含实例路径）能命中。
      这里用 `sleep 3600 <实例路径>` 复现那个形状 ——
      否则测的是「一个 cmdline 里什么都没有的进程」，与真实启动不符。
    """
    inst = INST if tag == "instA" else INST2
    # ★ 用**真 java** 起一个最小常驻进程（-version 会立刻退出，
    #   所以用一个只读 stdin 的小程序让它活着）。
    #   真实 backend 的 cmdline 形如
    #     java -Xmx… -cp … Main --gameDir <实例目录>
    #   ⇒ 身份校验查 cmdline 里的实例路径能命中。
    #   早先我用 `sleep 3600 <path>` / `bash -c "sleep … # <path>"` 都失败了：
    #     sleep 拒绝多余参数；bash 会 exec 掉自己，cmdline 变成裸 sleep。
    #   ⇒ 「模拟进程」的形状必须与真实启动一致，否则测的是假形状。
    argv = cmd_argv or [JAVA, "-cp", "/tmp", "Hold", str(inst)]
    p = subprocess.Popen(argv, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    _spawned.append(p)
    return p


def _cleanup_all():
    for p in _spawned:
        try:
            p.kill()
            p.wait(timeout=3)
        except Exception:
            pass
    _spawned.clear()
    for inst in (INST, INST2):
        try:
            release(inst, ROOT)
        except Exception:
            pass


# ---------- ① 三态必须分清（agent 应对完全不同） ----------


def test_three_states_distinguished():
    """★ `free` / `stale` / `running` 三态语义不同，绝不能混。

    - free    没跑过 / 已正常退出
    - stale   **有记录但进程已死**（上次崩溃残留）⇒ 该清理后重启
    - running 真在跑 ⇒ 重复启动必须被拦住
    """
    release(INST, ROOT)
    st = status(INST, ROOT)
    assert st["state"] == "free", st

    # 起一个真进程并登记
    p = _spawn()
    register(INST, p.pid, runtime_root=ROOT, generation="gen_1_21", version="1.21.1")
    st = status(INST, ROOT)
    assert st["state"] == "running", st
    assert st["running"] is True
    assert st["generation"] == "gen_1_21"
    assert st["started_at"], "必须记录启动时间（报错时要给）"

    # 杀掉进程 ⇒ 应变 stale（而不是继续说 running）
    p.kill(); p.wait(timeout=5)
    st = status(INST, ROOT)
    assert st["state"] == "stale", f"进程已死应报 stale，实得 {st}"
    assert st["running"] is False
    assert "cleanup" in st["how_to"], st
    release(INST, ROOT)
    print("  ★ 三态：free / running / stale 分清")


# ---------- ② 重复启动必须被拦住，且报错能指导下一步 ----------


def test_second_launch_is_blocked():
    """★★ 同实例重复启动 ⇒ **启动前**就报，且信息够定位。

    ★ 这是本模块存在的核心理由：
      不拦的话两个 JVM 会抢同一个 run/ 与同一个 X display，
      报错会指向渲染异常，**完全指不到「重复启动」**。
    """
    release(INST, ROOT)
    p = _spawn()
    register(INST, p.pid, runtime_root=ROOT)
    try:
        claim(INST, runtime_root=ROOT)
        raise AssertionError("重复启动应该被拦住")
    except AlreadyRunning as e:
        info = e.info
        assert info["pid"] == p.pid, info
        assert info["started_at"], "报错要带启动时间"
        assert "已在运行" in str(e), str(e)
    # 异常类型要能让上层**区分**（该类不该重试）
    assert issubclass(AlreadyRunning, RuntimeError)
    release(INST, ROOT)
    print("  ★★ 重复启动被拦（带 pid 与启动时间）")


def test_claim_clears_stale_and_then_proceeds():
    """★ stale 残留应被自动清掉并放行（不能让垃圾状态挡住下一次启动）。"""
    release(INST, ROOT)
    p = _spawn()
    register(INST, p.pid, runtime_root=ROOT)
    p.kill(); p.wait(timeout=5)
    assert status(INST, ROOT)["state"] == "stale"
    r = claim(INST, runtime_root=ROOT)
    assert r["ok"] is True, r
    assert r["cleared_stale"] is True, r
    assert r["cleared_pid"] == p.pid, r
    assert status(INST, ROOT)["state"] == "free", "清完应是 free"
    print("  ★ stale 残留自动清理后放行")


# ---------- ③ 多实例互不干扰 ----------


def test_two_instances_are_independent():
    """★ 实例 A 在跑**不影响**实例 B（这才是多实例的意义）。"""
    release(INST, ROOT); release(INST2, ROOT)
    pa = _spawn()
    register(INST, pa.pid, runtime_root=ROOT, generation="gen_1_21")
    # B 可以正常 claim
    r = claim(INST2, runtime_root=ROOT)
    assert r["ok"] is True, r
    pb = _spawn(tag="instB")
    register(INST2, pb.pid, runtime_root=ROOT, generation="gen_1_20")
    assert status(INST, ROOT)["state"] == "running"
    assert status(INST2, ROOT)["state"] == "running"
    # 同版本不同实例也不冲突（多实例同版本是正常用法）
    assert status(INST, ROOT)["pid"] != status(INST2, ROOT)["pid"]
    print("  ★ 两个实例互不干扰（同版本多实例可用）")


def test_runtime_dir_is_third_layer():
    """★ 运行时目录必须在**第三层**（与 store/ instances/ 同级）。

    放内容根或实例目录都违反 `test_version_isolation` 守的分层。
    """
    d = runtime_dir_for(INST, ROOT)
    assert d.parent == ROOT, d
    assert d.name == "instA", d
    # 不得落在 store 或 instances 里
    assert "/store/" not in str(d), d
    assert "/instances/" not in str(d), d
    print(f"  ★ 运行时目录在第三层（{d}）")


# ---------- ④ 防误杀：pid 复用 ----------


def test_foreign_pid_is_not_killed():
    """★★ pid 存活但**不属于本实例** ⇒ 绝不能杀，且要说清原因。

    ★ pid 会被系统回收复用：一个早已结束的 JVM 的 pid 可能已属于别的程序。
      不校验就 kill ⇒ 误杀无关进程。
    """
    release(INST, ROOT)
    # 登记一个**真实存活**但 cmdline 里没有实例路径的进程
    # （模拟「pid 被系统回收后分给了别的程序」）
    p = _spawn(cmd_argv=["tail", "-f", "/dev/null"])   # 非 java，且 cmdline 不含 instA
    register(INST, p.pid, runtime_root=ROOT)
    st = status(INST, ROOT)
    assert st["state"] == "foreign", f"应识别为外来 pid，实得 {st['state']}"
    assert st["running"] is False, "外来 pid 不算本实例在跑"
    assert "不属于" in st["why"], st

    r = stop(INST, runtime_root=ROOT)
    assert r["ok"] is False, r
    assert r["killed"] is False, r
    assert "误杀" in r["why"] or "不属于" in r["why"], r
    # ★ 关键：那个进程必须还活着
    assert p.poll() is None, "★ 被误杀了（这正是要防的事故）"
    p.kill(); p.wait(timeout=5)
    print("  ★★ 外来 pid 拒绝终止（未误杀）")


def test_cleanup_does_not_kill_process():
    """★ cleanup 只清**记录**，不杀进程（状态是 foreign 时尤其不能杀）。"""
    release(INST, ROOT)
    p = _spawn(cmd_argv=["tail", "-f", "/dev/null"])
    register(INST, p.pid, runtime_root=ROOT)
    r = cleanup(INST, runtime_root=ROOT)
    assert r["ok"] is True, r
    assert p.poll() is None, "cleanup 不该杀进程"
    p.kill(); p.wait(timeout=5)
    print("  ✓ cleanup 只清记录，不杀进程")


def test_cleanup_refuses_while_running():
    """★ 正在跑时 cleanup 要拒绝（否则会留下「在跑但无记录」的野状态）。"""
    release(INST, ROOT)
    p = _spawn()
    register(INST, p.pid, runtime_root=ROOT)
    r = cleanup(INST, runtime_root=ROOT)
    assert r["ok"] is False, r
    assert r["state"] == "running", r
    assert "先 stop" in r["how_to"], r
    release(INST, ROOT)
    print("  ✓ 运行中拒绝 cleanup（避免野状态）")


# ---------- ⑤ 正常退出路径 ----------


def test_release_then_relaunch_clean():
    """★ 完整循环：起 → 登记 → 停 → 释放 → 再起（重复启动的正常形态）。"""
    release(INST, ROOT)
    p1 = _spawn()
    register(INST, p1.pid, runtime_root=ROOT, generation="gen_1_21")
    assert status(INST, ROOT)["state"] == "running"
    r = stop(INST, runtime_root=ROOT, timeout=10)
    assert r["ok"] is True and r["killed"] is True, r
    assert status(INST, ROOT)["state"] == "free", "停完应是 free"

    # 第二次启动必须顺畅
    r2 = claim(INST, runtime_root=ROOT)
    assert r2["ok"] is True, r2
    p2 = _spawn()
    register(INST, p2.pid, runtime_root=ROOT, generation="gen_1_21")
    assert status(INST, ROOT)["pid"] == p2.pid
    release(INST, ROOT)
    print("  ★ 完整循环：起→停→再起（重复启动顺畅）")


def test_session_file_is_readable_json():
    """★ 记录是给人/agent 看的 JSON（排障时直接 cat）。"""
    release(INST, ROOT)
    p = _spawn()
    register(INST, p.pid, runtime_root=ROOT, generation="gen_1_21",
             version="1.21.1", loader="vanilla", extra={"display": ":98"})
    d = runtime_dir_for(INST, ROOT)
    f = d / "mcfold.session.json"
    assert f.exists(), f
    data = json.loads(f.read_text(encoding="utf-8"))
    for k in ("pid", "started_at", "generation", "version", "loader", "display"):
        assert k in data, f"{k} 缺失：{data}"
    assert data["pid"] == p.pid
    assert (d / "mcfold.pid").read_text().strip() == str(p.pid)
    release(INST, ROOT)
    print("  ★ 会话记录是可读 JSON（含 pid/时间/世代/版本/加载器）")


if __name__ == "__main__":
    _ensure_hold_class()
    try:
        for fn in [test_three_states_distinguished,
                   test_second_launch_is_blocked,
                   test_claim_clears_stale_and_then_proceeds,
                   test_two_instances_are_independent,
                   test_runtime_dir_is_third_layer,
                   test_foreign_pid_is_not_killed,
                   test_cleanup_does_not_kill_process,
                   test_cleanup_refuses_while_running,
                   test_release_then_relaunch_clean,
                   test_session_file_is_readable_json]:
            fn()
        print("\n启动生命周期：全部通过")
    finally:
        _cleanup_all()