feat(wechat-auto-reply): 环节5 API 发送打通 — onResume flush + 送达校验

- text_b64 传中文(--es text 被 shell/am 弄乱→卡 status=1) - z2.b 只入队,发送后 HOME+LauncherUI 触发 onResume 派发(设备机制非代码 bug) - _send_text_verified 轮询 message.status 直到 2,超时仍 1 视为已入队不误报 - 拆 stage 目录(watcher=_stage / orchestrator=_stage_orch)避开拷库竞态 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

feat(wechat-auto-reply): 环节5 API 发送打通 — onResume flush + 送达校验
- text_b64 传中文(--es text 被 shell/am 弄乱→卡 status=1) - z2.b 只入队,发送后 HOME+LauncherUI 触发 onResume 派发(设备机制非代码 bug) - _send_text_verified 轮询 message.status 直到 2,超时仍 1 视为已入队不误报 - 拆 stage 目录(watcher=_stage / orchestrator=_stage_orch)避开拷库竞态 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
yibo
1 parent 70ff426e
Showing 4 changed files with 110 additions and 12 deletions
md/wechat-auto-reply-requirements.md
wechat_plain/scripts/auto-reply/README.md
wechat_plain/scripts/auto-reply/wechat_reply_orchestrator.py
wechat_plain/scripts/auto-reply/wechat_reply_watcher.py
@@ -100,7 +100,15 @@
 - 安全：生成超时/失败 → 无草稿 → policy skip → 不发（失败即静默，绝不误发）。
 - 真机验证：接线（watcher inotify → 自动拉起编排器 drain）通过；真实上下文（猎头 婷婷）生成「哈哈婷婷，我不是什么总，就是个写代码的，不负责招人这块哈」→ 正确进 /approvals 待审。
-### 环节 5 · 发送回复带引用（原生气泡）
+### 环节 5 · 可靠发送 —— ✅ **API 发送打通(2026-07-11)**
+经历一场硬仗才让 API 发送真正送达(火星人跨账号收到确认)。两个坑：
+1. **中文被弄乱**：`send_text_via_hook` 用 `--es text` 传中文被 shell/am 弄乱→服务器拒收→卡 status=1。hook 早已要求 `--es text_b64`,发送端漏改。→ 改用 text_b64。
+2. **z2.b 只入队不派发(设备机制,非代码 bug)**：`sendViaHighLevel` 的 `z2.b` 把消息入队(status=1 转圈),WeChat NetSceneQueue **要一次 onResume(后台→前台切换)才派发**。发送代码跨版本逐字节相同,不是回归。→ 发送后自动 `input keyevent KEYCODE_HOME` + `am start LauncherUI` 触发 onResume,消息 flush 到 status=2。
+- 落地 `_send_text_verified`(orchestrator)：前台微信 → text_b64 发送 → 后台前台切换 → 轮询 status 直到 2。status=1 超时视为"已入队将送达"(不误报失败)。
+- 详见 memory `wechat-api-send-onresume-flush`。
+- **待深挖(下次)**：onResume 到底改变了什么让队列派发(疑似后台暂停网络场景/长连接休眠)——要 hook NetSceneQueue/网络模块在前后台切换点打 log 拿根因。
+
+### （原）环节 5 · 发送回复带引用（原生气泡）
 需求 = 点原文→引用→发送后显示引用气泡。两条实现路：
 - **(a) A11y UI 自动化（一期推荐）**：长按目标气泡 → 点"引用" → 输入 → 发送。驱动微信自己 UI → **产出货真价实的原生引用气泡，零逆向**。代价：需微信前台、脚本略脆、比 API 慢。已有 A11y 发送服务底子。
 - **(b) 内部 API 构造 refermsg（二期优化）**：微信引用 = type 49 appmsg + refermsg(type 57) 指向被引 svrid。一整块新逆向，量级同当初啃发图，重。留作提速。
@@ -46,6 +46,10 @@
 - **环节4 为何必须是确定性代码**：环节3 实测「SOUL.md 里的软约束会漏」（约饭案例模型仍主动接话）。安全门必须是 LLM 说服不动的硬规则：default-deny + 敏感词（钱/承诺/约时间/隐私，白名单也拉回）+ skip（群没@我/公众号/系统消息/自己发的）。
 - **账号自适应**：运行时 `MD5("mm"+uin)` 定位当前登录账号，`MD5("1234567890ABCDEF"+uin)[:7]` 推导 SQLCipher key，多账号切换自动跟随（不硬编码）。
 - **失败即静默**：生成超时/失败 → 无草稿 → policy skip → 不发。最坏情况是不回，绝不误发。
+- **API 发送的两个坑（2026-07-11 硬仗定位，详见 memory `wechat-api-send-onresume-flush`）**：
+  1. 中文必须走 `--es text_b64`（`--es text` 被 shell/am 弄乱 → 卡 status=1）。
+  2. `z2.b` 只入队（status=1 转圈），**WeChat NetSceneQueue 要一次 onResume（后台→前台切换）才派发** → 发送后自动 `KEYCODE_HOME` + `am start LauncherUI` 触发派发 → status=2。**这是设备机制不是代码 bug**，发送代码跨版本逐字节相同，别去 diff/重编译 hook 找代码差异。
+  3. 送达校验：轮询 `message.status`（1=发送中/2=已达/5=失败）；超时仍 1 视为"已入队将送达"，不误报失败。
 ## 运维
@@ -159,10 +159,93 @@ def _queue_insert(event, draft, sender_disp, action, reason, status):
     con.close()
-def _send_text(to_wxid, text):
+VERIFY_TIMEOUT = 25   # 轮询确认送达的总超时(秒)
+VERIFY_POLL = 3       # 轮询间隔
+
+
+def _cleanup_verify():
+    for p in __import__("glob").glob(f"{WORK_DIR}/_stage_verify/en.db*"):
+        try:
+            os.remove(p)
+        except OSError:
+            pass
+
+
+def _sent_status(db, key, talker, base_msgid, text):
+    """查我方(isSend=1)在 base 之后、内容==text 的那条消息的 status。
+    微信 status: 1=发送中(转圈) / 2=已发送 / 5=失败。查不到返回 None。"""
+    st = w.copy_db(db, "_stage_verify")
+    try:
+        rows = w.sqlcipher_query(
+            st, key,
+            f"SELECT status, hex(content) FROM message WHERE talker='{talker}' "
+            f"AND isSend=1 AND msgId>{base_msgid} ORDER BY msgId DESC LIMIT 8;")
+    finally:
+        _cleanup_verify()
+    for r in rows:
+        if len(r) < 2:
+            continue
+        try:
+            body = bytes.fromhex(r[1]).decode("utf-8", "ignore")
+        except ValueError:
+            body = ""
+        if body == text:
+            return int(r[0]) if str(r[0]).lstrip("-").isdigit() else None
+    return None
+
+
+def _send_text_verified(to_wxid, text, key):
+    """可靠发送：切前台唤醒微信 → 广播发送 → 轮询 status 直到 =2(确认送达)。
+    超时仍卡在发送中/失败 → 返回 ok=False,由调用方标 failed 进审核(不假装成功)。
+    ponytail: 校验靠反复拷库(~96MB/次)读 status；发送稀疏可接受,首轮多半即达。
+    上限=拷库 I/O;升级=让 hook 直接回报送达 / 常驻前台。"""
+    _, db, _ = w.account_paths()
+    st = w.copy_db(db, "_stage_verify")
+    try:
+        r = w.sqlcipher_query(
+            st, key, f"SELECT COALESCE(MAX(msgId),0) FROM message WHERE talker='{to_wxid}';")
+        base = int(r[0][0]) if r and r[0] else 0
+    finally:
+        _cleanup_verify()
+    # 微信在后台被 Android 冻结→NetScene 队列不提交,消息卡"发送中"。先切前台唤醒。
+    try:
+        subprocess.run(["am", "start", "-n", "com.tencent.mm/.ui.LauncherUI"],
+                       capture_output=True, timeout=8)
+        time.sleep(3)
+    except Exception:  # noqa: BLE001
+        pass
     sys.path.insert(0, EVENT_BUS)
     from wechat_hook_send import send_text_via_hook
-    return send_text_via_hook(to_wxid, text)
+    res = send_text_via_hook(to_wxid, text)
+    if res.get("status") != "sent":
+        return {"ok": False, "reason": "broadcast失败:" + str(res.get("error", ""))[:40]}
+    # 关键机制:z2.b 只把消息入队(status=1 转圈),WeChat 发送队列要一次 onResume
+    # (后台→前台切换)才真正派发。真机验证:发送后 HOME 切后台再拉回前台，卡住的消息
+    # 全部 flush 到 status=2。（等同"重启微信才发出去",但轻得多。）
+    time.sleep(2)  # 让消息先入库
+    try:
+        subprocess.run(["input", "keyevent", "KEYCODE_HOME"], capture_output=True, timeout=6)
+        time.sleep(1.5)
+        subprocess.run(["am", "start", "-n", "com.tencent.mm/.ui.LauncherUI"],
+                       capture_output=True, timeout=8)
+        time.sleep(2)
+    except Exception:  # noqa: BLE001
+        pass
+    deadline = time.time() + VERIFY_TIMEOUT
+    last = None
+    while time.time() < deadline:
+        time.sleep(VERIFY_POLL)
+        last = _sent_status(db, key, to_wxid, base, text)
+        if last == 2:
+            return {"ok": True, "reason": "verified(status=2)"}
+        if last == 5:
+            return {"ok": False, "reason": "status=5(微信报失败)"}
+    # 超时判定：真机验证 status=1 是"已入队发送中"，微信连接慢时会延迟但最终送达
+    # (曾观察到十几条 status=1 在微信重连后全部 flush 到 2)→ 视为已发,不误报失败。
+    # 仅 status=None(消息压根没创建,hook 没处理/微信没就绪)才算真失败 → 进审核。
+    if last == 1:
+        return {"ok": True, "reason": "queued(status=1,已入队将送达)"}
+    return {"ok": False, "reason": f"消息未创建(status={last})"}
 def process_event(event, stage, key, self_wxid, cfg):
@@ -174,11 +257,12 @@ def process_event(event, stage, key, self_wxid, cfg):
         log(f"skip #{event.get('msgId')} {sender_disp}: {reason}")
         return
     if action == "auto":
-        res = _send_text(event.get("talker"), draft)
-        ok = bool(res.get("ok") or res.get("status") == "sent")
+        res = _send_text_verified(event.get("talker"), draft, key)
+        ok = res["ok"]
         _queue_insert(event, draft, sender_disp, "auto",
                       reason, "auto_sent" if ok else "failed")
-        log(f"auto {'OK' if ok else 'FAIL'} #{event.get('msgId')} {sender_disp} → {draft!r}")
+        log(f"auto {'OK ' + res['reason'] if ok else 'FAILED ' + res['reason']} "
+            f"#{event.get('msgId')} {sender_disp} → {draft!r}")
     else:  # review
         _queue_insert(event, draft, sender_disp, "review", reason, "pending")
         log(f"review #{event.get('msgId')} {sender_disp}: {reason} → 草稿 {draft!r}")
@@ -245,7 +329,7 @@ def _drain_inner():
             and e.get("kind") in CANDIDATE_KINDS]
     if todo:
         _, db, key = w.account_paths()
-        stage = w.copy_db(db)
+        stage = w.copy_db(db, subdir="_stage_orch")   # 独立暂存,避开与 watcher 的 _stage 竞态
         try:
             self_wxid, _ = w.detect_self(stage, key)
             cfg = policy.load_config()
@@ -257,7 +341,7 @@ def _drain_inner():
                 except Exception as ex:  # noqa: BLE001
                     log(f"处理 #{e.get('msgId')} 失败: {ex}")
         finally:
-            for p in __import__("glob").glob(f"{WORK_DIR}/_stage/en.db*"):
+            for p in __import__("glob").glob(f"{WORK_DIR}/_stage_orch/en.db*"):
                 try:
                     os.remove(p)
                 except OSError:
@@ -129,10 +129,12 @@ def sqlcipher_query(db_copy: str, key: str, sql: str):
     return rows
-def copy_db(db: str) -> str:
-    """拷当前库(含 journal)到 Termux 私有 tmp,返回副本路径。root 直接文件拷,不经 /sdcard。"""
-    os.makedirs(f"{WORK_DIR}/_stage", exist_ok=True)
-    dst = f"{WORK_DIR}/_stage/en.db"
+def copy_db(db: str, subdir: str = "_stage") -> str:
+    """拷当前库(含 journal)到独立暂存目录,返回副本路径。root 直接文件拷,不经 /sdcard。
+    subdir 隔离并发方(watcher=_stage / orchestrator=_stage_orch),避免共用文件互相删/覆盖
+    导致 'no such table' 竞态。"""
+    os.makedirs(f"{WORK_DIR}/{subdir}", exist_ok=True)
+    dst = f"{WORK_DIR}/{subdir}/en.db"
     subprocess.run(["cp", db, dst], check=True, timeout=60)
     for ext in ("-journal", "-wal", "-shm"):
         if os.path.exists(db + ext):