宜博 / pixelphone

#!/usr/bin/env python3
"""环节 1→5 编排器 —— 把整条自动回复链串起来

消费 watcher 的 incoming.jsonl（cursor 增量）→ 对候选事件（私聊/群@我/群引用我）：
  env2 查上下文（发送者身份 + 最近历史）→ env3 hermes 生成草稿 →
  env4 policy.decide()（白名单读 wechat-viewer 的 circle.db）→
  env5 落地：auto 直接 send_text_via_hook + 入队(auto_sent) / review 入队(pending)

策略权威参考：wechat-auto-reply skill (v3.0)。代码是参考实现，skill 定义逻辑。
"""
import json
import os
import sqlite3
import subprocess
import sys
import time

_HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, _HERE)
import wechat_reply_watcher as w   # 复用解密/查询/账号推导
import wechat_reply_policy as policy

WORK_DIR = w.WORK_DIR
QUEUE_FILE = f"{WORK_DIR}/incoming.jsonl"
CURSOR_FILE = f"{WORK_DIR}/orch_state.json"
ORCH_LOG = f"{WORK_DIR}/orchestrator.log"
CIRCLE_DB = "/data/data/com.termux/files/home/wechat-viewer/circle.db"
EVENT_BUS = "/data/data/com.termux/files/home/.hermes/event_bus"
HERMES = "/data/data/com.termux/files/usr/bin/hermes"

CANDIDATE_KINDS = {"private", "group_at_me", "group_quote_me"}
HISTORY_N = 12
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):
    line = f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}"
    print(line, flush=True)
    try:
        with open(ORCH_LOG, "a") as f:
            f.write(line + "\n")
    except OSError:
        pass


# ── env2：查上下文 ─────────────────────────────────────
def _display_name(stage, key, wxid):
    if not wxid:
        return ""
    rows = w.sqlcipher_query(
        stage, key,
        f"SELECT conRemark, nickname, alias FROM rcontact WHERE username='{wxid}';")
    if rows and rows[0]:
        r = rows[0] + ["", "", ""]
        return r[0] or r[1] or r[2] or wxid
    return wxid


def _placeholder(mtype, text):
    if mtype == 1:
        return text
    return {3: "[图片]", 34: "[语音]", 43: "[视频]", 47: "[表情]",
            42: "[名片]", 49: "[链接/文件]", 48: "[位置]"}.get(mtype, text[:40] if text else "[消息]")


def build_context(stage, key, event, self_wxid):
    talker = event.get("talker", "")
    is_group = event.get("isGroup") or talker.endswith("@chatroom")
    sender_wxid = event.get("sender") or talker
    sender_disp = _display_name(stage, key, sender_wxid)
    lines = [f"对方：{sender_disp}"]
    if is_group:
        gname = _display_name(stage, key, talker)
        lines.insert(0, f"[群] {gname}")
    # 最近历史
    rows = w.sqlcipher_query(
        stage, key,
        "SELECT isSend, type, createTime, hex(content) FROM message "
        f"WHERE talker='{talker}' ORDER BY msgId DESC LIMIT {HISTORY_N};")
    hist = []
    for r in reversed(rows):
        if len(r) < 4:
            continue
        is_send, mtype, _, chex = r
        try:
            raw = bytes.fromhex(chex).decode("utf-8", "ignore")
        except ValueError:
            raw = ""
        if is_send == "1":
            who, body = "我", raw
        else:
            s, body = w.parse_group_content(raw) if is_group else (sender_wxid, raw)
            who = _display_name(stage, key, s) if is_group and s else sender_disp
        txt = _placeholder(int(mtype) if mtype.isdigit() else 0, body).replace("\n", " ").strip()
        if txt:
            hist.append(f"  {who}: {txt}")
    ctx = "\n".join(lines)
    if hist:
        ctx += "\n最近对话：\n" + "\n".join(hist[-HISTORY_N:])
    ctx += f"\n本条：{sender_disp}: {(event.get('body') or '').strip()[:200]}"
    return ctx, sender_disp


# ── env3：hermes 生成 ─────────────────────────────────
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",
        "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=GEN_TIMEOUT, env=env)
        # 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 "", -1, "timeout", result_file if role == "admin" else ""
    except Exception as e:
        log(f"生成异常: {e}")
        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():
    """返回 {wxid: role} 映射。role: 'admin' | 'member'。"""
    roles = {}
    try:
        con = sqlite3.connect(CIRCLE_DB)
        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 roles


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, 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()


VERIFY_TIMEOUT = 60   # 送达校验总超时(秒)
VERIFY_POLL = 3       # 前台停留期每次查 status 的间隔
DWELL_POLLS = 4       # 一次 onResume 后停留前台查几次(给网络发送时间,不立刻再背景化打断)


def _onresume_cycle():
    """一次真实的 onPause→onResume:HOME 背景化 → 停顿 → 拉回微信前台。
    触发 WeChat NetSceneQueue flush 卡在 status=1 的发送。切换要给足时间——太快
    (背景/前台 <2s)微信来不及走完 onPause/onResume+网络 tick,flush 不了。"""
    try:
        subprocess.run(["input", "keyevent", "KEYCODE_HOME"], capture_output=True, timeout=6)
        time.sleep(2.5)   # 让 onPause 走完、微信真的进后台
        subprocess.run(["am", "start", "-n", "com.tencent.mm/.ui.LauncherUI"],
                       capture_output=True, timeout=8)
        time.sleep(3)     # onResume 后停一下,让网络 scene 开始派发
    except Exception:  # noqa: BLE001
        pass


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 _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 进审核(不假装成功)。
    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
    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 发送队列要一次真实的 app 切换
    # (切走→切回微信 = onPause→onResume)才真正派发。真机确认:发送后等几秒 → 切到别的 app
    # → 切回微信,卡住的消息就 flush 到 status=2。这里:先等消息落库,再做完整切换 cycle,
    # 切回后停在前台多轮查(不立刻再切走,否则会打断正在进行的网络发送)。
    time.sleep(4)  # 让消息先稳稳落库(太快切换时微信还没登记这条,flush 不到它)
    last = None
    deadline = time.time() + VERIFY_TIMEOUT
    while time.time() < deadline:
        _onresume_cycle()                 # 一次真实切走→切回
        for _ in range(DWELL_POLLS):      # 切回后停留前台查几次,给网络发送时间
            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(微信报失败)"}
            if time.time() >= deadline:
                break
            time.sleep(VERIFY_POLL)
    # 超时判定：多轮切换仍卡 status=1 = 微信连接慢/离线,最终多半会 flush(网络恢复时批量
    # 派发)→ 视为已入队不误报失败。仅 status=None(消息没创建,hook 没处理)才算真失败。
    if last == 1:
        return {"ok": True, "reason": "queued(status=1,多轮切换仍未flush,网络恢复后送达)"}
    return {"ok": False, "reason": f"消息未创建(status={last})"}


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)
    # 查 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 上跑过,前缀只影响发送/展示的文本。
    # ponytail: 只是文字 @（看得见）,不带 atuserlist → 未必真 ping 到对方红点;真通知要改
    # SEND_MSG_INTERNAL hook 加 at 列表(见 README 环节5 待收尾)。
    send_text = draft
    if event.get("isGroup") and sender_disp:
        send_text = f"@{sender_disp} {draft}"
    if action == "auto":
        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}")


# ── drain：消费 incoming.jsonl ─────────────────────────
def _read_events():
    evs = []
    try:
        with open(QUEUE_FILE) as f:
            for line in f:
                line = line.strip()
                if line:
                    try:
                        evs.append(json.loads(line))
                    except ValueError:
                        pass
    except OSError:
        pass
    return evs


def drain():
    """非阻塞单实例：已有 drain 在跑就跳过（后续消息的 inotify 会再触发）。"""
    lock = f"{WORK_DIR}/orch.lock"
    try:
        fd = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
        os.close(fd)
    except FileExistsError:
        try:  # 5min 以上视为残留锁，抢占
            if time.time() - os.path.getmtime(lock) > 300:
                os.remove(lock)
                os.close(os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY))
            else:
                return
        except OSError:
            return
    try:
        _drain_inner()
    finally:
        try:
            os.remove(lock)
        except OSError:
            pass


def _drain_inner():
    evs = _read_events()
    if not evs:
        return
    try:
        with open(CURSOR_FILE) as f:
            cursor = json.load(f).get("last_msgId", 0)
    except (OSError, ValueError):
        cursor = None
    max_id = max((int(e.get("msgId", 0)) for e in evs), default=0)
    if cursor is None:  # 首跑：基线，不回放
        with open(CURSOR_FILE, "w") as f:
            json.dump({"last_msgId": max_id}, f)
        w.chown_to_hermes(CURSOR_FILE)
        log(f"编排器首跑基线 last_msgId={max_id}")
        return
    todo = [e for e in evs if int(e.get("msgId", 0)) > cursor
            and e.get("kind") in CANDIDATE_KINDS]
    if todo:
        _, db, key = w.account_paths()
        stage = w.copy_db(db, subdir="_stage_orch")   # 独立暂存,避开与 watcher 的 _stage 竞态
        try:
            self_wxid, _ = w.detect_self(stage, key)
            cfg = policy.load_config()
            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:
                    process_event(e, stage, key, self_wxid, cfg)
                except Exception as ex:  # noqa: BLE001
                    log(f"处理 #{e.get('msgId')} 失败: {ex}")
        finally:
            for p in __import__("glob").glob(f"{WORK_DIR}/_stage_orch/en.db*"):
                try:
                    os.remove(p)
                except OSError:
                    pass
    with open(CURSOR_FILE, "w") as f:
        json.dump({"last_msgId": max_id}, f)
    w.chown_to_hermes(CURSOR_FILE)


def main():
    args = sys.argv[1:]
    if "--drain" in args:
        drain()
    else:
        print("用法: wechat_reply_orchestrator.py --drain")


if __name__ == "__main__":
    main()