chore: M3 验收监督脚本 scripts/m3_supervise.py — cron 10min no_agent 状态机(idle→30x20→50x100→done), pid/cmdline 防重复拉起, 运行中零输出, PASS 链式推进/FAIL 停机待人工, --selftest(9 项解析+转移断言); .gitignore 补 m3_logs/m3_state/.opencode-tmp

1 files changed, 154 insertions, 0 deletions

diff --git a/scripts/m3_supervise.py b/scripts/m3_supervise.py
new file mode 100755
index 0000000..3975b81
--- /dev/null
+++ b/scripts/m3_supervise.py
@@ -0,0 +1,154 @@
+#!/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())