宜博 / pixelphone

#!/usr/bin/env python3
"""VERIFY: Step5 双轴观测量（World Model / Actor）+ 兼容性检验 gap 登记
（PhAI Labs RSI-Anything, learn_queue #1388, 2026-10-13 Step6 审计）

学到的知识（原文 L9-L13）：
  RSI-Anything 的核心机制 = 策略（Actor）与世界模型（World Model）**允许异步更新**，
  但只有「通过组件评估 + 兼容性检验」的组合才被保留；候选筛选质量可独立于模型能力提升
  （固定 Coder + 固定候选预算，仅换筛选器 26.3%→31.2%，在线测试调用 ~3.7 次不变）。

病理 WHERE=step5_verify gate 决策
     WHY=Step5 逐轴验证（skill 存在 / memory 指纹），但 trace **不记录本次吸收了哪几根轴**
          → 双轴组合不可观测 ⇒ RSI-Anything 的「兼容性检验门」无输入可用（连「两侧是否
          同时被改过」都判不出来，更谈不上判「改后是否互相一致」）。
          同 effort_alloc / chain 口径：观测量，不评分、不作 Gate（CLM 检测与存储解耦）。

修复：
  1) `_verify_and_apply` 的 verify_log 增加 `axis=wm|actor|both|none` 观测量
  2) VerifyReason 登记两个 L2 gap：WM_ACTOR_COMPAT_UNCHECKED / CANDIDATE_FILTER_AXIS_BLIND

TDD：
TC1 (RED/复现旧病理) 旧 verify_log 无 axis 字段 → 双轴不可观测
TC2 (GREEN/真实数据) 本轮 #1388 的真实 verified 列表（reference 型，落点 SKILL.md）→ axis=actor
TC3 (GREEN/判别力) memory-only → wm；两条轴都有 → both；负对照空列表 → none
TC4 (源码守卫/防回退) app.py 含 axis 观测 + 两个 reason code 的定义与注册
TC5 (冻结重跑 spot check) 同一输入 3 次采样，结果一致（确定性纯函数）
"""
import pathlib
import sqlite3
import sys

APP = pathlib.Path("/data/data/com.termux/files/home/wechat-viewer/app.py")
DB = pathlib.Path("/data/data/com.termux/files/home/wechat-viewer/circle.db")
FAIL = 0


def old_verify_log(verified):
    """复现旧逻辑：只记 gate + verified 文案，无轴信息。"""
    return f"[node=step5_verify,attempt=1,gate=PASS,action=CONTINUE] {'; '.join(verified)}"


def axis_of(verified, cskill=""):
    """复现新逻辑：从 verified 证据列表判「有证据落盘的轴」。

    wm  = memory.json / MEMORY.md（World Model：对环境/记忆的理解）
    actor = skill: 前缀 或 SKILL.md（Actor：执行逻辑）
    """
    wm = any(v.startswith("memory:") or "memory.json" in v or "MEMORY.md" in v
             for v in verified)
    actor = bool(cskill) or any(v.startswith("skill:") or "SKILL.md" in v
                                or "skill '" in v for v in verified)
    if wm and actor:
        return "both"
    if wm:
        return "wm"
    if actor:
        return "actor"
    return "none"


# ── 取本轮真实吸收数据（learning_results #1388）──
con = sqlite3.connect(DB)
row = con.execute(
    "SELECT result_type, created_skill, learn_target FROM learning_results lr "
    "LEFT JOIN learn_queue q ON q.id = lr.learn_queue_id "
    "WHERE lr.learn_queue_id=1388 ORDER BY lr.id DESC LIMIT 1").fetchone()
REAL_TYPE, REAL_CSKILL, REAL_TARGET = (row[0] or "", row[1] or "", row[2] or "")
# 本轮 Step5 实际产出的 verified 列表（见 verify_log）
REAL_VERIFIED = [f"reference:{REAL_TARGET} SKILL.md confirmed (phrase)"]
con.close()

print("=== TC1(RED/复现旧病理): 旧 verify_log 无轴信息 ===")
old = old_verify_log(REAL_VERIFIED)
print(f"  旧 verify_log: {old}")
print(f"  {'✅' if 'axis=' not in old else '❌'} 旧 trace 不含 axis → 无法判「本次动了哪几根轴」")
if "axis=" in old:
    FAIL += 1

print("\n=== TC2(GREEN/真实数据): #1388 真实吸收的轴判定 ===")
print(f"  type={REAL_TYPE} target={REAL_TARGET} created_skill={REAL_CSKILL!r}")
print(f"  verified={REAL_VERIFIED}")
real_axis = axis_of(REAL_VERIFIED, REAL_CSKILL)
print(f"  判定: axis={real_axis}")
print(f"  {'✅' if real_axis == 'actor' else '❌'} reference→SKILL.md 吸收 = Actor 轴（执行逻辑/知识落点）")
if real_axis != "actor":
    FAIL += 1

print("\n=== TC3(GREEN/判别力): 四态判定 ===")
cases = [
    (["memory:content confirmed"], "", "wm", "只写 memory.json → World Model 轴"),
    (["skill:foo found"], "foo", "actor", "skill 存在 + created_skill → Actor 轴"),
    (["memory:MEMORY.md confirmed (fallback)", "reference:x SKILL.md confirmed (phrase)"],
     "", "both", "双轴同时落盘 → both（兼容性检验的适用情形）"),
    (["reference:confirmed (target skill 'foo' exists on disk)"], "",
     "actor", "旧确认性分支措辞（含空格无冒号）→ actor（冻结重跑 60 条历史发现：漏此措辞 ⇒ 21/60 误判 none）"),
    ([], "", "none", "负对照：无证据 → none（不得默认 both）"),
]
for verified, cskill, want, desc in cases:
    got = axis_of(verified, cskill)
    print(f"  {'✅' if got == want else '❌'} {desc}: axis={got} (期望 {want})")
    if got != want:
        FAIL += 1

print("\n=== TC4(源码守卫/防回退) ===")
src = APP.read_text(encoding="utf-8")
checks = [
    ("verify_log 含 axis 观测量", "axis={axis}" in src),
    ("axis 三态谓词 both/wm/actor", '"both" if (_axis_wm and _axis_actor)' in src),
    ("wm 轴谓词（memory.json/MEMORY.md）", 'v.startswith("memory:")' in src),
    ("actor 轴谓词（skill:/SKILL.md）", '"SKILL.md" in v' in src),
    ("actor 轴谓词覆盖旧措辞 skill '", '''or "skill '" in v for v in verified''' in src),
    ("reason code 定义 WM_ACTOR_COMPAT_UNCHECKED",
     "WM_ACTOR_COMPAT_UNCHECKED = " in src),
    ("reason code 定义 CANDIDATE_FILTER_AXIS_BLIND",
     "CANDIDATE_FILTER_AXIS_BLIND = " in src),
    ("reason code 注册 WM_ACTOR_COMPAT_UNCHECKED",
     "VerifyReason.WM_ACTOR_COMPAT_UNCHECKED:" in src),
    ("reason code 注册 CANDIDATE_FILTER_AXIS_BLIND",
     "VerifyReason.CANDIDATE_FILTER_AXIS_BLIND:" in src),
    ("观测量不参与 Gate 决策（口径注释）", "不评分、不作 Gate" in src),
]
for label, ok in checks:
    print(f"  {'✅' if ok else '❌'} {label}")
    if not ok:
        FAIL += 1

print("\n=== TC5(冻结重跑 spot check): 真实历史 trace 重放 3 次采样 ===")
runs = [axis_of(REAL_VERIFIED, REAL_CSKILL) for _ in range(3)]
stable = len(set(runs)) == 1
print(f"  本轮 #1388: 3 次结果 {runs} → {'✅ 3/3 一致' if stable else '❌ 漂移'}")
if not stable:
    FAIL += 1
# 冻结当前环境原样重放：最近 60 条 applied 的 Step5 trace（真实数据，非构造）
con = sqlite3.connect(DB)
hist = [r[0] or "" for r in con.execute(
    "SELECT error_msg FROM learn_queue WHERE status='applied' ORDER BY id DESC LIMIT 60")]
con.close()
samples = []
for _ in range(3):
    got = [axis_of([t.split("] ", 1)[-1]] if "]" in t else [t]) for t in hist]
    samples.append((got.count("wm"), got.count("actor"), got.count("both"), got.count("none")))
print(f"  历史 60 条 applied 重放（wm/actor/both/none）: {samples}")
print(f"  {'✅' if len(set(samples)) == 1 else '❌'} 3/3 采样一致（确定性重放）")
if len(set(samples)) != 1:
    FAIL += 1
cov = samples[0][0] + samples[0][1] + samples[0][2]
print(f"  轴可判覆盖率 {cov}/60（none={samples[0][3]}）"
      f" → {'✅' if cov >= 55 else '❌'} 谓词覆盖真实措辞（修复前 none=21）")
if cov < 55:
    FAIL += 1
print(f"  双轴(both) 历史发生数 = {samples[0][2]}/60 → 兼容性检验门当前无历史触发实例"
      f"（观测量是可落地前置，不是已修好的门）")

print(f"\n{'✅ ALL PASS' if FAIL == 0 else f'❌ {FAIL} FAILED'}")
sys.exit(0 if FAIL == 0 else 1)