"""Step 6 源码守卫:阶段→档位(reasoning-tier-allocation, learn_queue #1384, 2026-10-13)。 WHY(病理,HGB WHERE×WHY): ① WHERE=_spawn_hermes_job 的 DSH 日志快照环节 / WHY=快照块引用尚未绑定的 log_file (定义顺序倒置)→ NameError 被 `except: pass` 静默吞掉 → 声明的 CMD/ENV_KEYS/TIMEOUT 字段从未落盘(实测 3 份 job log CMD=0)。病理类别=silent-swallow of declared audit field。 ② WHERE=管线各阶段 spawn 配置 / WHY=无「阶段→档位」维度 → 验证阶段与执行阶段同档, 复查杠杆未启用且不可观测;补 PHASE/TIER_EXPECTED/MODEL 观测字段。 本测试防两类回退:字段被再次静默吞掉 / 观测字段被删。 """ from pathlib import Path SRC = Path(__file__).with_name("app.py").read_text(encoding="utf-8") # 顺序守卫:快照块(用 log_file)必须在 log_file 绑定之后 _bind_at = SRC.index("log_file = str(HERMES_JOBS_DIR") _snapshot_at = SRC.index("PHASE:{_phase} TIER_EXPECTED:{_tier_expected}") checks = [ # ① 静默吞错回归:快照块必须在 log_file 绑定之后(否则 NameError 被吞,字段永不落盘) ("快照块位于 log_file 绑定之后", _snapshot_at > _bind_at), # 快照失败必须留痕(禁止 silent pass) ("快照失败写 SNAPSHOT_ERROR 留痕", "SNAPSHOT_ERROR:" in SRC), # "w" 只能出现在快照块(作为 log 头部);_run 内必须全为 "a",否则快照被覆盖 ('log 写入仅快照块用 "w"(不被后续覆盖)', SRC.count('with open(log_file, "w")') == 1), ("CMD 快照字段保留", "CMD:{' '.join(cmd)}" in SRC), ("ENV_KEYS 快照字段保留", "ENV_KEYS:" in SRC), # ② 阶段→档位观测字段 ("PHASE 观测字段", "PHASE:{_phase} TIER_EXPECTED:{_tier_expected}" in SRC), ("MODEL pin/unpinned 观测字段", "MODEL:{_model_pin or 'DEFAULT(unpinned)'}" in SRC), ("PHASE_TIER_PLAN 映射表定义", "PHASE_TIER_PLAN = {" in SRC), ("验证阶段映射为高档(learn_apply → verify/high)", '"learn_apply": ("verify", "high")' in SRC), ("执行阶段映射为中档(learn_absorb → execute/mid)", '"learn_absorb": ("execute", "mid")' in SRC), # reason code 定义 + 注册 ("阶段档位 reason code 定义", 'PHASE_TIER_UNMAPPED = "PHASE_TIER_UNMAPPED"' in SRC), ("阶段档位 reason code 注册", "VerifyReason.PHASE_TIER_UNMAPPED:" in SRC), # N4:本修复不得改动任何 job 的模型 pin(用户定版配置) ("learn_absorb spawn 未被加 -m(N4:不擅自改用户定版档位)", '"learn_absorb", f"{ltype}: {target[:40]}",\n' ' ["hermes", "chat", "-q", prompt, "--yolo", "-t", "terminal,file,web",\n' ' "-s", "extract-learning-outline,learn-and-build"]' in SRC), ] failed = [label for label, ok in checks if not ok] for label, ok in checks: print(("PASS " if ok else "FAIL ") + label) if failed: print(f"\n{len(failed)}/{len(checks)} 检查未通过") raise SystemExit(1) print(f"\n{len(checks)}/{len(checks)} 检查通过")