app.py 179 KB

Edit Raw Blame History

"""WeChat Viewer Flask 主应用"""

import subprocess
import json
import time
import os
import threading
from datetime import datetime, timedelta
from pathlib import Path
from flask import Flask, jsonify, request, render_template, send_file, redirect, url_for, Response
import config
import wechat_db
import local_db


def _su_exists(path: str) -> bool:
    try:
        r = subprocess.run(
            ["su", "-c", f"test -f {path} && echo yes || echo no"],
            capture_output=True, text=True, timeout=5
        )
        return "yes" in r.stdout
    except Exception:
        return False


def _safe_exists(path: str) -> bool:
    try:
        if Path(path).exists():
            return True
    except PermissionError:
        return _su_exists(path)
    # WeChat data dir: Path.exists() returns False (not PermissionError) on Android
    # when the process lacks read permission on parent directories.
    if path.startswith("/data/data/") and _su_exists(path):
        return True
    return False

import re
import hashlib
import html as html_module

app = Flask(__name__)
app.jinja_env.filters['from_json'] = lambda s: json.loads(s) if s else {}
local_db.init_db()


# ── 统一 Hermes 子进程启动 ──────────────────────────

HERMES_JOBS_DIR = Path("/data/data/com.termux/files/home/tmp/hermes_jobs")


def _spawn_hermes_job(job_type: str, job_name: str, cmd: list[str],
                       timeout: int = 300, on_done=None, task_id: str = "", env: dict = None) -> str:
    """启动 Hermes CLI 子进程并追踪。
    返回 task_id（12 位 hex）。完成后调用 on_done(task_id, success)。
    如果传入 task_id 则使用，否则自动生成。
    
    全局并发控制：使用 _hermes_semaphore（上限 3）限制同时运行的 hermes chat 进程数，
    避免 Pixel 内存溢出导致 Termux 崩溃。
    """
    # 防御：所有 hermes 子进程必须指定 skill
    if "-s" not in cmd and "--skills" not in cmd:
        raise ValueError(f"_spawn_hermes_job({job_type}): 缺少 -s/--skills，拒绝执行。所有 hermes 子进程必须指定 skill。")
    import uuid, threading
    if not hasattr(_spawn_hermes_job, "_hermes_semaphore"):
        _spawn_hermes_job._hermes_semaphore = threading.Semaphore(3)
    herm_sem = _spawn_hermes_job._hermes_semaphore
    if not task_id:
        task_id = uuid.uuid4().hex[:12]
    HERMES_JOBS_DIR.mkdir(parents=True, exist_ok=True)
    log_file = str(HERMES_JOBS_DIR / f"{task_id}.log")
    local_db.job_add(task_id, job_type, job_name, log_file)

    # ── Harness Trace: 结构化错误码 ──
    # 对齐 Harness Trace 四步诊断（Model Input / Tool Selection / Runtime / Context Assembly）
    # 每个失败原因映射到 reason_code，方便离线归因而非盲目重试
    # HydraMind Gate trace quad: [node=spawn,attempt=1,gate=<reason_code>,action=BLOCK]
    # reason codes 统一在 VerifyReason 类中定义

    def _run():
        success = False
        reason_code = None
        # ── 预算闸门（最上游：不占 semaphore，不写完整 log）──
        # Inspired by Paperclip: 预算硬上限，达上限自动停止
        budget_allowed, budget_rc = _check_budget(job_type)
        if not budget_allowed:
            reason_code = budget_rc  # BUDGET_EXHAUSTED
            try:
                with open(log_file, "w") as lf:
                    month_key = datetime.now().strftime("%Y-%m")
                    lf.write(f"=== Hermes Job BLOCKED: budget exhausted ===\n"
                             f"=== Task ID: {task_id} ===\n"
                             f"=== Type: {job_type} - {job_name} ===\n"
                             f"=== Month: {month_key} ===\n"
                             f"=== Gate: [node=spawn,attempt=1,gate={reason_code},"
                             f"action={GateAction.BLOCK}] ===\n")
            except:
                pass
        else:
            # ── 全局并发控制：获取 semaphore 后才能启动 hermes chat ──
            acquired = herm_sem.acquire(blocking=False)
            if not acquired:
                # 不能阻塞等待 → 快速失败，让 cron 下一轮重试
                reason_code = "CONCURRENCY_LIMIT"
                try:
                    with open(log_file, "w") as lf:
                        lf.write(f"=== Hermes Job SKIPPED: concurrency limit reached ===\n"
                                 f"=== Task ID: {task_id} ===\n"
                                 f"=== Type: {job_type} - {job_name} ===\n")
                except:
                    pass
            if acquired:
                try:
                    with open(log_file, "w") as lf:
                        lf.write(f"=== Hermes Job Started: {job_type} - {job_name} ===\n")
                        lf.write(f"=== Task ID: {task_id} ===\n")
                        lf.write(f"=== Command: {' '.join(cmd)} ===\n")
                        lf.write(f"=== Timeout: {timeout}s ===\n")
                        lf.write(f"=== Started at: {time.strftime('%Y-%m-%d %H:%M:%S')} ===\n\n")
                        lf.flush()
                        # merge provided env on top of current process env so PATH/HOME survive
                        merged_env = os.environ.copy()
                        if env:
                            merged_env.update(env)
                        proc = subprocess.Popen(cmd, stdout=lf,
                            stderr=subprocess.STDOUT, text=True,
                            cwd=str(Path(__file__).parent), env=merged_env)
                        try:
                            proc.wait(timeout=timeout)
                            if proc.returncode == 0:
                                success = True
                                # ── HydraMind Gate: post-execution content thin check ──
                                try:
                                    log_size = Path(log_file).stat().st_size
                                    if log_size < 200:
                                        lf.write(
                                            f"\n=== [node=spawn,attempt=1,gate={VerifyReason.CONTENT_THIN},"
                                            f"action=CONTINUE] "
                                            f"log file only {log_size} bytes, job may have "
                                            f"produced no meaningful output ===\n")
                                except Exception:
                                    pass
                            else:
                                reason_code = VerifyReason.SPAWN_NONZERO
                            lf.write(f"\n=== [node=spawn,attempt=1,gate={reason_code},action=BLOCK] "
                                     f"Exit code: {proc.returncode} ===\n")
                        except subprocess.TimeoutExpired:
                            proc.kill()
                            proc.wait()
                            reason_code = VerifyReason.SPAWN_TIMEOUT
                            lf.write(f"\n=== [node=spawn,attempt=1,gate={reason_code},action=BLOCK] "
                                     f"job killed after {timeout}s timeout ===\n")
                        lf.write(f"=== Finished at: {time.strftime('%Y-%m-%d %H:%M:%S')} ===\n")
                except Exception as e:
                    reason_code = VerifyReason.SPAWN_FAILED
                    try:
                        with open(log_file, "a") as lf:
                            lf.write(f"\n=== [node=spawn,attempt=1,gate={reason_code},action=BLOCK] {e} ===\n")
                    except Exception:
                        pass
                finally:
                    herm_sem.release()
        status = "completed" if success else "failed"
        local_db.job_update_status(task_id, status)
        if on_done:
            try:
                on_done(task_id, success, reason_code)
            except TypeError:
                # Backward-compat: old on_done with 2 params
                try:
                    on_done(task_id, success)
                except Exception as e2:
                    try:
                        with open(log_file, "a") as lf:
                            lf.write(f"\n=== [node=spawn,attempt=1,gate={VerifyReason.SPAWN_ON_DONE_ERR},action=BLOCK] "
                                     f"on_done callback failed: {e2} ===\n")
                    except Exception:
                        pass
            except Exception as e:
                # Harness Trace: callback 失败也记录，不静默吞
                try:
                    with open(log_file, "a") as lf:
                        lf.write(f"\n=== [node=spawn,attempt=1,gate={VerifyReason.SPAWN_ON_DONE_ERR},action=BLOCK] "
                                 f"on_done callback failed: {e} ===\n")
                except Exception:
                    pass

    threading.Thread(target=_run, daemon=True).start()
    return task_id


def _extract_dedup_key(url: str) -> str:
    if not url:
        return ""
    if "mp.weixin.qq.com" in url:
        biz = re.search(r"__biz=([^&]+)", url)
        mid = re.search(r"mid=(\d+)", url)
        idx = re.search(r"idx=(\d+)", url)
        sn = re.search(r"sn=([a-f0-9]+)", url)
        if biz and mid and idx and sn:
            return f"{biz.group(1)}|{mid.group(1)}|{idx.group(1)}|{sn.group(1)}"
    return url


# ── Page: 文章列表 ──────────────────────────────────

@app.route("/articles")
def articles_page():
    return render_template("articles.html")


@app.route("/api/articles")
def api_articles():
    try:
        page = max(1, int(request.args.get("page", "1")))
        per_page = max(1, min(500, int(request.args.get("per_page", "20"))))
        date_from = request.args.get("date_from", "")
        date_to = request.args.get("date_to", "")
        total, articles = local_db.article_list_new(page, per_page, date_from, date_to)
        return jsonify({"ok": True, "total": total, "page": page,
            "per_page": per_page, "data": articles})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/articles/extract", methods=["POST"])
def api_article_extract():
    """Extract articles from WeChat DB into articles + article_shares (two-table schema)."""
    try:
        sql = """SELECT m.msgId, m.talker, m.content, m.createTime
        FROM message m WHERE m.type = 49 ORDER BY m.createTime DESC LIMIT 1000"""
        rows = wechat_db.query(sql)
        new_articles, new_shares, total = 0, 0, 0
        for row in rows:
            content = row.get("content", "") or ""
            url_match = re.search(r"<url>(.*?)</url>", content)
            if not url_match:
                continue
            url = url_match.group(1)
            url = re.sub(r"<!\[CDATA\[(.*?)\]\]>", r"\1", url, flags=re.DOTALL)
            url = html_module.unescape(url)
            if not url or url.startswith("https://support.weixin.qq.com"):
                continue
            total += 1
            title_match = re.search(r"<title>(.*?)</title>", content)
            title = title_match.group(1) if title_match else ""
            title = html_module.unescape(re.sub(r"<!\[CDATA\[(.*?)\]\]>", r"\1", title, flags=re.DOTALL))
            # Extract actual sender from content prefix (group msgs: "sender:\n<xml>")
            raw_talker = row.get("talker", "")
            sender_match = re.match(r"^([^:\n<]+):\n", content)
            actual_sender = sender_match.group(1) if sender_match else raw_talker
            chatroom_id = raw_talker if "@chatroom" in raw_talker else ""
            msg_create_time = row.get("createTime", 0)
            source_msg_id = row.get("msgId", 0)
            # Upsert article (dedup by URL → biz|mid|idx|sn)
            was_new = local_db.article_upsert(url, title, return_was_new=True)
            article_id = was_new[0] if isinstance(was_new, tuple) else was_new
            if isinstance(was_new, tuple) and was_new[1]:
                new_articles += 1
            # Record share (dedup by source_msg_id)
            if not local_db.share_exists(source_msg_id):
                local_db.article_share_add(article_id, chatroom_id or raw_talker,
                    actual_sender, source_msg_id, msg_create_time)
                new_shares += 1
        return jsonify({"ok": True, "data": {"new_articles": new_articles, "new_shares": new_shares, "total": total}})
    except Exception as e:
        import traceback
        traceback.print_exc()
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/articles/refresh", methods=["POST"])
def api_article_refresh():
    """Manual refresh: run scan_articles.py via Hermes CLI to extract new articles."""
    import uuid
    task_id = uuid.uuid4().hex[:12]
    scan_script = "/data/data/com.termux/files/home/.hermes/skills/wechat-group-digest/scripts/scan_articles.py"
    prompt = f"执行 python3 {scan_script} --extract-only。只做提取不进摘要。完成后报告结果。"
    _spawn_hermes_job(
        "article_refresh", "手动刷新文章列表",
        ["hermes", "chat", "-q", prompt, "-s", "wechat-group-digest",
         "--yolo", "-t", "terminal"],
        timeout=120, task_id=task_id
    )
    return jsonify({"ok": True, "data": {"task_id": task_id, "message": "提取任务已启动"}})


@app.route("/api/articles/unsummarized")
def api_article_unsummarized():
    """Get articles without summaries for cron job."""
    try:
        limit = min(int(request.args.get("limit", "20")), 200)
        rows = local_db.article_get_unsummarized_new(limit)
        return jsonify({"ok": True, "data": list(rows)})
    except Exception as e:
        import traceback
        traceback.print_exc()
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/articles/summarize", methods=["POST"])
def api_article_summarize():
    data = request.json or {}
    url = data.get("url", "")
    title = data.get("title", "")
    if not url:
        return jsonify({"ok": False, "error": "需要 url"}), 400
    dedup_key = _extract_dedup_key(url)
    if not dedup_key:
        return jsonify({"ok": False, "error": "无法提取去重标识"}), 400
    existing = local_db.find_article_summary(dedup_key, "summary")
    if existing:
        return jsonify({"ok": True, "data": {"status": "cached", "report_id": existing["id"],
            "content": existing["content"], "summary": (existing.get("content", "") or "")[:200],
            "created_at": existing.get("created_at", "")}})
    import uuid
    task_id = uuid.uuid4().hex[:12]
    result_file = f"/data/data/com.termux/files/home/tmp/article_summary_{task_id}.md"
    prompt = (f"请抓取并阅读以下文章，生成一份核心观点摘要。\n\n"
        f"文章标题: {title}\n文章链接: {url}\n\n"
        f"要求:\n1. 用 terminal curl 或 web 抓取文章全文\n"
        f"2. 提取 3-5 个核心观点，每个观点用一句话概括\n"
        f"3. 列出文章中的关键数据或案例（如有）\n"
        f"4. 格式为 Markdown，标题: # {title} 摘要\n"
        f"5. 将结果写入文件: {result_file}\n6. 确认文件已写入\n\n"
        f"重要: 只做抓取和摘要，完成后不要做其他操作。")
    def _on_summary_done(tid, ok):
        try:
            if Path(result_file).exists():
                md = Path(result_file).read_text(encoding="utf-8", errors="replace")
                # 检查是否生成成功：内容足够长且不含失败标记
                if not md or len(md.strip()) < 50 or "生成失败" in md or "重试" in md:
                    try: Path(result_file).unlink()
                    except: pass
                    return  # 失败不写 DB，用户可重试
            else:
                return  # 没有结果文件，不写 DB
            local_db.report_add("article", title, "summary", md, 0, 0, 0, dedup_key=dedup_key, article_url=url)
            try: Path(result_file).unlink()
            except: pass
        except Exception:
            pass
    _spawn_hermes_job("article_summary", title[:50], ["hermes", "chat", "-q", prompt, "-s", "article-summarizer", "--yolo", "-t", "terminal,web,file"],
        timeout=180, on_done=_on_summary_done, task_id=task_id,
        env={"HOME": "/data/data/com.termux/files/home", "HERMES_HOME": "/data/data/com.termux/files/home/.hermes"})
    return jsonify({"ok": True, "data": {"status": "started", "task_id": task_id}})

@app.route("/api/articles/learn", methods=["POST"])
def api_article_learn():
    data = request.json or {}
    url = data.get("url", "")
    title = data.get("title", "")
    if not url:
        return jsonify({"ok": False, "error": "需要 url"}), 400
    dedup_key = _extract_dedup_key(url)
    if not dedup_key:
        return jsonify({"ok": False, "error": "无法提取去重标识"}), 400
    existing = local_db.find_article_summary(dedup_key, "learn")
    if existing:
        return jsonify({"ok": True, "data": {"status": "cached", "report_id": existing["id"],
            "content": existing["content"], "summary": (existing.get("content", "") or "")[:200],
            "created_at": existing.get("created_at", "")}})
    import uuid
    task_id = uuid.uuid4().hex[:12]
    result_file = f"/data/data/com.termux/files/home/tmp/article_learn_{task_id}.md"
    prompt = (f"请学习以下文章，按照 learn-and-build skill 的流程处理。\n\n"
        f"文章标题: {title}\n文章链接: {url}\n\n"
        f"要求:\n1. 抓取文章全文\n2. 判断文章类型（开源代码/论文/方法论）\n"
        f"3. 按类型执行对应的学习+构建流程\n"
        f"4. 将学习结果和构建内容写入文件: {result_file}\n5. 确认文件已写入\n\n"
        f"完成后不要做其他操作。")
    def _on_learn_done(tid, ok):
        try:
            if Path(result_file).exists():
                md = Path(result_file).read_text(encoding="utf-8", errors="replace")
                if not md or len(md.strip()) < 50 or "生成失败" in md or "重试" in md:
                    try: Path(result_file).unlink()
                    except: pass
                    return
            else:
                return
            local_db.report_add("article", title, "learn", md, 0, 0, 0, dedup_key=dedup_key, article_url=url)
            try: Path(result_file).unlink()
            except: pass
        except Exception:
            pass
    _spawn_hermes_job("article_learn", title[:50], ["hermes", "chat", "-q", prompt, "-s", "learn-and-build", "--yolo", "-t", "terminal,web,file"],
        timeout=300, on_done=_on_learn_done, task_id=task_id)
    return jsonify({"ok": True, "data": {"status": "started", "task_id": task_id}})


@app.route("/api/articles/check-summaries", methods=["POST"])
def api_article_check_summaries():
    data = request.json or {}
    urls = data.get("urls", [])
    report_type = data.get("type", "summary")
    if not urls:
        return jsonify({"ok": False, "error": "需要 urls 数组"}), 400
    url_map = {}
    for item in urls:
        url = item.get("url", "") if isinstance(item, dict) else str(item)
        title = item.get("title", "") if isinstance(item, dict) else ""
        dk = _extract_dedup_key(url)
        if dk:
            url_map[dk] = {"url": url, "title": title}
    if not url_map:
        return jsonify({"ok": True, "data": {"results": {}, "missing": urls}})
    found = local_db.find_article_summaries_batch(list(url_map.keys()), report_type)
    missing = []
    for dk, info in url_map.items():
        if dk not in found:
            missing.append({"url": info["url"], "title": info["title"], "dedup_key": dk})
    return jsonify({"ok": True, "data": {"results": found, "missing": missing}})


@app.route("/api/articles/generate-image", methods=["POST"])
def api_article_generate_image():
    data = request.json or {}
    title = data.get("title","")
    url = data.get("url","")
    if not title:
        return jsonify({"ok":False,"error":"需要 title"}), 400
    import uuid
    task_id = uuid.uuid4().hex[:12]
    img_file = f"/data/data/com.termux/files/home/tmp/article_img_{task_id}.png"
    prompt = f"为以下文章内容生成一张 19 世纪复古铜版画风格图片。\n\n文章标题: {title}\n\n要求:\n1. 生成中文 prompt: 复古老旧报纸版面，19世纪黑白铜版画蚀刻风格，泛黄做旧牛皮纸，折痕毛边污渍，顶部大号粗黑体报纸刊头={title[:20]}，主标题加粗醒目，油墨印刷颗粒质感，黑白单色调，高细节，8K\n2. 用 GPT Image 2 API 生成图片(model=GPT Image 2, n=1, size=1024x1024)\n3. 将 base64 解码保存到 {img_file}\n4. 确认文件已写入\n\n重要: 完成后不要做其他操作。"
    def _on_img_done(tid, ok):
        try:
            if Path(img_file).exists():
                local_db.report_image_add(0, img_file, "wsj")
            try: Path(img_file).unlink()
            except: pass
        except: pass
    _spawn_hermes_job("article_image", title[:50], ["hermes","chat","-q",prompt,"-s","gpt-image-gen","--yolo","-t","terminal,web,file"],
        timeout=300, on_done=_on_img_done, task_id=task_id)
    return jsonify({"ok":True,"data":{"status":"started","task_id":task_id}})


# ── 账号切换 ──────────────────────────────────────

@app.route("/api/accounts")
def api_accounts():
    """返回所有已发现账号"""
    return jsonify({
        "ok": True,
        "data": {
            "accounts": config.ACCOUNTS,
            "current": config.get_current_account_id()
        }
    })

@app.route("/api/accounts/switch", methods=["POST"])
def api_accounts_switch():
    """切换当前账号"""
    data = request.json or {}
    aid = data.get("account_id", "")
    if not aid:
        return jsonify({"ok": False, "error": "需要 account_id"}), 400
    if not any(a["id"] == aid for a in config.ACCOUNTS):
        return jsonify({"ok": False, "error": "无效的账号"}), 400
    config.set_current_account_id(aid)
    return jsonify({"ok": True, "data": {"current": aid}})


# ── Page 1: 消息浏览器 ──────────────────────────────

@app.route("/")
def index():
    return render_template("index.html")


@app.route("/chat/<talker>")
def chat_page(talker):
    return render_template("chat.html", talker=talker)


@app.route("/messages")
def messages_page():
    return render_template("index.html")


@app.route("/message/<talker>")
def message_redirect(talker):
    return redirect(url_for("chat_page", talker=talker))


@app.route("/api/sessions")
def api_sessions():
    try:
        page = max(1, int(request.args.get("page", "1")))
        per_page = max(1, min(500, int(request.args.get("per_page", "10"))))
        sort = request.args.get("sort", "time_desc")
        if sort not in ("time_desc", "time_asc", "name"):
            sort = "time_desc"

        total = wechat_db.count_sessions()
        rows = wechat_db.get_session_list(page=page, per_page=per_page, sort=sort)
        for r in rows:
            # 从 content 提取纯文本预览
            raw = (r.get("content", "") or "")
            # 群聊消息去掉 "sender:\n" 前缀
            if ":\n" in raw and raw.split(":\n")[0]:
                raw = raw.split(":\n", 1)[1] if len(raw.split(":\n", 1)) > 1 else raw
            # 去掉 XML 标签 + HTML 实体
            clean = re.sub(r"<[^>]+>", "", raw).strip()
            clean = html_module.unescape(clean)
            r["preview"] = clean[:100]
        return jsonify({"ok": True, "data": rows, "total": total,
                        "page": page, "per_page": per_page})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/talker/<talker>")
def api_talker_info(talker):
    try:
        info = wechat_db.get_talker_info(talker)
        if info:
            return jsonify({"ok": True, "data": info})
        return jsonify({"ok": False, "error": "not found"}), 404
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/messages")
def api_messages():
    talker = request.args.get("talker", "")
    limit = int(request.args.get("limit", "40"))
    offset = int(request.args.get("offset", "0"))
    date = request.args.get("date", "")
    if not talker:
        return jsonify({"ok": False, "error": "需要 talker"}), 400
    try:
        rows = wechat_db.get_messages(talker, limit=limit, offset=offset, date=date)
        # Batch lookup sender nicknames
        sender_usernames = set()
        for row in rows:
            su = row.get("sender_username")
            if su:
                sender_usernames.add(su)
        nickname_map = wechat_db.get_talker_infos_batch(list(sender_usernames)) if sender_usernames else {}

        parsed = []
        for row in rows:
            p = _parse_message_content(row.get("content", "") or "", row.get("imgPath", "") or "", row.get("type", 0))
            p["msgId"] = row.get("msgId"); p["type"] = row.get("type")
            p["isSend"] = row.get("isSend"); p["createTime"] = row.get("createTime")
            p["talker"] = row.get("talker")
            su = row.get("sender_username")
            p["sender_username"] = su
            p["sender_nickname"] = nickname_map.get(su) if su else None
            parsed.append(p)
        return jsonify({"ok": True, "data": parsed})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# ── 消息内容解析 ────────────────────────────────────

def _parse_message_content(content: str, img_path: str, msg_type: int) -> dict:
    result = {"text": "", "images": [], "voice": None, "articles": []}
    clean_content = content or ""
    clean_content = re.sub(r"<!\[CDATA\[(.*?)\]\]>", r"\1", clean_content, flags=re.DOTALL)
    display_text = clean_content
    if clean_content and "\n" in clean_content and ":" in clean_content.split("\n")[0]:
        parts = clean_content.split("\n", 1)
        if len(parts) == 2:
            display_text = parts[1]

    if msg_type == 1:
        result["text"] = re.sub(r"<[^>]+>", "", display_text).strip() if "<msg>" in display_text else display_text.strip()
    elif msg_type == 3:
        if img_path:
            result["images"].append({"src": img_path})
        result["text"] = "[图片]"
    elif msg_type == 34:
        if img_path:
            result["voice"] = {"path": img_path}
    elif msg_type == 49 or msg_type == 285212721:
        url_match = re.search(r"<url>(.*?)</url>", clean_content)
        title_match = re.search(r"<title>(.*?)</title>", clean_content)
        desc_match = re.search(r"<des>(.*?)</des>", clean_content)
        thumb_match = re.search(r"<thumburl>(.*?)</thumburl>", clean_content)
        if url_match or title_match:
            result["articles"].append({"url": url_match.group(1) if url_match else "",
                "title": html_module.unescape(title_match.group(1)) if title_match else "",
                "description": html_module.unescape(desc_match.group(1)) if desc_match else "",
                "thumbnail": thumb_match.group(1) if thumb_match else ""})
        txt = re.sub(r"<[^>]+>", "", display_text).strip()
        if txt:
            result["text"] = txt
    elif msg_type == 47:
        result["text"] = "[表情]"
    elif msg_type == 43:
        result["text"] = "[视频]"
    elif msg_type == 754974769 or "<finderFeed>" in clean_content:
        cdata_clean = re.sub(r"<!\[CDATA\[(.*?)\]\]>", r"\1", clean_content)
        fnick = re.search(r"<nickname>(.*?)</nickname>", cdata_clean)
        favatar = re.search(r"<avatar>(.*?)</avatar>", cdata_clean)
        fdesc = re.search(r"<desc>(.*?)</desc>", cdata_clean)
        fcover = re.search(r"<coverUrl>(.*?)</coverUrl>", cdata_clean)
        fthumb = re.search(r"<thumbUrl>(.*?)</thumbUrl>", cdata_clean)
        fvideo = re.search(r"<media>.*?<url>(.*?)</url>", cdata_clean, re.DOTALL)
        fdur = re.search(r"<videoPlayDuration>(\d+)</videoPlayDuration>", cdata_clean)
        result["finder"] = {"nickname": html_module.unescape(fnick.group(1).strip()) if fnick else "",
            "avatar": favatar.group(1).strip() if favatar else "",
            "desc": html_module.unescape(fdesc.group(1).strip()) if fdesc else "",
            "cover": (fcover.group(1) or fthumb.group(1)).strip() if (fcover or fthumb) else "",
            "video_url": fvideo.group(1).strip() if fvideo else "", "duration": fdur.group(1) if fdur else "0"}
    elif msg_type == 822083633:
        display_name = re.search(r"<displayname>(.*?)</displayname>", clean_content)
        refer_title = re.search(r"<title>(.*?)</title>", clean_content)
        result["text"] = f"[引用 {display_name.group(1)}: {refer_title.group(1)}]" if display_name and refer_title else "[引用消息]"
    elif msg_type == 10000:
        result["text"] = "[系统消息]"
    return result


# ── 媒体文件代理 ────────────────────────────────────

@app.route("/media/image")
def serve_image():
    path = request.args.get("path", "")
    if not path:
        return "no path", 400
    real_path = _resolve_image_path(path)
    if not real_path:
        return "image not found", 404
    tmp_path = f"/tmp/img_{abs(hash(real_path))}"
    if not _safe_exists(tmp_path):
        subprocess.run(["su", "-c", f"cp {real_path} {tmp_path} && chmod 644 {tmp_path}"],
            capture_output=True, text=True, timeout=10)
    if _safe_exists(tmp_path):
        return send_file(tmp_path)
    return "image not found", 404


@app.route("/media/image/full")
def serve_image_full():
    path = request.args.get("path", "")
    if not path:
        return "no path", 400
    hash_part = path.replace("THUMBNAIL_DIRPATH://", "").replace("THUMBNAIL_DIR_PATH://", "")
    clean = hash_part.replace("th_", "")
    full_path = _resolve_image_path(f"THUMBNAIL_DIRPATH://{clean}")
    if full_path:
        tmp_path = f"/tmp/img_full_{abs(hash(full_path))}"
        if not _safe_exists(tmp_path):
            subprocess.run(["su", "-c", f"cp {full_path} {tmp_path} && chmod 644 {tmp_path}"],
                capture_output=True, text=True, timeout=10)
        if _safe_exists(tmp_path):
            return send_file(tmp_path)
    return serve_image()


@app.route("/media/voice")
def serve_voice():
    path = request.args.get("path", "")
    if not path:
        return "no path", 400
    real_path = _resolve_voice_path(path)
    if not real_path:
        return "voice not found", 404
    tmp_amr = f"/tmp/voice_{abs(hash(real_path))}.amr"
    if not _safe_exists(tmp_amr):
        subprocess.run(["su", "-c", f"cp {real_path} {tmp_amr} && chmod 644 {tmp_amr}"],
            capture_output=True, text=True, timeout=10)
    if not _safe_exists(tmp_amr):
        return "voice not found", 404
    mp3_path = tmp_amr.replace(".amr", ".mp3")
    if not _safe_exists(mp3_path):
        subprocess.run(["ffmpeg", "-y", "-i", tmp_amr, "-acodec", "libmp3lame", "-ab", "32k", mp3_path],
            capture_output=True, text=True, timeout=30)
    if _safe_exists(mp3_path):
        return send_file(mp3_path, mimetype="audio/mpeg")
    return send_file(tmp_amr)


def _resolve_image_path(path: str) -> str | None:
    """THUMBNAIL_DIRPATH://th_<hash> → image2/XX/YY/<hash>.jpg (原图优先) 或 th_<hash> (缩略图)"""
    if not path:
        return None
    if path.startswith("/") and _safe_exists(path):
        return path
    path2 = path.replace("THUMBNAIL_DIRPATH://", "").replace("THUMBNAIL_DIR_PATH://", "")
    if not path2:
        return None
    hash_part = path2.replace("th_", "")
    if len(hash_part) >= 4:
        sub1, sub2 = hash_part[:2], hash_part[2:4]
        base = f"{config.get_current_account()['data_root']}/image2/{sub1}/{sub2}"
        for c in [f"{base}/{hash_part}.jpg", f"{base}/{hash_part}.png", f"{base}/{hash_part}", f"{base}/{path2}"]:
            if _safe_exists(c):
                return c
        for ext in [".jpg", ".png", ""]:
            c = f"{config.get_current_account()['data_root']}/image2/{hash_part}{ext}"
            if _safe_exists(c):
                return c
    return None


def _resolve_voice_path(path: str) -> str | None:
    if not path:
        return None
    if _safe_exists(path):
        return path
    clean = path.replace("THUMBNAIL_DIRPATH://", "").replace("THUMBNAIL_DIR_PATH://", "")
    try:
        r = subprocess.run(["su", "-c", f"find {config.get_current_account()['data_root']}/voice2 -name '*{clean}*' 2>/dev/null | head -1"],
            capture_output=True, text=True, timeout=10)
        found = r.stdout.strip()
        if found:
            return found
    except Exception:
        pass
    return None


# ── Avatar 服务 ──────────────────────────────────────

def _resolve_avatar_path(username: str) -> str | None:
    """username → MD5 → avatar/{hash[:2]}/{hash[2:4]}/user_hd_{hash}.png"""
    if not username:
        return None
    h = hashlib.md5(username.encode()).hexdigest()
    base = f"{config.get_current_account()['data_root']}/avatar/{h[:2]}/{h[2:4]}"
    # Try HD first, then regular, then inside hash directory
    for fname in [f"user_hd_{h}.png", f"user_{h}.png"]:
        path = f"{base}/{fname}"
        if _safe_exists(path):
            return path
    # Also try inside the full hash subdirectory
    for fname in [f"user_hd_{h}.png", f"user_{h}.png"]:
        path = f"{base}/{h}/{fname}"
        if _safe_exists(path):
            return path
    return None


@app.route("/media/avatar/<username>")
def serve_avatar(username):
    real_path = _resolve_avatar_path(username)
    if not real_path:
        return "avatar not found", 404
    tmp_path = f"/tmp/avatar_{abs(hash(real_path))}.png"
    if not _safe_exists(tmp_path):
        subprocess.run(["su", "-c", f"cp {real_path} {tmp_path} && chmod 644 {tmp_path}"],
            capture_output=True, text=True, timeout=10)
    if _safe_exists(tmp_path):
        return send_file(tmp_path, mimetype="image/png")
    return "avatar not accessible", 404


# ── Page 2: 数据中心 ────────────────────────────────

@app.route("/circle")
def circle_page():
    return render_template("circle.html")


@app.route("/api/circle/stats")
def api_circle_stats():
    try:
        rows = wechat_db.query("SELECT COUNT(*) as cnt FROM message")
        messages_total = rows[0]["cnt"] if rows else 0
        rows = wechat_db.query("SELECT COUNT(DISTINCT talker) as cnt FROM message WHERE talker LIKE '%@chatroom'")
        groups_total = rows[0]["cnt"] if rows else 0
        conn = local_db.get_conn()
        images_total = conn.execute("SELECT COUNT(*) FROM image_analysis").fetchone()[0]
        reports_total = conn.execute("SELECT COUNT(*) FROM daily_reports WHERE report_date NOT IN ('summary','learn')").fetchone()[0]
        articles_total = conn.execute("SELECT COUNT(*) FROM articles").fetchone()[0]
        summaries_total = conn.execute("SELECT COUNT(*) FROM chat_summaries").fetchone()[0]
        conn.close()
        return jsonify({"ok": True, "data": {"messages_total": messages_total, "groups_total": groups_total,
            "images_total": images_total, "reports_total": reports_total, "articles_total": articles_total,
            "summaries_total": summaries_total}})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/circle/articles")
def api_circle_articles():
    try:
        status = request.args.get("status", "pending")
        conn = local_db.get_conn()
        # 兼容 article_queue 旧表和 articles 主表
        rows = conn.execute(
            "SELECT id, source_msg_id, chatroom_id, talker, url, title, status, extracted_at FROM article_queue WHERE status = ? ORDER BY extracted_at DESC LIMIT 200",
            (status,)
        ).fetchall()
        data = [dict(r) for r in rows]
        conn.close()
        return jsonify({"ok": True, "data": data})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/circle/articles/<int:article_id>/status", methods=["POST"])
def api_circle_article_status(article_id):
    try:
        body = request.get_json() or {}
        new_status = body.get("status", "read")
        conn = local_db.get_conn()
        conn.execute("UPDATE article_queue SET status = ? WHERE id = ?", (new_status, article_id))
        conn.commit()
        conn.close()
        return jsonify({"ok": True})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/circle/images")
def api_circle_images():
    try:
        conn = local_db.get_conn()
        rows = conn.execute(
            "SELECT id, source_msg_id, chatroom_id, talker, image_path, ocr_text, description, analyzed_at FROM image_analysis ORDER BY analyzed_at DESC LIMIT 200"
        ).fetchall()
        data = [dict(r) for r in rows]
        conn.close()
        return jsonify({"ok": True, "data": data})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/circle/summaries")
def api_circle_summaries():
    try:
        conn = local_db.get_conn()
        rows = conn.execute(
            "SELECT id, chatroom_id, date, summary, msg_count, created_at FROM chat_summaries ORDER BY date DESC, created_at DESC LIMIT 200"
        ).fetchall()
        data = [dict(r) for r in rows]
        conn.close()
        return jsonify({"ok": True, "data": data})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# ── Page 3: 日报系统 ─────────────────────────────────

@app.route("/reports")
def reports_page():
    return render_template("reports.html")

@app.route("/summaries")
def summaries_page():
    return render_template("reports.html", page_type="summary")

@app.route("/learn-records")
def learn_records_page():
    return render_template("reports.html", page_type="learn")


@app.route("/report/<int:report_id>")
def report_detail(report_id):
    return render_template("report_detail.html", report_id=report_id)


@app.route("/api/reports")
def api_reports():
    try:
        report_type = request.args.get("type", "digest")
        rows = local_db.report_list(report_type=report_type)
        for r in rows:
            r["summary"] = (r.get("content", "") or "")[:200]
            r.pop("content", None)
        return jsonify({"ok": True, "data": rows})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/report/<int:report_id>")
def api_report_detail(report_id):
    try:
        row = local_db.report_get(report_id)
        if row:
            return jsonify({"ok": True, "data": row})
        return jsonify({"ok": False, "error": "not found"}), 404
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/reports/generate", methods=["POST"])
def api_report_generate():
    data = request.json or {}
    chatroom_id = data.get("chatroom_id", "")
    date_str = data.get("date", "")
    if not chatroom_id or not date_str:
        return jsonify({"ok": False, "error": "需要 chatroom_id 和 date"}), 400
    try:
        import uuid
        talker_info = wechat_db.get_talker_info(chatroom_id)
        chatroom_name = str(talker_info.get("displayname", chatroom_id)) if talker_info else chatroom_id
        if isinstance(chatroom_name, dict):
            chatroom_name = chatroom_id
        task_id = uuid.uuid4().hex[:12]
        task_file = f"/data/data/com.termux/files/home/tmp/report_task_{task_id}.json"
        result_file = f"/data/data/com.termux/files/home/tmp/report_result_{task_id}.md"
        with open(task_file, "w") as f:
            json.dump({"chatroom_id": chatroom_id, "date": date_str, "result_file": result_file}, f)
        prompt = (f"你的任务是生成一份微信群的日报。\n\n群聊 ID: {chatroom_id}\n日期: {date_str}\n\n"
            f"## 步骤（简洁高效，不要过度解析）\n\n"
            f"### 1. 查消息列表\n"
            f"用 terminal curl 调 http://127.0.0.1:5001/api/messages?talker={chatroom_id}&date={date_str}\n"
            f"拿到消息列表即可，**不要逐条解析 finder/voice/image 内容**。\n"
            f"消息类型标注为文本/图片/语音/文章/视频号即可，不需要提取具体内容。\n\n"
            f"### 2. 提取文章并查摘要\n"
            f"从消息中提取所有 type=49（文章）的 URL 和标题。\n"
            f"用 terminal curl 调 POST http://127.0.0.1:5001/api/articles/check-summaries\n"
            f"批量查哪些已有摘要（JSON: {{\"urls\":[{{\"url\":\"...\",\"title\":\"...\"}}], \"type\":\"summary\"}}）\n\n"
            f"### 3. 生成日报\n"
            f"- 已有摘要的文章：直接引用摘要内容\n"
            f"- **没有摘要的文章：只列标题和链接，标注「⏳ 待摘要」，不要调 summarize API，不要等待**\n"
            f"- 按主题分组 + 讨论亮点\n"
            f"- 写入 {result_file}\n\n"
            f"**重要：不要解析图片/视频号/语音/文件的具体内容。不要逐条写 Python 脚本解析消息。用 API 就够了。**")
        def _on_report_done(tid, ok):
            try:
                if Path(result_file).exists():
                    md_content = Path(result_file).read_text(encoding="utf-8", errors="replace")
                else:
                    log_path = str(HERMES_JOBS_DIR / f"{tid}.log")
                    md_content = "# 生成失败\n\n请查看日志了解详情。"
                local_db.report_add(chatroom_id, chatroom_name, date_str, md_content, 0, 0, 0)
            except Exception:
                pass
            try: Path(task_file).unlink()
            except: pass
            try: Path(result_file).unlink()
            except: pass
        _spawn_hermes_job("daily_report", f"{chatroom_name} {date_str}", ["hermes", "chat", "-q", prompt, "-s", "wechat-group-digest", "--yolo",
            "-t", "terminal,file,web", "-m", "deepseek-v4-pro", "--provider", "deepseek"],
            timeout=600, on_done=_on_report_done, task_id=task_id)
        return jsonify({"ok": True, "data": {"status": "started", "task_id": task_id}})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/report/<int:report_id>/delete", methods=["POST"])
def api_report_delete(report_id):
    try:
        local_db.report_delete(report_id)
        return jsonify({"ok": True})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/reports/generate/status")
def api_report_status():
    try:
        rows = local_db.report_list(1)
        if rows:
            return jsonify({"ok": True, "data": rows[0]})
        return jsonify({"ok": True, "data": None})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# ── Page 4: 日报图片 ─────────────────────────────────

@app.route("/report-images")
def report_images_page():
    return render_template("report_images.html")


@app.route("/api/reports", methods=["POST"])
def api_reports_create():
    """外部调用（如 cron job）注册日报到数据库"""
    data = request.json or {}
    chatroom_id = data.get("chatroom_id", "")
    chatroom_name = data.get("chatroom_name", chatroom_id)
    report_date = data.get("report_date", "")
    content = data.get("content", "")
    msg_count = data.get("msg_count", 0)
    image_count = data.get("image_count", 0)
    article_count = data.get("article_count", 0)
    if not chatroom_id or not report_date or not content:
        return jsonify({"ok": False, "error": "需要 chatroom_id, report_date, content"}), 400
    try:
        report_id = local_db.report_add(chatroom_id, chatroom_name, report_date, content,
                                        msg_count, image_count, article_count)
        return jsonify({"ok": True, "data": {"report_id": report_id}})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/report-images", methods=["POST"])
def api_report_images_create():
    """外部调用（如 cron job）注册日报图片到数据库"""
    data = request.json or {}
    report_id = data.get("report_id", 0)
    image_path = data.get("image_path", "")
    style = data.get("style", "wsj")
    if not report_id or not image_path:
        return jsonify({"ok": False, "error": "需要 report_id 和 image_path"}), 400
    try:
        image_id = local_db.report_image_add(report_id, image_path, style)
        return jsonify({"ok": True, "data": {"image_id": image_id}})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/report-images")
def api_report_images():
    try:
        rows = local_db.report_image_list()
        return jsonify({"ok": True, "data": rows})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/media/report-image/<int:image_id>")
def serve_report_image(image_id):
    conn = local_db.get_conn()
    row = conn.execute("SELECT image_path FROM report_images WHERE id = ?", (image_id,)).fetchone()
    conn.close()
    if row and row["image_path"] and _safe_exists(row["image_path"]):
        return send_file(row["image_path"])
    return "image not found", 404


@app.route("/api/reports/<int:report_id>/generate-image", methods=["POST"])
def api_report_generate_image(report_id):
    try:
        report = local_db.report_get(report_id)
        if not report:
            return jsonify({"ok": False, "error": "日报不存在"}), 404
        import uuid
        task_id = uuid.uuid4().hex[:12]
        content_file = f"/data/data/com.termux/files/home/tmp/report_content_{task_id}.md"
        img_file = f"/data/data/com.termux/files/home/tmp/report_img_{task_id}.png"
        Path(content_file).write_text(report["content"] or "", encoding="utf-8")
        chatroom_name = report.get("chatroom_name", "群聊")
        report_date = report.get("report_date", "")
        prompt = (
            f"按 gpt-image-gen skill 的日报图片 Prompt 生成报纸图片。\n"
            f"日报文件: {content_file}\n输出图片: {img_file}\n"
            f"群名: {chatroom_name}\n日期: {report_date}\n\n"
            f"步骤:\n"
            f"1. read_file 读完整日报内容\n"
            f"2. 按 skill 提炼: 群名+日期(报头)、选定头条主题+配大图、所有文章的标题+摘要(多栏)、群友金句(原文)、供稿名单\n"
            f"3. 按 skill 的 Prompt 限制规则生成图片 prompt: 英文描述结构，关键内容(标题/金句/人名)用中文\n"
            f"4. GPT Image 2 API (model='GPT Image 2', n=1, size='1024x1024')\n"
            f"5. base64解码保存到 {img_file}, ls 确认\n\n"
            f"禁止: 第X版、详见、参见、X版——这是单页群日报"
        )
        STORAGE_DIR = Path("/data/data/com.termux/files/home/wechat-viewer/static/report_images")
        STORAGE_DIR.mkdir(parents=True, exist_ok=True)
        def _on_img_done(tid, ok):
            try:
                if Path(img_file).exists():
                    # 移到永久存储，不在 DB 入库后删除
                    dest = STORAGE_DIR / f"report_{report_id}_{tid}.png"
                    import shutil
                    shutil.move(img_file, str(dest))
                    local_db.report_image_add(report_id, str(dest), "wsj")
                try: Path(content_file).unlink()
                except: pass
            except Exception:
                pass
        _spawn_hermes_job("report_image", f"{chatroom_name} {report_date}", ["hermes", "chat", "-q", prompt, "-s", "gpt-image-gen", "--yolo", "-t", "terminal,web,file"],
            timeout=300, on_done=_on_img_done, task_id=task_id)
        return jsonify({"ok": True, "data": {"status": "started", "task_id": task_id}})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/report-image/<int:image_id>/delete", methods=["POST"])
def api_report_image_delete(image_id):
    try:
        conn = local_db.get_conn()
        row = conn.execute("SELECT image_path FROM report_images WHERE id = ?", (image_id,)).fetchone()
        if row and row["image_path"]:
            try: Path(row["image_path"]).unlink()
            except: pass
        conn.close()
        local_db.report_image_delete(image_id)
        return jsonify({"ok": True})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/report-image/<int:image_id>/send", methods=["POST"])
def api_report_image_send(image_id):
    data = request.json or {}
    wxid = data.get("to", "")
    if not wxid:
        return jsonify({"ok": False, "error": "需要 to (wxid)"}), 400
    try:
        conn = local_db.get_conn()
        row = conn.execute("SELECT image_path FROM report_images WHERE id = ?", (image_id,)).fetchone()
        conn.close()
        if not row or not row["image_path"]:
            return jsonify({"ok": False, "error": "图片不存在"}), 404
        import shutil
        dest = f"/sdcard/Pictures/report_{image_id}.png"
        shutil.copy2(row["image_path"], dest)
        import sys
        def _send():
            try:
                sys.path.insert(0, "/data/data/com.termux/files/home/.hermes/event_bus")
                from wechat_hook_send import send_image_via_hook
                send_image_via_hook(wxid, dest)
            except Exception:
                pass
        threading.Thread(target=_send, daemon=True).start()
        return jsonify({"ok": True, "data": {"status": "sent"}})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# ── Page 5: 任务监控 ─────────────────────────────────

@app.route("/jobs")
def jobs_page():
    return render_template("jobs.html")


@app.route("/job/<task_id>")
def job_detail_page(task_id):
    return render_template("job_detail.html", task_id=task_id)


@app.route("/api/jobs")
def api_jobs():
    try:
        rows = local_db.job_list(200)
        return jsonify({"ok": True, "data": rows})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/job/<task_id>")
def api_job_detail(task_id):
    try:
        row = local_db.job_get(task_id)
        if row:
            return jsonify({"ok": True, "data": row})
        return jsonify({"ok": False, "error": "not found"}), 404
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/job/<task_id>/log")
def api_job_log(task_id):
    try:
        row = local_db.job_get(task_id)
        if not row or not row.get("log_file"):
            return jsonify({"ok": False, "error": "not found"}), 404
        log_path = row["log_file"]
        offset_val = request.args.get("offset", "0")
        # offset can be line number or byte position (for polling)
        try:
            with open(log_path, "r") as f:
                content = f.read()
        except FileNotFoundError:
            return jsonify({"ok": True, "data": {"content": "", "size": 0,
                "status": row.get("status", "unknown")}})
        return jsonify({"ok": True, "data": {
            "content": content,
            "size": len(content),
            "status": row.get("status", "unknown"),
            "started_at": row.get("started_at", ""),
            "completed_at": row.get("completed_at", "")
        }})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/job/<task_id>/delete", methods=["POST"])
def api_job_delete(task_id):
    try:
        row = local_db.job_get(task_id)
        if row and row.get("log_file"):
            try: Path(row["log_file"]).unlink()
            except: pass
        local_db.job_delete(task_id)
        return jsonify({"ok": True})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# ── Page 6: 学习审核 ─────────────────────────────────

@app.route("/learn")
def learn_page():
    return render_template("learn.html")


@app.route("/learn/<int:item_id>")
def learn_detail_page(item_id):
    """学习项详情页 — 展示 Step 4 吸收报告 + Step 6 应用审计（如有）"""
    item = local_db.learn_get(item_id)
    if not item:
        return "Not Found", 404
    # Step 4: 取最近的 非apply_audit 类型结果（吸收报告）
    result = local_db.learn_result_get_step4(item_id)
    # Step 6: 取 apply_audit 类型结果（知识应用审计）
    apply_result = local_db.learn_result_get_apply_audit(item_id)
    return render_template("learn_detail.html", item=item, result=result, apply_result=apply_result)


@app.route("/api/learn-queue")
def api_learn_queue():
    try:
        status = request.args.get("status", "pending_review")
        rows = local_db.learn_list(status)
        return jsonify({"ok": True, "data": rows})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/learn/classify", methods=["POST"])
def api_learn_classify():
    """对未分类的文章摘要运行分类器"""
    try:
        summaries = local_db.learn_get_unclassified_summaries(20)
        if not summaries:
            return jsonify({"ok": True, "data": {"classified": 0, "message": "无未分类摘要"}})

        import uuid
        task_id = uuid.uuid4().hex[:12]
        summaries_file = f"/data/data/com.termux/files/home/tmp/classify_input_{task_id}.json"
        output_file = f"/data/data/com.termux/files/home/tmp/classify_output_{task_id}.json"

        input_data = [{"dedup_key": s["dedup_key"], "title": s.get("chatroom_name", ""),
                        "article_url": s.get("article_url", ""),
                        "summary": (s.get("content", "") or "")[:1000]}
                      for s in summaries if s.get("dedup_key")]

        import json
        Path(summaries_file).write_text(json.dumps(input_data, ensure_ascii=False, indent=2))

        prompt = f"""你是文章评分分类器。阅读文件 {summaries_file} 中的文章摘要，对每篇判断学习类型并写入 {output_file}。

## 评分标准 (score: 1-10)
评估这篇文章对 Hermes AI Agent 的价值：
- 9-10: 直接可用的工具/API/代码，显著提升 Hermes 能力
- 7-8: 高质量方法论/架构，值得深入学习后构建
- 5-6: 行业动态/趋势，有助于决策上下文
- 3-4: 一般资讯，可能有参考价值
- 1-2: 低价值，不值得花时间

## 分类规则
- skill: 有 GitHub/PyPI 链接、可安装运行的开源工具/框架
- memory: 行业趋势、市场数据、竞品动态等需要记住的事实
- todo: 待跟进的项目、产品发布、需要采取行动的事项
- reference: 技术架构、设计模式、方法论等值得存档的知识
- skip: 泛泛资讯，不值得深入学习

## 输出格式 (JSON)
{{"results":[{{"dedup_key":"...", "learn_type":"skill", "score":8, "confidence":"high",
  "learn_target":"skill-name", "learn_content":"# 要学习的内容\\n\\n核心要点..."}}, ...]}}

## learn_target 规则
- skill: 建议的 skill 名称 (小写, 用连字符)
- memory: 记忆键名 (描述性)
- todo: 待办标题
- reference: 目标 skill 名或 "new:topic-name"
- skip: 空字符串

完成后不要做其他操作。"""

        def _on_classify_done(tid, ok, spawn_reason=None):
            try:
                if not ok:
                    # ── HydraMind Gate: 分类器失败 trace ──
                    # 对齐四元组: node/attempt/gate/reason_code
                    # 与 spawn gate 不同：分类器失败需要 REPLAN（人工判断是否重试），而非 BLOCK
                    classify_trace = (
                        f"[node=classify,attempt=1,gate=RUNTIME_CLASSIFY_FAILED,"
                        f"action={GateAction.REPLAN}] "
                        f"分类子进程异常退出 task_id={tid}")
                    # 写入数据库供 UI 展示（learn_queue 中最后一条相关记录）
                    try:
                        conn = local_db.get_conn()
                        latest = conn.execute(
                            "SELECT id FROM learn_queue WHERE status='unclassified' OR status='pending_review' "
                            "ORDER BY id DESC LIMIT 1").fetchone()
                        if latest:
                            conn.execute("UPDATE learn_queue SET error_msg=? WHERE id=?",
                                         (classify_trace, latest["id"]))
                        conn.commit()
                        conn.close()
                    except Exception:
                        pass
                    # 同时保留文件日志用于离线排查
                    try:
                        with open(f"/data/data/com.termux/files/home/tmp/classify_error_{tid}.log", "w") as ef:
                            ef.write(f"=== {classify_trace} ===\n")
                    except Exception:
                        pass
                if Path(output_file).exists():
                    results = json.loads(Path(output_file).read_text())
                    for r in results.get("results", []):
                        dedup = r.get("dedup_key", "")
                        ltype = r.get("learn_type", "skip")
                        if ltype == "skip":
                            continue
                        # Find original summary
                        orig = next((s for s in summaries if s.get("dedup_key") == dedup), None)
                        if not orig:
                            continue
                        # Dedup: skip if already in learn_queue for this type
                        existing = local_db.learn_find_by_dedup(dedup, ltype)
                        if existing:
                            continue
                        # Score-based routing: ≥7 cron自动学, 4-6 人工审, 1-3 丢弃
                        score = int(r.get("score", 0) or 0)
                        if score >= 7:
                            status = "pending_learn"
                            confidence = r.get("confidence", "high")
                        elif score >= 4:
                            status = "pending_review"
                            confidence = r.get("confidence", "medium")
                        else:
                            continue  # 低分不写入
                        # ── HydraMind Gate: 分类器输出质量门控 ──
                        # 依据可验证事实（learn_content 长度）决策，非 LLM 评分
                        learn_content = r.get("learn_content", "") or ""
                        if not learn_content or len(learn_content.strip()) < 20:
                            # 空内容不能进学习队列——写 REPLAN gate 让人工判断
                            classify_bad_trace = (
                                f"[node=classify,attempt=1,gate=CONTENT_EMPTY,"
                                f"action={GateAction.REPLAN}] "
                                f"type={ltype} target={r.get('learn_target','?')} score={score}"
                                f" — learn_content 为空或过短，分类器可能幻觉")
                            try:
                                with open(f"/data/data/com.termux/files/home/tmp/classify_content_empty_{tid}.log", "a") as ef:
                                    ef.write(f"=== {classify_bad_trace} ===\n    dedup={dedup}\n    content={learn_content[:100]}\n")
                            except Exception:
                                pass
                            continue  # 不入库，等下一次分类器重试\n                        # ── HydraMind Gate: 分类路由 trace ──
                        # 补齐四元组：分类成功也有 gate trace，不只是失败路径
                        route_reason = "pending_learn" if score >= 7 else "pending_review"
                        route_trace = (
                            f"[node=classify,attempt=1,gate=ROUTED,"
                            f"action={GateAction.CONTINUE}] "
                            f"score={score} type={ltype} → {route_reason}")
                        new_id = local_db.learn_add(
                            article_url=orig.get("article_url", ""),
                            title=orig.get("chatroom_name", ""),
                            dedup_key=dedup,
                            learn_type=ltype,
                            learn_target=r.get("learn_target", ""),
                            learn_content=r.get("learn_content", ""),
                            confidence=confidence,
                            score=score,
                            status=status
                        )
                        local_db.learn_update_status(new_id, status, route_trace)
                try: Path(summaries_file).unlink()
                except: pass
                try: Path(output_file).unlink()
                except: pass
            except Exception as e:
                # ── HydraMind Gate: 分类回调异常 ──
                classify_cb_trace = (
                    f"[node=classify,attempt=1,gate=CLASSIFY_CALLBACK_ERR,"
                    f"action={GateAction.REPLAN}] "
                    f"分类回调异常: {str(e)[:200]}")
                try:
                    with open(f"/data/data/com.termux/files/home/tmp/classify_error_{tid}.log", "a") as ef:
                        ef.write(f"=== {classify_cb_trace} ===\n")
                except Exception:
                    pass

        _spawn_hermes_job("learn_classify", f"分类 {len(input_data)} 篇",
            ["hermes", "chat", "-q", prompt, "-s", "learn-and-build", "--yolo", "-t", "terminal,file,web"],
            timeout=180, on_done=_on_classify_done, task_id=task_id)
        return jsonify({"ok": True, "data": {"classified": len(input_data), "task_id": task_id}})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


def _run_absorb(item_id: int):
    """统一吸收函数：6 步 Loop — Step 1-4: 学习摘要 → 消化吸收 → 验证测试 → 生成报告; Step 5: 吸收验证; Step 6: 知识应用审计"""
    try:
        item = local_db.learn_get(item_id)
        if not item:
            return
        import uuid
        tid = uuid.uuid4().hex[:12]
        content_file = f"/data/data/com.termux/files/home/tmp/learn_absorb_{tid}.md"
        Path(content_file).write_text(item.get("learn_content", "") or "", encoding="utf-8")

        ltype = item["learn_type"]
        target = item.get("learn_target", "")
        prompt = f"""你是 Hermes 的知识吸收引擎。按 extract-learning-outline skill 的四步 Loop 执行吸收。

**铁律：没有系统影响的吸收等于垃圾。吸收完成后，必须能回答：Hermes 下次遇到什么场景会有什么不同的行为？如果答不上来 → 标记 rejected，不吸收。**

文章学习内容: {content_file}
吸收类型: {ltype}
吸收目标: {target}

## Step 1: 制作学习摘要
1. 读 {content_file} 的原文内容
2. 扫描 Hermes 当前状态 — skill_view 看相关 skill、search_files 搜代码、memory 查记忆
3. 写一份「学习摘要」到 {content_file}.summary.md：
   - 这篇文章的核心知识点（1-5 条，每条一句话）
   - Hermes 已有多少覆盖？新知识占比？
   - **关键判断：这篇文章能让 Hermes 做什么之前做不到的事？如果不能 → 直接 rejected**

**覆盖度判断铁律：文档里有 ≠ 实现了。用以下量表强制打分：**

| 判据 | 加分 |
|------|:--:|
| SKILL.md 提到概念 | +1 |
| 代码有对应模块 | +3 |
| memory 有结构化记录 | +2 |
| 有可执行测试 | +2 |
| 近7天实际使用过 | +2 |

**总分 ≤4 → 必须吸收，不能 reject。** 常见误判：SKILL.md 写了「Memory五类分类」→ 只得1分，应吸收。

**吸收深度分级**（不是吸/不吸二值）：
- L1: 文档追加（skill patch / memory add）— 当前工具能做
- L2: 结构化（需改代码/memory schema）— 标「部分吸收:需L2」传给Step6
- L3: 代码重构（改core逻辑）— 标「部分吸收:需L3」传给Step6

## Step 2: 消化吸收
基于学习摘要，制定 Planning 到 {content_file}.plan.md：
(a) 吸收目标 — 具体学什么、达到什么效果
(b) 测试用例 — 至少 2 个场景，描述吸收后 Hermes 应有的不同反应
(c) 关键要点 — 需要记住的 API/方法/结论
(d) 引用路径 — 代码路径、skill 名、外部链接

然后**实际调用工具**执行吸收（不能用文字替代）：

### 所有类型统一规则：必须产出可验证的系统变更
- **skill**: skill_manage(action='create'/'patch')，完整 SKILL.md。创建后用 skill_view 确认。
- **memory**: memory(action='add', target='memory')。写入后验证。
- **reference**: 不是写孤立 MD 文件！必须：
  1. 扫描所有可用 skill，找到最相关的一个
  2. skill_manage(action='patch') 把核心知识作为新章节加入目标 skill 的 SKILL.md
  3. 如果涉及跨 skill 的通用认知 → memory add
  4. **如果完全无法匹配任何 skill 或 memory → 标记 rejected（reason: "无可内化目标"），不吸收**

## Step 3: 验证测试
执行 Benchmark：
1. 对 Step 2 的测试用例，逐一验证
2. skill 创建/更新后用 skill_view 确认
3. memory 写入后确认存在
4. 如果吸收后的 Hermes 能通过测试用例 → 成功
5. 如果失败 → 记录失败原因，标记 failed

**Step 级 trace**：每一步结束时写入 checkpoint 文件供外部 Gate 审计：
- Step 1 完成 → 写 `[node=absorb_step1,gate=STEP_DONE,action=CONTINUE]` 到 `{content_file}.step1.done`
- Step 2 完成 → 写 `[node=absorb_step2,gate=STEP_DONE,action=CONTINUE]` 到 `{content_file}.step2.done`
- Step 3 完成 → 写 `[node=absorb_step3,gate=STEP_DONE,action=CONTINUE]` 到 `{content_file}.step3.done`
- 任一步失败 → 写 `[node=absorb_stepN,gate=STEP_FAILED,action=REVISE] reason:...` 到 `{content_file}.stepN.failed`
这样即使吸收中途崩溃，也能通过 checkpoint 文件定位失败步骤。

## Step 4: 生成报告
写入 {content_file}.result.md，严格按此结构（前端依赖这些字段渲染卡片）：
```
# 学习完成报告

## 学习内容
- 标题: [文章标题]
- 来源: [URL 或文件路径]
- 类型: {ltype}
- 日期: [YYYY-MM-DD]

## 吸收产出
### 创建/更新 Skill
- **[skill-name]**: create/patch +N行 | 新增「章节名」
（如无则写「无」）

### 写入 Memory
- "关键认知" → memory.json
（如无则写「无」）

### 代码变更
- [path/to/file.py]: +N行 -M行 | 改了什么
（如无则写「无」）

### 安装/集成
- pip install xxx → 成功/失败
（如无则写「无」）

## 验证结果
- TC1: [PASS/FAIL] — [场景 → 验证了什么]
- TC2: [PASS/FAIL] — [场景 → 验证了什么]

## 行为变化
- 下次遇到 [X 场景] → Hermes 会 [Y 不同反应]

## 未完成/待跟进
- [原因]
（如无则写「无」）
```
如果没有产出任何 skill/memory → 报告标题写「REJECTED: 无可内化内容」。

注意：每一步都要实际执行工具调用，不能用文字描述代替。用 --yolo 模式。"""

        def _on_absorb_done(tid2, ok, spawn_reason=None):
            try:
                result_file = f"{content_file}.result.md"
                plan_file = f"{content_file}.plan.md"
                summary_file = f"{content_file}.summary.md"
                is_rejected = False  # default: not rejected
                content_thin = False  # Gate: content quality check
                if ok and Path(result_file).exists():
                    result_text = Path(result_file).read_text()[:1200]
                    # ── Gate 密度加强: content quality gate ──
                    # 依据可验证事实（文本长度）决策，非 LLM 评分
                    if len(result_text.strip()) < 50:
                        content_thin = True
                        # HydraMind REVISE: content thin → auto-retry absorb (not BLOCK)
                        # Must set status='pending_learn' so cron picks it up for retry
                        local_db.learn_update_status(item_id, "pending_learn",
                            f"[node=absorb,attempt=1,gate={VerifyReason.CONTENT_THIN},"
                            f"action={GateAction.REVISE}] "
                            f"{VERIFY_REASON_CODE[VerifyReason.CONTENT_THIN]}")
                    # Gate: 行为变化缺失 — 吸收没有产生系统影响
                    elif "## 行为变化" not in result_text and "REJECTED" not in result_text:
                        # 宽松处理：如果「吸收产出」章节有内容，仍放行但加 gate trace
                        if "## 吸收产出" in result_text:
                            # Soft-pass: 有产出但缺行为变化声明 → CONTINUE with trace
                            soft_trace = (
                                f"[node=absorb,attempt=1,gate=NO_BEHAVIOR_CHANGE_SOFT,"
                                f"action={GateAction.CONTINUE}] "
                                f"报告有吸收产出但缺「行为变化」章节，放行但需人工复核")
                            local_db.learn_update_status(item_id, "applied",
                                soft_trace + " | " + result_text[:300])
                            # Build outline + reflection for learning_results (needed by Step 5)
                            outline_parts = []
                            for f in [summary_file, plan_file]:
                                if Path(f).exists():
                                    outline_parts.append(Path(f).read_text()[:500])
                            outline = "\n\n".join(outline_parts)
                            local_db.learn_result_add(
                                learn_queue_id=item_id,
                                article_url=item.get("article_url", ""),
                                article_title=item.get("title", ""),
                                outline=outline,
                                result_type=ltype,
                                result_summary=result_text[:500],
                                created_skill=target if ltype == "skill" else "",
                                auto_applied=1
                            )
                        else:
                            local_db.learn_update_status(item_id, "failed",
                                f"[node=absorb,attempt=1,gate={VerifyReason.NO_BEHAVIOR_CHANGE},"
                                f"action={GateAction.REPLAN}] "
                                f"吸收报告缺少「行为变化」章节，无法确认系统影响（建议人工审核后 /relearn）")
                    # 检测 REJECTED：无系统影响的吸收不应该标 applied
                    elif "REJECTED" in result_text[:100]:
                        is_rejected = True
                        local_db.learn_update_status(item_id, "rejected",
                            f"[node=absorb,attempt=1,gate=REJECTED,"
                            f"action={GateAction.BLOCK}] "
                            f"{result_text[:300] or '无可内化内容'}")
                    else:
                        # 仅在正常成功路径（非 soft-pass）执行
                        # ── Gate 密度加强: Step 1 摘要质量 gate ──
                        # 吸收子进程不可见，事后检查 Step 1 产物是否存在且有实质内容
                        summary_ok = (Path(summary_file).exists()
                                      and len(Path(summary_file).read_text()[:100].strip()) >= 30)
                        absorb_trace = (
                            f"[node=absorb,attempt=1,gate=PASS,"
                            f"action={GateAction.CONTINUE}] "
                            f"summary={'ok' if summary_ok else 'thin'}")
                        # Prepend gate trace to error_msg (replaces the result_text-only msg)
                        local_db.learn_update_status(item_id, "applied",
                            absorb_trace + " | " + result_text[:300])
                        # Build outline from summary + plan + reflection
                        outline_parts = []
                        for f in [summary_file, plan_file]:
                            if Path(f).exists():
                                outline_parts.append(Path(f).read_text()[:500])
                        outline = "\n\n".join(outline_parts)
                        reflection = result_text[:500]
                        local_db.learn_result_add(
                            learn_queue_id=item_id,
                            article_url=item.get("article_url", ""),
                            article_title=item.get("title", ""),
                            outline=outline,
                            result_type=ltype,
                            result_summary=reflection,
                            created_skill=target if ltype == "skill" else "",
                            auto_applied=1
                        )
                else:
                    # ── HydraMind Gate: spawn failure → check Step-level checkpoints ──
                    # Step trace files let us pinpoint which step the LLM failed at
                    step_checks = []
                    for step_n in [1, 2, 3]:
                        sf = Path(f"{content_file}.step{step_n}.failed")
                        df = Path(f"{content_file}.step{step_n}.done")
                        if sf.exists():
                            step_checks.append(sf.read_text()[:80])
                            try: sf.unlink()
                            except: pass
                        elif df.exists():
                            try: df.unlink()
                            except: pass
                    if step_checks:
                        rc = VerifyReason.NO_OUTPUT
                        local_db.learn_update_status(item_id, "failed",
                            f"[node=spawn,attempt=1,gate={rc},"
                            f"action={GateAction.REPLAN}] "
                            f"Step-level failures: {'; '.join(step_checks)[:300]}")
                    else:
                        # No step checkpoints → spawn-layer diagnostic fallback
                        rc = spawn_reason if spawn_reason else VerifyReason.NO_OUTPUT
                        ga = GateAction.REVISE if rc in (VerifyReason.SPAWN_TIMEOUT,) else GateAction.BLOCK
                        local_db.learn_update_status(item_id, "failed",
                            f"[node=spawn,attempt=1,gate={rc},"
                            f"action={ga}] "
                            f"{VERIFY_REASON_CODE.get(rc, '子进程执行失败')}")

                # ── Step 5+6: 验证吸收 → 知识应用（必须在 cleanup 之前！）──
                should_apply = ok and not is_rejected and not content_thin
                # Cleanup（先保留 result_text 引用，cleanup 后再用）
                for f in [content_file, plan_file, summary_file]:
                    try: Path(f).unlink()
                    except: pass
                # Cleanup step checkpoint files
                for step_n in [1, 2, 3]:
                    for suffix in [".done", ".failed"]:
                        try: Path(f"{content_file}.step{step_n}{suffix}").unlink()
                        except: pass
                # result_file 最后删——_verify_and_apply 不需要它
                if should_apply:
                    # ── HydraMind Gate: absorb→step5 transition gate trace ──
                    # Emit explicit gate at the between-node boundary for auditability
                    gate_rc = VerifyReason.PASS if not ("BEHAVIOR_CHANGE_SOFT" in (item.get("error_msg","") or "")) else "NO_BEHAVIOR_CHANGE_SOFT"
                    local_db.learn_update_status(item_id, "applied",
                        f"{item.get('error_msg','') or ''} | "
                        f"[node=absorb_to_step5,attempt=1,gate={gate_rc},"
                        f"action={GateAction.CONTINUE}] 进入 Step5 验证")
                    _verify_and_apply(item_id, ltype, target)
                try: Path(result_file).unlink()
                except: pass
            except Exception as e:
                local_db.learn_update_status(item_id, "failed",
                    f"[node=absorb,attempt=1,gate={VerifyReason.NO_OUTPUT},"
                    f"action={GateAction.BLOCK}] "
                    f"on_done 异常: {str(e)[:150]}")

        _spawn_hermes_job("learn_absorb", f"{ltype}: {target[:40]}",
            ["hermes", "chat", "-q", prompt, "--yolo", "-t", "terminal,file,web",
             "-s", "extract-learning-outline,learn-and-build"],
            timeout=600, on_done=_on_absorb_done, task_id=tid)
    except Exception:
        local_db.learn_update_status(item_id, "failed", "吸收启动失败")


# ── Step 5+6: 吸收验证 + 知识应用 ──────────────────────

# ── HydraMind Gate Audit: Gate 动作枚举 ──
# 每个 Gate 不只 binary pass/fail，应该产出控制动作
# 对应 Gate 动作审计第二层：continue/revise/replan/block
class GateAction:
    CONTINUE = "CONTINUE"   # 放行下一步
    REVISE   = "REVISE"     # 可自动修复 → 重试当前步（带修正参数）
    REPLAN   = "REPLAN"     # 需人工判断 → 暂停当前，写 Feishu task
    BLOCK    = "BLOCK"      # 不可恢复 → 标记 failed，终止

# Gate reason codes — HydraMind-inspired structured failure attribution
# Maps to execution trace quad: node/attempt/gate/reason_code
# Format: [node=<node_name>,attempt=<N>,gate=<reason_code>,action=<GateAction>]
class VerifyReason:
    NO_OUTPUT        = "NO_OUTPUT"         # 无吸收产出记录
    SKILL_MISSING    = "SKILL_MISSING"     # 声称创建的 skill 不存在
    MEMORY_MISSING   = "MEMORY_MISSING"    # memory.json 文件缺失
    NO_VERIFIABLE    = "NO_VERIFIABLE"     # 无可验证项
    FINGERPRINT_FAIL = "FINGERPRINT_FAIL"  # 指纹不匹配（可能空内容）
    CONTENT_THIN     = "CONTENT_THIN"      # 吸收产出内容过短（可能假成功）
    NO_BEHAVIOR_CHANGE = "NO_BEHAVIOR_CHANGE"  # 吸收报告缺少「行为变化」章节
    APPLY_CALLBACK_ERR = "APPLY_CALLBACK_ERR"  # Step6 应用审计回调异常
    PASS             = "PASS"              # 通过
    PASS_LAX         = "PASS_LAX"          # 柔性通过（memory unconfirmed）
    # ── Spawn/Harness 层 reason codes（原 HARNESS_REASON，统一到此）──
    SPAWN_TIMEOUT    = "RUNTIME_TIMEOUT"      # 子进程超时
    SPAWN_NONZERO    = "RUNTIME_NONZERO_EXIT" # 子进程非零退出
    SPAWN_FAILED     = "RUNTIME_SPAWN_FAILED" # 子进程启动失败
    SPAWN_ON_DONE_ERR = "CONTEXT_CALLBACK_FAILED"  # 完成回调异常
    # ── 预算控制（Inspired by Paperclip budget hard limits）──
    BUDGET_EXHAUSTED = "BUDGET_EXHAUSTED"   # 月度预算达上限，自动停止
    BUDGET_WARNING   = "BUDGET_WARNING"     # 预算使用超80%，预警

VERIFY_REASON_CODE = {
    VerifyReason.NO_OUTPUT:      "吸收子进程未产出任何结果记录",
    VerifyReason.SKILL_MISSING:  "声称创建的 skill 文件不存在于文件系统",
    VerifyReason.MEMORY_MISSING: "memory.json 文件缺失",
    VerifyReason.NO_VERIFIABLE:  "吸收产出中无可验证的实体项",
    VerifyReason.FINGERPRINT_FAIL: "content 指纹不在 memory.json 中",
    VerifyReason.CONTENT_THIN:   "吸收产出内容过短（<50字符），疑似假成功",
    VerifyReason.NO_BEHAVIOR_CHANGE: "吸收报告缺少「行为变化」章节，无法确认系统影响",
    VerifyReason.APPLY_CALLBACK_ERR: "Step6 应用审计回调异常",
    # Spawn 层
    VerifyReason.SPAWN_TIMEOUT:  "Hermes 子进程超时被杀",
    VerifyReason.SPAWN_NONZERO:  "Hermes 子进程非零退出码",
    VerifyReason.SPAWN_FAILED:   "Hermes 子进程启动失败",
    VerifyReason.SPAWN_ON_DONE_ERR: "Hermes 子进程完成后回调异常",
    # 预算控制 (Inspired by Paperclip)
    VerifyReason.BUDGET_EXHAUSTED: "月度预算已耗尽，自动停止当前 job_type 的新任务",
    VerifyReason.BUDGET_WARNING:   "月度预算使用超 80%，建议人工审核",
}

# ── 预算控制模块（Inspired by Paperclip budget hard limits）──
# 文件级预算追踪：避免 API 费用失控。当前为 L1 实现（JSON 文件），
# L2 路线：DB schema + 月度自动重置 + 管理 UI。
# 设计原则：闸门在最上游——预算检查在 semaphore 之前，不占并发槽位。

import dataclasses as _dc
from typing import Optional as _Opt

@_dc.dataclass
class BudgetConfig:
    """单 job_type 的预算配置"""
    monthly_limit: float       # 月度预算上限（美元）
    cost_per_call: float       # 单次调用估算成本（美元）
    warn_threshold: float = 0.8  # 预警阈值（默认 80%）

BUDGET_FILE = Path("/data/data/com.termux/files/home/tmp/hermes_budget.json")

# 默认预算配置 — 按 job_type 分
# 估算基础：hermes chat --yolo 单次 ~0.03-0.50 USD（取决于 prompt 长度和模型）
# 用户应在 BUDGET_FILE 中覆盖这些默认值
DEFAULT_BUDGETS: dict[str, BudgetConfig] = {
    "learn_absorb":   BudgetConfig(monthly_limit=20.0, cost_per_call=0.50),
    "learn_classify": BudgetConfig(monthly_limit=5.0,  cost_per_call=0.10),
    "learn_apply":    BudgetConfig(monthly_limit=5.0,  cost_per_call=0.30),
    "chat_reply":     BudgetConfig(monthly_limit=10.0, cost_per_call=0.05),
    "hermes_chat":    BudgetConfig(monthly_limit=30.0, cost_per_call=0.20),
}


def _load_budget_state() -> dict:
    """加载预算使用状态。格式: {month_key: {job_type: total_cost}}"""
    try:
        if BUDGET_FILE.exists():
            return json.loads(BUDGET_FILE.read_text())
    except Exception:
        pass
    return {}


def _save_budget_state(state: dict) -> None:
    """持久化预算状态"""
    BUDGET_FILE.parent.mkdir(parents=True, exist_ok=True)
    BUDGET_FILE.write_text(json.dumps(state, indent=2))


def _check_budget(job_type: str) -> tuple[bool, str | None]:
    """检查 job_type 是否有剩余预算。
    返回 (allowed, reason_code_or_None)。
    如果 allowed=False，reason_code 是 BUDGET_EXHAUSTED 或 BUDGET_WARNING。
    BUDGET_WARNING 不阻止执行，只写 trace。
    """
    cfg = DEFAULT_BUDGETS.get(job_type)
    if not cfg:
        return True, None  # 无配置的 job_type 不受预算限制

    month_key = datetime.now().strftime("%Y-%m")
    state = _load_budget_state()

    # 读取或初始化当前月份的 usage
    month_state = state.get(month_key, {})
    current_cost = month_state.get(job_type, 0.0)

    # 检查预算耗尽
    if current_cost >= cfg.monthly_limit:
        return False, VerifyReason.BUDGET_EXHAUSTED

    # 检查预算预警
    if current_cost / max(cfg.monthly_limit, 0.01) >= cfg.warn_threshold:
        return True, VerifyReason.BUDGET_WARNING

    return True, None


def _record_budget_spend(job_type: str) -> None:
    """记录一次调用的估算成本"""
    cfg = DEFAULT_BUDGETS.get(job_type)
    if not cfg:
        return

    month_key = datetime.now().strftime("%Y-%m")
    state = _load_budget_state()

    if month_key not in state:
        state[month_key] = {}
    state[month_key][job_type] = state[month_key].get(job_type, 0.0) + cfg.cost_per_call

    _save_budget_state(state)


def _verify_and_apply(item_id: int, ltype: str, target: str):
    """吸收完成后的 Step 5（确定性验证）+ Step 6（知识应用审计）。

    Step 5 Gate: 确定性检查点，依据可验证事实产生控制动作。
      动作: CONTINUE (放行 Step 6) / REVISE (自动重试) / REPLAN (人工判断) / BLOCK (标记 failed)
      输出: reason_code + GateAction 用于离线评估归因。
    Step 6: Hermes 子进程加载新知识 → 扫描相关系统 → 审计 → 自动 patch 或写 Feishu。
    """
    item = local_db.learn_get(item_id)
    if not item:
        return
    result = local_db.learn_result_get_by_learn_id(item_id)
    if not result:
        local_db.learn_update_status(item_id, "failed",
            f"[node=step5_verify,attempt=1,gate={VerifyReason.NO_OUTPUT},"
            f"action={GateAction.BLOCK}] {VERIFY_REASON_CODE[VerifyReason.NO_OUTPUT]}")
        return

    # ── Step 5 Gate: 确定性验证 ──
    # Gate consumes boundary facts (file existence, content fingerprint),
    # NOT LLM scores — per HydraMind principle
    verified = []
    reason_code = None
    # skill 验证
    cskill = result.get("created_skill", "") or ""
    otype = result.get("result_type", "") or ""
    if cskill:
        skill_dir = Path("/data/data/com.termux/files/home/.hermes/skills")
        found = list(skill_dir.rglob(f"*/{cskill}/SKILL.md")) or \
                list(skill_dir.rglob(f"{cskill}/SKILL.md"))
        if found:
            verified.append(f"skill:{cskill} found")
        else:
            # HydraMind REVISE: one-shot retry — skill 可能还在磁盘写入中
            import time as _time
            _time.sleep(0.5)
            found2 = list(skill_dir.rglob(f"*/{cskill}/SKILL.md")) or \
                     list(skill_dir.rglob(f"{cskill}/SKILL.md"))
            if found2:
                verified.append(f"skill:{cskill} found (retry)")
            else:
                # REVISE path: reset to pending_learn for one more cron-managed retry
                # (before BLOCK), consistent with MEMORY_MISSING/FINGERPRINT_FAIL behavior
                import re as _re_sk
                cur = local_db.learn_get(item_id) or {}
                prev_msgs = (cur.get("error_msg") or "")
                attempts = [int(m) for m in _re_sk.findall(r"attempt=(\d+)", prev_msgs)]
                next_att = max(attempts) + 1 if attempts else 2
                reason_code = VerifyReason.SKILL_MISSING
                if next_att <= 3:
                    local_db.learn_update_status(item_id, "pending_learn",
                        f"[node=step5_verify,attempt={next_att},gate={reason_code},"
                        f"action={GateAction.REVISE}] "
                        f"{VERIFY_REASON_CODE[reason_code]}: {cskill}")
                else:
                    local_db.learn_update_status(item_id, "failed",
                        f"[node=step5_verify,attempt={next_att},gate={reason_code},"
                        f"action={GateAction.BLOCK}] "
                        f"{VERIFY_REASON_CODE[reason_code]}: {cskill}，已重试 {next_att-1} 次")
                return
    # memory 验证：检查 learn_content 里是否有实际产出标记
    if otype in ("memory", "reference"):
        content = item.get("learn_content", "") or ""
        fingerprint = content.strip()[:30]
        verified_fingerprint = False

        if otype == "reference" and target:
            # reference → content was patched into target skill's SKILL.md
            skill_dir = Path("/data/data/com.termux/files/home/.hermes/skills")
            skill_files = list(skill_dir.rglob(f"*/{target}/SKILL.md")) or \
                          list(skill_dir.rglob(f"{target}/SKILL.md"))
            for sf in skill_files:
                try:
                    if fingerprint and fingerprint in sf.read_text():
                        verified.append(f"reference:{target} SKILL.md confirmed")
                        verified_fingerprint = True
                        break
                except:
                    pass
            if not verified_fingerprint:
                # fallback: also check memory.json (some reference absorbs write to both)
                mem_path = Path("/data/data/com.termux/files/home/.hermes/memory.json")
                if mem_path.exists():
                    mem_text = mem_path.read_text()
                    if fingerprint and fingerprint in mem_text:
                        verified.append("reference:memory.json confirmed (fallback)")
                        verified_fingerprint = True
        elif otype == "memory":
            # memory → check memory.json, then MEMORY.md
            mem_path = Path("/data/data/com.termux/files/home/.hermes/memory.json")
            if mem_path.exists():
                mem_text = mem_path.read_text()
                if fingerprint and fingerprint in mem_text:
                    verified.append("memory:content confirmed")
                    verified_fingerprint = True
                elif len(content.strip()) < 100:
                    verified.append("memory:unconfirmed (pass)")
                    verified_fingerprint = True
                else:
                    mem_md_path = Path("/data/data/com.termux/files/home/.hermes/memories/MEMORY.md")
                    if mem_md_path.exists():
                        if fingerprint and fingerprint in mem_md_path.read_text():
                            verified.append("memory:MEMORY.md confirmed (fallback)")
                            verified_fingerprint = True

        if not verified_fingerprint and len(content.strip()) >= 100:
            reason_code = VerifyReason.FINGERPRINT_FAIL
            import re as _re2
            cur = local_db.learn_get(item_id) or {}
            prev_msgs = (cur.get("error_msg") or "")
            attempts = [int(m) for m in _re2.findall(r"attempt=(\d+)", prev_msgs)]
            next_att = max(attempts) + 1 if attempts else 2
            if next_att <= 3:
                local_db.learn_update_status(item_id, "pending_learn",
                    f"[node=step5_verify,attempt={next_att},gate={reason_code},"
                    f"action={GateAction.REVISE}] "
                    f"{VERIFY_REASON_CODE[reason_code]}: "
                    f"content 指纹 '{fingerprint}' 不在 target skill SKILL.md 或 memory.json 中（len={len(content.strip())}, type={otype}, target={target}）")
            else:
                local_db.learn_update_status(item_id, "failed",
                    f"[node=step5_verify,attempt={next_att},gate={reason_code},"
                    f"action={GateAction.BLOCK}] "
                    f"{VERIFY_REASON_CODE[reason_code]}: "
                    f"content 指纹 '{fingerprint}' 不在目标中，已重试 {next_att-1} 次")
            return

    if not verified:
        reason_code = VerifyReason.NO_VERIFIABLE
        local_db.learn_update_status(item_id, "failed",
            f"[node=step5_verify,attempt=1,gate={reason_code},"
            f"action={GateAction.REPLAN}] "
            f"{VERIFY_REASON_CODE[reason_code]}")
        return

    # Gate decision: CONTINUE — all boundary facts verified
    rc = VerifyReason.PASS if "unconfirmed" not in verified[0] else VerifyReason.PASS_LAX
    ga = GateAction.CONTINUE
    verify_log = (f"[node=step5_verify,attempt=1,gate={rc},action={ga}] "
                  f"{'; '.join(verified)}")
    # 追加验证日志到 error_msg（保留原状态 + reason_code）
    item_cur = local_db.learn_get(item_id) or {}
    prev = (item_cur.get("error_msg") or "")[:200]
    local_db.learn_update_status(item_id, item_cur.get("status", "applied"),
        f"{prev} | {verify_log}"[:400])

    # ── Step 6: 知识应用审计 ──
    _run_apply(item_id, ltype, target, verify_log)


def _run_apply(item_id: int, ltype: str, target: str, verify_log: str):
    """Step 6: spawn Hermes 子进程 → 加载新知识 → 审计相关系统 → 产出改进。

    用 learn-and-build skill 让 Hermes 扫描刚学的知识关联的系统，
    用学到的框架/方法论审计，自动修复可自动的 gap，写 Feishu task 给人工判断的。
    """
    item = local_db.learn_get(item_id)
    if not item:
        return
    import uuid
    tid = uuid.uuid4().hex[:12]
    content_file = f"/data/data/com.termux/files/home/tmp/learn_apply_{tid}.md"

    # 准备应用内容：吸收摘要 + 产物信息
    apply_content = item.get("learn_content", "") or ""
    if len(apply_content) < 50:
        Path(content_file).write_text(
            f"# 吸收产物\n- 类型: {ltype}\n- 目标: {target}\n- 验证: {verify_log}",
            encoding="utf-8")
    else:
        Path(content_file).write_text(apply_content, encoding="utf-8")

    prompt = f"""你是 Hermes 的知识应用引擎。Step 5 验证已通过，现在执行 Step 6：把刚学的知识应用到 Hermes 系统中。

## 刚吸收的知识
文件: {content_file}
类型: {ltype}
目标: {target}
验证结果: {verify_log}
{'**⚠️ WARNING: memory 验证为 PASS_LAX（指纹未匹配），吸收内容可能未真正写入 memory.json。审计时请 double-check 目标系统。**' if 'PASS_LAX' in verify_log else ''}

## 审计框架：多层栅格（HydraMind Gate Audit）+ Graph Engineering 扩展维度

用刚学的知识当「验收标准」，逐层审计相关系统。先加载 learn-and-build skill 的 references/etclovg-audit-framework.md 获取完整审计表；审计时把 Graph / Loop / Harness 三者分开检查，并逐条验证四类边（数据边、权限边、验证边、恢复边）。

### 第一层：Gate 密度审计
- 相关系统在关键步骤间有几个运行时门控（Gate）？
- 每个 Gate 是否依据可验证事实（文件存在/测试通过/diff合理）决策，而非 LLM 评分？
- Gate 间距是否过大导致失败定位退化到 Agent 级（14% 步级准确率）而非步级？
- **Graph 视角**：步骤间依赖/并行/汇合是否显式表达？是否能在运行前审计拓扑？

### 第二层：Gate 动作审计
- 每个 Gate 支持哪些控制动作？（continue/revise/replan/block）
- 是否只有 binary pass/fail？缺 revise/replan 路径意味着什么？
- **Loop 视角**：失败后回退到哪个节点？收敛条件是否可验证？

### 第三层：执行轨迹审计
- 失败时能否追溯到四元组：node/attempt/gate/reason_code？
- 如果只能看到"失败了"而不知道哪个节点、第几次尝试、哪个 gate 拦截、什么原因码 → O（可观测性）gap
- **Harness 视角**：证据/日志放哪？能否暂停/恢复/回滚？

### 第四层：过程评测闭环
- 是否有：运行时 Gate → 离线评估识别失败模式 → 修订 Gate 规则 → 回归测试？
- 缺失 → L（生命周期）gap，系统不会自我改进
- **四类边审计**：数据边（下游字段结构）、权限边（只读→写入放大）、验证边（谁有资格宣布通过）、恢复边（失败后从哪继续/幂等键）

## 任务
1. 读 {content_file} 的完整内容，理解刚学到的知识
2. 用 search_files / read_file 扫描 Hermes 中与这些知识相关的系统：
   - 相关的 skill（读取 SKILL.md）
   - 相关的代码（search_files 找 .py 文件中的 gate/verify/validate/check 模式）
   - 相关的配置
3. 用上述四层栅格审计这些系统：
   - 当前实现 vs 新知识的差距在哪？
   - 哪些 gap 可以立即自动修复（加 reason_code、加中间 gate、改 prompt）？
   - 哪些需要人工判断（改 DB schema、加离线评估管线）？
4. **执行修复**：
   - 可自动的 gap → 直接 patch 代码或 skill_manage
   - 需人工的 → 记入报告，标注「建议 Feishu task」
5. 写报告到 {content_file}.apply.md：
```
# Step 6 知识应用报告
- 学习内容: [一句话]
- 审计了哪些系统: [skill名 / 文件路径]
- Gate 密度评分: [X/10] — [说明]
- Gate 动作评分: [X/10] — [说明]
- 执行轨迹评分: [X/10] — [说明]
- 过程评测闭环评分: [X/10] — [说明]
- 自动修复: [改了什么，具体到行]
- 发现但未修复: [gap 描述 + 为什么不能自动修]
- 建议: [下一步行动]
- Human gate: [哪些需要人看一眼再执行]
```

注意：实际执行工具调用，不要只用文字描述。超时 5 分钟。"""

    def _on_apply_done(tid2, ok, spawn_reason=None):
        try:
            result_file = f"{content_file}.apply.md"
            if ok and Path(result_file).exists():
                apply_text = Path(result_file).read_text()[:1500]
                local_db.learn_result_add(
                    learn_queue_id=item_id,
                    article_url=item.get("article_url", ""),
                    article_title=item.get("title", ""),
                    outline=f"Step6 知识应用审计\n{verify_log}",
                    result_type="apply_audit",
                    result_summary=apply_text,
                    created_skill="",
                    auto_applied=1
                )
                # Trace quad: gate action emitted for offline audit
                has_human = "Human gate" in apply_text or "建议 Feishu" in apply_text
                item_cur = local_db.learn_get(item_id)
                if item_cur:
                    ga = GateAction.REPLAN if has_human else GateAction.CONTINUE
                    local_db.learn_update_status(item_id, item_cur.get("status", "applied"),
                        f"{verify_log} | "
                        f"[node=step6_apply,attempt=1,gate={VerifyReason.PASS},action={ga}] "
                        f"{'人工事项待处理' if has_human else '应用审计完成，无待办'}")
                # 如果有 Human gate 项 → 写 Feishu
                if has_human:
                    try:
                        import requests
                        requests.post("http://127.0.0.1:5001/api/feishu/todo",
                            json={"type": "learn_apply", "item_id": item_id,
                                  "title": f"学习应用: {target[:30]}",
                                  "content": apply_text[:800]},
                            timeout=10)
                    except Exception:
                        pass
            else:
                # apply 失败不改变 learn status——吸收本身是成功的
                # ── HydraMind Gate: propagate spawn reason_code ──
                rc = spawn_reason if spawn_reason else VerifyReason.NO_OUTPUT
                ga = GateAction.REVISE if rc in (VerifyReason.SPAWN_TIMEOUT,) else GateAction.REPLAN
                item_cur = local_db.learn_get(item_id)
                if item_cur:
                    local_db.learn_update_status(item_id, item_cur.get("status", "applied"),
                        f"{verify_log} | "
                        f"[node=step6_apply,attempt=1,gate={rc},"
                        f"action={ga}] "
                        f"Step6: 应用审计执行失败(吸收仍有效，可 /relearn 重试)")
            for f in [content_file, result_file]:
                try: Path(f).unlink()
                except: pass
        except Exception as e:
            item_cur = local_db.learn_get(item_id)
            if item_cur:
                local_db.learn_update_status(item_id, item_cur.get("status", "applied"),
                    f"[node=step6_apply,attempt=1,gate=APPLY_CALLBACK_ERR,"
                    f"action={GateAction.BLOCK}] "
                    f"Step6: 异常 {str(e)[:100]}")

    _spawn_hermes_job("learn_apply", f"apply: {target[:40]}",
        ["hermes", "chat", "-q", prompt, "--yolo", "-t", "terminal,file,web",
         "-s", "learn-and-build"],
        timeout=300, on_done=_on_apply_done, task_id=tid)


@app.route("/api/learn/<int:item_id>/approve", methods=["POST"])
def api_learn_approve(item_id):
    """人工点「吸收」→ 进入 pending_learn 队列，等 Cron 学习执行器处理。"""
    try:
        item = local_db.learn_get(item_id)
        if not item:
            return jsonify({"ok": False, "error": "not found"}), 404
        local_db.learn_update_status(item_id, "pending_learn")
        return jsonify({"ok": True, "data": {"status": "pending_learn"}})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/learn/<int:item_id>/reject", methods=["POST"])
def api_learn_reject(item_id):
    try:
        item = local_db.learn_get(item_id)
        if not item:
            return jsonify({"ok": False, "error": "not found"}), 404
        local_db.learn_update_status(item_id, "rejected")
        return jsonify({"ok": True})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


VALID_LEARN_STATUSES = {"pending_review", "pending_learn", "applied", "approved", "failed", "rejected"}


@app.route("/api/learn/<int:item_id>/priority", methods=["PUT"])
def api_learn_update_priority(item_id):
    """手动调整学习项优先级（支持拖拽排序）"""
    try:
        item = local_db.learn_get(item_id)
        if not item:
            return jsonify({"ok": False, "error": "not found"}), 404
        body = request.get_json(silent=True)
        if not body or "priority" not in body:
            return jsonify({"ok": False, "error": "missing priority field"}), 400
        try:
            prio = int(body["priority"])
        except (ValueError, TypeError):
            return jsonify({"ok": False, "error": "priority must be integer"}), 400
        conn = local_db.get_conn()
        conn.execute("UPDATE learn_queue SET priority = ? WHERE id = ?", (prio, item_id))
        conn.commit()
        conn.close()
        return jsonify({"ok": True, "data": {"id": item_id, "priority": prio}})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/learn/<int:item_id>/status", methods=["PUT"])
def api_learn_update_status(item_id):
    """手动修改学习项状态"""
    try:
        item = local_db.learn_get(item_id)
        if not item:
            return jsonify({"ok": False, "error": "not found"}), 404
        body = request.get_json(silent=True)
        if not body or "status" not in body:
            return jsonify({"ok": False, "error": "missing status field"}), 400
        new_status = body["status"]
        if new_status not in VALID_LEARN_STATUSES:
            return jsonify({"ok": False, "error": f"invalid status: {new_status}",
                           "valid": sorted(VALID_LEARN_STATUSES)}), 400
        local_db.learn_update_status(item_id, new_status)
        return jsonify({"ok": True, "data": {"id": item_id, "status": new_status}})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/learn/<int:item_id>/relearn", methods=["POST"])
def api_learn_relearn(item_id):
    """重新触发学习：保留历史 trace（带 attempt 递增），重置状态 → pending_learn"""
    try:
        item = local_db.learn_get(item_id)
        if not item:
            return jsonify({"ok": False, "error": "not found"}), 404
        # ── HydraMind Gate: attempt 计数器 ──
        # 解析现有 error_msg 中的 attempt 计数，递增后再重置
        # 格式: [node=xxx,attempt=N,...]  → attempt 取最大值 + 1
        import re as _re
        prev_msg = item.get("error_msg", "") or ""
        attempts = [int(m) for m in _re.findall(r"attempt=(\d+)", prev_msg)]
        next_attempt = max(attempts) + 1 if attempts else 2
        # 保留历史 trace（带 attempt 标签）追加到 error_msg
        history_note = (
            f"[retry,attempt={next_attempt}] "
            f"重新触发吸收（共 {len(attempts)} 次历史 gate 记录）| ")
        # 清除旧 learning_results
        conn = local_db.get_conn()
        conn.execute("DELETE FROM learning_results WHERE learn_queue_id = ?", (item_id,))
        conn.commit()
        conn.close()
        # 重置状态：pending_learn + 保留历史 error_msg + 新 attempt
        local_db.learn_update_status(item_id, "pending_learn", history_note + prev_msg[:300])
        _run_absorb(item_id)
        return jsonify({"ok": True, "data": {"status": "pending_learn", "attempt": next_attempt}})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/learn/execute", methods=["POST"])
def api_learn_execute():
    """Cron 调用的学习执行器：扫描 pending_learn 队列，逐条执行 4 步吸收 Loop。
    
    并发安全：使用文件锁确保同一时间只有一个吸收在执行，避免多个 hermes chat
    进程同时运行撑爆内存（每个 ~150MB，3 个并发 = 450MB+ 加上其他 agent 直接 OOM）。
    """
    import fcntl as _fcntl_exec
    LOCK_FILE = "/data/data/com.termux/files/home/tmp/learn_executor.lock"
    lock_fd = open(LOCK_FILE, "w")
    try:
        _fcntl_exec.flock(lock_fd, _fcntl_exec.LOCK_EX | _fcntl_exec.LOCK_NB)
    except (IOError, OSError):
        lock_fd.close()
        return jsonify({"ok": True, "data": {"processed": 0, "message": "another absorption in progress"}})
    try:
        conn = local_db.get_conn()
        items = conn.execute(
            "SELECT id, error_msg FROM learn_queue WHERE status='pending_learn' ORDER BY priority DESC, score DESC LIMIT 3"
        ).fetchall()
        conn.close()
        if not items:
            return jsonify({"ok": True, "data": {"processed": 0, "message": "无待学习项"}})
        import re as _re_exec
        skipped = 0
        MAX_RETRY = 5
        # 并发安全：一次只处理 1 条（原来的 for 循环会 spawn 多个后台 hermes chat，撑爆内存）
        row = items[0]
        rid = row["id"]
        error_msg = row["error_msg"] or ""
        # ── HydraMind Gate: pre-execution attempt limit ──
        attempts = [int(m) for m in _re_exec.findall(r"attempt=(\d+)", error_msg)]
        max_att = max(attempts) if attempts else 0
        if max_att >= MAX_RETRY:
            gate_trace = (
                f"[node=executor,attempt={max_att+1},gate=MAX_RETRY_EXCEEDED,"
                f"action={GateAction.REPLAN}] "
                f"已重试 {max_att} 次，超过上限 {MAX_RETRY}，需人工判断")
            local_db.learn_update_status(rid, "failed", gate_trace)
            return jsonify({"ok": True, "data": {"processed": 0, "skipped": 1,
                "message": f"task {rid} exceeded {MAX_RETRY} retries, moved to failed"}})
        _run_absorb(rid)
        more = len(items) - 1
        return jsonify({"ok": True, "data": {"processed": 1, "more_pending": more}})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500
    finally:
        try:
            _fcntl_exec.flock(lock_fd, _fcntl_exec.LOCK_UN)
            lock_fd.close()
        except:
            pass


@app.route("/api/learn/articles")
def api_learn_articles():
    """返回分类器处理过的所有文章及评分"""
    try:
        conn = local_db.get_conn()
        rows = conn.execute("""
            SELECT id, title, learn_type, learn_target, score, confidence, status,
                   created_at, reviewed_at, error_msg,
                   (SELECT result_summary FROM learning_results lr WHERE lr.learn_queue_id = learn_queue.id LIMIT 1) as result_summary,
                   (SELECT created_skill FROM learning_results lr WHERE lr.learn_queue_id = learn_queue.id LIMIT 1) as created_skill
            FROM learn_queue
            ORDER BY score DESC, created_at DESC
        """).fetchall()
        conn.close()
        return jsonify({"ok": True, "data": [dict(r) for r in rows]})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/learn/stats")
def api_learn_stats():
    try:
        conn = local_db.get_conn()
        pending = conn.execute("SELECT COUNT(*) FROM learn_queue WHERE status='pending_review'").fetchone()[0]
        pending_learn = conn.execute("SELECT COUNT(*) FROM learn_queue WHERE status='pending_learn'").fetchone()[0]
        approved = conn.execute("SELECT COUNT(*) FROM learn_queue WHERE status='approved'").fetchone()[0]
        applied = conn.execute("SELECT COUNT(*) FROM learn_queue WHERE status='applied'").fetchone()[0]
        rejected = conn.execute("SELECT COUNT(*) FROM learn_queue WHERE status='rejected'").fetchone()[0]
        failed = conn.execute("SELECT COUNT(*) FROM learn_queue WHERE status='failed'").fetchone()[0]
        unclassified = len(local_db.learn_get_unclassified_summaries(999))
        high = conn.execute("SELECT COUNT(*) FROM learn_queue WHERE score >= 7 AND status IN ('pending_review','pending_learn')").fetchone()[0]
        results_count = conn.execute("SELECT COUNT(*) FROM learning_results").fetchone()[0]
        conn.close()
        return jsonify({"ok": True, "data": {
            "pending": pending, "pending_learn": pending_learn,
            "approved": approved, "applied": applied,
            "rejected": rejected, "failed": failed,
            "unclassified": unclassified, "high_score": high,
            "results_count": results_count
        }})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# ── HydraMind Layer 4: Gate Audit (过程评测闭环) ──
@app.route("/api/learn/gate-audit")
def api_learn_gate_audit():
    """扫描 learn_queue 的 error_msg 字段，提取所有 gate trace 四元组，产出聚合统计。
    用于离线评估：识别失败模式 → 修订 Gate 规则 → 回归测试。
    """
    try:
        import re
        conn = local_db.get_conn()
        rows = conn.execute(
            "SELECT id, status, error_msg FROM learn_queue WHERE error_msg IS NOT NULL AND error_msg != ''"
        ).fetchall()
        conn.close()

        # Parse gate traces: [node=X,attempt=N,gate=RC,action=A]
        trace_pat = re.compile(
            r'\[node=([^,]+),attempt=(\d+),gate=([^,]+),action=([^\]]+)\]'
        )
        gate_stats = {}  # {reason_code: {count, actions: {action: count}, nodes: {node: count}}}
        item_traces = {}  # {item_id: [traces]}
        total_gates = 0

        for row in rows:
            rid, status, msg = row["id"], row["status"], row["error_msg"] or ""
            matches = trace_pat.findall(msg)
            if not matches:
                continue
            item_traces[rid] = []
            for node, att, rc, action in matches:
                total_gates += 1
                item_traces[rid].append({
                    "node": node.strip(),
                    "attempt": int(att),
                    "gate": rc.strip(),
                    "action": action.strip()
                })
                key = rc.strip()
                if key not in gate_stats:
                    gate_stats[key] = {"count": 0, "actions": {}, "nodes": {}}
                gs = gate_stats[key]
                gs["count"] += 1
                gs["actions"][action.strip()] = gs["actions"].get(action.strip(), 0) + 1
                gs["nodes"][node.strip()] = gs["nodes"].get(node.strip(), 0) + 1

        # Layer 4 health indicators
        has_offline_eval = False  # no automated eval pipeline yet
        has_regression = False    # no regression tests for gate rules
        has_feedback_loop = False # gate rules are hardcoded, don't evolve
        has_graph_loop_harness = False  # Graph/Loop/Harness dimensions not yet explicitly audited
        has_edge_audit = False    # data/permission/verification/recovery edges not systematically checked

        return jsonify({"ok": True, "data": {
            "total_items_with_traces": len(item_traces),
            "total_gate_events": total_gates,
            "gate_stats_by_reason_code": gate_stats,
            "layer4_health": {
                "offline_evaluation": has_offline_eval,
                "regression_testing": has_regression,
                "feedback_loop": has_feedback_loop,
                "graph_loop_harness_explicit": has_graph_loop_harness,
                "edge_audit": has_edge_audit,
                "score": "3/10 — traces exist but no automated evaluation, feedback loop, or Graph/Loop/Harness edge audit"
            },
            "recommendation": "建议：1) 加 cron job 定期调此 API → 分析 TOP N 失败 reason_code → 人工审核 gate 规则是否需要修订 → 回归验证；"
                             "2) 在 Step 6 审计中显式拆分 Graph/Loop/Harness 和四类边；"
                             "3) 将 gate trace 结构化存储（DB schema 新增 node/attempt/reason_code/action 字段）以便离线评估。"
        }})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# ── Doctor: 全系统健康自检（Inspired by agent-reach `doctor`）──
# agent-reach 的 doctor 命令一次性诊断所有平台通道状态。
# Hermes 对等物：跨技能/DB/预算/内存/cron 的全系统健康检查，
# 一行 curl 即可判断全系统是否健康。
@app.route("/api/doctor")
def api_doctor():
    """全系统健康自检。返回各子系统状态和聚合健康评分。
    用法: curl http://127.0.0.1:5001/api/doctor
    """
    import os as _os_doctor, json as _json_doctor
    checks = {}
    all_ok = True

    # 1. DB 连通性
    try:
        conn = local_db.get_conn()
        conn.execute("SELECT 1")
        conn.close()
        checks["db"] = {"ok": True, "detail": "SQLite reachable"}
    except Exception as e:
        checks["db"] = {"ok": False, "detail": str(e)[:100]}
        all_ok = False

    # 2. Skill 目录完整性
    try:
        skill_dir = Path("/data/data/com.termux/files/home/.hermes/skills")
        if skill_dir.exists():
            skills = list(skill_dir.rglob("*/SKILL.md"))
            checks["skills"] = {"ok": True, "total": len(skills),
                "detail": f"{len(skills)} skills found"}
        else:
            checks["skills"] = {"ok": False, "detail": "skills directory missing"}
            all_ok = False
    except Exception as e:
        checks["skills"] = {"ok": False, "detail": str(e)[:100]}
        all_ok = False

    # 3. Memory 文件完整性
    try:
        mem_path = Path("/data/data/com.termux/files/home/.hermes/memory.json")
        if mem_path.exists():
            mem_size = mem_path.stat().st_size
            try:
                _json_doctor.loads(mem_path.read_text())
                checks["memory"] = {"ok": True, "size_bytes": mem_size,
                    "detail": "memory.json exists and is valid JSON"}
            except Exception:
                checks["memory"] = {"ok": False, "size_bytes": mem_size,
                    "detail": "memory.json exists but is not valid JSON"}
                all_ok = False
        else:
            checks["memory"] = {"ok": False, "detail": "memory.json missing"}
            all_ok = False
    except Exception as e:
        checks["memory"] = {"ok": False, "detail": str(e)[:100]}
        all_ok = False

    # 4. 预算状态
    try:
        month_key = datetime.now().strftime("%Y-%m")
        budget_state = _load_budget_state()
        month_usage = budget_state.get(month_key, {})
        budget_summary = {}
        for jt, cfg in DEFAULT_BUDGETS.items():
            used = month_usage.get(jt, 0.0)
            pct = round(used / max(cfg.monthly_limit, 0.01) * 100, 1)
            budget_summary[jt] = {"used": used, "limit": cfg.monthly_limit, "pct": pct,
                "exhausted": used >= cfg.monthly_limit}
        checks["budget"] = {"ok": True, "month": month_key,
            "detail": budget_summary}
    except Exception as e:
        checks["budget"] = {"ok": False, "detail": str(e)[:100]}
        all_ok = False

    # 5. 学习队列健康
    try:
        conn = local_db.get_conn()
        pending = conn.execute(
            "SELECT COUNT(*) FROM learn_queue WHERE status IN ('pending_learn','pending_review')"
        ).fetchone()[0]
        failed = conn.execute(
            "SELECT COUNT(*) FROM learn_queue WHERE status='failed'"
        ).fetchone()[0]
        applied_7d = conn.execute(
            "SELECT COUNT(*) FROM learn_queue WHERE status='applied' AND reviewed_at > datetime('now','-7 days')"
        ).fetchone()[0]
        conn.close()
        checks["learn_queue"] = {"ok": True, "pending": pending, "failed": failed,
            "applied_7d": applied_7d,
            "detail": f"pending={pending}, failed={failed}, applied(7d)={applied_7d}"}
        if failed > 10:
            checks["learn_queue"]["ok"] = False
            checks["learn_queue"]["detail"] += " — WARN: >10 failed items"
            all_ok = False
    except Exception as e:
        checks["learn_queue"] = {"ok": False, "detail": str(e)[:100]}
        all_ok = False

    # 6. Cron 调度器状态（检查 cron 是否有最近执行记录）
    try:
        cron_log_dir = Path("/data/data/com.termux/files/home/tmp/cron_logs")
        recent_runs = 0
        if cron_log_dir.exists():
            cutoff = time.time() - 3600  # 1 hour
            for f in cron_log_dir.iterdir():
                if f.is_file() and f.stat().st_mtime > cutoff:
                    recent_runs += 1
        checks["cron"] = {"ok": recent_runs > 0,
            "detail": f"{recent_runs} log files written in last hour" if recent_runs > 0
            else "no cron output in last hour — scheduler may be down"}
        if recent_runs == 0:
            all_ok = False
    except Exception as e:
        checks["cron"] = {"ok": False, "detail": str(e)[:100]}
        all_ok = False

    # 7. Hermes CLI 可达性
    try:
        result = subprocess.run(["hermes", "--version"], capture_output=True,
            text=True, timeout=10)
        checks["hermes_cli"] = {"ok": result.returncode == 0,
            "detail": result.stdout.strip()[:80] if result.returncode == 0
            else f"exit={result.returncode} {result.stderr[:80]}"}
        if result.returncode != 0:
            all_ok = False
    except FileNotFoundError:
        checks["hermes_cli"] = {"ok": False, "detail": "hermes binary not found in PATH"}
        all_ok = False
    except Exception as e:
        checks["hermes_cli"] = {"ok": False, "detail": str(e)[:100]}
        all_ok = False

    # 8. 模型 API（快速连通性检查 — 仅检查配置文件，不打实际 API 调用）
    try:
        config_path = Path("/data/data/com.termux/files/home/.hermes/config.yaml")
        if config_path.exists():
            checks["model_api"] = {"ok": True, "detail": "config.yaml exists"}
        else:
            alt_config = Path("/data/data/com.termux/files/home/.hermes/config.json")
            if alt_config.exists():
                checks["model_api"] = {"ok": True, "detail": "config.json exists"}
            else:
                checks["model_api"] = {"ok": False, "detail": "no config file found — model API may be unreachable"}
                all_ok = False
    except Exception as e:
        checks["model_api"] = {"ok": False, "detail": str(e)[:100]}
        all_ok = False

    return jsonify({
        "ok": all_ok,
        "timestamp": datetime.now().isoformat(),
        "checks": checks,
        "summary": f"{sum(1 for c in checks.values() if c['ok'])}/{len(checks)} subsystems healthy",
        "gate_trace": f"[node=doctor,attempt=1,gate={'PASS' if all_ok else 'PARTIAL_FAIL'},"
                      f"action={'CONTINUE' if all_ok else 'REPLAN'}]"
    })


# ── Page: 学习结果 ─────────────────────────────────

@app.route("/learn-results")
def learn_results_page():
    return render_template("learn_results.html")


@app.route("/api/learn-results")
def api_learn_results():
    try:
        results = local_db.learn_result_list(limit=200)
        return jsonify({"ok": True, "data": results})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/learn-result/<int:item_id>/feedback", methods=["POST"])
def api_learn_result_feedback(item_id):
    try:
        data = request.get_json()
        feedback = (data or {}).get("feedback", "")
        if not feedback:
            return jsonify({"ok": False, "error": "feedback is required"}), 400
        local_db.learn_result_add_feedback(item_id, feedback)
        return jsonify({"ok": True})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# ── Page: 自动回复白名单 ─────────────────────────────

@app.route("/whitelist")
def whitelist_page():
    return render_template("whitelist.html")


@app.route("/api/whitelist")
def api_whitelist_list():
    try:
        return jsonify({"ok": True, "data": local_db.whitelist_list()})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/whitelist", methods=["POST"])
def api_whitelist_add():
    """支持单条 {wxid,name,kind,role} 或批量 {items:[...]}（弹窗多选）。"""
    try:
        d = request.json or {}
        items = d.get("items") if isinstance(d.get("items"), list) else [d]
        n = 0
        for it in items:
            wxid = (it.get("wxid") or "").strip()
            if not wxid:
                continue
            kind = "group" if wxid.endswith("@chatroom") else (it.get("kind") or "contact")
            local_db.whitelist_add(kind, wxid, (it.get("name") or "").strip(),
                                   role=(it.get("role") or "member"))
            n += 1
        if n == 0:
            return jsonify({"ok": False, "error": "没有有效条目"}), 400
        return jsonify({"ok": True, "data": {"added": n}})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


POLICY_CFG = "/data/data/com.termux/files/home/.hermes/wechat_reply/policy_config.json"


def _read_auto_send():
    """读策略总开关。文件 root 属主但 644，termux 用户可直接读。"""
    try:
        with open(POLICY_CFG) as f:
            return bool(json.load(f).get("auto_send_enabled", False))
    except Exception:
        return False


def _write_auto_send(enabled):
    """只翻转 auto_send_enabled，保留其余字段。文件 root 属主，viewer 经 su 写回。"""
    try:
        with open(POLICY_CFG) as f:
            cfg = json.load(f)
    except Exception:
        cfg = {}
    cfg["auto_send_enabled"] = bool(enabled)
    blob = json.dumps(cfg, ensure_ascii=False, indent=2)
    tmp = "/data/data/com.termux/files/home/.hermes/wechat_reply/.policy_config.tmp"
    with open(tmp, "w") as f:
        f.write(blob)
    # viewer 是 termux 用户无权直接覆盖 root 属主文件 → 写 tmp 再 su cp（沿用本 app su 模式）
    subprocess.run(["su", "-c", f"cp {tmp} {POLICY_CFG}"],
                   capture_output=True, timeout=8)
    try:
        os.remove(tmp)
    except OSError:
        pass


@app.route("/api/auto_send")
def api_auto_send_get():
    return jsonify({"ok": True, "enabled": _read_auto_send()})


@app.route("/api/auto_send", methods=["POST"])
def api_auto_send_set():
    """总开关：True=白名单可自动发（仍受敏感词等硬规则拦截），False=一切进人工审批。"""
    try:
        d = request.json or {}
        _write_auto_send(bool(d.get("enabled")))
        return jsonify({"ok": True, "enabled": _read_auto_send()})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


def _talker_row(wxid, name, ttype):
    return {"wxid": wxid, "name": name or wxid,
            "kind": "group" if ttype == "group" else "contact"}


@app.route("/api/contacts")
def api_contacts():
    """弹窗默认列表：会话中出现过的联系人 + 群（排除公众号），去重带昵称，按最近排序。"""
    try:
        sql = """
        SELECT wxid, name, talker_type FROM (
          SELECT m.talker AS wxid,
            CASE WHEN m.talker LIKE '%@chatroom'
                 THEN COALESCE(NULLIF(r.nickname,''), c.displayname, m.talker)
                 ELSE COALESCE(NULLIF(r.conRemark,''), NULLIF(r.nickname,''), m.talker) END AS name,
            CASE WHEN m.talker LIKE '%@chatroom' THEN 'group'
                 WHEN m.talker LIKE 'gh_%' THEN 'official' ELSE 'personal' END AS talker_type,
            MAX(m.createTime) AS t
          FROM message m
          LEFT JOIN rcontact r ON m.talker = r.username
          LEFT JOIN chatroom c ON m.talker = c.chatroomname
          GROUP BY m.talker
        ) WHERE talker_type != 'official' ORDER BY t DESC LIMIT 500
        """
        rows = wechat_db.query(sql)
        return jsonify({"ok": True, "data": [
            _talker_row(r.get("wxid"), r.get("name"), r.get("talker_type")) for r in rows]})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/contacts/search")
def api_contacts_search():
    """按名字/wxid 搜全量 rcontact + 群（弹窗搜索框）。"""
    try:
        q = (request.args.get("q") or "").strip().replace("'", "").replace('"', "")
        if not q:
            return jsonify({"ok": True, "data": []})
        rows = wechat_db.search_talker(q, limit=50)
        out = [_talker_row(r.get("talker"), r.get("display_name"), r.get("talker_type"))
               for r in rows if r.get("talker_type") != "official"]
        return jsonify({"ok": True, "data": out})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/whitelist/<int:item_id>/delete", methods=["POST"])
def api_whitelist_delete(item_id):
    try:
        local_db.whitelist_remove(item_id)
        return jsonify({"ok": True})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/whitelist/<int:item_id>/role", methods=["POST"])
def api_whitelist_set_role(item_id):
    """设置白名单角色：{"role": "admin"|"member"}"""
    try:
        role = (request.json or {}).get("role", "member")
        if role not in ("admin", "member"):
            return jsonify({"ok": False, "error": "role 必须是 admin 或 member"}), 400
        local_db.whitelist_set_role(item_id, role)
        return jsonify({"ok": True})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# ── Page: 微信自动回复审批 ───────────────────────────

@app.route("/approvals")
def approvals_page():
    return render_template("approvals.html")


@app.route("/api/replies")
def api_replies_list():
    """待审批 + 最近已处理（发送/自动/取消）。"""
    try:
        status = request.args.get("status", "")
        return jsonify({"ok": True, "data": local_db.reply_list(status, limit=200)})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


def _send_wechat_text(to_wxid: str, text: str) -> dict:
    import sys
    sys.path.insert(0, "/data/data/com.termux/files/home/.hermes/event_bus")
    from wechat_hook_send import send_text_via_hook
    return send_text_via_hook(to_wxid, text)


@app.route("/api/reply/<int:item_id>/send", methods=["POST"])
def api_reply_send(item_id):
    """人工点发送：用（可编辑后的）草稿发到该会话，标记 sent。"""
    try:
        item = local_db.reply_get(item_id)
        if not item:
            return jsonify({"ok": False, "error": "not found"}), 404
        text = ((request.json or {}).get("text") or item.get("draft") or "").strip()
        if not text:
            return jsonify({"ok": False, "error": "回复内容为空"}), 400
        to = item.get("talker") or ""
        res = _send_wechat_text(to, text)
        ok = bool(res.get("ok") or res.get("success") or res.get("status") == "sent")
        local_db.reply_update(item_id, "sent" if ok else "failed", draft=text)
        return jsonify({"ok": ok, "data": res})
    except Exception as e:
        try:
            local_db.reply_update(item_id, "failed")
        except Exception:
            pass
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/reply/<int:item_id>/cancel", methods=["POST"])
def api_reply_cancel(item_id):
    try:
        local_db.reply_update(item_id, "cancelled")
        return jsonify({"ok": True})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# ── AI 聊天 API ──────────────────────────────────────

def call_hermes(user_input: str, timeout: int = 120) -> str:
    """调用 Hermes CLI 生成 AI 回复。

    Args:
        user_input: 用户输入文本（可包含历史上下文）
        timeout: 超时秒数（默认 120）

    Returns:
        Hermes 的 stdout 输出，或错误提示字符串
    """
    try:
        r = subprocess.run(
            ["hermes", "chat", "-q", user_input, "-Q"],
            capture_output=True, text=True,
            timeout=timeout,
            cwd="/data/data/com.termux/files/home"
        )
        ai_reply = (r.stdout or "").strip()
        if not ai_reply and r.stderr:
            ai_reply = f"[Hermes 错误] {r.stderr[:200]}"
        elif not ai_reply:
            ai_reply = "[Hermes 返回空回复]"
        return ai_reply
    except subprocess.TimeoutExpired:
        return "[Hermes:gate=TIMEOUT] 调用超时"
    except FileNotFoundError:
        return "[Hermes:gate=CMD_NOT_FOUND] 命令未找到"
    except Exception as hermes_err:
        return f"[Hermes:gate=RUNTIME_ERR] 调用失败: {str(hermes_err)[:200]}"


def call_hermes_stream(user_input: str, timeout: int = 120, profile: str = ""):
    """调用 Hermes CLI 并以生成器方式流式输出。

    使用 subprocess.Popen 逐行读取 stdout，yield 每个文本块。

    Args:
        user_input: 用户输入文本（可包含历史上下文）
        timeout: 超时秒数（默认 120）
        profile: Hermes profile 名称（空字符串=默认 default profile）

    Yields:
        str: 文本块（逐行或逐词）
    """
    from threading import Timer
    proc = None
    try:
        cmd = ["hermes"]
        if profile and profile != "default":
            cmd.extend(["-p", profile])
        cmd.extend(["chat", "-q", user_input, "-Q"])
        proc = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE, stderr=subprocess.PIPE,
            text=True, cwd="/data/data/com.termux/files/home"
        )
        # Read line by line with timeout
        start_time = time.time()
        for line in proc.stdout:
            elapsed = time.time() - start_time
            if elapsed > timeout:
                proc.kill()
                yield "\n[Hermes:gate=TIMEOUT] 调用超时"
                return
            if line:
                yield line
        # Check for stderr if no stdout
        if proc.poll() is not None and proc.returncode != 0:
            err = (proc.stderr.read() or "")[:200]
            if err:
                yield f"\n[Hermes 错误: {err}]"
    except FileNotFoundError:
        yield "[Hermes:gate=CMD_NOT_FOUND] 命令未找到"
    except Exception as e:
        yield f"\n[Hermes:gate=RUNTIME_ERR] 调用失败: {str(e)[:200]}]"
    finally:
        if proc is not None:
            try:
                proc.kill()
            except Exception:
                pass


@app.route("/api/conversations/<int:conv_id>/stream")
def api_conversations_stream(conv_id):
    """SSE 流式对话：发送用户消息并流式返回 AI 回复。"""
    from flask import Response

    conv = local_db.conv_get(conv_id)
    if not conv:
        return jsonify({"ok": False, "error": "会话不存在"}), 404

    message = request.args.get("message", "").strip()
    if not message:
        return jsonify({"ok": False, "error": "需要 message 参数"}), 400

    agent = request.args.get("agent", "default").strip()

    # 保存用户消息
    local_db.msg_add(conv_id, "user", message)

    # 如果是首条消息，自动更新标题
    if local_db.msg_count(conv_id) == 1:
        title = message[:30] + ("..." if len(message) > 30 else "")
        local_db.conv_update_title(conv_id, title)

    # 构建上下文
    history = local_db.msg_list(conv_id)
    context_lines = []
    for msg in history[:-1]:  # exclude just-added user message
        role_label = "用户" if msg["role"] == "user" else "助手"
        context_lines.append(f"{role_label}: {msg['content']}")

    hermes_input = message
    if context_lines:
        hermes_input = "\n".join(context_lines[-10:]) + f"\n用户: {message}"

    def generate():
        """SSE generator：流式输出 AI 回复并累积全文。"""
        full_reply = []
        try:
            for chunk in call_hermes_stream(hermes_input, profile=agent):
                full_reply.append(chunk)
                yield f"data: {chunk}\n\n"
            yield "data: [DONE]\n\n"
        except GeneratorExit:
            pass
        finally:
            # 流结束后保存完整回复到 DB
            reply_text = "".join(full_reply).strip()
            if reply_text:
                try:
                    local_db.msg_add(conv_id, "assistant", reply_text)
                except Exception:
                    pass

    return Response(
        generate(),
        mimetype="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        }
    )


@app.route("/api/conversations")
def api_conversations():
    """列出所有会话，按 updated_at 倒序"""
    try:
        convs = local_db.conv_list()
        return jsonify({"ok": True, "data": convs})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/conversations", methods=["POST"])
def api_conversations_create():
    """新建会话"""
    try:
        data = request.get_json(silent=True) or {}
        title = data.get("title", "新对话")
        cid = local_db.conv_create(title)
        conv = local_db.conv_get(cid)
        return jsonify({"ok": True, "data": conv}), 201
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/conversations/<int:conv_id>", methods=["DELETE"])
def api_conversations_delete(conv_id):
    """删除会话（级联删除消息）"""
    try:
        conv = local_db.conv_get(conv_id)
        if not conv:
            return jsonify({"ok": False, "error": "会话不存在"}), 404
        local_db.conv_delete(conv_id)
        return jsonify({"ok": True})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/conversations/<int:conv_id>", methods=["PUT"])
def api_conversations_rename(conv_id):
    """重命名会话"""
    try:
        conv = local_db.conv_get(conv_id)
        if not conv:
            return jsonify({"ok": False, "error": "会话不存在"}), 404
        data = request.get_json(silent=True) or {}
        title = (data.get("title") or "").strip()
        if not title:
            return jsonify({"ok": False, "error": "标题不能为空"}), 400
        local_db.conv_update_title(conv_id, title)
        return jsonify({"ok": True})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/conversations/<int:conv_id>/messages")
def api_messages_list(conv_id):
    """获取某会话的所有消息（按 created_at 正序）"""
    try:
        conv = local_db.conv_get(conv_id)
        if not conv:
            return jsonify({"ok": False, "error": "会话不存在"}), 404
        msgs = local_db.msg_list(conv_id)
        return jsonify({"ok": True, "data": msgs})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/conversations/<int:conv_id>/messages", methods=["POST"])
def api_messages_send(conv_id):
    """发送用户消息并获取 AI 回复"""
    try:
        conv = local_db.conv_get(conv_id)
        if not conv:
            return jsonify({"ok": False, "error": "会话不存在"}), 404

        data = request.get_json(silent=True) or {}
        content = data.get("content", "").strip()
        if not content:
            return jsonify({"ok": False, "error": "消息内容不能为空"}), 400

        # 保存用户消息
        user_msg_id = local_db.msg_add(conv_id, "user", content)

        # 如果是首条消息，自动更新标题
        if local_db.msg_count(conv_id) == 1:
            title = content[:30] + ("..." if len(content) > 30 else "")
            local_db.conv_update_title(conv_id, title)

        # 构建 Hermes 上下文（历史消息）
        history = local_db.msg_list(conv_id)
        context_lines = []
        for msg in history[:-1]:  # exclude the just-added user message
            role_label = "用户" if msg["role"] == "user" else "助手"
            context_lines.append(f"{role_label}: {msg['content']}")

        # 构建 Hermes 上下文
        hermes_input = content
        if context_lines:
            hermes_input = "\n".join(context_lines[-10:]) + f"\n用户: {content}"

        # 调用 Hermes（使用封装的 call_hermes 函数）
        ai_reply = call_hermes(hermes_input)

        # 保存 AI 回复
        ai_msg_id = local_db.msg_add(conv_id, "assistant", ai_reply)

        return jsonify({
            "ok": True,
            "data": {
                "user_message": {"id": user_msg_id, "role": "user", "content": content},
                "assistant_message": {"id": ai_msg_id, "role": "assistant", "content": ai_reply}
            }
        }), 201
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# ── S6: 消息编辑 ──

@app.route("/api/conversations/<int:conv_id>/messages/<int:msg_id>", methods=["PUT"])
def api_messages_edit(conv_id, msg_id):
    """编辑消息内容（仅限 user 角色）"""
    try:
        conv = local_db.conv_get(conv_id)
        if not conv:
            return jsonify({"ok": False, "error": "会话不存在"}), 404

        msg = local_db.msg_get(msg_id)
        if not msg or msg["conversation_id"] != conv_id:
            return jsonify({"ok": False, "error": "消息不存在"}), 404

        if msg["role"] != "user":
            return jsonify({"ok": False, "error": "只能编辑用户消息"}), 400

        data = request.get_json(silent=True) or {}
        content = (data.get("content") or "").strip()
        if not content:
            return jsonify({"ok": False, "error": "内容不能为空"}), 400

        local_db.msg_update(msg_id, content)
        local_db.conv_update_time(conv_id)
        return jsonify({"ok": True, "data": {"id": msg_id, "content": content}})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# ── S6: 重新生成 ──

@app.route("/api/conversations/<int:conv_id>/regenerate", methods=["POST"])
def api_regenerate(conv_id):
    """重新生成 AI 回复：删除最后一条 assistant 消息 → 重新调 LLM → SSE 流式返回"""
    try:
        conv = local_db.conv_get(conv_id)
        if not conv:
            return jsonify({"ok": False, "error": "会话不存在"}), 404

        # 找到最后一条消息
        msgs = local_db.msg_list(conv_id)
        if not msgs:
            return jsonify({"ok": False, "error": "没有消息可以重新生成"}), 400

        last_msg = msgs[-1]
        if last_msg["role"] != "assistant":
            return jsonify({"ok": False, "error": "最后一条不是 AI 回复"}), 400

        # 找到对应的用户消息（最后一条 assistant 之前的第一条 user）
        last_user_content = ""
        for m in reversed(msgs[:-1]):
            if m["role"] == "user":
                last_user_content = m["content"]
                break

        if not last_user_content:
            return jsonify({"ok": False, "error": "没有找到对应的用户消息"}), 400

        # 删除最后一条 assistant 消息
        local_db.msg_delete_since(conv_id, last_msg["id"] - 1)

        # 重新构建上下文并调 Hermes
        history = local_db.msg_list(conv_id)
        context_lines = []
        for msg in history:
            role_label = "用户" if msg["role"] == "user" else "助手"
            context_lines.append(f"{role_label}: {msg['content']}")

        hermes_input = "\n".join(context_lines[-10:]) + f"\n用户: {last_user_content}"

        def generate():
            proc = subprocess.Popen(
                ["hermes", "ask", hermes_input],
                stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
                env={**os.environ, "NO_COLOR": "1"}
            )
            full_reply = ""
            if proc.stdout:
                for line in proc.stdout:
                    line = line.rstrip("\n")
                    if line:
                        full_reply += line
                        yield f"data: {line}\n\n"
            proc.wait()
            # Save the new AI reply
            if full_reply.strip():
                local_db.msg_add(conv_id, "assistant", full_reply.strip())
            local_db.conv_update_time(conv_id)
            yield "data: [DONE]\n\n"

        return Response(
            generate(),
            mimetype="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "X-Accel-Buffering": "no",
                "Connection": "keep-alive",
            }
        )
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# ── Dashboard Page ──────────────────────────────────

@app.route("/dashboard")
def dashboard_page():
    return render_template("dashboard.html")


# ── Dashboard API ────────────────────────────────────

@app.route("/api/dashboard")
def api_dashboard():
    """Aggregate metrics from all modules for the data dashboard."""
    try:
        conn = local_db.get_conn()
        data = {
            "chats": {},
            "articles": {},
            "learn": {},
            "reports": {},
            "jobs": {},
            "auto_reply": {},
            "circle": {},
        }

        # ── Chats (WeChat DB + local) ──
        try:
            chats_rows = wechat_db.query(
                "SELECT COUNT(DISTINCT talker) as groups_count FROM message WHERE talker LIKE '%@chatroom'")
            data["chats"]["groups"] = chats_rows[0]["groups_count"] if chats_rows else 0
            msg_rows = wechat_db.query("SELECT COUNT(*) as cnt FROM message")
            data["chats"]["messages"] = msg_rows[0]["cnt"] if msg_rows else 0
            speaker_rows = wechat_db.query("SELECT COUNT(DISTINCT talker) as cnt FROM message")
            data["chats"]["speakers"] = speaker_rows[0]["cnt"] if speaker_rows else 0
        except Exception:
            data["chats"]["groups"] = None
            data["chats"]["messages"] = None
            data["chats"]["speakers"] = None

        data["chats"]["shares"] = conn.execute(
            "SELECT COUNT(*) FROM article_shares").fetchone()[0]

        # ── Articles ──
        total_articles = conn.execute("SELECT COUNT(*) FROM articles").fetchone()[0]
        summarized = conn.execute(
            "SELECT COUNT(DISTINCT article_url) FROM daily_reports WHERE report_date='summary' AND article_url IS NOT NULL"
        ).fetchone()[0]
        learned = conn.execute(
            "SELECT COUNT(DISTINCT article_url) FROM daily_reports WHERE report_date='learn' AND article_url IS NOT NULL"
        ).fetchone()[0]
        data["articles"] = {
            "total": total_articles,
            "summarized": summarized,
            "unsummarized": max(0, total_articles - summarized),
            "learned": learned,
            "unlearned": max(0, total_articles - learned),
        }

        # ── Learn ──
        data["learn"] = {
            "pending_review": conn.execute(
                "SELECT COUNT(*) FROM learn_queue WHERE status='pending_review'").fetchone()[0],
            "approved": conn.execute(
                "SELECT COUNT(*) FROM learn_queue WHERE status='approved'").fetchone()[0],
            "applied": conn.execute(
                "SELECT COUNT(*) FROM learn_queue WHERE status='applied'").fetchone()[0],
            "rejected": conn.execute(
                "SELECT COUNT(*) FROM learn_queue WHERE status='rejected'").fetchone()[0],
            "unclassified": conn.execute(
                """SELECT COUNT(*) FROM daily_reports dr
                   WHERE dr.report_date='summary' AND dr.dedup_key IS NOT NULL
                   AND dr.dedup_key NOT IN (
                       SELECT DISTINCT lq.dedup_key FROM learn_queue lq WHERE lq.dedup_key IS NOT NULL
                   )""").fetchone()[0],
            "by_type": {
                "skill": conn.execute(
                    "SELECT COUNT(*) FROM learn_queue WHERE status='applied' AND learn_type='skill'").fetchone()[0],
                "memory": conn.execute(
                    "SELECT COUNT(*) FROM learn_queue WHERE status='applied' AND learn_type='memory'").fetchone()[0],
                "todo": conn.execute(
                    "SELECT COUNT(*) FROM learn_queue WHERE status='applied' AND learn_type='todo'").fetchone()[0],
                "reference": conn.execute(
                    "SELECT COUNT(*) FROM learn_queue WHERE status='applied' AND learn_type='reference'").fetchone()[0],
            }
        }

        # ── Reports ──
        last_report = conn.execute(
            "SELECT created_at FROM daily_reports WHERE report_date NOT IN ('summary','learn') ORDER BY id DESC LIMIT 1"
        ).fetchone()
        data["reports"] = {
            "daily": conn.execute(
                "SELECT COUNT(*) FROM daily_reports WHERE report_date NOT IN ('summary','learn')").fetchone()[0],
            "summaries": conn.execute(
                "SELECT COUNT(*) FROM daily_reports WHERE report_date='summary'").fetchone()[0],
            "learn_records": conn.execute(
                "SELECT COUNT(*) FROM daily_reports WHERE report_date='learn'").fetchone()[0],
            "images": conn.execute("SELECT COUNT(*) FROM report_images").fetchone()[0],
            "last_generated": last_report["created_at"] if last_report else None,
        }

        # ── Jobs ──
        jobs_total = conn.execute("SELECT COUNT(*) FROM hermes_jobs").fetchone()[0]
        jobs_completed = conn.execute(
            "SELECT COUNT(*) FROM hermes_jobs WHERE status='completed'").fetchone()[0]
        jobs_failed = conn.execute(
            "SELECT COUNT(*) FROM hermes_jobs WHERE status='failed'").fetchone()[0]
        jobs_running = conn.execute(
            "SELECT COUNT(*) FROM hermes_jobs WHERE status='running'").fetchone()[0]
        last_job = conn.execute(
            "SELECT completed_at FROM hermes_jobs WHERE completed_at IS NOT NULL ORDER BY id DESC LIMIT 1"
        ).fetchone()
        done_total = jobs_completed + jobs_failed
        data["jobs"] = {
            "total": jobs_total,
            "running": jobs_running,
            "completed": jobs_completed,
            "failed": jobs_failed,
            "success_rate": round(jobs_completed / done_total * 100, 1) if done_total > 0 else 0,
            "last_completed": last_job["completed_at"] if last_job else None,
        }

        # ── Auto Reply ──
        data["auto_reply"] = {
            "whitelist": conn.execute("SELECT COUNT(*) FROM reply_whitelist").fetchone()[0],
            "pending": conn.execute(
                "SELECT COUNT(*) FROM reply_queue WHERE status='pending'").fetchone()[0],
            "sent": conn.execute(
                "SELECT COUNT(*) FROM reply_queue WHERE status IN ('sent','auto_sent')").fetchone()[0],
        }

        # ── Circle (AI Chat) ──
        data["circle"] = {
            "conversations": conn.execute("SELECT COUNT(*) FROM conversations").fetchone()[0],
            "messages": conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0],
        }

        conn.close()
        return jsonify({"ok": True, "data": data})
    except Exception as e:
        import traceback
        traceback.print_exc()
        return jsonify({"ok": False, "error": str(e)}), 500


# ── AI Chat Page ────────────────────────────────────

@app.route("/chat")
def ai_chat_page():
    """AI 聊天页面：左侧会话列表 + 右侧消息区域（无 conv_id，默认最新或新建）"""
    return render_template("aichat.html")


@app.route("/ai-chat/<int:conv_id>")
def ai_chat_page_with_id(conv_id):
    """AI 聊天页面：带会话 ID，可直接定位到特定会话"""
    return render_template("aichat.html", initial_conv_id=conv_id)


# ── Cron Management APIs ─────────────────────────────

CRON_JOBS_FILE = Path("/data/data/com.termux/files/home/.hermes/cron/jobs.json")
CRON_OUTPUT_DIR = Path("/data/data/com.termux/files/home/.hermes/cron/output")

_cron_cache = {"ts": 0, "data": []}


def _load_cron_jobs(force=False):
    """Load cron jobs from jobs.json with 30s cache."""
    global _cron_cache
    now = time.time()
    if not force and (now - _cron_cache["ts"]) < 30 and _cron_cache["data"]:
        return _cron_cache["data"]

    if not CRON_JOBS_FILE.exists():
        return []

    try:
        with open(CRON_JOBS_FILE) as f:
            raw = json.load(f)
    except Exception:
        return _cron_cache["data"] if _cron_cache["data"] else []

    jobs = []
    for job in raw.get("jobs", []):
        jobs.append({
            "id": job.get("id", ""),
            "name": job.get("name", ""),
            "schedule": job.get("schedule_display", job.get("schedule", {}).get("display", "")),
            "enabled": job.get("enabled", False),
            "state": job.get("state", "unknown"),
            "last_run": job.get("last_run_at"),
            "last_status": job.get("last_status"),
            "next_run": job.get("next_run_at"),
            "skills": job.get("skills", []),
            "deliver": job.get("deliver", "local"),
            "no_agent": job.get("no_agent", False),
        })
    _cron_cache = {"ts": now, "data": jobs}
    return jobs


@app.route("/cron")
def cron_page():
    """Cron 管理页面"""
    return render_template("cron.html")


@app.route("/api/cron/list")
def api_cron_list():
    """返回所有 cron job 列表"""
    try:
        jobs = _load_cron_jobs()
        return jsonify({"ok": True, "data": jobs})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/cron/<job_id>")
def cron_job_page(job_id):
    """独立全屏时间线日志详情页"""
    jobs = _load_cron_jobs()
    job = next((j for j in jobs if j["id"] == job_id), None)
    if not job:
        return "Job not found", 404
    return render_template("cron_job.html", job=job)


@app.route("/api/cron/logs/<job_id>")
def api_cron_logs(job_id):
    """返回指定 job 的执行历史"""
    try:
        out_dir = CRON_OUTPUT_DIR / job_id
        if not out_dir.is_dir():
            return jsonify({"ok": False, "error": f"Job {job_id} not found or no logs"}),
            404

        logs = []
        for fpath in sorted(out_dir.iterdir(), reverse=True):
            if not fpath.suffix == ".md":
                continue
            stat = fpath.stat()
            try:
                full_content = fpath.read_text(encoding="utf-8", errors="replace")
                summary = full_content[:500] if len(full_content) > 500 else full_content
                if len(full_content) > 500:
                    summary += f"\n\n... (共 {len(full_content)} 字符，截取前 500)"
            except Exception:
                full_content = ""
                summary = "[读取失败]"

            logs.append({
                "timestamp": fpath.stem,
                "filename": fpath.name,
                "content": summary,
                "full_content": full_content,
                "size": stat.st_size,
                "mtime": stat.st_mtime,
            })

        return jsonify({"ok": True, "data": logs})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# ── Page: Skill List ─────────────────────────────────

SKILLS_DIR = Path("/data/data/com.termux/files/home/.hermes/skills")
USAGE_PATH = SKILLS_DIR / ".usage.json"


def _scan_skills():
    """Scan skills directory recursively for SKILL.md files, merge with .usage.json."""
    usage = {}
    if USAGE_PATH.exists():
        try:
            usage = json.loads(USAGE_PATH.read_text())
        except Exception:
            pass

    skills = []
    if SKILLS_DIR.exists():
        # Walk all subdirectories; any dir with a SKILL.md is a skill
        for skill_md in sorted(SKILLS_DIR.rglob("SKILL.md")):
            skill_dir = skill_md.parent
            name = skill_dir.name
            # Skip if this is a category dir (no SKILL.md at this level
            # and we already process child dirs)
            if name == "skills":
                continue
            # Use directory name as skill name
            meta = usage.get(name, {})
            job_count = 0
            try:
                conn = local_db.get_conn()
                row = conn.execute(
                    "SELECT COUNT(*) FROM hermes_jobs WHERE job_name LIKE ?",
                    (f"%{name}%",)
                ).fetchone()
                conn.close()
                job_count = row[0] if row else 0
            except Exception:
                pass

            skills.append({
                "name": name,
                "created_at": meta.get("created_at", ""),
                "use_count": meta.get("use_count", 0),
                "last_used_at": meta.get("last_used_at", ""),
                "state": meta.get("state", "active"),
                "pinned": meta.get("pinned", False),
                "patch_count": meta.get("patch_count", 0),
                "view_count": meta.get("view_count", 0),
                "job_count": job_count,
            })
    return skills


@app.route("/skills")
def skills_page():
    return render_template("skills.html")


@app.route("/skill/<name>")
def skill_detail_page(name):
    return render_template("skill_detail.html", skill_name=name)


@app.route("/api/skills")
def api_skills():
    try:
        skills = _scan_skills()
        return jsonify({"ok": True, "data": skills})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/skill/<name>")
def api_skill_detail(name):
    try:
        # Find SKILL.md recursively (skill may be in category subdir)
        matches = list(SKILLS_DIR.rglob(f"{name}/SKILL.md"))
        if not matches or not matches[0].exists():
            return jsonify({"ok": False, "error": "skill not found"}), 404

        raw = matches[0].read_text()
        # Strip YAML frontmatter
        content = raw
        if raw.startswith("---"):
            parts = raw.split("---", 2)
            if len(parts) >= 3:
                content = parts[2].strip()

        # Get related jobs
        jobs = []
        try:
            conn = local_db.get_conn()
            rows = conn.execute(
                "SELECT * FROM hermes_jobs WHERE job_name LIKE ? ORDER BY id DESC LIMIT 50",
                (f"%{name}%",)
            ).fetchall()
            conn.close()
            jobs = [dict(r) for r in rows]
        except Exception:
            pass

        return jsonify({"ok": True, "data": {
            "name": name,
            "content": content,
            "jobs": jobs,
        }})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/skill/<name>/toggle", methods=["POST"])
def api_skill_toggle(name):
    """Toggle skill state active ↔ disabled in .usage.json."""
    try:
        body = request.get_json(force=True) or {}
        new_state = body.get("state", "")
        if new_state not in ("active", "disabled"):
            return jsonify({"ok": False, "error": "state must be 'active' or 'disabled'"}), 400

        usage = {}
        if USAGE_PATH.exists():
            try:
                usage = json.loads(USAGE_PATH.read_text())
            except Exception:
                pass

        if name not in usage:
            usage[name] = {
                "created_at": datetime.now().isoformat(),
                "use_count": 0,
                "state": new_state,
                "pinned": False,
                "patch_count": 0,
                "last_used_at": None,
                "last_viewed_at": None,
                "view_count": 0,
                "created_by": None,
                "archived_at": None,
            }
        else:
            usage[name]["state"] = new_state

        USAGE_PATH.parent.mkdir(parents=True, exist_ok=True)
        USAGE_PATH.write_text(json.dumps(usage, indent=2, ensure_ascii=False))
        return jsonify({"ok": True, "data": {"name": name, "state": new_state}})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# ─── Agents (multi-profile) ─────────────────────────────────────────────────

HERMES_HOME = Path("/data/data/com.termux/files/home/.hermes")
PROFILES_DIR = HERMES_HOME / "profiles"


def _parse_yaml_simple(text):
    """Parse minimal YAML for model/provider/base_url from config.yaml."""
    result = {}
    in_model = False
    for line in text.splitlines():
        stripped = line.strip()
        if stripped.startswith("model:"):
            in_model = True
            rest = stripped[len("model:"):].strip()
            if rest:
                result["model"] = rest
        elif stripped.startswith("default:") and in_model:
            result["model"] = stripped[len("default:"):].strip()
        elif stripped.startswith("provider:") and in_model:
            result["provider"] = stripped[len("provider:"):].strip()
        elif stripped.startswith("base_url:") and in_model:
            result["base_url"] = stripped[len("base_url:"):].strip()
        elif stripped and not stripped.startswith("-") and not stripped.startswith("#"):
            if not line.startswith(" "):
                in_model = False
    return result


def _scan_profiles():
    """Scan ~/.hermes/ (default) + ~/.hermes/profiles/<name>/ for all profiles."""
    profiles = []

    def _scan_one(profile_dir, name):
        """Scan a single profile directory for metadata."""
        config_path = profile_dir / "config.yaml"
        soul_path = profile_dir / "SOUL.md"
        env_path = profile_dir / ".env"
        memory_path = profile_dir / "memory.json"
        skills_dir = profile_dir / "skills"

        config = {}
        if config_path.exists():
            try:
                config = _parse_yaml_simple(config_path.read_text())
            except Exception:
                pass

        soul_preview = ""
        if soul_path.exists():
            try:
                soul_preview = soul_path.read_text()[:200]
            except Exception:
                pass

        skills_count = 0
        if skills_dir.exists():
            try:
                skills_count = len(list(skills_dir.rglob("SKILL.md")))
            except Exception:
                pass

        sessions_dir = profile_dir / "sessions"
        sessions_count = 0
        if sessions_dir.exists():
            try:
                sessions_count = len(list(sessions_dir.iterdir()))
            except Exception:
                pass

        return {
            "name": name,
            "model": config.get("model", "unknown"),
            "provider": config.get("provider", "unknown"),
            "base_url": config.get("base_url", ""),
            "skills_count": skills_count,
            "memory_exists": memory_path.exists(),
            "memory_size": memory_path.stat().st_size if memory_path.exists() else 0,
            "env_exists": env_path.exists(),
            "soul_preview": soul_preview,
            "soul_exists": soul_path.exists(),
            "sessions_count": sessions_count,
            "description": "",  # filled below if profile.yaml exists
            "is_default": name == "default",
        }

    # Default profile
    if HERMES_HOME.exists():
        info = _scan_one(HERMES_HOME, "default")
        # Try to get description from hermes profile show
        try:
            r = subprocess.run(
                ["hermes", "profile", "show", "default"],
                capture_output=True, text=True, timeout=10
            )
            if r.returncode == 0:
                for line in r.stdout.splitlines():
                    if line.lower().startswith("model:"):
                        parts = line.split(":", 1)
                        if len(parts) > 1:
                            model_part = parts[1].strip()
                            if "(" in model_part:
                                info["model"] = model_part.split("(")[0].strip()
                            else:
                                info["model"] = model_part
        except Exception:
            pass
        profiles.append(info)

    # Other profiles under ~/.hermes/profiles/
    if PROFILES_DIR.exists():
        for p in sorted(PROFILES_DIR.iterdir()):
            if p.is_dir() and not p.name.startswith("."):
                info = _scan_one(p, p.name)
                profiles.append(info)

    return profiles


@app.route("/api/profiles")
def api_profiles():
    try:
        profiles = _scan_profiles()
        return jsonify({"ok": True, "data": profiles})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/profile/<name>")
def api_profile_detail(name):
    try:
        if name == "default":
            profile_dir = HERMES_HOME
        else:
            profile_dir = PROFILES_DIR / name
            if not profile_dir.exists():
                return jsonify({"ok": False, "error": "profile not found"}), 404

        # SOUL.md full text
        soul = ""
        soul_path = profile_dir / "SOUL.md"
        if soul_path.exists():
            try:
                soul = soul_path.read_text()
            except Exception:
                pass

        # Config summary
        config = {}
        config_path = profile_dir / "config.yaml"
        if config_path.exists():
            try:
                config = _parse_yaml_simple(config_path.read_text())
            except Exception:
                pass

        # Skills list
        skills = []
        skills_dir = profile_dir / "skills"
        if skills_dir.exists():
            for skill_md in sorted(skills_dir.rglob("SKILL.md")):
                skills.append({"name": skill_md.parent.name})

        # Sessions count
        sessions_dir = profile_dir / "sessions"
        sessions_count = 0
        if sessions_dir.exists():
            try:
                sessions_count = len(list(sessions_dir.iterdir()))
            except Exception:
                pass

        # Memory info
        memory_path = profile_dir / "memory.json"
        memory_exists = memory_path.exists()
        memory_size = memory_path.stat().st_size if memory_exists else 0

        # .env existence
        env_exists = (profile_dir / ".env").exists()

        return jsonify({"ok": True, "data": {
            "name": name,
            "soul": soul,
            "config": config,
            "skills": skills,
            "skills_count": len(skills),
            "sessions_count": sessions_count,
            "memory_exists": memory_exists,
            "memory_size": memory_size,
            "env_exists": env_exists,
            "path": str(profile_dir),
        }})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/profile", methods=["POST"])
def api_profile_create():
    try:
        body = request.get_json(force=True) or {}
        name = body.get("name", "").strip()
        description = body.get("description", "")

        if not name or not name.replace("-", "").replace("_", "").isalnum():
            return jsonify({"ok": False, "error": "invalid profile name"}), 400

        if name == "default":
            return jsonify({"ok": False, "error": "cannot create profile named 'default'"}), 400

        # --no-skills: empty skills + default SOUL.md template (not cloned)
        # Then manually copy .env and config.yaml from default
        target_dir = PROFILES_DIR / name
        if target_dir.exists():
            return jsonify({"ok": False, "error": f"profile '{name}' already exists"}), 409

        cmd = ["hermes", "profile", "create", name, "--no-skills"]
        if description:
            cmd.extend(["--description", description])

        r = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
        if r.returncode != 0:
            return jsonify({"ok": False, "error": r.stderr.strip() or r.stdout.strip() or "create failed"}), 500

        # Copy .env and config.yaml from default profile
        default_home = HERMES_HOME  # ~/.hermes/
        import shutil
        for fname in [".env", "config.yaml"]:
            src = default_home / fname
            dst = target_dir / fname
            if src.exists():
                shutil.copy2(src, dst)

        # Overwrite SOUL.md with blank template (not default's personality)
        blank_soul = "# SOUL.md for profile: " + name + "\n\n# Define this profile's personality, role, and instructions here.\n# Example:\n# 你是一个专注于任务规划和审核的 Agent。你的职责是...\n"
        (target_dir / "SOUL.md").write_text(blank_soul)

        return jsonify({"ok": True, "data": {"name": name, "created": True}})
    except subprocess.TimeoutExpired:
        return jsonify({"ok": False, "error": "timeout creating profile"}), 500
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/profile/<name>", methods=["DELETE"])
def api_profile_delete(name):
    try:
        if name == "default":
            return jsonify({"ok": False, "error": "cannot delete default profile"}), 403

        # Check exists
        target_dir = PROFILES_DIR / name
        if not target_dir.exists():
            # Maybe it's a hermes profile not in profiles/ dir
            pass

        cmd = ["hermes", "profile", "delete", name, "-y"]
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
        if r.returncode != 0:
            return jsonify({"ok": False, "error": r.stderr.strip() or r.stdout.strip() or "delete failed"}), 500

        return jsonify({"ok": True, "data": {"name": name, "deleted": True}})
    except subprocess.TimeoutExpired:
        return jsonify({"ok": False, "error": "timeout deleting profile"}), 500
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/profile/<name>/chat", methods=["POST"])
def api_profile_chat(name):
    try:
        body = request.get_json(force=True) or {}
        message = body.get("message", "").strip()

        if not message:
            return jsonify({"ok": False, "error": "message is required"}), 400

        cmd = ["hermes", "-p", name, "chat", "-q", message]
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
        if r.returncode != 0:
            err = r.stderr.strip() or r.stdout.strip() or "chat failed"
            return jsonify({"ok": False, "error": err}), 500

        response = r.stdout.strip()
        if not response:
            return jsonify({"ok": False, "error": "empty response from model"}), 500

        return jsonify({"ok": True, "data": {"response": response}})
    except subprocess.TimeoutExpired:
        return jsonify({"ok": False, "error": "timeout (60s) waiting for model response"}), 500
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# --- Agent execution logs from Hermes state.db ---

AGENT_JOB_IDS = {
    "plan": "9f6ca9d65936",
    "dev": "dccc73082265",
    "test": "de73ff46d901",
}

STATE_DB_PATH = HERMES_HOME / "state.db"


def _state_db():
    """Open Hermes state.db read-only with timeout."""
    import sqlite3 as _sq
    db = _sq.connect(f"file:{STATE_DB_PATH}?mode=ro", uri=True, timeout=5.0)
    db.row_factory = _sq.Row
    return db


def _format_time(ts):
    if not ts:
        return None
    try:
        return datetime.fromtimestamp(float(ts)).strftime("%Y-%m-%d %H:%M:%S")
    except Exception:
        return None


def _is_session_silent(db, sid):
    """Return True if the last assistant message content is exactly [SILENT]."""
    try:
        row = db.execute(
            "SELECT content FROM messages WHERE session_id = ? AND role = 'assistant' ORDER BY timestamp DESC LIMIT 1",
            (sid,),
        ).fetchone()
        return bool(row and row["content"] == "[SILENT]")
    except Exception:
        return False


@app.route("/api/agent/<name>/sessions")
def api_agent_sessions(name):
    """Return recent execution sessions for an agent (cron job)."""
    try:
        job_id = AGENT_JOB_IDS.get(name.lower())
        if not job_id:
            # For unknown agent names, try profile_name match if available
            pass

        if not STATE_DB_PATH.exists():
            return jsonify({"ok": True, "data": []}), 200

        db = _state_db()
        try:
            if job_id:
                rows = db.execute(
                    """SELECT id, source, title, started_at, ended_at, end_reason,
                              message_count, tool_call_count, api_call_count
                       FROM sessions
                       WHERE id LIKE ?
                       ORDER BY started_at DESC LIMIT 20""",
                    (f"cron_{job_id}_%",),
                ).fetchall()
            else:
                rows = db.execute(
                    """SELECT id, source, title, started_at, ended_at, end_reason,
                              message_count, tool_call_count, api_call_count
                       FROM sessions
                       WHERE profile_name = ? OR source = ?
                       ORDER BY started_at DESC LIMIT 20""",
                    (name, name),
                ).fetchall()

            sessions = []
            for r in rows:
                sid = r["id"]
                started = r["started_at"]
                ended = r["ended_at"]
                status = "running"
                if ended:
                    status = (r["end_reason"] or "completed") if r["end_reason"] else "completed"
                sessions.append({
                    "session_id": sid,
                    "title": r["title"] or sid,
                    "source": r["source"],
                    "started_at": started,
                    "ended_at": ended,
                    "started_at_str": _format_time(started),
                    "ended_at_str": _format_time(ended),
                    "status": status,
                    "message_count": r["message_count"] or 0,
                    "tool_call_count": r["tool_call_count"] or 0,
                    "api_call_count": r["api_call_count"] or 0,
                    "is_silent": _is_session_silent(db, sid),
                })
            return jsonify({"ok": True, "data": sessions})
        finally:
            db.close()
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@app.route("/api/agent/<name>/session/<sid>")
def api_agent_session_detail(name, sid):
    """Return messages for a specific session."""
    try:
        if not STATE_DB_PATH.exists():
            return jsonify({"ok": False, "error": "state db not found"}), 404

        db = _state_db()
        try:
            # Verify session belongs to this agent (job prefix or profile_name/source)
            job_id = AGENT_JOB_IDS.get(name.lower())
            sess = db.execute("SELECT id, profile_name, source FROM sessions WHERE id = ?", (sid,)).fetchone()
            if not sess:
                return jsonify({"ok": False, "error": "session not found"}), 404

            belongs = False
            if job_id and isinstance(sess["id"], str) and sess["id"].startswith(f"cron_{job_id}_"):
                belongs = True
            if sess["profile_name"] and sess["profile_name"].lower() == name.lower():
                belongs = True
            if sess["source"] and sess["source"].lower() == name.lower():
                belongs = True
            if not belongs:
                return jsonify({"ok": False, "error": "session does not belong to agent"}), 404

            rows = db.execute(
                """SELECT id, role, content, tool_call_id, tool_calls, tool_name, timestamp
                   FROM messages
                   WHERE session_id = ?
                   ORDER BY timestamp ASC
                   LIMIT 50""",
                (sid,),
            ).fetchall()

            messages = []
            for r in rows:
                content = r["content"] or ""
                summary = content[:500]
                if len(content) > 500:
                    summary += "…"
                tool_calls_summary = None
                if r["tool_calls"]:
                    try:
                        tc = json.loads(r["tool_calls"])
                        if isinstance(tc, list):
                            tool_calls_summary = [t.get("function", {}).get("name") or t.get("name") for t in tc]
                    except Exception:
                        tool_calls_summary = ["parse_error"]
                messages.append({
                    "id": r["id"],
                    "role": r["role"],
                    "content": content,
                    "content_summary": summary,
                    "tool_call_id": r["tool_call_id"],
                    "tool_name": r["tool_name"],
                    "tool_calls_summary": tool_calls_summary,
                    "timestamp": r["timestamp"],
                    "timestamp_str": _format_time(r["timestamp"]),
                })
            return jsonify({"ok": True, "data": {"session_id": sid, "messages": messages}})
        finally:
            db.close()
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# --- Agents page routes ---

@app.route("/agents")
def agents_page():
    return render_template("agents.html")


@app.route("/agent/<name>")
def agent_detail_page(name):
    try:
        if name == "default":
            profile_dir = HERMES_HOME
        else:
            profile_dir = PROFILES_DIR / name
            if not profile_dir.exists():
                return render_template("agent_detail.html", profile_name=name,
                                     is_default=False, soul_text="Profile not found",
                                     config={}, skills=[], skills_count=0,
                                     sessions_count=0, memory_exists=False,
                                     memory_size=0, env_exists=False), 404

        soul = ""
        soul_path = profile_dir / "SOUL.md"
        if soul_path.exists():
            try:
                soul = soul_path.read_text()
            except Exception:
                pass

        config = {}
        config_path = profile_dir / "config.yaml"
        if config_path.exists():
            try:
                config = _parse_yaml_simple(config_path.read_text())
            except Exception:
                pass

        skills = []
        skills_dir = profile_dir / "skills"
        if skills_dir.exists():
            for skill_md in sorted(skills_dir.rglob("SKILL.md")):
                skills.append({"name": skill_md.parent.name})

        sessions_dir = profile_dir / "sessions"
        sessions_count = 0
        if sessions_dir.exists():
            try:
                sessions_count = len(list(sessions_dir.iterdir()))
            except Exception:
                pass

        memory_path = profile_dir / "memory.json"
        memory_exists = memory_path.exists()
        memory_size = memory_path.stat().st_size if memory_exists else 0
        env_exists = (profile_dir / ".env").exists()

        return render_template("agent_detail.html", profile_name=name,
                             is_default=(name == "default"), soul_text=soul,
                             config=config, skills=skills, skills_count=len(skills),
                             sessions_count=sessions_count, memory_exists=memory_exists,
                             memory_size=memory_size, env_exists=env_exists)
    except Exception as e:
        return render_template("agent_detail.html", profile_name=name,
                             is_default=False, soul_text=f"Error: {e}",
                             config={}, skills=[], skills_count=0,
                             sessions_count=0, memory_exists=False,
                             memory_size=0, env_exists=False), 500


# ── 简历筛选 ──────────────────────────────────────

RESUMES_DB = "/sdcard/bosszhipin/data/resumes.db"
RUBRIC_PATH = "/sdcard/bosszhipin/data/scoring_rubric_v2.json"
SCORE_SCRIPT = "/data/data/com.termux/files/home/.hermes/skills/android/resume-screener/scripts/score_resumes.py"
EXTRACT_SCRIPT = "/sdcard/bosszhipin/extract_resumes.py"
IMPORT_SCRIPT = "/sdcard/bosszhipin/import_resumes.py"


def _resumes_db():
    import sqlite3 as _sq
    db = _sq.connect(RESUMES_DB)
    db.row_factory = _sq.Row
    return db


@app.route("/resumes")
def resumes_page():
    return render_template("resumes.html")


@app.route("/resume/<int:resume_id>")
def resume_detail_page(resume_id):
    db = _resumes_db()
    r = db.execute("SELECT * FROM resumes WHERE id = ?", (resume_id,)).fetchone()
    if not r:
        db.close()
        return "Not Found", 404
    exps = json.loads(r["experiences"]) if r["experiences"] else []
    raw = r["raw_json"] or ""
    score = db.execute(
        "SELECT * FROM scores WHERE resume_id = ? ORDER BY rubric_version DESC LIMIT 1",
        (resume_id,),
    ).fetchone()
    db.close()
    return render_template(
        "resume_detail.html",
        resume=dict(r),
        experiences=exps,
        raw_json=raw,
        score=dict(score) if score else None,
    )


@app.route("/api/resumes")
def api_resumes():
    page = int(request.args.get("page", 1))
    per_page = int(request.args.get("per_page", 20))
    offset = (page - 1) * per_page
    db = _resumes_db()
    total = db.execute("SELECT count(*) FROM resumes").fetchone()[0]
    rows = db.execute(
        """SELECT r.id, r.friendId, r.name, r.education, r.secondText,
                  r.jobPosition, r.applyStatus, r.city, r.age,
                  r.workAge, r.expSalary, r.gender,
                  s.total_score, s.tier, s.rubric_version,
                  s.eliminated, s.eliminated_reason
           FROM resumes r
           LEFT JOIN scores s ON s.resume_id = r.id
           AND s.id = (SELECT max(id) FROM scores WHERE resume_id = r.id)
           ORDER BY r.id LIMIT ? OFFSET ?""",
        (per_page, offset),
    ).fetchall()
    db.close()
    data = [dict(r) for r in rows]
    return jsonify(ok=True, data=data, total=total, page=page, per_page=per_page)


@app.route("/api/resumes/score", methods=["POST"])
def api_resumes_score():
    """对未打分的简历批量调用 score_resumes.py"""
    import subprocess as _sp
    try:
        result = _sp.run(
            ["python3", SCORE_SCRIPT,
             "--rubric", RUBRIC_PATH,
             "--db", RESUMES_DB,
             "--only-unscored"],
            capture_output=True, text=True, timeout=120,
        )
        if result.returncode != 0:
            return jsonify(ok=False, error=result.stderr[:500]), 500
        # 解析输出里的打分数量
        scored = 0
        for line in result.stdout.split("\n"):
            if "打分完成" in line:
                import re as _re
                m = _re.search(r"(\d+)\s*份", line)
                if m:
                    scored = int(m.group(1))
        # 同步打分结果到 scores 表
        _sync_scores_to_db()
        return jsonify(ok=True, data={"scored": scored, "stdout": result.stdout[-300:]})
    except _sp.TimeoutExpired:
        return jsonify(ok=False, error="打分超时"), 500
    except Exception as e:
        return jsonify(ok=False, error=str(e)), 500


def _sync_scores_to_db():
    """从最新版本 scoring_result JSONL 同步打分结果到 resumes.db scores 表"""
    import sqlite3 as _sq
    # 找最新版本的 JSONL
    import glob as _glob
    files = sorted(_glob.glob("/sdcard/bosszhipin/data/scoring_result_v*.jsonl"))
    if not files:
        return
    result_path = files[-1]  # 最新版本
    if not os.path.exists(result_path):
        return
    db = _resumes_db()
    with open(result_path, "r", encoding="utf-8") as f:
        for line in f:
            try:
                r = json.loads(line)
                row = db.execute(
                    "SELECT id FROM resumes WHERE friendId = ?", (r["friendId"],)
                ).fetchone()
                if not row:
                    continue
                rid = row[0]
                db.execute(
                    """INSERT OR REPLACE INTO scores
                       (resume_id, friendId, rubric_version, total_score, tier,
                        dimension_scores, scored_at, eliminated, eliminated_reason)
                       VALUES (?,?,?,?,?,?,?,?,?)""",
                    (rid, r["friendId"], r.get("rubric_version", 1),
                     r["total_score"], r["tier"],
                     json.dumps(r.get("dimension_scores", {}), ensure_ascii=False),
                     r.get("scored_at", ""),
                     r.get("eliminated", 0),
                     r.get("eliminated_reason")),
                )
            except Exception:
                pass
    db.commit()
    db.close()


if __name__ == "__main__":
    app.config['TEMPLATES_AUTO_RELOAD'] = True
    app.run(host="0.0.0.0", port=5001, debug=False)