"""真机正向验证：主动制造输出冻结，证明守卫**真的**会触发，且 60s 内不第二次触发。 这套判据历史上"修过一次又复发"，而复发时页面全绿、wr 恒 0，没有任何测试断言它 必须触发。纯逻辑门槛（test_guard_logic）已经能离线判死，但它证明的是"函数"， 这条证明的是"函数确实挂在这台机器的总线上工作"。 流程（两处物理动作都要人在场，脚本会停下来等）： 1. 前置：bus/op/loop=1、dc=250us、wkc>=3、pm_hold=0、recovering=0、心跳在涨； 并要求 since_full_s>61（60s 冷却是进程全局的，早先任何一次恢复都会吃掉预算） 2. 证明通路活着：发一帧，出口 tx_count/ack 必须动 3. → 请按 GD32 的 RESET：主站继续打包（cycles 涨）而出口冻住 = 现场症状复现 4. 只发一次（就是"点一下"），然后盯 guard：armed 必须在 1s 内起来， tx_stale_s 必须爬过 8s，fires 必须 +1（预算 25s：按 GD32 复位会连输入一起冻， 输入侧判据可能先做一次 OP-only 恢复，其 10s 冷却会把本判据的补触发往后推） 5. → 请让 GD32 重新跑起 APP：出口必须重新前进、dc/wkc/loop 恢复、再发一帧拿回显 6. 冷却：从触发那一刻起每 5s 一帧共 70s，fires 增量必须仍为 1 7.（可选 --with-pm-hold）再来一轮，但先 /api/pm/hold=1：要求 fires 零增量**且心跳 在涨** —— 证明"静音"是设计而不是线程死亡 用法：python scripts/check_guard_fires_hw.py [--with-pm-hold] 退出码：0=PASS 1=FAIL 2=BLOCKED（前置/不在场）。末行恒为 VERDICT。 """ import os import sys import time HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) from guard_probe import (Blocked, ask, bridge_of, guard_of, open_link, # noqa: E402 preflight, verdict) NATIVE_LOG = os.path.abspath(os.path.join(HERE, os.pardir, os.pardir, "ecat", "build_native", "native_stdout.log")) def ck(fails, cond, msg): print((" OK " if cond else " FAIL ") + msg) if not cond: fails.append(msg) def log_tail(n=400): try: with open(NATIVE_LOG, "rb") as f: f.seek(max(0, os.path.getsize(NATIVE_LOG) - 65536)) return f.read().decode("utf-8", "replace").splitlines()[-n:] except OSError: return [] def wait_freeze(lk, fails, budget=12.0): """等"主站还在打包、出口冻住"——现场症状的定义。""" t0 = time.time() seen = None while time.time() - t0 < budget: h = lk.health() st = lk.state() br = bridge_of(st) now = (br.get("tx_count"), br.get("ack"), h.get("cycles")) if seen and seen[0] == now[0] and seen[1] == now[1] and now[2] > seen[2]: print(" 出口已冻 %s（cycles 仍在涨 %s→%s），耗时 %.1fs" % (now[0], seen[2], now[2], time.time() - t0)) return True seen = now time.sleep(0.4) ck(fails, False, "%.0fs 内没出现「cycles 涨而出口不动」——冻结没制造成功" % budget) return False def run_round(lk, fails, tag, pm_hold=False): g0 = guard_of(lk.health()) st0 = lk.state() fires0 = g0["fires"] wr0 = st0.get("wr", 0) if pm_hold: lk.api("/api/pm/hold", {"hold": 1}) print(" pm_hold=1（守卫按设计静音；心跳必须仍在涨）") ask("请按 GD32 板上 RESET（让 APP 停止消费输出 PDO）") if not wait_freeze(lk, fails): ask("请让 GD32 重新跑起 APP，然后继续") return _r, ms = lk.send() print(" 单次发送完成（HTTP %.1fms）；开始盯 guard" % ms) t_fire = None armed_at = None stale_max = 0.0 t0 = time.time() while time.time() - t0 < 25.0: g = guard_of(lk.health()) if g["armed"] and armed_at is None: armed_at = time.time() - t0 stale_max = max(stale_max, g.get("tx_stale_s") or 0) if g["fires"] > fires0: t_fire = time.time() - t0 break time.sleep(0.5) hit = [l for l in log_tail() if "GUARD" in l and "全量重连" in l] if pm_hold: ck(fails, t_fire is None, "pm_hold 期间不得触发（%s）" % ("没触发 ✔" if t_fire is None else "%.1fs 就触发了 ✗" % t_fire)) g1 = guard_of(lk.health()) ticks0 = g0["alive_ticks"] time.sleep(1.2) g2 = guard_of(lk.health()) ck(fails, g2["alive_ticks"] - g1["alive_ticks"] >= 3, "静音中心跳仍在涨（+%d/1.2s）—— 证明是设计静音而不是线程死了" % (g2["alive_ticks"] - g1["alive_ticks"])) lk.api("/api/pm/hold", {"hold": 0}) ask("请让 GD32 重新跑起 APP，然后继续") return ck(fails, armed_at is not None and armed_at <= 1.5, "armed 在 %.2fs 内起来（旧判据在这里就已经断了）" % (armed_at if armed_at is not None else -1)) ck(fails, stale_max >= 8.0, "tx_stale_s 爬过 8s（峰值 %.1fs）" % stale_max) ck(fails, t_fire is not None, "fires %d→%d：守卫在 %.1fs 时触发（预算 25s）" % (fires0, guard_of(lk.health())["fires"], t_fire if t_fire is not None else -1)) ck(fails, bool(hit), "native_stdout.log 里有 [GUARD] … 全量重连 这一行（最近命中 %d 条）" % len(hit)) if t_fire is None: print(" ⇒ 判据没触发：先 curl /api/native/health 看 guard 块，" "再确认跑的是新 exe（tools/ecat/check_build_provenance.py）") ask("请让 GD32 重新跑起 APP（守卫已做过 close→open，恢复应该很快）") t1 = time.time() resumed = False while time.time() - t1 < 30.0: h = lk.health() if h.get("loop") == 1 and (h.get("wkc") or 0) >= 3 and h.get("dc_cycle_us") == 250: pre = bridge_of(lk.state()) _r, ms2 = lk.send() lag, ok = None, False t2 = time.perf_counter() while time.perf_counter() - t2 < 2.0: bb = bridge_of(lk.state()) if (bb.get("tx_count") != pre.get("tx_count") or bb.get("ack") != pre.get("ack")): ok, lag = True, (time.perf_counter() - t2) * 1000.0 break time.sleep(0.002) if ok: resumed = True print(" 恢复：出口重新前进（发送→跟上 %.1fms），probe HTTP %.1fms" % (lag, ms2)) break time.sleep(0.5) ck(fails, resumed, "松手后出口重新前进且总线回到 250us/OP") st1 = lk.state() ck(fails, st1.get("wr", 0) > wr0, "wr %d→%d（全量重连计一次；wr 汇总所有恢复路径，" "不能单独当本判据的凭据）" % (wr0, st1.get("wr", 0))) print(" 冷却观察：每 5s 一帧共 70s，期间 fires 不得再涨") f_now = guard_of(lk.health())["fires"] t3 = time.time() extra = 0 while time.time() - t3 < 70.0: lk.send() time.sleep(5.0) g = guard_of(lk.health()) if g["fires"] > f_now and time.time() - t3 < 60.0: extra += 1 f_now = g["fires"] g_end = guard_of(lk.health()) ck(fails, extra == 0 and g_end["fires"] <= fires0 + 2, "冷却守住：本轮 fires %d→%d，60s 内额外触发 %d 次" % (fires0, g_end["fires"], extra)) def main(): fails = [] rounds_pm = "--with-pm-hold" in sys.argv lk = open_link() preflight(lk) g = guard_of(lk.health()) since_full = g.get("since_full_s", -1) if since_full is not None and since_full >= 0 and since_full < 61: wait = 62 - since_full print("上次全量重连在 %.0fs 前，等 %.0fs 让 60s 冷却走完" % (since_full, wait)) time.sleep(wait) st = lk.state() act = [m for m in (st.get("motors") or []) if m.get("online")] if act: print(" 注意：有 %d 个电机槽位 online。本测试会让总线断流并强制重连，" % len(act)) ask("请确认电机已停止/未使能（否则输入 s 放弃）") print("\n=== 主轮：制造冻结，要求守卫触发 ===") run_round(lk, fails, "main") if rounds_pm: print("\n=== 对照轮：pm_hold=1 时同样的冻结不得触发，但心跳必须仍在涨 ===") run_round(lk, fails, "pm_hold", pm_hold=True) print("\n日志：%s" % NATIVE_LOG) return verdict(not fails, "; ".join(fails[:4])) if __name__ == "__main__": try: sys.exit(main()) except Blocked as e: print("BLOCKED %s" % e) print("VERDICT BLOCKED") sys.exit(2) except KeyboardInterrupt: print("VERDICT BLOCKED 中断（别忘了 GD32 可能还在 RESET 状态）") sys.exit(2)