DCU_DesktopController/scripts/check_guard_fires_hw.py

"""真机正向验证：主动制造输出冻结，证明守卫**真的**会触发，且 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)