From ef1f6b3d27e292a6b5e005411844eb8e14cbe4b2 Mon Sep 17 00:00:00 2001 From: "Somhairle H. Marisol" Date: Sat, 19 Sep 2026 19:20:07 +0800 Subject: infra: P0 监督加固 — m3_supervise.py flock防重入/原子状态/冻结Release全依赖+sha256清单/真实child退出码(sh -c exitfile)/PID+starttime身份/收紧PASS判据(批次参数+world行数全OK+batch_summary+verdict token+清单hash+commit绑定)/失败停机并写diagnosis报告, selftest 33项合成断言; completion_notify.py 新completion-notice.md hash仅打印一次缺失静默, selftest 7项; Headless --batch 世界级并行(LV_BATCH_WORKERS 默认modest 4, 逐世界确定性, 输出按序, world行与串行逐字节一致); .gitignore 补冻结/锁/通知状态文件 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/completion_notify.py | 132 +++++++++++ scripts/m3_supervise.py | 508 ++++++++++++++++++++++++++++++++++++------- 2 files changed, 559 insertions(+), 81 deletions(-) create mode 100644 scripts/completion_notify.py (limited to 'scripts') diff --git a/scripts/completion_notify.py b/scripts/completion_notify.py new file mode 100644 index 0000000..e18796b --- /dev/null +++ b/scripts/completion_notify.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python3 +"""completion-notice.md 变更通知：供 Hermes cron 调用。 + +行为（严格）: +- 通知文件缺失: 静默退出 0，无任何输出 +- 文件存在: 计算 sha256；与上次已通知 hash 相同 -> 静默退出 0 +- hash 为新值: 把文件内容原样打印到 stdout 一次（供 cron 捕获后发 Slack），并原子记录该 hash +- 每个新内容 hash 只打印一次；内容再次变化会再次打印 +- 状态文件损坏/不可读: 视为未通知过（打印一次） +- 不依赖任何模型凭据；纯本地文件与 hash + +用法: + completion_notify.py [--notice PATH] [--state PATH] [--selftest] +默认 notice=/docs/completion-notice.md, state=<脚本所在目录>/completion_notify_state.json。 +脚本放项目中，由 Hermes 复制到 cron 脚本目录时建议显式传 --notice/--state。 + +--selftest: 全部合成测试（临时文件模拟 缺失/新hash/重复/变更/损坏状态），不伪造真实结果。 +""" +import argparse +import hashlib +import json +import os +import sys +import tempfile +from pathlib import Path + +SCRIPT_DIR = Path(__file__).resolve().parent +DEFAULT_NOTICE = Path.cwd() / "docs/completion-notice.md" +DEFAULT_STATE = SCRIPT_DIR / "completion_notify_state.json" + + +def content_hash(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def load_state(path: Path): + try: + v = json.loads(Path(path).read_text()) + if isinstance(v, dict) and isinstance(v.get("hash"), str): + return v["hash"] + except Exception: + pass + return None + + +def save_state(path: Path, h: str): + p = Path(path) + p.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=str(p.parent), prefix=".notify_state.", suffix=".tmp") + try: + with os.fdopen(fd, "w") as fh: + fh.write(json.dumps({"hash": h}, indent=1) + "\n") + os.replace(tmp, p) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise + + +def run_once(notice_path, state_path, out=sys.stdout): + """返回 'printed' | 'silent' | 'missing'。打印发生在 out 上。""" + p = Path(notice_path) + try: + data = p.read_bytes() + except OSError: + return "missing" # 缺失静默 + h = content_hash(data) + if load_state(state_path) == h: + return "silent" + out.write(data.decode("utf-8", "replace")) + save_state(state_path, h) + return "printed" + + +def run_once_str(notice_path, state_path): + """测试用: 返回 (action, printed_text)。""" + import io + buf = io.StringIO() + action = run_once(notice_path, state_path, out=buf) + return action, buf.getvalue() + + +def selftest(): + ok = True + + def check(name, cond): + nonlocal ok + print(("PASS " if cond else "FAIL ") + name) + ok = ok and cond + + import tempfile + with tempfile.TemporaryDirectory() as td: + td = Path(td) + notice = td / "completion-notice.md" + state = td / "state" / "notify.json" + action, text = run_once_str(notice, state) + check("missing -> silent no output", action == "missing" and text == "") + notice.write_text("# 完成\n- 测试内容\n") + action, text = run_once_str(notice, state) + check("new content printed once", action == "printed" and text == notice.read_text()) + action, text = run_once_str(notice, state) + check("same hash silent", action == "silent" and text == "") + notice.write_text("# 完成 v2\n") + action, text = run_once_str(notice, state) + check("changed content printed again", action == "printed" and text == "# 完成 v2\n") + action, text = run_once_str(notice, state) + check("second view silent", action == "silent" and text == "") + state.write_text("not-json{{{") + action, text = run_once_str(notice, state) + check("corrupt state treated as new", action == "printed" and text == "# 完成 v2\n") + check("state saved after corrupt", load_state(state) == content_hash("# 完成 v2\n".encode())) + + print("SELFTEST=" + ("PASS" if ok else "FAIL")) + return 0 if ok else 1 + + +def main(): + ap = argparse.ArgumentParser(description="completion-notice.md new-hash notifier") + ap.add_argument("--notice", default=str(DEFAULT_NOTICE)) + ap.add_argument("--state", default=str(DEFAULT_STATE)) + ap.add_argument("--selftest", action="store_true") + args = ap.parse_args() + if args.selftest: + return selftest() + action = run_once(args.notice, args.state) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/m3_supervise.py b/scripts/m3_supervise.py index 3975b81..62fe94d 100755 --- a/scripts/m3_supervise.py +++ b/scripts/m3_supervise.py @@ -1,13 +1,32 @@ #!/usr/bin/env python3 -"""M3 验收监督：供 Hermes cron 每 10 分钟调用（no_agent）。 +"""M3 验收监督（加固版）：供 Hermes cron 每 5 分钟调用（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/_.log -- --selftest: 仅验证解析与状态机转移, 不启动模拟 +- flock 单实例防重入；抢不到锁静默退出 +- 冻结运行: 首次启动把 Release 输出目录整体复制到 scripts/m3_frozen/_/， + 记录全部文件 sha256 清单；两阶段都从冻结副本运行，后续构建不会覆盖运行中二进制 +- 真实退出码: 子进程为 sh -c 'dotnet --batch K D; echo $? > exitfile'， + 结束后读 exitfile 得到真实退出码（含信号 128+sig；shell 被杀则无 exitfile -> interrupted） +- PID 身份: 记录 /proc//stat 字段22 starttime，存活判定要求 pid+starttime+cmdline 全匹配（防 PID 复用） +- PASS 判定收紧（全部满足才 PASS，任一不满足即 FAIL/停机）: + 1) exitfile 退出码 == 0 + 2) 日志含 `batch worlds=K days=D` 且 K/D 与该阶段参数完全一致 + 3) `world=` 行数 == K 且每行末 token 为 OK + 4) `batch_summary worlds=K passed=K failed=0` + 5) 行 `M3_ACCEPTANCE=PASS` + 6) 冻结清单重新校验 sha256 全部一致，且清单 commit == 启动时记录 commit（commit 绑定） +- FAIL/无结论/校验失败: 停机不自动重启，并在日志旁写 .diagnosis.txt 可诊断报告 +- 状态文件 scripts/m3_state.json 原子写（tmp + os.replace） +- --selftest: 全部合成测试（解析/判据校验/原子状态/锁互斥/PID身份解析/诊断报告/退出码读取）， + 不启动任何真实模拟，不伪造真实结果 """ +import fcntl +import hashlib import json +import os +import re +import shlex +import shutil import subprocess import sys import tempfile @@ -15,89 +34,297 @@ import time from pathlib import Path ROOT = Path(__file__).resolve().parent.parent -BINARY = ROOT / "src/LivingVillage.Headless/bin/Release/net8.0/LivingVillage.Headless.dll" +SRC_BINARY = ROOT / "src/LivingVillage.Headless/bin/Release/net8.0/LivingVillage.Headless.dll" STATE_PATH = ROOT / "scripts/m3_state.json" +LOCK_PATH = ROOT / "scripts/m3_supervise.lock" LOG_DIR = ROOT / "scripts/m3_logs" -PHASES = {"phase1": ["--batch", "30", "20"], "phase2": ["--batch", "50", "100"]} -NEXT = {"phase1": "phase2", "phase2": "done"} +FROZEN_ROOT = ROOT / "scripts/m3_frozen" +PHASES = {"phase1": (30, 20), "phase2": (50, 100)} +NEXT = {"phase1": "phase2", "phase2": None} +FRAG = "LivingVillage.Headless.dll" -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 git_commit(root=ROOT): + try: + r = subprocess.run(["git", "rev-parse", "HEAD"], cwd=root, + capture_output=True, text=True, timeout=10) + c = r.stdout.strip() + return c if r.returncode == 0 and c else None + except Exception: + return None -def save_state(state): - STATE_PATH.write_text(json.dumps(state, indent=1, ensure_ascii=False) + "\n") +def sha256_file(path): + h = hashlib.sha256() + with open(path, "rb") as fh: + for chunk in iter(lambda: fh.read(1 << 20), b""): + h.update(chunk) + return h.hexdigest() -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 freeze_release(commit, src_binary=SRC_BINARY, frozen_root=None): + """复制 Release 输出目录全部文件到冻结目录并生成 sha256 清单。""" + frozen_root = Path(frozen_root) if frozen_root else FROZEN_ROOT + 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) + files = {} + for f in sorted(Path(src_binary).parent.iterdir()): + 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} + mpath = dst / "manifest.json" + mpath.write_text(json.dumps(manifest, indent=1, ensure_ascii=False) + "\n") + return dst, mpath, manifest -def pid_matches(pid, cmd): - if not pid: - return False +def verify_manifest(frozen_dir, manifest_path, expected_commit): + """重校验冻结目录: commit 绑定 + 全部文件 sha256。返回问题列表（空=通过）。""" + problems = [] + try: + m = json.loads(Path(manifest_path).read_text()) + except Exception as e: + return [f"manifest unreadable: {e}"] + if m.get("commit") != expected_commit: + problems.append(f"commit mismatch manifest={m.get('commit')} expected={expected_commit}") + files = m.get("files") or {} + if not files: + problems.append("manifest has no files") + for name, want in sorted(files.items()): + p = Path(frozen_dir) / name + if not p.exists(): + problems.append(f"frozen file missing: {name}") + continue + got = sha256_file(p) + if got != want: + problems.append(f"hash mismatch {name}: got {got[:16]} want {want[:16]}") + return problems + + +def load_state(path=STATE_PATH): + if Path(path).exists(): + try: + return json.loads(Path(path).read_text()) + except Exception: + return {"phase": "corrupt_state"} + return {"phase": "idle"} + + +def save_state(state, path=STATE_PATH): + """原子写: 同目录 tmp + os.replace。""" + p = Path(path) + p.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=str(p.parent), prefix=".m3_state.", suffix=".tmp") + try: + with os.fdopen(fd, "w") as fh: + fh.write(json.dumps(state, indent=1, ensure_ascii=False) + "\n") + os.replace(tmp, p) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise + + +def read_starttime(pid, proc_root="/proc"): + """/proc//stat 字段22 starttime（boot 后时钟滴答）。解析失败返回 None。""" try: - with open(f"/proc/{pid}/cmdline", "rb") as f: - argv = f.read().decode("utf-8", "replace").replace("\0", " ") + data = (Path(proc_root) / str(pid) / "stat").read_text() except OSError: + return None + rp = data.rfind(")") + 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[19] + + +def read_cmdline(pid, proc_root="/proc"): + try: + return (Path(proc_root) / str(pid) / "cmdline").read_bytes().decode("utf-8", "replace").replace("\0", " ") + except OSError: + return "" + + +def proc_alive(pid, start, frag, proc_root="/proc"): + """pid + starttime + cmdline 片段三者全匹配才算同一进程（防 PID 复用）。""" + if not pid or start is None: return False - return "LivingVillage.Headless.dll" in argv and (not cmd or cmd in argv) + now = read_starttime(pid, proc_root) + if now != str(start): + return False + return frag in read_cmdline(pid, proc_root) + + +def read_exitcode(exitfile): + p = Path(exitfile) + if not p.exists(): + return None + try: + return int(p.read_text().strip()) + except ValueError: + return None + + +def verify_batch(text, worlds, days): + """完整批次判据校验。返回问题列表（空=满足全部收紧条件）。""" + problems = [] + m = re.search(r"^batch worlds=(\d+) days=(\d+) ", text, re.M) + if not m: + problems.append("missing 'batch worlds=K days=D' header line") + else: + if int(m.group(1)) != worlds or int(m.group(2)) != days: + problems.append(f"batch params mismatch: log worlds={m.group(1)} days={m.group(2)}, expected {worlds}x{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}") + 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 ''}") + s = re.search(r"^batch_summary worlds=(\d+) passed=(\d+) failed=(\d+) ", text, re.M) + if not s: + problems.append("missing batch_summary line") + else: + if (int(s.group(1)), int(s.group(2)), int(s.group(3))) != (worlds, worlds, 0): + problems.append(f"batch_summary mismatch: worlds={s.group(1)} passed={s.group(2)} failed={s.group(3)}, expected {worlds}/{worlds}/0") + if not re.search(r"^M3_ACCEPTANCE=PASS$", text, re.M): + problems.append("missing line M3_ACCEPTANCE=PASS") + return problems + + +def generate_diagnosis(state, problems, tag): + """可诊断报告文本（纯函数，便于合成测试）。""" + log = state.get("log") + tail = "" + if log and Path(log).exists(): + lines = Path(log).read_text(errors="replace").splitlines() + tail = "\n".join(lines[-60:]) + exitcode = state.get("exitcode") + 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"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"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") + ) + + +def write_diagnosis(state, problems, tag): + log = state.get("log") + if not log: + return None + rep = Path(str(log) + ".diagnosis.txt") + rep.write_text(generate_diagnosis(state, problems, tag)) + return rep + + +def build_sh_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): + worlds, days = PHASES[phase] + 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" - cmd = ["dotnet", str(BINARY)] + PHASES[phase] + exitfile = Path(str(log) + ".exit") + cmd = ["sh", "-c", build_sh_cmd(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) + proc = subprocess.Popen(cmd, stdout=fh, stderr=subprocess.STDOUT, + stdin=subprocess.DEVNULL, start_new_session=True, cwd=ROOT) + start = read_starttime(proc.pid) + if start is None: + # 无法建立进程身份: 立即终止并按失败停机，不进入无身份运行 + try: + proc.kill() + except OSError: + pass + state.update({"phase": "failed", "worlds": worlds, "days": days, "pid": proc.pid, + "log": str(log), "verdict": None, "exitcode": None, + "problems": [f"cannot read starttime for new pid {proc.pid}; launch aborted"]}) + write_diagnosis(state, state["problems"], "LAUNCH_ABORTED") + save_state(state) + return f"launch-aborted {phase} pid={proc.pid} (no proc identity); stopped" state.update({ - "phase": phase, "pid": proc.pid, "cmd": str(BINARY), - "log": str(log), "started": time.strftime("%Y-%m-%dT%H:%M:%S"), - "verdict": None}) + "phase": phase, "worlds": worlds, "days": days, + "pid": proc.pid, "start": start, "frag": FRAG, + "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": []}) save_state(state) - return f"started {phase} pid={proc.pid} log={log}" + 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 注入便于合成测试。""" 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")): + if state.get("pid") and alive(state["pid"], state.get("start"), state.get("frag") or FRAG): return None, state, None + # 进程结束: 收集全部证据 + 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 "" - verdict = parse_acceptance(text) - state["verdict"] = verdict + 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_manifest(state.get("frozen"), state.get("manifest"), state.get("commit")) + state["problems"] = problems state["pid"] = None - if verdict == "PASS": + if state["exitcode"] is None: + state["phase"] = "interrupted" + return "save", state, f"{phase} interrupted without exit code (log={log}); stopped, awaiting human" + if state["exitcode"] == 0 and not problems: nxt = NEXT[phase] - if nxt == "done": + state["verdict"] = "PASS" + if nxt is None: state["phase"] = "done" - return "save", state, f"{phase} PASS -> M3 acceptance complete (30x20 + 50x100)" + return "save", state, f"{phase} PASS (commit={state.get('commit')}, frozen={state.get('frozen')}) -> 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" + state["verdict"] = "FAIL" + state["phase"] = "failed" + why = f"exit={state['exitcode']}" + (f" problems={problems}" if problems else "") + return "save", state, f"{phase} FAIL ({why}; log={log}); stopped, awaiting human" if phase == "idle": return "launch:phase1", state, None 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) + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + os.close(fd) + return None + return fd + + def selftest(): ok = True @@ -106,29 +333,119 @@ def selftest(): 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) + # ---- 解析 ---- + 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("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))) + 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))) + 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))) + + # ---- 冻结清单（合成目录） ---- + 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("{}") + 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("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))) + + # ---- 原子状态 ---- + with tempfile.TemporaryDirectory() as td: + sp = Path(td) / "state.json" + save_state({"phase": "phase1", "n": 1}, sp) + check("atomic roundtrip", load_state(sp) == {"phase": "phase1", "n": 1}) + check("no tmp leftovers", not [q for q in Path(td).iterdir() if q.name.endswith(".tmp")]) + + # ---- flock 互斥（同进程第二个 fd 抢锁必须失败） ---- + with tempfile.TemporaryDirectory() as td: + l1 = acquire_lock(Path(td) / "l.lock") + l2 = acquire_lock(Path(td) / "l.lock") + check("flock blocks second holder", l1 is not None and l2 is None) + if l1 is not None: + fcntl.flock(l1, fcntl.LOCK_UN) + l3 = acquire_lock(Path(td) / "l.lock") + check("flock released", l3 is not None) + + # ---- PID 身份（合成 /proc） ---- + with tempfile.TemporaryDirectory() as td: + 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 / "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")) + 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")) + + # ---- 退出码读取 ---- + with tempfile.TemporaryDirectory() as td: + ef = Path(td) / "x.exit" + check("exit missing -> None", read_exitcode(ef) is None) + ef.write_text("3\n") + check("exit read", read_exitcode(ef) == 3) + 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'")) + + # ---- 状态机（合成日志/注入存活函数） ---- + 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": []} + (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()}})) + 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) + check("phase2 PASS -> done", s["phase"] == "done" and m and "complete" in m) + (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") + (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()}})) + 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) + check("done -> silent", a is None) + 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) + print("SELFTEST=" + ("PASS" if ok else "FAIL")) return 0 if ok else 1 @@ -136,18 +453,47 @@ def selftest(): 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 + lock = acquire_lock() + if lock is None: + return 0 # 已有实例在运行: 静默退出 + try: + state = load_state() + if state.get("phase") == "corrupt_state": + print("state file corrupt; stopped, awaiting human") + return 1 + if state.get("phase") == "idle": + if not SRC_BINARY.exists(): + print(f"binary missing: {SRC_BINARY}; run dotnet build src/LivingVillage.Headless -c Release") + return 1 + commit = git_commit() + if not commit: + print("cannot determine git commit; refusing to launch unbound run") + return 1 + frozen, manifest, _ = freeze_release(commit) + state.update({"frozen": str(frozen), "manifest": str(manifest), "commit": commit}) + action, state, msg = decide(state, proc_alive) + if action and action.startswith("launch:"): + r = launch(action.split(":", 1)[1], state) + if r == "freeze-missing": + print("frozen release missing at launch; stopped, awaiting human") + save_state({**state, "phase": "failed", + "problems": ["frozen release/manifest/commit missing at launch"]}) + write_diagnosis(state, ["frozen release missing"], "LAUNCH_ABORTED") + return 1 + msg = r + elif action == "save": + save_state(state) + if state.get("phase") in ("failed", "interrupted"): + write_diagnosis(state, state.get("problems") or [], state.get("phase", "").upper()) + if msg: + print(msg) + return 0 + finally: + try: + fcntl.flock(lock, fcntl.LOCK_UN) + os.close(lock) + except OSError: + pass if __name__ == "__main__": -- cgit v1.2.3