test_skill_denominator_probe.py
5.43 KB
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
#!/usr/bin/env python3
"""VERIFY: 分母观测（RSI 三面镜子①，learn_queue #1426, 2026-10-14）
病理（HGB WHERE×WHY）：WHERE=Step6 apply-done 的观测登记分支；
WHY=观测登记只覆盖分子侧（bet 契约 / 收尾态 / 人工事项），分母侧两个 reason code
（SKILL_BLOAT / SKILL_ABLATION_DUE）定义了却 0 调用方 ⇒ Hermes 永远拿不到
「性能涨、成本不涨」这条判据的自读数（实测 570 个 SKILL.md / 125,634 行越阈无 trace）。
TDD：
TC1 空目录 → 0 个 skill、0 行、无 bloat
TC2 计数与头部集中度（多 skill，取最大者份额）
TC3 越阈 → bloat=True（用 patch 阈值常量注入，不建 201 个目录）
TC4 消融到期：无 ledger ⇒ due=True；ledger 新鲜 ⇒ due=False 且天数正确
TC5 源码守卫：`_on_apply_done` 内必须调用 `_skill_denominator`（防回退成死码）
"""
import json
import pathlib
import re
import tempfile
import time
APP = pathlib.Path("/data/data/com.termux/files/home/wechat-viewer/app.py")
SRC = APP.read_text(encoding="utf-8")
FAIL = 0
def check(name, cond, detail=""):
global FAIL
print(("PASS" if cond else "FAIL") + f" {name}" + (f" — {detail}" if detail else ""))
if not cond:
FAIL += 1
def load_denominator():
"""从源码抽出 _skill_denominator（避免 import app 的重量级依赖），补齐它用到的全局。"""
start = SRC.find("def _skill_denominator(")
end = SRC.find("def _run_absorb(", start)
ns = {"Path": pathlib.Path, "json": json,
"SKILLS_ROOT": pathlib.Path("/data/data/com.termux/files/home/.dsh/skills"),
"ABLATION_LEDGER": pathlib.Path("/data/data/com.termux/files/home/pixelphone/skill_ablation_ledger.json"),
"SKILL_BLOAT_COUNT": 200, "SKILL_BLOAT_LINES": 50000, "ABLATION_STALE_DAYS": 180}
exec(SRC[start:end], ns) # noqa: S102
return ns["_skill_denominator"], ns
def mk(root, rel, nlines):
p = root / rel
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(("\n".join(f"line{i}" for i in range(nlines)) + "\n") if nlines else "")
return p
def main():
fn, ns = load_denominator()
# TC1 空目录
with tempfile.TemporaryDirectory() as td:
r = fn(root=td, ledger=pathlib.Path(td) / "nope.json")
check("TC1 空目录 → 0/0/无bloat",
r["n_skills"] == 0 and r["total_lines"] == 0 and r["bloat"] is False
and r["ablation_due"] is True, str(r))
# TC2 计数 + 头部集中度
with tempfile.TemporaryDirectory() as td:
root = pathlib.Path(td)
mk(root, "a/x/SKILL.md", 10)
mk(root, "b/y/SKILL.md", 90)
mk(root, "b/z/SKILL.md", 0)
r = fn(root=root, ledger=pathlib.Path(td) / "nope.json")
check("TC2 计数=3 / 行数=100 / 头部=b/y 占 90%",
r["n_skills"] == 3 and r["total_lines"] == 100
and r["top_skill"] == "b/y/SKILL.md" and r["top_share_pct"] == 90, str(r))
# TC2b 备份目录不计入运行时分母
with tempfile.TemporaryDirectory() as td:
root = pathlib.Path(td)
mk(root, "a/x/SKILL.md", 5)
mk(root, ".curator_backups/2026-01-01/a/x/SKILL.md", 500)
r = fn(root=root, ledger=pathlib.Path(td) / "nope.json")
check("TC2b .curator_backups 被排除", r["n_skills"] == 1 and r["total_lines"] == 5, str(r))
# TC3 越阈（注入小阈值，避免建 201 个目录）
with tempfile.TemporaryDirectory() as td:
root = pathlib.Path(td)
mk(root, "a/x/SKILL.md", 30)
ns["SKILL_BLOAT_LINES"] = 10
r = fn(root=root, ledger=pathlib.Path(td) / "nope.json")
check("TC3 行数越阈 → bloat=True", r["bloat"] is True, str(r))
ns["SKILL_BLOAT_LINES"] = 50000
# TC4 消融到期
with tempfile.TemporaryDirectory() as td:
led = pathlib.Path(td) / "ledger.json"
now = time.time()
led.write_text(json.dumps({"last_ablation_at": now - 10 * 86400}))
r_fresh = fn(root=td, ledger=led, now=now)
check("TC4a ledger 新鲜(10天) → 未到期、天数=10",
r_fresh["ablation_due"] is False and r_fresh["last_ablation_days"] == 10, str(r_fresh))
led.write_text(json.dumps({"last_ablation_at": now - 200 * 86400}))
r_stale = fn(root=td, ledger=led, now=now)
check("TC4b ledger 过旧(200天) → 到期",
r_stale["ablation_due"] is True and r_stale["last_ablation_days"] == 200, str(r_stale))
# TC5 源码守卫：接线点必须调用
apply_block = SRC[SRC.find("def _on_apply_done("):]
apply_block = apply_block[:apply_block.find("def _run_apply(") if apply_block.find("def _run_apply(") > 0 else len(apply_block)]
check("TC5 _on_apply_done 内调用 _skill_denominator（防死码回退）",
"_skill_denominator()" in apply_block, f"block_len={len(apply_block)}")
check("TC5b 分母观察用既有 reason code SKILL_BLOAT",
re.search(r"step6_apply\.denominator[\s\S]{0,200}SKILL_BLOAT", apply_block) is not None)
# TC6 真实读数（冻结重跑样本）：对本机 skills 目录跑 3 次，须完全一致（确定性 Oracle）
samples = [fn() for _ in range(3)]
stable = all(s == samples[0] for s in samples)
check("TC6 本机读数 3/3 完全一致（确定性）", stable, str(samples[0]))
print(f"\n=== denominator probe: {'ALL PASS' if FAIL == 0 else f'{FAIL} FAILED'} ===")
return 1 if FAIL else 0
if __name__ == "__main__":
raise SystemExit(main())