diff options
Diffstat (limited to 'scripts/m3_supervise.py')
| -rwxr-xr-x | scripts/m3_supervise.py | 509 |
1 files changed, 419 insertions, 90 deletions
diff --git a/scripts/m3_supervise.py b/scripts/m3_supervise.py index 62fe94d..e5a2315 100755 --- a/scripts/m3_supervise.py +++ b/scripts/m3_supervise.py @@ -3,30 +3,36 @@ 状态机: idle -> phase1(--batch 30 20) -> phase2(--batch 50 100) -> done - flock 单实例防重入；抢不到锁静默退出 -- 冻结运行: 首次启动把 Release 输出目录整体复制到 scripts/m3_frozen/<stamp>_<commit>/， - 记录全部文件 sha256 清单；两阶段都从冻结副本运行，后续构建不会覆盖运行中二进制 +- 冻结运行: 首次启动把 Release 输出目录递归复制到 scripts/m3_frozen/<stamp>_<commit>/， + v2 清单覆盖全部子目录文件 sha256；两阶段都从冻结副本运行，后续构建不会覆盖 - 真实退出码: 子进程为 sh -c 'dotnet <frozen.dll> --batch K D; echo $? > exitfile'， - 结束后读 exitfile 得到真实退出码（含信号 128+sig；shell 被杀则无 exitfile -> interrupted） -- PID 身份: 记录 /proc/<pid>/stat 字段22 starttime，存活判定要求 pid+starttime+cmdline 全匹配（防 PID 复用） + 读 exitfile 得真实退出码（信号 128+sig；shell 被杀无 exitfile -> interrupted） +- PID 身份: pid + /proc starttime + cmdline 片段全匹配（防 PID 复用） - PASS 判定收紧（全部满足才 PASS，任一不满足即 FAIL/停机）: 1) exitfile 退出码 == 0 - 2) 日志含 `batch worlds=K days=D` 且 K/D 与该阶段参数完全一致 - 3) `world=` 行数 == K 且每行末 token 为 OK + 2) `batch worlds=K days=D` 与固定 PHASES[phase] 完全一致（不用 state 里可篡改值） + 3) `world=` 行: 行数==K、索引恰为 0..K-1 无重复无缺失、seed==42+i 4) `batch_summary worlds=K passed=K failed=0` 5) 行 `M3_ACCEPTANCE=PASS` - 6) 冻结清单重新校验 sha256 全部一致，且清单 commit == 启动时记录 commit（commit 绑定） -- FAIL/无结论/校验失败: 停机不自动重启，并在日志旁写 <log>.diagnosis.txt 可诊断报告 -- 状态文件 scripts/m3_state.json 原子写（tmp + os.replace） -- --selftest: 全部合成测试（解析/判据校验/原子状态/锁互斥/PID身份解析/诊断报告/退出码读取）， - 不启动任何真实模拟，不伪造真实结果 + 6) 冻结清单 sha256 全部一致 + commit 绑定；v2 清单另查额外/缺失文件 +- shell 死亡留下存活 dotnet(孤儿): interrupted 停机并记录孤儿 pid，绝不自动重启新 batch +- FAIL/无结论/校验失败: 停机不自动重启，写 <log>.diagnosis.txt 可诊断报告 +- 状态文件原子写（tmp + os.replace）；旧 v1 清单只校验所列文件（兼容正在运行的旧冻结副本， + 不修改任何旧冻结目录） +- --selftest: 合成断言 + 真实隔离生命周期测试（临时目录 + 短真实子进程: + 退出非零/重复调用/控制器重启/子进程中断/孤儿防重启），不伪造任何 M3 结果， + 不触碰项目真实 state/log/frozen """ +import contextlib import fcntl import hashlib +import io import json import os import re import shlex import shutil +import signal import subprocess import sys import tempfile @@ -44,9 +50,10 @@ NEXT = {"phase1": "phase2", "phase2": None} FRAG = "LivingVillage.Headless.dll" -def git_commit(root=ROOT): +def git_commit(root=None): + root = ROOT if root is None else Path(root) try: - r = subprocess.run(["git", "rev-parse", "HEAD"], cwd=root, + r = subprocess.run(["git", "rev-parse", "HEAD"], cwd=str(root), capture_output=True, text=True, timeout=10) c = r.stdout.strip() return c if r.returncode == 0 and c else None @@ -62,26 +69,40 @@ def sha256_file(path): return h.hexdigest() -def freeze_release(commit, src_binary=SRC_BINARY, frozen_root=None): - """复制 Release 输出目录全部文件到冻结目录并生成 sha256 清单。""" +def freeze_release(commit, src_binary=None, frozen_root=None): + """递归复制 Release 输出目录（含子目录）到冻结目录，生成 v2 sha256 清单。""" frozen_root = Path(frozen_root) if frozen_root else FROZEN_ROOT + src = Path(src_binary) if src_binary else SRC_BINARY stamp = time.strftime("%Y%m%dT%H%M%S") - dst = frozen_root / f"{stamp}_{(commit or 'nocommit')[:8]}" - dst.mkdir(parents=True, exist_ok=False) + base = frozen_root / f"{stamp}_{(commit or 'nocommit')[:8]}" + dst = base + suffix = 1 + while True: + try: + dst.mkdir(parents=True, exist_ok=False) + break + except FileExistsError: + dst = base.parent / f"{base.name}_{suffix}" + suffix += 1 files = {} - for f in sorted(Path(src_binary).parent.iterdir()): + for f in sorted(src.parent.rglob("*")): if f.is_file(): - shutil.copy2(f, dst / f.name) - files[f.name] = sha256_file(dst / f.name) - manifest = {"commit": commit, "created": time.strftime("%Y-%m-%dT%H:%M:%S"), - "source": str(src_binary), "files": files} + rel = f.relative_to(src.parent).as_posix() + target = dst / rel + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(f, target) + files[rel] = sha256_file(target) + manifest = {"version": 2, "commit": commit, "created": time.strftime("%Y-%m-%dT%H:%M:%S"), + "source": str(src), "files": files} mpath = dst / "manifest.json" mpath.write_text(json.dumps(manifest, indent=1, ensure_ascii=False) + "\n") return dst, mpath, manifest def verify_manifest(frozen_dir, manifest_path, expected_commit): - """重校验冻结目录: commit 绑定 + 全部文件 sha256。返回问题列表（空=通过）。""" + """重校验冻结目录。v2: commit 绑定 + 所列文件 sha256 + 额外/缺失文件检测 + （manifest.json 自身除外）。旧 v1 清单(无 version): 只校验所列文件，兼容运行中旧副本。 + 返回问题列表（空=通过）。""" problems = [] try: m = json.loads(Path(manifest_path).read_text()) @@ -100,21 +121,31 @@ def verify_manifest(frozen_dir, manifest_path, expected_commit): got = sha256_file(p) if got != want: problems.append(f"hash mismatch {name}: got {got[:16]} want {want[:16]}") + if m.get("version") == 2: + for f in Path(frozen_dir).rglob("*"): + if not f.is_file(): + continue + rel = f.relative_to(Path(frozen_dir)).as_posix() + if rel == "manifest.json" or rel in files: + continue + problems.append(f"extra file not in manifest: {rel}") return problems -def load_state(path=STATE_PATH): - if Path(path).exists(): +def load_state(path=None): + # path 延迟到调用时解析（模块级常量可能被测试替换; 默认参数在 def 时绑定是已知事故源） + p = Path(path) if path is not None else STATE_PATH + if p.exists(): try: - return json.loads(Path(path).read_text()) + return json.loads(p.read_text()) except Exception: return {"phase": "corrupt_state"} return {"phase": "idle"} -def save_state(state, path=STATE_PATH): - """原子写: 同目录 tmp + os.replace。""" - p = Path(path) +def save_state(state, path=None): + """原子写: 同目录 tmp + os.replace。path 延迟到调用时解析。""" + p = Path(path) if path is not None else STATE_PATH p.parent.mkdir(parents=True, exist_ok=True) fd, tmp = tempfile.mkstemp(dir=str(p.parent), prefix=".m3_state.", suffix=".tmp") try: @@ -129,8 +160,8 @@ def save_state(state, path=STATE_PATH): raise -def read_starttime(pid, proc_root="/proc"): - """/proc/<pid>/stat 字段22 starttime（boot 后时钟滴答）。解析失败返回 None。""" +def _read_proc_stat(pid, proc_root="/proc"): + """读取 /proc/<pid>/stat 的字段（字段3之后从 parts[0] 开始）。""" try: data = (Path(proc_root) / str(pid) / "stat").read_text() except OSError: @@ -139,9 +170,17 @@ def read_starttime(pid, proc_root="/proc"): if rp < 0: return None parts = data[rp + 1:].split() - # parts[i] 对应字段 i+3；字段22 -> parts[19] if len(parts) < 20: return None + return parts + + +def read_starttime(pid, proc_root="/proc"): + """/proc/<pid>/stat 字段22 starttime（boot 后时钟滴答）。解析失败返回 None。""" + parts = _read_proc_stat(pid, proc_root) + if parts is None: + return None + # parts[i] 对应字段 i+3；字段22 -> parts[19] return parts[19] @@ -156,12 +195,40 @@ def proc_alive(pid, start, frag, proc_root="/proc"): """pid + starttime + cmdline 片段三者全匹配才算同一进程（防 PID 复用）。""" if not pid or start is None: return False - now = read_starttime(pid, proc_root) - if now != str(start): + parts = _read_proc_stat(pid, proc_root) + if parts is None or parts[0] == "Z": + return False + if parts[19] != str(start): return False return frag in read_cmdline(pid, proc_root) +def find_orphans(marker, proc_root="/proc"): + """扫描 /proc，返回 cmdline 含 marker 的存活 pid（孤儿检测）。""" + out = [] + if not marker: + return out + for d in Path(proc_root).iterdir(): + if not d.name.isdigit(): + continue + if marker in read_cmdline(int(d.name), proc_root): + out.append(int(d.name)) + return out + + +def live_batch_orphans(proc_root="/proc"): + """state 丢失/_idle 时兜底: 存活 cmdline 同时含冻结 dll 片段与 --batch 的进程。 + 命中则绝不允许再拉起新批次（防止 state 被破坏后重复 launch）。排除自身。""" + out = [] + for d in Path(proc_root).iterdir(): + if not d.name.isdigit() or int(d.name) == os.getpid(): + continue + cl = read_cmdline(int(d.name), proc_root) + if FRAG in cl and "--batch" in cl: + out.append(int(d.name)) + return out + + def read_exitcode(exitfile): p = Path(exitfile) if not p.exists(): @@ -173,7 +240,8 @@ def read_exitcode(exitfile): def verify_batch(text, worlds, days): - """完整批次判据校验。返回问题列表（空=满足全部收紧条件）。""" + """完整批次判据校验（固定 worlds/days 由调用方从 PHASES 取）。 + 返回问题列表（空=满足全部收紧条件）。""" problems = [] m = re.search(r"^batch worlds=(\d+) days=(\d+) ", text, re.M) if not m: @@ -184,6 +252,25 @@ def verify_batch(text, worlds, days): wlines = [l for l in text.splitlines() if l.startswith("world=")] if len(wlines) != worlds: problems.append(f"world line count {len(wlines)} != expected {worlds}") + parsed = [] + malformed = [] + for l in wlines: + mw = re.match(r"^world=(\d+) seed=(\d+) ", l) + if not mw: + malformed.append((l.split() or ["?"])[0]) + else: + parsed.append((int(mw.group(1)), int(mw.group(2)))) + if malformed: + problems.append(f"malformed world lines: {malformed[:5]}") + idxs = [i for i, _ in parsed] + if idxs != list(range(worlds)): + dup = sorted({i for i in idxs if idxs.count(i) > 1}) + missing = sorted(set(range(worlds)) - set(idxs)) + extra = sorted(set(idxs) - set(range(worlds))) + problems.append(f"world index set mismatch: duplicates={dup[:5]} missing={missing[:5]} extra={extra[:5]} (count {len(idxs)} vs {worlds})") + badseed = [i for i, s in parsed if s != 42 + i] + if badseed: + problems.append(f"seed binding mismatch (expect seed=42+i): worlds={badseed[:8]}") bad = [l.split()[0] for l in wlines if not l.split() or l.split()[-1] != "OK"] if bad: problems.append(f"non-OK world lines: {bad[:8]}{'...' if len(bad) > 8 else ''}") @@ -205,17 +292,18 @@ def generate_diagnosis(state, problems, tag): if log and Path(log).exists(): lines = Path(log).read_text(errors="replace").splitlines() tail = "\n".join(lines[-60:]) - exitcode = state.get("exitcode") + orphans = state.get("orphans") or [] return ( f"M3 DIAGNOSIS {tag}\n" f"time: {time.strftime('%Y-%m-%dT%H:%M:%S')}\n" f"phase: {state.get('phase')}\n" - f"expected batch: {state.get('worlds')}x{state.get('days')}\n" + f"expected batch: fixed PHASES={PHASES.get(state.get('phase'))} state={state.get('worlds')}x{state.get('days')}\n" f"commit: {state.get('commit')}\n" f"frozen dir: {state.get('frozen')}\n" f"manifest: {state.get('manifest')}\n" f"pid: {state.get('pid')} starttime: {state.get('start')}\n" - f"exit code: {exitcode}\n" + f"exit code: {state.get('exitcode')}\n" + f"orphan child pids (shell dead, child alive): {orphans if orphans else 'none'}\n" f"log: {log}\n" f"problems:\n" + "".join(f" - {p}\n" for p in problems) + (f"\nlog tail (last 60 lines):\n{tail}\n" if tail else "\nlog missing or empty\n") @@ -231,24 +319,34 @@ def write_diagnosis(state, problems, tag): return rep -def build_sh_cmd(dll, batch_args, exitfile): +def runner_cmd(dll, batch_args, exitfile): return f"exec 2>&1; {shlex.quote('dotnet')} {shlex.quote(str(dll))} {batch_args}; echo $? > {shlex.quote(str(exitfile))}" -def launch(phase, state): +def launch(phase, state, runner=None): worlds, days = PHASES[phase] + if runner is None: + runner = runner_cmd frozen, manifest, commit = state.get("frozen"), state.get("manifest"), state.get("commit") dll = Path(frozen) / FRAG if frozen else None if not dll or not dll.exists() or not manifest or not commit: return "freeze-missing" LOG_DIR.mkdir(parents=True, exist_ok=True) stamp = time.strftime("%Y%m%dT%H%M%S") - log = LOG_DIR / f"{phase}_{stamp}.log" + base_log = LOG_DIR / f"{phase}_{stamp}.log" + log = base_log + suffix = 1 + # A controller restart can happen within the same second; never reuse a + # previous log or its exit marker when that happens. + while (log.exists() or Path(str(log) + ".exit").exists() + or Path(str(log) + ".diagnosis.txt").exists()): + log = LOG_DIR / f"{phase}_{stamp}_{suffix}.log" + suffix += 1 exitfile = Path(str(log) + ".exit") - cmd = ["sh", "-c", build_sh_cmd(dll, f"--batch {worlds} {days}", exitfile)] + cmd = ["sh", "-c", runner(dll, f"--batch {worlds} {days}", exitfile)] with open(log, "w") as fh: proc = subprocess.Popen(cmd, stdout=fh, stderr=subprocess.STDOUT, - stdin=subprocess.DEVNULL, start_new_session=True, cwd=ROOT) + stdin=subprocess.DEVNULL, start_new_session=True, cwd=str(ROOT)) start = read_starttime(proc.pid) if start is None: # 无法建立进程身份: 立即终止并按失败停机，不进入无身份运行 @@ -268,13 +366,14 @@ def launch(phase, state): "cmd": " ".join(cmd), "log": str(log), "exitfile": str(exitfile), "frozen": str(frozen), "manifest": str(manifest), "commit": commit, "started": time.strftime("%Y-%m-%dT%H:%M:%S"), - "verdict": None, "exitcode": None, "problems": []}) + "verdict": None, "exitcode": None, "problems": [], "orphans": []}) save_state(state) return f"started {phase} pid={proc.pid} start={start} batch={worlds}x{days} frozen={frozen} log={log}" def decide(state, alive): - """纯状态机转移。alive(pid,start,frag)->bool 注入便于合成测试。""" + """纯状态机转移。alive(pid,start,frag)->bool 注入便于合成测试。 + 批次参数固定取 PHASES[phase]；state 中的 worlds/days 仅作篡改检测，不作判据。""" phase = state.get("phase", "idle") if phase == "done": return None, state, None @@ -282,22 +381,25 @@ def decide(state, alive): if state.get("pid") and alive(state["pid"], state.get("start"), state.get("frag") or FRAG): return None, state, None # 进程结束: 收集全部证据 + problems = [] + worlds, days = PHASES[phase] + if state.get("worlds") != worlds or state.get("days") != days: + problems.append(f"state phase params tampered: state={state.get('worlds')}x{state.get('days')}, expected fixed {worlds}x{days}") state["exitcode"] = read_exitcode(state.get("exitfile")) log = state.get("log") p = Path(log) if log else None text = p.read_text(errors="replace") if p and p.exists() else "" - worlds, days = state.get("worlds"), state.get("days") - problems = [] - if worlds is None or days is None: - problems.append("state missing worlds/days (legacy or corrupt state)") - else: - problems += verify_batch(text, worlds, days) + problems += verify_batch(text, worlds, days) problems += verify_manifest(state.get("frozen"), state.get("manifest"), state.get("commit")) state["problems"] = problems state["pid"] = None if state["exitcode"] is None: + state["verdict"] = "FAIL" + state["problems"].insert(0, "missing exit code: process ended or disappeared before exitfile was written") + state["orphans"] = find_orphans(state.get("frozen") or "") state["phase"] = "interrupted" - return "save", state, f"{phase} interrupted without exit code (log={log}); stopped, awaiting human" + extra = f"; orphan child pids alive: {state['orphans']}" if state["orphans"] else "" + return "save", state, f"{phase} interrupted without exit code (log={log}){extra}; stopped, awaiting human" if state["exitcode"] == 0 and not problems: nxt = NEXT[phase] state["verdict"] = "PASS" @@ -314,9 +416,10 @@ def decide(state, alive): return None, state, None -def acquire_lock(path=LOCK_PATH): - Path(path).parent.mkdir(parents=True, exist_ok=True) - fd = os.open(str(path), os.O_RDWR | os.O_CREAT, 0o644) +def acquire_lock(path=None): + p = Path(path) if path is not None else LOCK_PATH + p.parent.mkdir(parents=True, exist_ok=True) + fd = os.open(str(p), os.O_RDWR | os.O_CREAT, 0o644) try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: @@ -325,6 +428,187 @@ def acquire_lock(path=LOCK_PATH): return fd +def make_batch_log(worlds, days, ok_token="OK"): + lines = [f"batch worlds={worlds} days={days} half_life_ticks=1 rel_threshold=0.50"] + for i in range(worlds): + ok = "OK" if ok_token == "OK" else "BAD" + lines.append(f"world={i} seed={42 + i} chats_max=10 chats_avg=1.00 ratio=3.000 gini=0.100 relcnt_max=2 relcnt_min=0 relconn=5/5 nonfinite=0 check_a=PASS check_b=PASS check_c=PASS old_crit=PASS {ok}") + lines.append(f"batch_summary worlds={worlds} passed={worlds if ok_token == 'OK' else 0} failed={0 if ok_token == 'OK' else worlds} old_passed={worlds}/{worlds} ratio_min=1.0 ratio_max=1.0 gini_min=0.1 gini_max=0.1 elapsed_s=1.0") + lines.append(f"M3_ACCEPTANCE={'PASS' if ok_token == 'OK' else 'FAIL'}") + return "\n".join(lines) + "\n" + + +def _run_main(): + buf = io.StringIO() + argv_saved = sys.argv + sys.argv = [argv_saved[0]] # 剥离 --selftest, 防止 main() 递归重入 selftest + try: + with contextlib.redirect_stdout(buf): + code = main() + finally: + sys.argv = argv_saved + return code, buf.getvalue() + + +def lifecycle_section(check): + """真实隔离生命周期测试: 临时目录 + 短真实子进程。不伪造 M3 结果。""" + global ROOT, SRC_BINARY, STATE_PATH, LOCK_PATH, LOG_DIR, FROZEN_ROOT, FRAG + with tempfile.TemporaryDirectory() as tds: + root = Path(tds) / "proj" + srcbin = root / "bin" + srcbin.mkdir(parents=True) + test_frag = f"M3SelftestHeadless_{os.getpid()}_{time.time_ns()}.dll" + (srcbin / test_frag).write_text("fake dll payload marker") + (srcbin / "runtimes").mkdir() + (srcbin / "runtimes/dep.dll").write_text("dep") + env = {**os.environ, "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t", + "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t"} + (root / "f.txt").write_text("x") + subprocess.run(["git", "init", "-q"], cwd=str(root), check=True) + subprocess.run(["git", "add", "-A"], cwd=str(root), check=True) + subprocess.run(["git", "commit", "-qm", "init"], cwd=str(root), check=True, env=env) + commit = git_commit(root) + check("lifecycle tmp git commit", bool(commit)) + + saved = (ROOT, SRC_BINARY, STATE_PATH, LOCK_PATH, LOG_DIR, FROZEN_ROOT, FRAG, runner_cmd) + try: + ROOT = root + FRAG = test_frag + SRC_BINARY = srcbin / FRAG + STATE_PATH = root / "scripts/m3_state.json" + LOCK_PATH = root / "scripts/lock" + LOG_DIR = root / "scripts/logs" + FROZEN_ROOT = root / "frozen" + + frozen, mpath, man = freeze_release(commit) + check("freeze recursive incl subdirs", (frozen / "runtimes/dep.dll").exists() + and man["files"].get("runtimes/dep.dll") is not None) + check("v2 manifest verify ok", verify_manifest(frozen, mpath, commit) == []) + same_second_root = root / "same-second-frozen" + old_strftime = time.strftime + same_second_error = None + same_second_dirs = [] + try: + time.strftime = lambda *args, **kwargs: "20260919T200000" + same_second_dirs.append(freeze_release(commit, frozen_root=same_second_root)[0]) + same_second_dirs.append(freeze_release(commit, frozen_root=same_second_root)[0]) + except Exception as exc: + same_second_error = exc + finally: + time.strftime = old_strftime + check("freeze same-second paths unique", + same_second_error is None and len(same_second_dirs) == 2 + and same_second_dirs[0] != same_second_dirs[1]) + (frozen / "extra.bin").write_text("e") + check("v2 extra file detected", any("extra file" in p for p in verify_manifest(frozen, mpath, commit))) + (frozen / "extra.bin").unlink() + (frozen / "runtimes/dep.dll").unlink() + check("v2 missing file detected", any("frozen file missing" in p + for p in verify_manifest(frozen, mpath, commit))) + # legacy v1 清单兼容: 所列文件通过即可，不做额外文件检测（兼容运行中的旧冻结副本） + (srcbin / "runtimes/dep.dll").write_text("dep") + leg = frozen / "legacy_dir" + leg.mkdir() + (leg / "a.dll").write_text("A") + legman = leg / "manifest.json" + legman.write_text(json.dumps({"commit": commit, + "files": {"a.dll": hashlib.sha256(b"A").hexdigest()}})) + (leg / "unlisted.txt").write_text("u") + check("legacy manifest verify ok (no extra check)", verify_manifest(leg, legman, commit) == []) + + def long_runner(dll, batch_args, exitfile): + q = shlex.quote + py = q(sys.executable) + code = q("import subprocess, time; subprocess.Popen(['sleep', '2']); time.sleep(2)") + return f"test -f {q(str(dll))}; {py} -c {code} {q(str(dll))}; echo $? > {q(str(exitfile))}" + + def fail_runner(dll, batch_args, exitfile): + q = shlex.quote + return f"sh -c 'exit 7'; echo $? > {q(str(exitfile))}" + + n_logs = lambda: len(list(LOG_DIR.glob("*.log"))) + + # (a) 控制器启动 phase1（真实 sh + Python 子进程, ~2s 存活） + save_state({"phase": "idle"}) + globals()["runner_cmd"] = long_runner + code, printed = _run_main() + check("controller launches phase1", code == 0 and "started phase1" in printed) + st = load_state() + check("launch state bound to frozen+commit", st["commit"] == commit and st["worlds"] == 30 and st["days"] == 20) + + # (b) 运行中重复调用: 静默, 不重复拉起 + code, printed = _run_main() + check("repeat call silent while alive", code == 0 and printed == "" and n_logs() == 1) + + # (c) 子进程正常退出但日志无有效批次 -> fail-closed FAIL 停机 + time.sleep(2.3) + code, printed = _run_main() + st = load_state() + diag = Path(str(st["log"]) + ".diagnosis.txt") + check("exit0 without valid batch -> FAIL stop", st["phase"] == "failed" and st["verdict"] == "FAIL" + and code == 0 and diag.exists()) + + # (d) 真实退出码 7 传播 + save_state({"phase": "idle"}) + globals()["runner_cmd"] = fail_runner + _run_main() + time.sleep(0.4) + _run_main() + st = load_state() + check("real child exit 7 propagated", st["exitcode"] == 7 and st["phase"] == "failed") + + # (e) 子进程被杀: shell 死 -> interrupted + 孤儿记录 + save_state({"phase": "idle"}) + globals()["runner_cmd"] = long_runner + _run_main() + st = load_state() + time.sleep(0.2) + os.kill(st["pid"], signal.SIGKILL) # 杀外层 sh, 内层 sleep sh 成为孤儿 + deadline = time.monotonic() + 1.5 + while proc_alive(st["pid"], st["start"], st["frag"]) and time.monotonic() < deadline: + time.sleep(0.02) + while not find_orphans(str(st["frozen"])) and time.monotonic() < deadline: + time.sleep(0.02) + _run_main() + st2 = load_state() + check("killed shell -> interrupted FAILED", + st2["phase"] == "interrupted" and st2["verdict"] == "FAIL") + check("orphan recorded", bool(st2.get("orphans"))) + check("interrupted diagnosis written", Path(str(st2["log"]) + ".diagnosis.txt").exists()) + + # (f) 控制器重启后绝不自动重启新 batch（孤儿/中断均终态） + logs_before = n_logs() + code, printed = _run_main() + st3 = load_state() + check("no relaunch after interrupted (controller restart)", + st3["phase"] == "interrupted" and n_logs() == logs_before and printed == "") + + # (g) state 丢失(idle) 但批次子进程存活: 拒绝重复拉起 + save_state({"phase": "idle"}) + guard = subprocess.Popen([ + sys.executable, "-c", "import time; time.sleep(2)", + str(SRC_BINARY), "--batch", "30", "20" + ]) + try: + logs_before = n_logs() + code, printed = _run_main() + check("idle refuses launch while batch child alive", + "refusing duplicate launch" in printed and n_logs() == logs_before) + finally: + if guard.poll() is None: + guard.kill() + guard.wait() + + for p in find_orphans(str(st3.get("frozen") or frozen)): + try: + os.kill(p, signal.SIGKILL) + except OSError: + pass + finally: + ROOT, SRC_BINARY, STATE_PATH, LOCK_PATH, LOG_DIR, FROZEN_ROOT, FRAG, rc = saved + globals()["runner_cmd"] = rc + + def selftest(): ok = True @@ -333,37 +617,58 @@ def selftest(): print(("PASS " if cond else "FAIL ") + name) ok = ok and cond - # ---- 解析 ---- - check("parse PASS", verify_batch("x\nM3_ACCEPTANCE=PASS\n", 1, 1) == ["missing 'batch worlds=K days=D' header line", "world line count 0 != expected 1", "missing batch_summary line"]) - check("parse empty", verify_batch("world=0 ...\n", 1, 1) != []) - check("verify full PASS log", verify_batch( - "batch worlds=2 days=20 half_life_ticks=1 rel_threshold=0.50\n" - "world=0 seed=42 ... OK\nworld=1 seed=43 ... OK\n" - "batch_summary worlds=2 passed=2 failed=0 old_passed=2/2 ratio_min=1.0 ratio_max=1.0 gini_min=0.1 gini_max=0.1 elapsed_s=1.0\n" - "M3_ACCEPTANCE=PASS\n", 2, 20) == []) - check("verify wrong params", any("params mismatch" in p for p in verify_batch( - "batch worlds=3 days=20 x\nworld=0 a OK\nworld=1 b OK\nworld=2 c OK\nbatch_summary worlds=3 passed=3 failed=0 x\nM3_ACCEPTANCE=PASS\n", 2, 20))) - check("verify BAD world", any("non-OK" in p for p in verify_batch( - "batch worlds=1 days=1 x\nworld=0 a BAD\nbatch_summary worlds=1 passed=0 failed=1 x\nM3_ACCEPTANCE=FAIL\n", 1, 1))) + # ---- 批次判据校验（固定参数注入） ---- + check("missing everything", verify_batch("x\nM3_ACCEPTANCE=PASS\n", 1, 1) == [ + "missing 'batch worlds=K days=D' header line", + "world line count 0 != expected 1", + "world index set mismatch: duplicates=[] missing=[0] extra=[] (count 0 vs 1)", + "missing batch_summary line"]) + check("verify full PASS log", verify_batch(make_batch_log(2, 20), 2, 20) == []) + check("verify wrong params", any("params mismatch" in p for p in verify_batch(make_batch_log(3, 20), 2, 20))) + check("1x1 log cannot certify phase2", any("params mismatch" in p for p in verify_batch(make_batch_log(1, 1), 50, 100))) + check("verify BAD world", any("non-OK" in p for p in verify_batch(make_batch_log(1, 1, ok_token="BAD"), 1, 1))) check("verify truncated worlds", any("world line count" in p for p in verify_batch( - "batch worlds=3 days=1 x\nworld=0 a OK\nbatch_summary worlds=3 passed=3 failed=0 x\nM3_ACCEPTANCE=PASS\n", 3, 1))) + "batch worlds=3 days=1 x\nworld=0 seed=42 a OK\nworld=1 seed=43 a OK\nbatch_summary worlds=3 passed=3 failed=0 x\nM3_ACCEPTANCE=PASS\n", 3, 1))) + check("duplicate world index", any("duplicates=[0]" in p for p in verify_batch( + "batch worlds=2 days=1 x\nworld=0 seed=42 a OK\nworld=0 seed=42 a OK\nbatch_summary worlds=2 passed=2 failed=0 x\nM3_ACCEPTANCE=PASS\n", 2, 1))) + check("wrong seed binding", any("seed binding mismatch" in p for p in verify_batch( + "batch worlds=2 days=1 x\nworld=0 seed=42 a OK\nworld=1 seed=99 a OK\nbatch_summary worlds=2 passed=2 failed=0 x\nM3_ACCEPTANCE=PASS\n", 2, 1))) + check("seed off-by-one rejected", any("seed binding mismatch" in p for p in verify_batch( + "batch worlds=1 days=1 x\nworld=0 seed=43 a OK\nbatch_summary worlds=1 passed=1 failed=0 x\nM3_ACCEPTANCE=PASS\n", 1, 1))) + check("malformed world line", any("malformed" in p for p in verify_batch( + "batch worlds=1 days=1 x\nworld=junk\nbatch_summary worlds=1 passed=1 failed=0 x\nM3_ACCEPTANCE=PASS\n", 1, 1))) check("verify summary mismatch", any("batch_summary mismatch" in p for p in verify_batch( - "batch worlds=2 days=1 x\nworld=0 a OK\nworld=1 b OK\nbatch_summary worlds=2 passed=1 failed=1 x\nM3_ACCEPTANCE=PASS\n", 2, 1))) + "batch worlds=2 days=1 x\nworld=0 seed=42 a OK\nworld=1 seed=43 a OK\nbatch_summary worlds=2 passed=1 failed=1 x\nM3_ACCEPTANCE=PASS\n", 2, 1))) check("verify missing verdict token", any("M3_ACCEPTANCE=PASS" in p for p in verify_batch( - "batch worlds=1 days=1 x\nworld=0 a OK\nbatch_summary worlds=1 passed=1 failed=0 x\nM3_ACCEPTANCE=FAIL\n", 1, 1))) + "batch worlds=1 days=1 x\nworld=0 seed=42 a OK\nbatch_summary worlds=1 passed=1 failed=0 x\nM3_ACCEPTANCE=FAIL\n", 1, 1))) - # ---- 冻结清单（合成目录） ---- + # ---- 冻结清单（合成目录, v2 递归 + 额外/缺失检测） ---- with tempfile.TemporaryDirectory() as td: td = Path(td) src = td / "bin"; src.mkdir() (src / "a.dll").write_bytes(b"hello") - (src / "b.json").write_text("{}") + (src / "b.json").write_bytes(b"{}") + (src / "runtimes").mkdir(); (src / "runtimes/r.dll").write_bytes(b"r") dst, mpath, man = freeze_release("deadbeef" * 8, src / "a.dll", td / "frozen") - check("freeze copies deps", (dst / "b.json").exists() and (dst / "a.dll").read_bytes() == b"hello" and man["files"]["b.json"]) + check("freeze copies deps recursively", (dst / "b.json").exists() + and (dst / "runtimes/r.dll").read_bytes() == b"r" + and man["files"].get("runtimes/r.dll") and man["version"] == 2) check("manifest verify ok", verify_manifest(dst, mpath, "deadbeef" * 8) == []) check("manifest commit bind", verify_manifest(dst, mpath, "f" * 64) != []) (dst / "a.dll").write_bytes(b"tampered") check("manifest detect tamper", any("hash mismatch" in p for p in verify_manifest(dst, mpath, "deadbeef" * 8))) + (dst / "sneaky.dll").write_bytes(b"x") + check("manifest detect extra", any("extra file" in p for p in verify_manifest(dst, mpath, "deadbeef" * 8))) + (dst / "sneaky.dll").unlink() + (dst / "runtimes/r.dll").unlink() + check("manifest detect missing", any("frozen file missing" in p for p in verify_manifest(dst, mpath, "deadbeef" * 8))) + # 旧 v1 清单: 只校验所列文件, 不做额外文件检测（兼容运行中的旧冻结副本） + leg = td / "leg"; leg.mkdir() + (leg / "a.dll").write_bytes(b"hello") + (leg / "unlisted.txt").write_text("u") + lm = leg / "manifest.json" + lm.write_text(json.dumps({"commit": "deadbeef" * 8, "files": {"a.dll": hashlib.sha256(b"hello").hexdigest()}})) + check("legacy manifest compatible", verify_manifest(leg, lm, "deadbeef" * 8) == []) # ---- 原子状态 ---- with tempfile.TemporaryDirectory() as td: @@ -387,13 +692,23 @@ def selftest(): td = Path(td) pr = td / "proc" / "4242" pr.mkdir(parents=True) - (pr / "stat").write_text("4242 (sh) S 1 4242 4242 0 -1 4194304 0 0 0 0 0 0 0 0 0 0 0 0 12345 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0\n") + (pr / "stat").write_text("4242 (sh) S 1 4242 4242 0 -1 4194304 0 0 0 0 0 0 0 0 0 0 0 0 12345 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0\n") (pr / "cmdline").write_bytes(b"sh -c dotnet LivingVillage.Headless.dll --batch 30 20\x00") check("starttime parsed", read_starttime(4242, td / "proc") == "12345") check("alive match", proc_alive(4242, "12345", "LivingVillage.Headless.dll", td / "proc")) + (pr / "stat").write_text((pr / "stat").read_text().replace("(sh) S ", "(sh) Z ")) + check("zombie rejected", not proc_alive(4242, "12345", "LivingVillage.Headless.dll", td / "proc")) + (pr / "stat").write_text((pr / "stat").read_text().replace("(sh) Z ", "(sh) S ")) check("starttime mismatch rejected", not proc_alive(4242, "99999", "LivingVillage.Headless.dll", td / "proc")) check("frag mismatch rejected", not proc_alive(4242, "12345", "other.dll", td / "proc")) check("missing proc rejected", not proc_alive(9999, "12345", "LivingVillage.Headless.dll", td / "proc")) + # state 丢失兜底: FRAG + --batch 存活进程必须被识别, 自身/无 --batch 排除 + (pr / "cmdline").write_bytes(b"dotnet /x/LivingVillage.Headless.dll --batch 30 20\x00") + pr2 = td / "proc" / "777"; pr2.mkdir() + (pr2 / "cmdline").write_bytes(b"sh -c sleep 9\x00") + pr3 = td / "proc" / str(os.getpid()); pr3.mkdir() + (pr3 / "cmdline").write_bytes(b"dotnet /x/LivingVillage.Headless.dll --batch 30 20\x00") + check("live batch orphan detected", live_batch_orphans(td / "proc") == [4242]) # ---- 退出码读取 ---- with tempfile.TemporaryDirectory() as td: @@ -404,41 +719,46 @@ def selftest(): ef.write_text("junk") check("exit junk -> None", read_exitcode(ef) is None) - # ---- sh 命令构造 ---- - c = build_sh_cmd("/tmp/a b/LivingVillage.Headless.dll", "--batch 30 20", "/tmp/x y.exit") - check("sh cmd quotes paths", "'/tmp/a b/LivingVillage.Headless.dll'" in c and "'/tmp/x y.exit'" in c and c.rstrip().endswith("> '/tmp/x y.exit'")) - - # ---- 状态机（合成日志/注入存活函数） ---- + # ---- 状态机（合成日志; worlds/days 以固定 PHASES 为准, state 篡改即 FAIL） ---- with tempfile.TemporaryDirectory() as td: td = Path(td) plog = td / "p.log" - plog.write_text( - "batch worlds=1 days=1 x\nworld=0 a OK\nbatch_summary worlds=1 passed=1 failed=0 x\nM3_ACCEPTANCE=PASS\n") - base = {"phase": "phase1", "worlds": 1, "days": 1, "pid": 123, "start": "1", - "frag": FRAG, "cmd": "c", "log": str(plog), - "exitfile": str(td / "e.exit"), "frozen": str(td / "fr"), - "manifest": str(td / "fr/manifest.json"), "commit": "c" * 40, - "verdict": None, "exitcode": None, "started": None, "problems": []} + plog.write_text(make_batch_log(30, 20)) (td / "fr").mkdir() (td / "fr/k.dll").write_bytes(b"kernel") (td / "fr/manifest.json").write_text(json.dumps( - {"commit": "c" * 40, "files": {"k.dll": hashlib.sha256(b"kernel").hexdigest()}})) + {"version": 2, "commit": "c" * 40, "files": {"k.dll": hashlib.sha256(b"kernel").hexdigest()}})) + base = {"phase": "phase1", "worlds": 30, "days": 20, "pid": 123, "start": "1", + "frag": FRAG, "cmd": "c", "log": str(plog), + "exitfile": str(td / "e.exit"), "frozen": str(td / "fr"), + "manifest": str(td / "fr/manifest.json"), "commit": "c" * 40, + "verdict": None, "exitcode": None, "started": None, "problems": [], "orphans": []} a, s, _ = decide(dict(base), lambda pid, st, fr: True) check("alive -> silent", a is None) (td / "e.exit").write_text("0\n") a, s, _ = decide(dict(base), lambda pid, st, fr: False) check("phase1 PASS -> launch phase2", a == "launch:phase2" and s["verdict"] == "PASS") - a, s, m = decide(dict(base, phase="phase2"), lambda pid, st, fr: False) + a, s, m = decide(dict(base, phase="phase2", log=str(plog)), lambda pid, st, fr: False) + check("phase2 with phase1 log -> FAIL (fixed PHASES)", s["phase"] == "failed" and s["verdict"] == "FAIL") + plog2 = td / "p2.log" + plog2.write_text(make_batch_log(50, 100)) + a, s, m = decide(dict(base, phase="phase2", worlds=50, days=100, log=str(plog2)), lambda pid, st, fr: False) check("phase2 PASS -> done", s["phase"] == "done" and m and "complete" in m) + (td / "e.exit").write_text("0\n") + tampered = dict(base, worlds=1, days=1) + a, s, _ = decide(tampered, lambda pid, st, fr: False) + check("state tamper fail-closed (1x1 state cannot pass phase1)", + s["phase"] == "failed" and any("tampered" in p for p in s["problems"])) (td / "e.exit").write_text("1\n") a, s, m = decide(dict(base), lambda pid, st, fr: False) check("exit 1 -> FAIL stop", s["phase"] == "failed" and s["verdict"] == "FAIL" and "FAIL" in m) (td / "e.exit").unlink() a, s, _ = decide(dict(base), lambda pid, st, fr: False) - check("no exitcode -> interrupted", s["phase"] == "interrupted") + check("no exitcode -> interrupted FAILED", + s["phase"] == "interrupted" and s["verdict"] == "FAIL") (td / "e.exit").write_text("0\n") (td / "fr/manifest.json").write_text(json.dumps( - {"commit": "d" * 40, "files": {"k.dll": hashlib.sha256(b"kernel").hexdigest()}})) + {"version": 2, "commit": "d" * 40, "files": {"k.dll": hashlib.sha256(b"kernel").hexdigest()}})) a, s, m = decide(dict(base), lambda pid, st, fr: False) check("commit mismatch -> FAIL even with token", s["phase"] == "failed" and any("commit" in p for p in s["problems"])) a, s, _ = decide({"phase": "done"}, lambda pid, st, fr: False) @@ -446,6 +766,9 @@ def selftest(): rep = generate_diagnosis(dict(base, phase="failed", exitcode=0, problems=["boom"]), ["boom"], "FAIL") check("diagnosis contains evidence", "M3 DIAGNOSIS FAIL" in rep and "boom" in rep and "exit code: 0" in rep) + # ---- 真实隔离生命周期测试（临时目录 + 短真实子进程） ---- + lifecycle_section(check) + print("SELFTEST=" + ("PASS" if ok else "FAIL")) return 0 if ok else 1 @@ -462,6 +785,12 @@ def main(): print("state file corrupt; stopped, awaiting human") return 1 if state.get("phase") == "idle": + live = live_batch_orphans() + if live: + # state 丢失/被清但批次子进程仍存活: 拒绝重复拉起, 保留现场等人 + print(f"live batch process(es) {live} found while state idle; " + "refusing duplicate launch, awaiting human") + return 0 if not SRC_BINARY.exists(): print(f"binary missing: {SRC_BINARY}; run dotnet build src/LivingVillage.Headless -c Release") return 1 |