test(e2e)+docs: 补上「平台侧积分电量」的**真实**端到端断言（此前只断言 suc=1）

## 起因

用户要求「`energy` 先实现自己积分，硬件能提供时再去掉」。**核实后：该积分早已实现**，
本轮真正缺的是**验证** —— 而且发现一个假覆盖。

## 核实结果：积分早已落地（不是新功能）

- **实现**：`software/api-server/src/workers/energy.rs`（12.9 KB / 8 个单测）——
  每次 `status_post` 按 `mWh = mV × mA × Δt / 3_600_000`（Δt 固定 60 秒）累加到
  **通道级** Redis 累加器（Lua 原子累加 + 最高温跟踪 + 连续无电池计数），
  建单 `reset`、结算 `take_and_reset` → 写 `charge_orders.energy` 与 `charge_records.energy`；
- **协议文档也已标注落地**：`docs/protocol/柜控板固件对接文档.md`「待确认事项」第 13 项
  写的是 **✅ 已落地（兜底方案已生效）**，并给出"硬件若能提供则平台改取硬件值"的口径；
- **`tasks/002` 把它记为「⏸ 阻塞中（等硬件答复）」是过时的** —— 已更正，并写明
  "硬件给值后怎么撤"：**单一入口 `energy::accumulate`**，落库字段/记录生成/前端展示都不用动。

## 本轮补的真东西：把假覆盖变成真验证

`test/tcp_e2e_smoke.py` 第 ⑥ 步标题一直写着"验证平台侧积分电量"，
**但只断言了 `suc=1`，从未检查电量值** —— 典型的"看起来验证了、其实没验证"。

新增第 ⑨ 步：`off` 结算后**读测试库**，断言：
1. `charge_orders.energy` == **由负载常量推导的期望值**
   （`CHARGING`=54.00V/8.00A × 60s × 3 次 ⇒ 21600 mWh ⇒ **22 Wh**）；
2. 已生成 `charge_records`，且其 `energy` 与订单**一致**（同一积分值写两处，分叉即为缺陷）；
3. `max_temperature` 被积分路径一并跟踪（29800 K×100）。

**实测结果（测试环境，18/18 通过、exit 0）**：
```
订单 id=58 status=1 energy=22Wh | 记录 energy=22Wh 最高温=29800K×100 | 期望=22Wh
```

**已自证能失败**：把 `CHARGING` 的电流由 8.00A 改为 4.00A（改**真实负载**而非期望值）
→ 实测 `11Wh ≠ 22Wh` → **FAILED / exit 1**；还原 → 18/18 通过。

## 顺带修掉一个"误导性红"

连续跑第 6 次本脚本会撞上 `auth_str` 的**「同一 IMEI 5 次 / 5 分钟」频率限制**（协议 §5.1），
于是第 ① 步被限流、后续 12 条断言**级联失败**，输出看起来像"登录/建单/结算全坏了"。
现第 ① 步即识别 `err == "rate limit exceeded"`，**明确说明是重复运行所致、不是功能缺陷**，
并提示 `systemctl restart pms-api-test` 清零进程内计数器，以 **exit 2（环境问题）**结束。

> 这个坑我实际踩到了：一度以为改动引入了回归，按「踩过的坑」第 7 条的思路先怀疑夹具/测量方式，
> 才定位到是限流。

## 退出码

新增"未验证"档：**读不到测试库 ⇒ exit 2**（需要 SSH 隧道），
**绝不当成通过**；有断言失败仍优先 exit 1。

## 验收

- `cargo check --all-targets`：0 error / 0 warning（本轮未改 Rust 代码）；
- `cargo test --bin pms-api`：**189 passed**；
- `python tools/db_schema_diff.py`：**无漂移**；
- `test/schema_alignment_e2e.py --env dev`：全部通过 exit 0；
- `test/tcp_e2e_smoke.py`：**18/18 exit 0**（含新增 5 条积分断言）；
- **发布**：本轮只改测试脚本与文档，**无需重新发布**；且实测生产与测试运行的是
  **逐字节相同的二进制**（api `f7e075a4a7a5983ebfe9b641e3d211cfdf7154361627858cbe68737211ff0f95`），
  即测试环境验证的积分链路**就是生产在跑的那份**。

## 文档

- `AGENTS.md` 验收流程第 5 项：注明该脚本现已含电量断言（**需要 SSH 隧道**）、
  以及 `auth_str` 限流的注意事项；
- `tasks/002`（不入库）：把 `energy` 从"阻塞中"更正为"已实现 + 已端到端验证"，
  并保留"硬件给值后如何撤"的口径。

AGENTS.md

@@ -318,6 +318,11 @@ python test/schema_alignment_e2e.py --env dev --jwt-secret <JWT_SECRET>
 python test/schema_alignment_e2e.py --env prod --jwt-secret <JWT_SECRET> --yes-prod
 
 # 5) TCP 协议端到端（模拟柜控板，打测试环境 device-server :3001）
+#    含 ⑨ 步「平台侧积分电量」断言：读测试库核对 charge_orders.energy 是否等于
+#    由负载常量推导的期望值（⇒ **需要 SSH 隧道**，连不上会 exit 2「未验证」而非通过）
+#    ⚠️ 前置：先 python test/reset_fixture.py；且 auth_str 有「同一 IMEI 5 次/5 分钟」限流，
+#       连续跑第 6 次会撞限流 —— 脚本会明确报"被频率限制/exit 2"，此时
+#       `systemctl restart pms-api-test` 清零进程内计数器即可（**不是功能缺陷**）
 python test/tcp_e2e_smoke.py
 
 # 6) 状态位解析两端一致性（逐位遍历，断言 PC 与小程序口径一致）

test/tcp_e2e_smoke.py

@@ -8,6 +8,7 @@
 import sys
 import hashlib
 import json
+import os
 import socket
 import time
 
@@ -15,6 +16,23 @@ HOST, PORT = "10.8.0.252", 3001
 IMEI = "860000000000001"
 BATTERY_ID = "SN-LEGACY-001"
 
+# ⚠️ **平台侧积分电量的期望值推导**（与 status_post 的负载常量绑定）：
+#   CHARGING = 54.00V / 8.00A ⇒ 电压 5400×10 = 54000 mV、电流 800×10 = 8000 mA
+#   积分口径（software/api-server/src/workers/energy.rs）：
+#       mWh = mV × mA × Δt / 3_600_000，Δt 固定 60 秒
+#   本脚本第 ⑥ 步发 3 次 status_post ⇒ 期望 Wh = (3×7200 + 500) / 1000 = 22（mWh→Wh 四舍五入）
+ENERGY_SAMPLES = 3
+ENERGY_VOLTAGE_MV = 5400 * 10
+ENERGY_CURRENT_MA = 800 * 10
+ENERGY_DT_SECS = 60
+
+
+def expected_energy_wh():
+    """按积分公式算出本脚本应落库的订单电量（Wh）。"""
+    per_sample_mwh = ENERGY_VOLTAGE_MV * ENERGY_CURRENT_MA * ENERGY_DT_SECS // 3_600_000
+    total_mwh = ENERGY_SAMPLES * per_sample_mwh
+    return (total_mwh + 500) // 1000
+
 
 def hex_channel(batt: int, power: int, soc_pct: int, voltage_001v: int,
                 current_001a: int, temp_k: int) -> str:
@@ -90,8 +108,18 @@ ok = []
 
 step("① auth_str（未登录可发）")
 cab.send({"act": "auth_str", "msg_id": cab.next_id(), "dev_id": IMEI})
-resp = cab.recv()
+resp = cab.recv() or {}
-auth_str = resp.get("auth_str") if resp else None
+# ⚠️ **限流要单独报**：协议 §5.1 规定同一 IMEI「5 次 / 5 分钟」。
+# 连续多跑几次本脚本就会撞上 —— 此时后续每一步都会级联失败，
+# 输出看起来像"登录/建单/结算全坏了"。必须与真正的功能缺陷区分开：
+# 明确说清原因并以**退出码 2（环境/夹具问题）**结束，别让人去查不存在的 bug。
+if resp.get("err") == "rate limit exceeded":
+    print("  ⚠️ 未验证：auth_str 被频率限制（协议 §5.1：同一 IMEI 5 次/5 分钟）")
+    print("     这是**重复运行本脚本**导致的，不是功能缺陷。")
+    print("     处理：等 5 分钟再跑；或重启测试环境 api-server 清零进程内计数器")
+    print("           （systemctl restart pms-api-test）；再跑前先 python test/reset_fixture.py")
+    sys.exit(2)
+auth_str = resp.get("auth_str")
 ok.append(("auth_str 返回 8 位安全码", bool(auth_str) and len(auth_str) == 8))
 print(f"  安全码 = {auth_str}")
 
@@ -151,6 +179,82 @@ except Exception as e:  # noqa: BLE001
 
 cab.sock.close()
 
+# ---------------------------------------------------------------------------
+# ⑨ 平台侧积分电量：**端到端断言**（读测试库，而不是只看 suc=1）
+# ---------------------------------------------------------------------------
+unverified = []
+step("⑨ 平台侧积分电量（读测试库：charge_orders.energy 应等于积分期望值）")
+
+
+def fetch_settled_order(timeout_secs=15):
+    """轮询测试库，返回 `(order_id, order_status, order_energy, compartment_id, record)`。
+
+    `record` = 该仓体最新 `charge_records` 行 `(energy, max_temperature)`，无则 None。
+
+    ⚠️ 列名以 information_schema 实测为准：**`max_temperature` 只在 `charge_records` 上**，
+    `charge_orders` 没有该列（早期版本这里写错，被"未验证"机制挡下而非静默通过）。
+    ⚠️ 结算在 report_worker 里是异步的（`off` 的 `suc` 可能在写回电量之前返回），
+    所以这里**轮询**而不是读一次。连不上库时抛异常，由调用方判为"未验证"。
+    """
+    import pymysql
+
+    host = os.environ.get("PMS_MYSQL_HOST", "127.0.0.1")
+    port = int(os.environ.get("PMS_MYSQL_PORT", "13306"))
+    db = os.environ.get("PMS_MYSQL_DB", "pms_dev")
+    conn = pymysql.connect(host=host, port=port, user="root",
+                           password="Hbhyg731024@", database=db, autocommit=True)
+    try:
+        cur = conn.cursor()
+        deadline = time.time() + timeout_secs
+        last = None
+        while time.time() < deadline:
+            cur.execute(
+                "SELECT co.id, co.status, co.energy, co.compartment_id "
+                "  FROM charge_orders co JOIN cabinets c ON c.id = co.cabinet_id "
+                " WHERE c.imei = %s ORDER BY co.id DESC LIMIT 1",
+                (IMEI,),
+            )
+            last = cur.fetchone()
+            if last and last[1] != 0:  # status <> 0 ⇒ 已结算
+                break
+            time.sleep(0.5)
+        if last is None:
+            return None
+        cur.execute(
+            "SELECT energy, max_temperature FROM charge_records "
+            " WHERE compartment_id = %s ORDER BY id DESC LIMIT 1",
+            (last[3],),
+        )
+        return last + (cur.fetchone(),)
+    finally:
+        conn.close()
+
+
+try:
+    row = fetch_settled_order()
+    if row is None:
+        ok.append(("积分电量：存在订单", False))
+    else:
+        order_id, order_status, energy_wh, _comp_id, rec = row
+        want = expected_energy_wh()
+        rec_energy, rec_temp = rec if rec else (None, None)
+        print(f"  订单 id={order_id} status={order_status} energy={energy_wh}Wh | "
+              f"记录 energy={rec_energy}Wh 最高温={rec_temp}K×100 | 期望={want}Wh")
+        ok.append(("积分电量：订单已结算", order_status != 0))
+        ok.append((f"积分电量：charge_orders.energy == 积分期望值 {want}Wh（实测 {energy_wh}）",
+                   energy_wh == want))
+        ok.append(("积分电量：已生成 charge_records", rec is not None))
+        # 同一积分值写两处，必须一致（两处不一致说明写回路径有分叉）
+        ok.append((f"积分电量：charge_records.energy 与订单一致（记录 {rec_energy}）",
+                   rec_energy == want))
+        # 最高温由积分路径一并跟踪（CHARGING 的 temp = 298K ⇒ K×100 = 29800）
+        ok.append((f"积分路径：最高温已跟踪（29800 K×100，实测 {rec_temp}）", rec_temp == 29800))
+except Exception as e:  # noqa: BLE001
+    print(f"  ⚠️ 未验证：连不上测试库（{e}）")
+    print("     需要先起 SSH 隧道：pwsh tools/dev-tunnel.ps1"
+          "（或用 PMS_MYSQL_HOST/PMS_MYSQL_PORT/PMS_MYSQL_DB 指向别处）")
+    unverified.append("平台侧积分电量（连不上测试库 —— 见上方 ⚠️，通常是没有 SSH 隧道）")
+
 print("\n========== 联调结果 ==========")
 for name, passed in ok:
     print(f"  {'✅' if passed else '❌'} {name}")
@@ -163,4 +267,9 @@ if failed:
     for n in failed:
         print(f"  ❌ {n}", file=sys.stderr)
     print(f"合计 {sum(1 for _, p in ok if p)}/{len(ok)} 通过，{len(failed)} 项失败", file=sys.stderr)
-sys.exit(0 if not failed else 1)
+if unverified:
+    # ⛔ 未验证 ≠ 通过：环境问题按仓库退出码约定走 **2**
+    print("未验证项（stderr，**不是通过**）：", file=sys.stderr)
+    for n in unverified:
+        print(f"  ⚠️ {n}", file=sys.stderr)
+sys.exit(1 if failed else (2 if unverified else 0))