#!/usr/bin/env python3
"""M3 验收监督：供 Hermes cron 每 10 分钟调用（no_agent）。

状态机: idle -> phase1(--batch 30 20) -> phase2(--batch 50 100) -> done
- 运行中（pid 存活且 cmdline 匹配）: 零输出静默退出
- 进程结束: 解析日志 M3_ACCEPTANCE -> PASS 则进入下一阶段并立即启动; FAIL/无结论则停机留给人工, 不自动重启
- 状态文件 scripts/m3_state.json 防重复拉起; 日志写 scripts/m3_logs/<phase>_<ts>.log
- --selftest: 仅验证解析与状态机转移, 不启动模拟
"""
import json
import subprocess
import sys
import tempfile
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
BINARY = ROOT / "src/LivingVillage.Headless/bin/Release/net8.0/LivingVillage.Headless.dll"
STATE_PATH = ROOT / "scripts/m3_state.json"
LOG_DIR = ROOT / "scripts/m3_logs"
PHASES = {"phase1": ["--batch", "30", "20"], "phase2": ["--batch", "50", "100"]}
NEXT = {"phase1": "phase2", "phase2": "done"}


def load_state():
    if STATE_PATH.exists():
        return json.loads(STATE_PATH.read_text())
    return {"phase": "idle", "pid": None, "cmd": None, "log": None, "started": None, "verdict": None}


def save_state(state):
    STATE_PATH.write_text(json.dumps(state, indent=1, ensure_ascii=False) + "\n")


def parse_acceptance(text):
    verdict = None
    for line in text.splitlines():
        if line.startswith("M3_ACCEPTANCE="):
            v = line.split("=", 1)[1].strip()
            verdict = v if v in ("PASS", "FAIL") else None
    return verdict


def pid_matches(pid, cmd):
    if not pid:
        return False
    try:
        with open(f"/proc/{pid}/cmdline", "rb") as f:
            argv = f.read().decode("utf-8", "replace").replace("\0", " ")
    except OSError:
        return False
    return "LivingVillage.Headless.dll" in argv and (not cmd or cmd in argv)


def launch(phase, state):
    LOG_DIR.mkdir(parents=True, exist_ok=True)
    stamp = time.strftime("%Y%m%dT%H%M%S")
    log = LOG_DIR / f"{phase}_{stamp}.log"
    cmd = ["dotnet", str(BINARY)] + PHASES[phase]
    with open(log, "w") as fh:
        proc = subprocess.Popen(
            cmd, stdout=fh, stderr=subprocess.STDOUT,
            stdin=subprocess.DEVNULL, start_new_session=True, cwd=ROOT)
    state.update({
        "phase": phase, "pid": proc.pid, "cmd": str(BINARY),
        "log": str(log), "started": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "verdict": None})
    save_state(state)
    return f"started {phase} pid={proc.pid} log={log}"


def decide(state, alive):
    phase = state.get("phase", "idle")
    if phase == "done":
        return None, state, None
    if phase in PHASES:
        if state.get("pid") and alive(state["pid"], state.get("cmd")):
            return None, state, None
        log = state.get("log")
        p = Path(log) if log else None
        text = p.read_text(errors="replace") if p and p.exists() else ""
        verdict = parse_acceptance(text)
        state["verdict"] = verdict
        state["pid"] = None
        if verdict == "PASS":
            nxt = NEXT[phase]
            if nxt == "done":
                state["phase"] = "done"
                return "save", state, f"{phase} PASS -> M3 acceptance complete (30x20 + 50x100)"
            return f"launch:{nxt}", state, None
        if verdict == "FAIL":
            state["phase"] = "failed"
            return "save", state, f"{phase} FAIL (log={log}); stopped, awaiting human"
        state["phase"] = "interrupted"
        return "save", state, f"{phase} interrupted without verdict (log={log}); stopped, awaiting human"
    if phase == "idle":
        return "launch:phase1", state, None
    return None, state, None


def selftest():
    ok = True

    def check(name, cond):
        nonlocal ok
        print(("PASS " if cond else "FAIL ") + name)
        ok = ok and cond

    check("parse PASS", parse_acceptance("x\nM3_ACCEPTANCE=PASS\n") == "PASS")
    check("parse FAIL", parse_acceptance("M3_ACCEPTANCE=FAIL") == "FAIL")
    check("parse empty", parse_acceptance("world=0 ...\n") is None)
    check("parse garbage", parse_acceptance("M3_ACCEPTANCE=MAYBE") is None)
    base = {"phase": "phase1", "pid": 123, "cmd": "b.dll", "log": "nope.log", "verdict": None, "started": None}
    a, s, _ = decide(dict(base), lambda pid, cmd: True)
    check("alive -> silent", a is None)
    a, s, _ = decide(dict(base), lambda pid, cmd: False)
    check("dead without verdict -> interrupted", a == "save" and s["phase"] == "interrupted")
    with tempfile.NamedTemporaryFile("w", suffix=".log", delete=False) as fh:
        fh.write("batch_summary ...\nM3_ACCEPTANCE=PASS\n")
        plog = fh.name
    a, s, _ = decide(dict(base, log=plog), lambda pid, cmd: False)
    check("phase1 PASS -> launch phase2", a == "launch:phase2")
    a, s, m = decide(dict(base, phase="phase2", log=plog), lambda pid, cmd: False)
    check("phase2 PASS -> done", s["phase"] == "done" and m and "complete" in m)
    with tempfile.NamedTemporaryFile("w", suffix=".log", delete=False) as fh:
        fh.write("M3_ACCEPTANCE=FAIL\n")
        flog = fh.name
    a, s, _ = decide(dict(base, log=flog), lambda pid, cmd: False)
    check("FAIL -> failed", s["phase"] == "failed")
    a, s, _ = decide({"phase": "done"}, lambda pid, cmd: False)
    check("done -> silent", a is None)
    print("SELFTEST=" + ("PASS" if ok else "FAIL"))
    return 0 if ok else 1


def main():
    if "--selftest" in sys.argv:
        return selftest()
    state = load_state()
    if state.get("phase") == "idle" and not BINARY.exists():
        print(f"binary missing: {BINARY}; run dotnet build src/LivingVillage.Headless -c Release")
        return 1
    action, state, msg = decide(state, pid_matches)
    if action and action.startswith("launch:"):
        msg = launch(action.split(":", 1)[1], state)
    elif action == "save":
        save_state(state)
    if msg:
        print(msg)
    return 0


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