fix(wechat-auto-reply): 看门狗单例保护 + 同步设备侧两周迭代

手机后台卡死到连电源键都无响应：wechat_reply_watcher.py 死循环+ wechat-viewer-watchdog.sh(Hermes cron 每 5 分钟跑一次)检测存活只信一次 su -c 'ps -A'|grep，系统过载时该探测本身会漏检，导致在卡死的旧实例上不断叠加新实例，6 天内滚雪球到 877 个同时存活进程，吃满内存 +swap 拖死整机。修复：watchdog 第 5 步改为每次都先 pkill 清场再起一个（与同文件里 Flask 重启逻辑一致的幂等单例模式），不再依赖探测结果是否可靠。已在设备上部署验证。顺带把 wechat_reply_watcher.py / orchestrator.py / policy.py 同步为设备上过去两周的最新版本（repo 里是旧的，device 端已经跑在更新版本上——非 root Termux su fallback、admin/member 双角色回复、去重、独立验证层等），此前只有 daemon.sh 和 service.d 脚本是同步的。 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

fix(wechat-auto-reply): 看门狗单例保护 + 同步设备侧两周迭代
手机后台卡死到连电源键都无响应：wechat_reply_watcher.py 死循环+ wechat-viewer-watchdog.sh(Hermes cron 每 5 分钟跑一次)检测存活只信一次 su -c 'ps -A'|grep，系统过载时该探测本身会漏检，导致在卡死的旧实例上不断叠加新实例，6 天内滚雪球到 877 个同时存活进程，吃满内存 +swap 拖死整机。修复：watchdog 第 5 步改为每次都先 pkill 清场再起一个（与同文件里 Flask 重启逻辑一致的幂等单例模式），不再依赖探测结果是否可靠。已在设备上部署验证。顺带把 wechat_reply_watcher.py / orchestrator.py / policy.py 同步为设备上过去两周的最新版本（repo 里是旧的，device 端已经跑在更新版本上——非 root Termux su fallback、admin/member 双角色回复、去重、独立验证层等），此前只有 daemon.sh 和 service.d 脚本是同步的。 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
yibo
1 parent d1b70de0
Showing 4 changed files with 500 additions and 54 deletions
wechat_plain/scripts/auto-reply/wechat-viewer-watchdog.sh
wechat_plain/scripts/auto-reply/wechat_reply_orchestrator.py
wechat_plain/scripts/auto-reply/wechat_reply_policy.py
wechat_plain/scripts/auto-reply/wechat_reply_watcher.py
+#!/data/data/com.termux/files/usr/bin/bash
+# WeChat Viewer 看门狗 — 检查 Flask + Hermes Gateway + 微信 Watcher 是否存活
+# 由 Hermes cron 每 5 分钟调用一次
+
+export PATH=/data/data/com.termux/files/usr/bin:$PATH
+
+FLASK_PORT=5001
+APP_DIR=/data/data/com.termux/files/home/wechat-viewer
+LOG=$APP_DIR/watchdog.log
+HOME_DIR=/data/data/com.termux/files/home
+
+# ── 1. Flask ──────────────────────────────────────
+
+ALIVE=$(su -c "ss -tlnp" 2>/dev/null | grep -c ":$FLASK_PORT")
+
+if [ "$ALIVE" -eq 0 ]; then
+    echo "[$(date)] Flask 不在线，重启中..." | tee -a "$LOG"
+    cd "$APP_DIR"
+    pkill -f 'python3 app.py' 2>/dev/null
+    sleep 1
+    fuser -k "$FLASK_PORT/tcp" 2>/dev/null
+    nohup python3 app.py > $APP_DIR/flask.log 2>&1 &
+    sleep 3
+    ALIVE2=$(su -c "ss -tlnp" 2>/dev/null | grep -c ":$FLASK_PORT")
+    if [ "$ALIVE2" -gt 0 ]; then
+        echo "[$(date)] Flask 重启成功" | tee -a "$LOG"
+    else
+        echo "[$(date)] Flask 重启失败！" | tee -a "$LOG"
+    fi
+fi
+
+# ── 2. Hermes Gateway ──────────────────────────────
+
+GATEWAY_ALIVE=$(pgrep -f "hermes.*gateway" 2>/dev/null | wc -l)
+
+if [ "$GATEWAY_ALIVE" -eq 0 ]; then
+    echo "[$(date)] Gateway 不在线，重启中..." | tee -a "$LOG"
+    nohup hermes gateway > /tmp/hermes-gateway.log 2>&1 &
+    sleep 3
+    GATEWAY_ALIVE2=$(pgrep -f "hermes.*gateway" 2>/dev/null | wc -l)
+    if [ "$GATEWAY_ALIVE2" -gt 0 ]; then
+        echo "[$(date)] Gateway 重启成功" | tee -a "$LOG"
+    else
+        echo "[$(date)] Gateway 重启失败！" | tee -a "$LOG"
+    fi
+fi
+
+# ── 3. Stuck Job Reaper ──────────────────────────
+# 清理因 Flask 重启而孤儿化的 hermes_jobs
+python3 "$APP_DIR/reap_stuck_jobs.py" >> "$LOG" 2>&1
+
+# ── 4. Orphan Report Reaper ─────────────────────
+# 清理因 Flask 重启丢回调而未入库的日报文件
+python3 "$APP_DIR/reap_orphan_reports.py" >> "$LOG" 2>&1
+
+# ── 5. 微信自动回复 Watcher ────────────────────────
+# 监听微信 DB 变化，写入 incoming.jsonl 队列
+# 单例保护：每次都先 pkill 清场再起一个，不能只信一次 ps|grep 的探测结果——
+# 探测在系统过载时会漏检/超时返回空，若不清场会在卡死的旧实例上再叠一个，
+# 雪球式堆积出几百个进程拖垮整机（2026-07-25 事故：堆到 877 个吃满内存+swap）。
+
+REPLY_DIR=$HOME_DIR/.hermes/wechat_reply
+WATCHER_COUNT=$(su -c 'ps -A' 2>/dev/null | grep -c '[w]echat_reply_watcher')
+
+if [ "$WATCHER_COUNT" -ne 1 ]; then
+    echo "[$(date)] 微信 Watcher 存活数=$WATCHER_COUNT(期望 1)，重置中..." | tee -a "$LOG"
+    su -c "pkill -f wechat_reply_watcher.py" 2>/dev/null
+    sleep 1
+    su -c "/data/data/com.termux/files/usr/bin/python3 $REPLY_DIR/wechat_reply_watcher.py >> $REPLY_DIR/watcher.log 2>&1 &"
+    sleep 2
+    WATCHER_COUNT2=$(su -c 'ps -A' 2>/dev/null | grep -c '[w]echat_reply_watcher')
+    if [ "$WATCHER_COUNT2" -eq 1 ]; then
+        echo "[$(date)] 微信 Watcher 重置成功" | tee -a "$LOG"
+    else
+        echo "[$(date)] 微信 Watcher 重置后存活数异常: $WATCHER_COUNT2" | tee -a "$LOG"
+    fi
+fi
+
+# ── 6. 微信自动回复 Orchestrator Drain ──────────────
+# 消费 incoming.jsonl 队列，处理候选事件
+
+ORCH_LOCK=$REPLY_DIR/orch.lock
+# 清理超过 5 分钟的残留锁
+if [ -f "$ORCH_LOCK" ]; then
+    LOCK_AGE=$(($(date +%s) - $(stat -c %Y "$ORCH_LOCK" 2>/dev/null || echo 0)))
+    if [ "$LOCK_AGE" -gt 300 ]; then
+        rm -f "$ORCH_LOCK"
+    fi
+fi
+
+su -c "/data/data/com.termux/files/usr/bin/python3 $REPLY_DIR/wechat_reply_orchestrator.py --drain >> $REPLY_DIR/orchestrator.log 2>&1"
@@ -2,17 +2,11 @@
 """环节 1→5 编排器 —— 把整条自动回复链串起来
  
 消费 watcher 的 incoming.jsonl（cursor 增量）→ 对候选事件（私聊/群@我/群引用我）：
-  env2 查上下文（发送者身份 + 最近历史）→ env3 hermes -z 生成草稿 →
+  env2 查上下文（发送者身份 + 最近历史）→ env3 hermes 生成草稿 →
   env4 policy.decide()（白名单读 wechat-viewer 的 circle.db）→
-  env5 落地：auto 直接 send_text_via_hook + 入队(auto_sent) / review 入队(pending，在 /approvals 等人工)
+  env5 落地：auto 直接 send_text_via_hook + 入队(auto_sent) / review 入队(pending)
  
-作为 root 在 Termux 跑（同 watcher 上下文）。由 watcher 每次读到新消息后触发 --drain。
-
-安全：policy 默认 auto_send_enabled=False → 一切进 review（不自动发）。生成超时/失败 →
-无草稿 → policy skip → 不发（失败即静默，绝不误发）。
-
-ponytail: 每次 drain 拷一次整库(~96MB)供该批所有事件查上下文；事件稀疏可接受。
-升级：与 watcher 共享一次拷贝；incoming.jsonl 增长需按 offset 游标+轮转。
+策略权威参考：wechat-auto-reply skill (v3.0)。代码是参考实现，skill 定义逻辑。
 """
 import json
 import os
@@ -36,7 +30,30 @@ HERMES = &quot;/data/data/com.termux/files/usr/bin/hermes&quot;
  
 CANDIDATE_KINDS = {"private", "group_at_me", "group_quote_me"}
 HISTORY_N = 12
-GEN_TIMEOUT = 160
+GEN_TIMEOUT = 600   # admin 任务超时（10分钟），GFW 下 GitHub 访问慢
+PROCESSED_FILE = f"{WORK_DIR}/processed_ids.json"
+# 去重：已处理的 msgId 集合。drain 时加载，处理后持久化
+_processed_ids = None
+
+
+def _load_processed():
+    global _processed_ids
+    if _processed_ids is not None:
+        return _processed_ids
+    try:
+        with open(PROCESSED_FILE) as f:
+            _processed_ids = set(json.load(f))
+    except (OSError, ValueError):
+        _processed_ids = set()
+    return _processed_ids
+
+
+def _save_processed():
+    try:
+        with open(PROCESSED_FILE, "w") as f:
+            json.dump(list(_processed_ids or set()), f)
+    except OSError:
+        pass
  
  
 def log(msg):
@@ -108,11 +125,43 @@ def build_context(stage, key, event, self_wxid):
  
  
 # ── env3：hermes 生成 ─────────────────────────────────
-def generate(context_text):
-    prompt = (
-        "【任务】下面是别人发到微信来的一条消息，请以你本人（火星人）的口吻，写一条你会真的发出去的"
-        "微信回复。只输出这条回复本身，不要解释、不要引号、不要用任何工具、不要真的发送。\n\n"
-        "【上下文】\n" + context_text)
+def generate(context_text, event=None, role="member"):
+    """调用 Hermes 生成回复。
+
+    role='admin' → hermes chat（完整 CLI，可调工具/skill/API，执行复杂任务）
+    role='member' → hermes -z（零工具纯文本补全，简单闲聊回复）
+    """
+    talker = (event or {}).get("talker", "")
+    body = (event or {}).get("body", "") or ""
+    is_group = talker.endswith("@chatroom")
+
+    if role == "admin":
+        is_task = _is_task_request(body)
+        result_file = f"{WORK_DIR}/_reply_result_{int(time.time())}.txt"
+        if is_task:
+            prompt = (
+                f"你是火星人的微信助手。用 reply-task-executor skill 执行任务。\n\n"
+                f"任务：{body}\n\n"
+                f"上下文：\n{context_text}\n\n"
+                f"重要：先执行任务（调工具、写文件、调API），不要先写报告。\n"
+                f"任务完成后，把结果报告写入 {result_file}。"
+            )
+        else:
+            prompt = (
+                f"你是火星人的微信助手。有人发了一条消息，分析内容，以火星人口吻回复。\n\n"
+                f"【消息内容】{body}\n\n"
+                f"【上下文】\n{context_text}\n\n"
+                f"把你要发送的回复文本写入文件 {result_file}。只写回复本身，不要解释。"
+            )
+        cmd = [HERMES, "chat", "-q", prompt, "-s", "reply-task-executor", "--yolo", "-t", "terminal,file,web"]
+    else:
+        # 简单模式：零工具，纯粹文本补全
+        prompt = (
+            "【任务】下面是别人发到微信来的一条消息，请以你本人（火星人）的口吻，写一条你会真的发出去的"
+            "微信回复。只输出这条回复本身，不要解释、不要引号、不要用任何工具、不要真的发送。\n\n"
+            "【上下文】\n" + context_text)
+        cmd = [HERMES, "-z", prompt]
+
     env = dict(os.environ)
     env.update({
         "HOME": "/data/data/com.termux/files/home",
@@ -121,40 +170,114 @@ def generate(context_text):
         "LD_LIBRARY_PATH": "/data/data/com.termux/files/usr/lib",
     })
     try:
-        p = subprocess.run([HERMES, "-z", prompt], capture_output=True, text=True,
+        p = subprocess.run(cmd, capture_output=True, text=True,
                            timeout=GEN_TIMEOUT, env=env)
-        out = (p.stdout or "").strip()
-        # 取最后一段非空文本（躲开可能的日志行）
-        return out.split("\n")[-1].strip() if out else ""
+        # admin 模式从文件读结果，不用 stdout
+        if role == "admin" and result_file:
+            try:
+                with open(result_file) as f:
+                    draft = f.read().strip()
+                try: os.remove(result_file)
+                except: pass
+            except OSError:
+                draft = ""
+        else:
+            # member 模式：从 stdout 取最后一行
+            out = (p.stdout or "").strip()
+            draft = out.split("\n")[-1].strip() if out else ""
+        stderr = (p.stderr or "")[:200]
+        return draft, p.returncode, stderr, result_file if role == "admin" else ""
     except subprocess.TimeoutExpired:
         log("生成超时")
-        return ""
-    except Exception as e:  # noqa: BLE001
+        return "", -1, "timeout", result_file if role == "admin" else ""
+    except Exception as e:
         log(f"生成异常: {e}")
-        return ""
-
-
-# ── env4→5：落地 ─────────────────────────────────────
+        return "", -1, str(e)[:200], result_file if role == "admin" else ""
+
+
+# ── V 层：独立验证 ──────────────────────────────────
+def _verify_draft(draft, role="member"):
+    """独立验证回复质量。不能由 agent 自己宣布完成。
+
+    admin 生成的回复需要更严格检查（因为可能调了错误工具/产生幻觉）。
+    member 的回复比较简单，主要防胡话。
+    返回 (ok: bool, reason: str)
+    """
+    if not draft or not draft.strip():
+        return False, "empty_draft"
+    text = draft.strip()
+    # 噪音行——hermes 输出里的元数据/分隔线
+    if all(c in "─═━═─ \t" for c in text):
+        return False, "noise_line"
+    for noise_start in ["Messages:", "Session:", "Duration:", "Tool:",
+                         "Exit code:", "📱", "Model:", "Provider:",
+                         "Tokens:", "Cost:", "$", "hermes ",
+                         "Resume this", "Session saved"]:
+        if text.startswith(noise_start):
+            return False, f"noise:{noise_start}"
+    # 错误标记——hermes 调用失败时可能输出的内容
+    for marker in ["[Hermes", "[调用失败", "[超时", "[错误", "[Error",
+                   "抱歉，我无法", "I cannot", "I'm unable"]:
+        if marker in text:
+            return False, f"error_marker:{marker}"
+    # 过短——不太可能是有效回复（admin 尤甚，它应该产出实质性内容）
+    if role == "admin" and len(text) < 5:
+        return False, "admin_too_short"
+    if len(text) < 2:
+        return False, "too_short"
+    # admin 回复不应该只是 emoji/单字（说明 hermes 没认真执行）
+    if role == "admin" and len(text) < 10 and not any('\u4e00' <= c <= '\u9fff' for c in text):
+        return False, "admin_no_chinese"
+    return True, "ok"
+
+
+# ── L 层：Task Contract ──────────────────────────────
+def _write_contract(event, role, exit_code, output, err):
+    """admin 的 hermes chat 失败时写 task contract——包含复现所需全部信息。"""
+    import __main__
+    contract_dir = os.path.join(WORK_DIR, "_q")
+    os.makedirs(contract_dir, exist_ok=True)
+    contract = {
+        "ts": time.time(),
+        "msgId": event.get("msgId"),
+        "talker": event.get("talker"),
+        "sender": event.get("sender"),
+        "body": (event.get("body") or "")[:300],
+        "role": role,
+        "exit_code": exit_code,
+        "output": (output or "")[:500],
+        "stderr": (err or "")[:200],
+    }
+    fname = os.path.join(contract_dir, f"ct_{int(time.time())}_{event.get('msgId')}.json")
+    try:
+        with open(fname, "w") as f:
+            json.dump(contract, f, ensure_ascii=False, indent=2)
+        w.chown_to_hermes(fname)
+        log(f"contract written: {fname}")
+    except Exception as e:  # noqa: BLE001
+        log(f"contract write failed: {e}")
 def _load_whitelist():
-    contacts, groups = [], []
+    """返回 {wxid: role} 映射。role: 'admin' | 'member'。"""
+    roles = {}
     try:
         con = sqlite3.connect(CIRCLE_DB)
-        for kind, wxid in con.execute("SELECT kind, wxid FROM reply_whitelist"):
-            (groups if kind == "group" else contacts).append(wxid)
+        for wxid, role in con.execute("SELECT wxid, COALESCE(role,'member') FROM reply_whitelist"):
+            roles[wxid] = role or "member"
         con.close()
     except Exception:
         pass
-    return contacts, groups
+    return roles
  
  
-def _queue_insert(event, draft, sender_disp, action, reason, status):
+def _queue_insert(event, draft, sender_disp, action, reason, status, result_file="", override_talker=""):
+    talker = override_talker or event.get("talker")
     con = sqlite3.connect(CIRCLE_DB)
     con.execute(
         """INSERT INTO reply_queue
-           (msg_id, talker, sender, sender_name, kind, incoming, draft, action, reason, status, sent_at)
-           VALUES (?,?,?,?,?,?,?,?,?,?, CASE WHEN ?='auto_sent' THEN datetime('now','localtime') ELSE NULL END)""",
-        (event.get("msgId"), event.get("talker"), event.get("sender"), sender_disp,
-         event.get("kind"), event.get("body"), draft, action, reason, status, status))
+           (msg_id, talker, sender, sender_name, kind, incoming, draft, result_file, action, reason, status, sent_at)
+           VALUES (?,?,?,?,?,?,?,?,?,?,?, CASE WHEN ?='auto_sent' THEN datetime('now','localtime') ELSE NULL END)""",
+        (event.get("msgId"), talker, event.get("sender"), sender_disp,
+         event.get("kind"), event.get("body"), draft, result_file, action, reason, status, status))
     con.commit()
     con.close()
  
@@ -209,6 +332,73 @@ def _sent_status(db, key, talker, base_msgid, text):
     return None
  
  
+def _run_hermes_z(prompt):
+    """快速 hermes -z 调用（零工具，纯文本补全），用于生成失败报告"""
+    cmd = [HERMES, "-z", prompt]
+    env = dict(os.environ)
+    env.update({
+        "HOME": "/data/data/com.termux/files/home",
+        "PREFIX": "/data/data/com.termux/files/usr",
+        "PATH": "/data/data/com.termux/files/usr/bin:/system/bin",
+        "LD_LIBRARY_PATH": "/data/data/com.termux/files/usr/lib",
+    })
+    try:
+        p = subprocess.run(cmd, capture_output=True, text=True,
+                           timeout=60, env=env)
+        return (p.stdout or "").strip(), p.returncode, (p.stderr or "")[:200]
+    except subprocess.TimeoutExpired:
+        return "", -1, "timeout"
+    except Exception as e:
+        return "", -1, str(e)[:200]
+
+
+# ── 两段式回复：任务识别 + ack 生成 ──────────────
+
+TASK_KEYWORDS = ["研究", "学习", "查一下", "分析", "帮我", "看看",
+                  "安装", "配置", "搭建", "写一个", "改一下",
+                  "怎么", "什么是", "解释", "总结", "调查",
+                  "research", "install", "setup", "analyze"]
+
+def _is_task_request(body):
+    """判断是否是需要后台执行的复杂任务（vs 简单闲聊）"""
+    text = body.lower()
+    score = sum(1 for kw in TASK_KEYWORDS if kw in text)
+    return score >= 1
+
+def _gen_ack(ctx, body, sender_disp):
+    """生成确认收到消息，根据任务内容个性化"""
+    if "研究" in body or "学习" in body or "看看" in body:
+        return "收到，我来研究，搞定了给你报告"
+    if "安装" in body or "配置" in body or "搭建" in body:
+        return "好的，我来处理，完了给你结果"
+    if "查" in body or "分析" in body:
+        return "收到，我查一下，稍等给你报告"
+    if "写" in body or "改" in body or "帮" in body:
+        return "好的，我来弄，弄好给你报告"
+    return "收到，我来处理，有结果跟你说"
+
+
+def _is_complete_report(draft):
+    """检查 admin 任务报告是否完整（不只是简短回复）"""
+    if not draft:
+        return False
+    text = draft.strip()
+    # 太短 → 不完整
+    if len(text) < 80:
+        return False
+    # 只有一句话没有结构 → 不完整
+    lines = [l for l in text.split("\n") if l.strip()]
+    if len(lines) < 2:
+        return False
+    # 至少含一项具体信息（数字/URL/skill名/文件路径/工具名）
+    has_specific = any(
+        kw in text
+        for kw in ["http", "github.com", "skill", "memory", "创建",
+                   "安装", "pip", "npm", "实现了", "测试", "验证",
+                   "Star", "star", "⭐"]
+    )
+    return has_specific
+
 def _send_text_verified(to_wxid, text, key):
     """可靠发送：切前台唤醒微信 → 广播发送 → 轮询 status 直到 =2(确认送达)。
     超时仍卡在发送中/失败 → 返回 ok=False,由调用方标 failed 进审核(不假装成功)。
@@ -260,12 +450,89 @@ def _send_text_verified(to_wxid, text, key):
  
  
 def process_event(event, stage, key, self_wxid, cfg):
+    # ═══ Step 0: 去重 ═══
+    msg_id = event.get("msgId")
+    processed = _load_processed()
+    if msg_id in processed:
+        return  # 已处理过，跳过
+
     ctx, sender_disp = build_context(stage, key, event, self_wxid)
-    draft = generate(ctx)
-    d = policy.decide(event, draft, self_wxid=self_wxid, cfg=cfg)
+    # 查 sender 的 role：admin → 完整 Hermes CLI，member → 简单 hermes -z
+    sender_wxid = event.get("sender") or event.get("talker") or ""
+    role = cfg.get("whitelist_roles", {}).get(sender_wxid, "member")
+
+    body = (event.get("body") or "").strip()
+    talker = event.get("talker", "")
+
+    # ═══ 两段式回复：admin 复杂任务先回 ack ═══
+    if role == "admin" and _is_task_request(body):
+        ack = _gen_ack(ctx, body, sender_disp)
+        ack_text = f"@{sender_disp} {ack}" if event.get("isGroup") and sender_disp else ack
+        _send_text_verified(talker, ack_text, key)
+        _queue_insert(event, ack_text, sender_disp, "auto", "ack", "auto_sent")
+        log(f"ack #{event.get('msgId')} {sender_disp}: {ack!r}")
+
+    # ── 生成回复（V 层：独立验证 + 回退）──
+    draft, exit_code, stderr, result_file = generate(ctx, event=event, role=role)
+    ok, vreason = _verify_draft(draft, role)
+
+    # ═══ 报告完整性检查（admin 任务必须产出完整报告） ═══
+    if ok and role == "admin" and _is_task_request(body):
+        if not _is_complete_report(draft):
+            ok = False
+            vreason = f"incomplete_report(len={len(draft)})"
+
+    if not ok:
+        log(f"V-verify FAIL #{event.get('msgId')} role={role}: {vreason} draft={draft!r}")
+        _write_contract(event, role, exit_code, draft, stderr)
+
+        if role == "admin":
+            is_task = _is_task_request(body)
+            if is_task:
+                # admin 任务失败 → 调 reply-failure-report skill 生成报告，写入队列
+                fail_prompt = (
+                    f"任务内容：{body}\\n"
+                    f"失败原因：{vreason}\\n"
+                    f"对方：{sender_disp}\\n\\n"
+                    f"按 reply-failure-report skill 生成一条要发给对方的失败报告。只输出消息本身。"
+                )
+                fail_draft, _, _ = _run_hermes_z(fail_prompt)
+                fail_report = (fail_draft or "").strip()
+                if not fail_report or len(fail_report) < 5:
+                    fail_report = f"这个任务没跑成（{vreason}），我加到 to-do list 了。"
+                fail_text = f"@{sender_disp} {fail_report}" if event.get("isGroup") and sender_disp else fail_report
+                # 写入队列，cron 会发送
+                _queue_insert(event, fail_text, sender_disp, "auto",
+                              f"task_failed:{vreason}", "failed")
+                log(f"task failed queued #{event.get('msgId')}: {vreason}")
+                if msg_id:
+                    loaded = _load_processed()
+                    loaded.add(msg_id)
+                    _save_processed()
+                return
+            # admin 闲聊失败 → 回退到 member 模式
+            log(f"admin fallback → member mode for #{event.get('msgId')}")
+            draft, exit_code, stderr, _ = generate(ctx, event=event, role="member")
+            ok, vreason = _verify_draft(draft, "member")
+            if not ok:
+                log(f"V-verify FAIL (member fallback) #{event.get('msgId')}: {vreason}")
+                _queue_insert(event, draft or "(验证失败)", sender_disp, "auto",
+                              f"verify_failed:{vreason}", "failed")
+                return
+        else:
+            _queue_insert(event, draft or "(验证失败)", sender_disp, "auto",
+                          f"verify_failed:{vreason}", "failed")
+            return
+
+    d = policy.decide(event, draft, self_wxid=self_wxid, cfg=cfg, sender_role=role)
     action, reason = d["action"], d["reason"]
     if action == "skip":
         log(f"skip #{event.get('msgId')} {sender_disp}: {reason}")
+        # ═══ 去重持久化（skip 也记录） ═══
+        if msg_id:
+            loaded = _load_processed()
+            loaded.add(msg_id)
+            _save_processed()
         return
     # 群里回复自动 @ 触发者：昵称前缀 + U+2005（微信原生 @ 分隔符）。安全判定(decide)已在
     # 原始 draft 上跑过,前缀只影响发送/展示的文本。
@@ -275,12 +542,28 @@ def process_event(event, stage, key, self_wxid, cfg):
     if event.get("isGroup") and sender_disp:
         send_text = f"@{sender_disp} {draft}"
     if action == "auto":
-        res = _send_text_verified(event.get("talker"), send_text, key)
-        ok = res["ok"]
-        _queue_insert(event, send_text, sender_disp, "auto",
-                      reason, "auto_sent" if ok else "failed")
-        log(f"auto {'OK ' + res['reason'] if ok else 'FAILED ' + res['reason']} "
-            f"#{event.get('msgId')} {sender_disp} → {send_text!r}")
+        is_task = role == "admin" and _is_task_request(body)
+        if is_task:
+            # admin 任务结果 → 私聊发 sender，直接发送
+            task_talker = sender_wxid if event.get("isGroup") else talker
+            res = _send_text_verified(task_talker, send_text, key)
+            ok_send = res["ok"]
+            _queue_insert(event, send_text, sender_disp, "auto",
+                          reason, "auto_sent" if ok_send else "failed",
+                          result_file=result_file, override_talker=task_talker)
+            log(f"task result sent #{event.get('msgId')} → {task_talker}: {'OK' if ok_send else 'FAILED'}")
+        else:
+            # 闲聊 / member → 直接发送
+            res = _send_text_verified(event.get("talker"), send_text, key)
+            ok_send = res["ok"]
+            _queue_insert(event, send_text, sender_disp, "auto",
+                          reason, "auto_sent" if ok_send else "failed")
+            log(f"auto {'OK ' + res['reason'] if ok_send else 'FAILED ' + res['reason']} "
+                f"#{event.get('msgId')} {sender_disp} → {send_text!r}")
+        if msg_id:
+            loaded = _load_processed()
+            loaded.add(msg_id)
+            _save_processed()
     else:  # review
         _queue_insert(event, send_text, sender_disp, "review", reason, "pending")
         log(f"review #{event.get('msgId')} {sender_disp}: {reason} → 草稿 {send_text!r}")
@@ -351,7 +634,10 @@ def _drain_inner():
         try:
             self_wxid, _ = w.detect_self(stage, key)
             cfg = policy.load_config()
-            cfg["whitelist_contacts"], cfg["whitelist_groups"] = _load_whitelist()
+            cfg["whitelist_roles"] = _load_whitelist()
+            # backward compat: policy.decide still checks these lists for whitelist membership
+            cfg["whitelist_contacts"] = [w for w, r in cfg["whitelist_roles"].items() if not w.endswith("@chatroom")]
+            cfg["whitelist_groups"] = [w for w, r in cfg["whitelist_roles"].items() if w.endswith("@chatroom")]
             log(f"drain：{len(todo)} 条候选（auto_send={cfg['auto_send_enabled']}）")
             for e in todo:
                 try:
@@ -64,8 +64,12 @@ def _hit_keyword(text, keywords):
     return None
  
  
-def decide(event, draft, confidence=None, should_reply=True, self_wxid="", cfg=None):
-    """核心判定。返回 {action, reason, hit?}。纯函数、无副作用（写队列在 route()）。"""
+def decide(event, draft, confidence=None, should_reply=True, self_wxid="", cfg=None, sender_role="member"):
+    """核心判定。返回 {action, reason, hit?}。纯函数、无副作用（写队列在 route()）。
+
+    sender_role='admin' → 允许执行命令/任务类消息（跳过部分敏感词、自动发送）
+    sender_role='member' → 默认保守策略
+    """
     cfg = cfg or load_config()
     kind = event.get("kind", "")
     sender = event.get("sender") or event.get("talker") or ""
@@ -89,15 +93,19 @@ def decide(event, draft, confidence=None, should_reply=True, self_wxid=&quot;&quot;, cfg=N
             return {"action": "skip", "reason": f"nontext:非文本(type={mtype})且无草稿"}
         return {"action": "skip", "reason": "empty:草稿为空"}
  
-    # ── 硬安全override：命中敏感词一律 review（白名单也拉回）──
+    # ── 硬安全override：命中敏感词一律 review（admin 豁免——可执行命令/任务）──
     hit = _hit_keyword(body + "\n" + (draft or ""), cfg["block_keywords"])
-    if hit:
+    if hit and sender_role != "admin":
         return {"action": "review", "reason": f"sensitive:命中敏感词「{hit}」", "hit": hit}
  
     # ── 低置信 → review ──
     if confidence is not None and confidence < cfg["min_confidence"]:
         return {"action": "review", "reason": f"lowconf:置信{confidence}<{cfg['min_confidence']}"}
  
+    # ── admin + 总开关 → 直接 auto（信任，无需白名单匹配）──
+    if sender_role == "admin" and cfg["auto_send_enabled"]:
+        return {"action": "auto", "reason": "admin:管理员信任→直接发"}
+
     # ── 白名单 + 总开关 → auto ──
     is_group = event.get("isGroup") or str(talker).endswith("@chatroom")
     whitelisted = (
@@ -73,6 +73,7 @@ def md5hex(s: str) -&gt; str:
  
 # ── 账号 / key 推导 ────────────────────────────────────────────────────────
 def read_uin() -> str:
+    # 先尝试直接 glob（root 环境下有效）
     for xml in glob.glob(f"{MM_BASE}/shared_prefs/*.xml"):
         try:
             with open(xml, "r", errors="ignore") as f:
@@ -81,6 +82,25 @@ def read_uin() -&gt; str:
                     return m.group(1)
         except OSError:
             continue
+
+    # fallback: su -c 读取（非root Termux 环境）
+    try:
+        result = subprocess.run(
+            ["su", "-c", f"grep -rl last_login_uin {MM_BASE}/shared_prefs/*.xml"],
+            capture_output=True, text=True, timeout=10
+        )
+        if result.returncode == 0 and result.stdout.strip():
+            xml_file = result.stdout.strip().split('\n')[0]
+            result2 = subprocess.run(
+                ["su", "-c", f"cat {xml_file}"],
+                capture_output=True, text=True, timeout=10
+            )
+            m = re.search(r'last_login_uin">(-?\d+)', result2.stdout)
+            if m:
+                return m.group(1)
+    except Exception:
+        pass
+
     raise RuntimeError("没读到微信 last_login_uin(未登录?没 root?)")
  
  
@@ -91,8 +111,28 @@ def account_paths():
     if not os.path.exists(db):                # 兜底:定位不到就取任一个
         cands = glob.glob(f"{MM_BASE}/MicroMsg/*/EnMicroMsg.db")
         if not cands:
-            raise RuntimeError("找不到 EnMicroMsg.db")
-        db = cands[0]
+            # fallback: su -c 找 db（非root Termux 环境）
+            try:
+                result = subprocess.run(
+                    ["su", "-c", f"ls {db}"],
+                    capture_output=True, text=True, timeout=10
+                )
+                if result.returncode == 0 and result.stdout.strip():
+                    return uin, result.stdout.strip(), md5hex(IMEI + uin)[:7]
+                # 兜底:用 su 找任一个 EnMicroMsg.db
+                result = subprocess.run(
+                    ["su", "-c", f"ls {MM_BASE}/MicroMsg/*/EnMicroMsg.db"],
+                    capture_output=True, text=True, timeout=10
+                )
+                if result.returncode == 0 and result.stdout.strip():
+                    cands = result.stdout.strip().split('\n')
+                    db = cands[0]
+                else:
+                    raise RuntimeError("找不到 EnMicroMsg.db")
+            except RuntimeError:
+                raise
+            except Exception as e:
+                raise RuntimeError(f"找不到 EnMicroMsg.db: {e}")
     key = md5hex(IMEI + uin)[:7]
     return uin, db, key
  
@@ -132,13 +172,27 @@ def sqlcipher_query(db_copy: str, key: str, sql: str):
 def copy_db(db: str, subdir: str = "_stage") -> str:
     """拷当前库(含 journal)到独立暂存目录,返回副本路径。root 直接文件拷,不经 /sdcard。
     subdir 隔离并发方(watcher=_stage / orchestrator=_stage_orch),避免共用文件互相删/覆盖
-    导致 'no such table' 竞态。"""
+    导致 'no such table' 竞态。非root Termux 环境用 su -c cp fallback + chown。"""
     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)
+    # 先尝试普通 cp（root 环境下有效）
+    p = subprocess.run(["cp", db, dst], capture_output=True, timeout=60)
+    if p.returncode != 0:
+        # fallback: su -c cp（非root Termux 环境）
+        subprocess.run(["su", "-c", f"cp {db} {dst}"], check=True, timeout=60)
+        # su cp 的文件属主是 root，需要 chown 给 hermes 用户
+        chown_to_hermes(dst)
     for ext in ("-journal", "-wal", "-shm"):
-        if os.path.exists(db + ext):
-            subprocess.run(["cp", db + ext, dst + ext], timeout=60)
+        src_ext = db + ext
+        dst_ext = dst + ext
+        if os.path.exists(src_ext):
+            subprocess.run(["cp", src_ext, dst_ext], timeout=60)
+        else:
+            # 非root 看不到 src_ext 时，用 su 探测并 cp
+            pe = subprocess.run(["su", "-c", f"test -f {src_ext} && cp {src_ext} {dst_ext}"],
+                               capture_output=True, timeout=60)
+            if pe.returncode == 0 and os.path.exists(dst_ext):
+                chown_to_hermes(dst_ext)
     return dst
  
  
@@ -193,7 +247,14 @@ def chown_to_hermes(path):
         try:
             os.chown(path, o[0], o[1])
         except OSError:
-            pass
+            # 非root 环境下 os.chown 无权限，fallback 到 su -c chown
+            try:
+                subprocess.run(
+                    ["su", "-c", f"chown {o[0]}:{o[1]} {path}"],
+                    capture_output=True, timeout=10
+                )
+            except Exception:
+                pass
  
  
 def save_state(state):