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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
|
#!/usr/bin/env python3
"""M3R 缩尺批验独立验收脚本（不依赖 writer 会话，可独立复跑）。
用法:
python3 scripts/verify-m3r.py <acceptance.log> <state.json> [--commit 895d792]
只读判定，不改任何产物。校验:
1) state.json 字段齐全且自洽（pid/unit/log/commit/workers/worlds/days/起止/exit）；
2) 日志恰有 50 条 `world=` 行，索引 0..49 无重无缺，seed==42+i；
3) 逐世界 ok 字段 == (check_a and check_b and check_c)；check_a==(ratio>2.0)；
check_b==(relconn 连通数 >= 达标数)；check_c==(nonfinite==0)；
4) batch_summary worlds=50 且 passed+failed=50，passed/failed 与逐世界统计一致，
ratio_min/max、gini_min/max 与逐世界极值一致，gini∈(0,1)；
5) M3_ACCEPTANCE=PASS 且 ratio_min>2.0；
6) 抽查 3 个 world 行（排序后首/中/尾）判据字段与 summary 一致；
7) [done k/50] 恰 50 条且末条为 50/50，存在步进 `[world k] progress` 行。
"""
import json
import re
import sys
WORLD_RE = re.compile(
r"^world=(?P<world>\d+) seed=(?P<seed>\d+) chats_max=(?P<cmax>\d+) "
r"chats_avg=(?P<cavg>[0-9.]+) ratio=(?P<ratio>[0-9.]+) gini=(?P<gini>[0-9.]+) "
r"relcnt_max=(?P<rmax>\d+) relcnt_min=(?P<rmin>\d+) relconn=(?P<conn>\d+)/(?P<need>\d+) "
r"nonfinite=(?P<nf>\d+) check_a=(?P<a>PASS|FAIL) check_b=(?P<b>PASS|FAIL) "
r"check_c=(?P<c>PASS|FAIL) old_crit=(?P<old>PASS|FAIL) (?P<ok>OK|BAD)"
r"(?: elapsed_s=(?P<elapsed>[0-9.]+))?$"
)
SUMMARY_RE = re.compile(
r"^batch_summary worlds=(?P<worlds>\d+) passed=(?P<passed>\d+) failed=(?P<failed>\d+) "
r"old_passed=(?P<op>\d+)/(?P<ow>\d+) ratio_min=(?P<rmin>[0-9.]+) ratio_max=(?P<rmax>[0-9.]+) "
r"gini_min=(?P<gmin>[0-9.]+) gini_max=(?P<gmax>[0-9.]+)"
)
DONE_RE = re.compile(r"^\[done (?P<done>\d+)/(?P<total>\d+)\] ")
PROGRESS_RE = re.compile(r"^\[world \d+\] progress day=\d+/\d+ elapsed=[0-9.]+s$")
REQUIRED_STATE = [
"pid", "unit", "log", "commit", "workers", "worlds",
"days", "started_at", "ended_at", "exit",
]
def main(argv):
if len(argv) < 3:
print("usage: verify-m3r.py <acceptance.log> <state.json> [--commit X]")
return 2
log_path, state_path = argv[1], argv[2]
commit_expect = None
if len(argv) >= 5 and argv[3] == "--commit":
commit_expect = argv[4]
problems = []
with open(log_path, encoding="utf-8", errors="replace") as fh:
lines = [line.rstrip("\n") for line in fh]
with open(state_path, encoding="utf-8") as fh:
state = json.load(fh)
# 1) state.json
for key in REQUIRED_STATE:
if key not in state:
problems.append(f"state.json 缺字段 {key}")
if state.get("unit") != "living-village-m3r.service":
problems.append(f"unit 非法: {state.get('unit')!r}")
if not isinstance(state.get("pid"), int) or state.get("pid", 0) <= 0:
problems.append(f"pid 非法: {state.get('pid')!r}")
if state.get("exit") != 0:
problems.append(f"exit 非 0: {state.get('exit')!r}")
for key in ("started_at", "ended_at"):
if not state.get(key):
problems.append(f"{key} 为空")
if state.get("started_at") and state.get("ended_at") and state["ended_at"] < state["started_at"]:
problems.append("ended_at 早于 started_at")
if commit_expect and state.get("commit") != commit_expect:
problems.append(f"commit 不匹配: {state.get('commit')!r} != {commit_expect!r}")
for key, want in (("worlds", 50), ("days", 15)):
if state.get(key) != want:
problems.append(f"{key} != {want}: {state.get(key)!r}")
if not isinstance(state.get("workers"), int) or state.get("workers", 0) <= 0:
problems.append(f"workers 非法: {state.get('workers')!r}")
# 2) world= 行
worlds = {}
for line in lines:
m = WORLD_RE.match(line)
if not m:
continue
d = m.groupdict()
idx = int(d["world"])
if idx in worlds:
problems.append(f"world={idx} 重复")
worlds[idx] = d
if sorted(worlds) != list(range(50)):
problems.append(f"world 索引非 0..49（实际 {len(worlds)} 条，缺失 {sorted(set(range(50)) - set(worlds))}）")
for idx, d in worlds.items():
if int(d["seed"]) != 42 + idx:
problems.append(f"world={idx} seed={d['seed']} != {42 + idx}")
# 3) 逐世界判据自洽
ok_count = 0
checks_a = checks_b = checks_c = 0
ratios, ginis = [], []
for idx in sorted(worlds):
d = worlds[idx]
ratio, gini = float(d["ratio"]), float(d["gini"])
conn, need = int(d["conn"]), int(d["need"])
nf = int(d["nf"])
a, b, c = d["a"] == "PASS", d["b"] == "PASS", d["c"] == "PASS"
derived = a and b and c
if (d["ok"] == "OK") != derived:
problems.append(f"world={idx} ok={d['ok']} 与 check_a/b/c 不一致")
if a != (ratio > 2.0):
problems.append(f"world={idx} check_a={d['a']} 与 ratio={ratio} 不一致")
if b != (conn >= need):
problems.append(f"world={idx} check_b={d['b']} 与 relconn={conn}/{need} 不一致")
if c != (nf == 0):
problems.append(f"world={idx} check_c={d['c']} 与 nonfinite={nf} 不一致")
if gini <= 0.0 or gini >= 1.0:
problems.append(f"world={idx} gini={gini} 不在 (0,1)")
ratios.append(ratio)
ginis.append(gini)
ok_count += 1 if d["ok"] == "OK" else 0
checks_a += 1 if a else 0
checks_b += 1 if b else 0
checks_c += 1 if c else 0
# 4) batch_summary
summary = None
for line in lines:
m = SUMMARY_RE.match(line)
if m:
summary = m.groupdict()
if summary is None:
problems.append("缺 batch_summary 行")
else:
sw, sp, sf = int(summary["worlds"]), int(summary["passed"]), int(summary["failed"])
if sw != 50:
problems.append(f"batch_summary worlds={sw} != 50")
if sp + sf != 50:
problems.append(f"batch_summary passed+failed={sp + sf} != 50")
if sp != ok_count or sf != 50 - ok_count:
problems.append(f"batch_summary passed={sp}/failed={sf} != 逐世界 ok={ok_count}/bad={50 - ok_count}")
if abs(float(summary["rmin"]) - min(ratios)) > 5e-4 or abs(float(summary["rmax"]) - max(ratios)) > 5e-4:
problems.append(f"batch_summary ratio 极值 [{summary['rmin']},{summary['rmax']}] != 逐世界 [{min(ratios)},{max(ratios)}]")
if abs(float(summary["gmin"]) - min(ginis)) > 5e-4 or abs(float(summary["gmax"]) - max(ginis)) > 5e-4:
problems.append(f"batch_summary gini 极值 [{summary['gmin']},{summary['gmax']}] != 逐世界 [{min(ginis)},{max(ginis)}]")
for key, val in (("gmin", float(summary["gmin"])), ("gmax", float(summary["gmax"]))):
if not 0.0 < val < 1.0:
problems.append(f"batch_summary {key}={val} 不在 (0,1)")
# 5) 总判据
acceptance = [line for line in lines if line.startswith("M3_ACCEPTANCE=")]
if not acceptance:
problems.append("缺 M3_ACCEPTANCE 行")
elif acceptance[-1] != "M3_ACCEPTANCE=PASS":
problems.append(f"末条 {acceptance[-1]} != M3_ACCEPTANCE=PASS")
ratio_min = min(ratios) if ratios else 0.0
if not ratio_min > 2.0:
problems.append(f"ratio_min={ratio_min} 不满足 >2.0")
# 6) 抽查 3 个 world 行（排序后首/中/尾）
if len(worlds) == 50:
sample = [0, 24, 49]
print("sample worlds: " + " ".join(f"{i}(ok={worlds[i]['ok']},ratio={worlds[i]['ratio']},gini={worlds[i]['gini']},relconn={worlds[i]['conn']}/{worlds[i]['need']})" for i in sample))
if summary:
for i in sample:
r, g = float(worlds[i]["ratio"]), float(worlds[i]["gini"])
if not (float(summary["rmin"]) <= r <= float(summary["rmax"])):
problems.append(f"sample world={i} ratio={r} 不在 summary 区间")
if not (float(summary["gmin"]) <= g <= float(summary["gmax"])):
problems.append(f"sample world={i} gini={g} 不在 summary 区间")
# 7) done / progress
done = [DONE_RE.match(line) for line in lines]
done = [m for m in done if m]
if len(done) != 50:
problems.append(f"[done] 行数 {len(done)} != 50")
elif int(done[-1].group("done")) != 50 or int(done[-1].group("total")) != 50:
problems.append(f"末条 [done] 非 50/50: {done[-1].group(0)!r}")
progress = sum(1 for line in lines if PROGRESS_RE.match(line))
if progress < 50:
problems.append(f"步进 progress 行数 {progress} < 50（逐世界至少一条）")
print(f"worlds={len(worlds)} ok={ok_count} bad={50 - ok_count} check_a_pass={checks_a} check_b_pass={checks_b} check_c_pass={checks_c}")
print(f"ratio_min={ratio_min:.3f} ratio_max={max(ratios) if ratios else 0:.3f} gini_min={min(ginis) if ginis else 0:.3f} gini_max={max(ginis) if ginis else 0:.3f} progress_lines={progress}")
if problems:
print(f"M3R_VERIFY=FAIL problems={len(problems)}")
for p in problems:
print(f" - {p}")
return 1
print("M3R_VERIFY=PASS")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))
|
