1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
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())
|
