auto_control/web/agent_api.py

"""AI 控制台 API：模型/Key 前端配置 + Agent 流式执行（SSE 事件流）。

流程：
  POST /api/agent/run {prompt}        → 启动 Agent 线程，返回 run_id
  GET  /api/agent/stream?run_id=      → SSE 事件流（EventSource 订阅）：
        event: delta  {text, kind: content|reasoning}   流式文本增量
        event: step   {tool, args, image?}               工具调用完成（MCP 步骤）
        event: done   {answer}                           完成
        event: error  {message}                          失败
  GET/POST /api/agent/config          → 配置读写（key 打码回显）

单实例：同时只允许一个 Agent 运行。
"""
import asyncio
import base64
import io
import json
import os
import queue
import re
import sys
import threading
import uuid

from flask import Blueprint, Response, jsonify, request

from core.logger import get_logger
from core.models import db
from web.auth import admin_required

_log = get_logger("web.agent")
bp = Blueprint("agent", __name__)

# ---------- 配置键（app_meta） ----------
_CFG_KEYS = {"api_base": "agent_api_base",
             "model": "agent_model",
             "api_key": "agent_api_key",
             "default_serial": "agent_default_serial",
             "max_steps": "agent_max_steps"}

# ---------- 运行状态（单实例 + 事件队列） ----------
_run = {"id": None, "state": "idle", "prompt": "", "serial": "",
        "answer": "", "error": "",
        "history": []}   # 多轮对话历史 [{role: user|assistant, content}]
_queues = {}            # run_id -> queue.Queue（SSE 消费者读取）
_stop_events = {}       # run_id -> threading.Event（用户中断）
_lock = threading.Lock()
_flask_app = None       # web_server 注册时注入（后台线程 db 操作需 app context）


def set_app(app):
    global _flask_app
    _flask_app = app


def _meta_get(key):
    return db.session.execute(
        db.text("SELECT value FROM app_meta WHERE key=:k"), {"k": key}).scalar() or ""


def _meta_put(key, value):
    db.session.execute(
        db.text("INSERT OR REPLACE INTO app_meta(key,value) VALUES(:k,:v)"),
        {"k": key, "v": str(value)})


def _read_cfg():
    return {k: _meta_get(v) for k, v in _CFG_KEYS.items()}


# ================== 经验记忆（自进化） ==================
# agent_experience：任务成功后的操作配方，下次相似任务检索注入 system prompt。
# 原始 SQLite（CREATE IF NOT EXISTS 幂等），不进模型层迁移。
_EXP_TABLE = """
CREATE TABLE IF NOT EXISTS agent_experience (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  task_prompt TEXT DEFAULT '',
  recipe TEXT DEFAULT '',
  tool_seq TEXT DEFAULT '',
  hits INTEGER DEFAULT 0,
  created_at VARCHAR(20) DEFAULT '')"""

# 经验巡检（AI 质检）：每日定时把经验交给模型评审，疑似问题标 pending 待人工确认。
# 删除只允许人工（前端 confirm / 本 API），巡检绝不自动删。
_AUDIT_TABLE = """
CREATE TABLE IF NOT EXISTS experience_audit (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  exp_id INTEGER NOT NULL,
  verdict TEXT DEFAULT '',       -- keep 保留 / delete 建议删除
  score REAL DEFAULT 0,          -- 0-10 可用性评分
  reason TEXT DEFAULT '',        -- 模型给出的理由
  hits INTEGER DEFAULT 0,        -- 巡检时的引用次数（hits>0 被删要更谨慎）
  action TEXT DEFAULT 'pending', -- pending 待人工确认 / kept 已保留 / deleted 已删除
  audited_at VARCHAR(20) DEFAULT '')"""

# 巡检评审提示词：模型只判「保留/建议删除」+ 评分 + 理由，不直接删。
_AUDIT_PROMPT = """你是「手机自动化经验库」质检员。经验库存放 AI 成功操作手机后提炼的
操作套路，下次相似任务会自动注入参考。请判断下面这条经验是否值得继续保留。

判断标准（全部满足才保留）：
1. 具体可执行：步骤是明确的工具操作（de_* 或清晰的中文步骤），不是「de_tap / de_input」
   这类空泛占位，也没有编造不存在的工具名
2. 不依赖易过时信息：不含写死的像素坐标（x,y 数字）这类界面一变就失效的步骤
3. 是独立任务的操作套路（如「打开X搜索Y并点赞」），不是对话追问/抱怨/单次性内容
4. 配方与下方实际操作序列大致一致，不是模型自由发挥编造

经验内容：
任务描述：{prompt}
操作配方：{recipe}
实际操作序列：{tool_seq}
被引用次数：{hits}

只输出 JSON：{{"keep": true或false, "score": 0到10的整数, "reason": "一句话中文理由"}}"""


def _ensure_exp_table():
    try:
        db.session.execute(db.text(_EXP_TABLE))
        db.session.commit()
    except Exception:
        pass


def _bigrams(text):
    """中文/英文文本 bigram 集合（无空格分词，粗粒度相似度）。"""
    t = "".join(c for c in (text or "").lower() if c.isalnum() or "\u4e00" <= c <= "\u9fff")
    return {t[i:i + 2] for i in range(len(t) - 1)}


# ---------- 经验质量门槛 ----------
# 自进化经验只应保存「独立操作任务」的成功套路。多轮对话里用户的短句
# （质疑/纠正/催促，如「继续啊」「你确定我是卡1吗」「还没有完成啊」）
# 不是新任务——把执行出错被纠正的轮次存成经验会教坏后续任务。
# 用启发式过滤（零成本，可解释），配方层再校验工具名真实性。

# 任务性动词：命中任一视为有明确操作诉求（疑问/纠错短句一般不含它们）
_TASK_VERBS = ("打开", "搜索", "查看", "找到", "截图", "输入", "点击", "点开",
               "发送", "安装", "卸载", "下载", "启动", "停止", "关闭", "退出",
               "登录", "切换", "设置", "删除", "清理", "复制", "粘贴", "读取",
               "剪贴板", "长按", "滑动", "播放", "发布", "检查", "看看",
               "帮我", "给我", "请", "拍张", "查一下")
# 对话续语开头：几乎只出现在承接上一轮（「继续啊」「还有吗」）
_CONTINUE_PREFIXES = ("继续", "还有", "然后呢", "再来", "快点", "刚才",
                      "接着", "上一步")
# 强质疑/纠错信号（不含「为什么」——「查一下为什么」是正当任务）
_DOUBT_MARKS = ("你确定", "是不是", "不是吧", "不是吗", "怎么都", "怎么还",
                "还没有", "没看到", "我说的是", "你听我说", "不对吧", "你又",
                "重新来", "错了", "你说得", "你回答")
# 全部真实 MCP 工具（配方里出现不存在的 de_* 说明蒸馏模型在编造，弃存）
_KNOWN_TOOLS = frozenset({
    "de_list_devices", "de_screenshot", "de_tap", "de_swipe", "de_ui_tree",
    "de_tap_element", "de_read_clipboard", "de_wake", "de_press_key",
    "de_open_app", "de_stop_app", "de_foreground_app", "de_type_text",
    "de_set_clipboard", "de_sleep", "de_ocr", "de_tap_text", "de_list_apps",
    "de_list_tasks"})


def _qualify_experience(prompt, recipe):
    """经验入库前质量门槛，返回 True=值得保存。

    1) prompt 太短 / 纯续语开头 / 质疑纠错 → 非独立任务，弃
    2) 疑问短句（≤30 字、以 吗/？ 结尾）且无任务动词 → 追问/反问，弃
    3) 配方含不存在的 de_* 工具（蒸馏模型自由发挥）→ 弃
    """
    t = "".join(c for c in (prompt or "") if not c.isspace())
    if len(t) < 6:
        _log.info("经验弃存：prompt 过短「%s」", t[:20])
        return False
    if any(t.startswith(p) for p in _CONTINUE_PREFIXES):
        _log.info("经验弃存：对话续语开头「%s」", t[:20])
        return False
    if any(m in t for m in _DOUBT_MARKS):
        _log.info("经验弃存：质疑/纠错语气「%s」", t[:20])
        return False
    if (t.endswith("吗") or t.endswith("？") or t.endswith("?")) \
            and len(t) <= 30 and not any(v in t for v in _TASK_VERBS):
        _log.info("经验弃存：无操作诉求的追问「%s」", t[:20])
        return False
    for name in re.findall(r"de_[a-z_]+", recipe or ""):
        if name not in _KNOWN_TOOLS:
            _log.info("经验弃存：配方含不存在的工具 %s", name)
            return False
    return True


def _find_experiences(prompt, limit=2, threshold=0.10):
    """按 bigram 重叠检索相似历史经验（prompt 与任务描述的字符相似度）。

    命中的经验 hits+1（回写），让被反复参考的有效经验浮到前面。
    """
    try:
        if _flask_app is None:
            return ""
        with _flask_app.app_context():
            _ensure_exp_table()
            rows = db.session.execute(db.text(
                "SELECT id, task_prompt, recipe, hits FROM agent_experience "
                "WHERE recipe != '' ORDER BY hits DESC, id DESC LIMIT 50")).fetchall()
    except Exception:
        return ""
    if not rows:
        return ""
    cur = _bigrams(prompt)
    if not cur:
        return ""
    scored = []
    for row_id, task_prompt, recipe, hits in rows:
        sim = len(cur & _bigrams(task_prompt)) / len(cur)
        if sim >= threshold:
            scored.append((sim, hits or 0, recipe, row_id))
    scored.sort(key=lambda x: (-x[0], -x[1]))
    if scored:
        # hits 回写（尽力而为，失败不影响检索）
        try:
            with _flask_app.app_context():
                for _sim, _hits, _recipe, row_id in scored:
                    db.session.execute(db.text(
                        "UPDATE agent_experience SET hits=hits+1 WHERE id=:i"),
                        {"i": row_id})
                db.session.commit()
        except Exception:
            pass
    parts = []
    for sim, _hits, recipe, _row_id in scored[:limit]:
        parts.append(f"- {recipe[:600]}")
    return "\n".join(parts)


def _distill_experience(cfg, prompt, tool_seq):
    """任务完成后用模型把操作序列提炼为可复用配方（失败静默，不阻塞）。"""
    try:
        import httpx
        body = {
            "model": cfg.get("model") or "deepseek-v4-flash-vision-exp",
            "messages": [{"role": "user",
                          "content": "以下是一次成功的手机自动化操作记录。请提炼成简洁的"
                                     "「操作配方」（2-6 步，每步：目标 → 用哪个工具），"
                                     "供下次同类任务参考。不要解释，直接输出配方。\n"
                                     f"任务：{prompt[:300]}\n操作序列：{tool_seq[:800]}"}],
            "max_tokens": 600,
        }
        headers = {"Authorization": f"Bearer {cfg.get('api_key', '')}",
                   "Content-Type": "application/json"}
        r = httpx.post(f"{(cfg.get('api_base') or 'https://api.deepseek.com').rstrip('/')}/chat/completions",
                       json=body, headers=headers, timeout=25)
        if r.status_code != 200:
            return ""
        j = r.json()
        recipe = ((j.get("choices") or [{}])[0].get("message") or {}).get("content") or ""
        return recipe.strip()[:1500]
    except Exception as e:
        _log.warning(f"经验提炼失败: {e}")
        return ""


def _save_experience(prompt, recipe, tool_seq):
    """保存经验（后台线程调用，包 app context）。返回是否保存成功。"""
    try:
        if _flask_app is None:
            return False
        with _flask_app.app_context():
            _ensure_exp_table()
            from datetime import datetime
            db.session.execute(db.text(
                "INSERT INTO agent_experience(task_prompt, recipe, tool_seq, hits, created_at) "
                "VALUES(:p, :r, :t, 0, :c)"),
                {"p": prompt[:500], "r": recipe, "t": tool_seq[:1000],
                 "c": datetime.now().strftime("%Y-%m-%d %H:%M")})
            db.session.commit()
            _log.info("经验已保存（配方 %d 字符）", len(recipe))
            return True
    except Exception as e:
        _log.warning(f"经验保存失败: {e}")
        return False


# ================== 经验巡检（AI 质检，删除需人工确认） ==================
_audit_state = {"running": False, "last": "", "last_summary": ""}


def _ensure_audit_table():
    try:
        db.session.execute(db.text(_AUDIT_TABLE))
        db.session.commit()
    except Exception:
        pass


def _review_one_exp(cfg, prompt, recipe, tool_seq, hits):
    """调模型评审单条经验，返回 (verdict, score, reason)。

    模型偶尔输出解释文字不带 JSON——重试一次（强调只输出 JSON），
    仍失败则保守返回保留。
    """
    import httpx
    base_prompt = _AUDIT_PROMPT.format(prompt=(prompt or "")[:400],
                                       recipe=(recipe or "")[:600],
                                       tool_seq=(tool_seq or "")[:600],
                                       hits=hits or 0)
    for attempt in (1, 2):
        try:
            content = base_prompt if attempt == 1 else (
                base_prompt + "\n注意：只输出一个 JSON 对象，不要输出任何其它文字、解释或代码块标记。")
            body = {
                "model": cfg.get("model") or "deepseek-v4-flash-vision-exp",
                "messages": [{"role": "user", "content": content}],
                "max_tokens": 200,
            }
            headers = {"Authorization": f"Bearer {cfg.get('api_key', '')}",
                       "Content-Type": "application/json"}
            r = httpx.post(f"{(cfg.get('api_base') or 'https://api.deepseek.com').rstrip('/')}/chat/completions",
                           json=body, headers=headers, timeout=20)
            if r.status_code != 200:
                raise RuntimeError(f"HTTP {r.status_code}")
            text = (((r.json() or {}).get("choices") or [{}])[0]
                    .get("message") or {}).get("content") or ""
            m = re.search(r"\{.*\}", text, re.S)
            if not m:
                raise RuntimeError("响应无 JSON")
            j = json.loads(m.group(0))
            keep = bool(j.get("keep", True))
            score = max(0, min(10, int(j.get("score", 5) or 5)))
            reason = str(j.get("reason", ""))[:200]
            return ("keep" if keep else "delete"), score, reason
        except Exception as e:
            _log.warning(f"经验评审第 {attempt} 次失败: {e}")
    return "keep", 0, "评审失败自动保留（两次尝试均无有效 JSON）"


def run_experience_audit():
    """巡检一轮：逐条评审经验 → 写 experience_audit。

    规则：
      - 最新已 kept / deleted 的经验不再重复评审（人工拍板过的尊重）
      - 评审 keep → 直接 action=kept；delete → action=pending（等人工确认）
      - 绝不自动删除；无 api_key 配置时跳过并记日志
    由 web_server 每日定时调度或 API 手动触发（后台线程）。
    """
    if _audit_state["running"]:
        _log.info("经验巡检已在运行，跳过本次触发")
        return
    _audit_state["running"] = True
    try:
        if _flask_app is None:
            return
        with _flask_app.app_context():
            _ensure_exp_table()
            _ensure_audit_table()
            cfg = _read_cfg()
            if not cfg.get("api_key"):
                _log.info("经验巡检跳过：未配置 API Key")
                return
            rows = db.session.execute(db.text(
                "SELECT id, task_prompt, recipe, tool_seq, hits FROM agent_experience "
                "ORDER BY id")).fetchall()
            if not rows:
                _log.info("经验巡检：经验库为空")
                return
            from datetime import datetime
            now = datetime.now().strftime("%Y-%m-%d %H:%M")
            reviewed = suggested = 0
            for exp_id, prompt, recipe, tool_seq, hits in rows:
                # 最新一次人工/评审结论：kept / deleted 的不再打扰
                last = db.session.execute(db.text(
                    "SELECT action FROM experience_audit WHERE exp_id=:e "
                    "ORDER BY id DESC LIMIT 1"), {"e": exp_id}).scalar()
                if last in ("kept", "deleted"):
                    continue
                verdict, score, reason = _review_one_exp(
                    cfg, prompt, recipe, tool_seq, hits)
                action = "kept" if verdict == "keep" else "pending"
                db.session.execute(db.text(
                    "INSERT INTO experience_audit"
                    "(exp_id, verdict, score, reason, hits, action, audited_at) "
                    "VALUES(:e, :v, :s, :r, :h, :a, :t)"),
                    {"e": exp_id, "v": verdict, "s": score, "r": reason,
                     "h": hits or 0, "a": action, "t": now})
                reviewed += 1
                suggested += 1 if action == "pending" else 0
            db.session.commit()
            summary = f"评审 {reviewed} 条，建议删除 {suggested} 条（待人工确认）"
            _audit_state.update(last=now, last_summary=summary)
            _log.info("经验巡检完成: %s", summary)
    except Exception as e:
        _log.warning(f"经验巡检异常: {e}")
    finally:
        _audit_state["running"] = False


# ================== 经验库管理 API ==================
@bp.route("/api/agent/experience")
@admin_required
def agent_experience_list():
    """经验库列表（含最近一次巡检结论），供 AI 控制台「🧠 经验库」面板管理。"""
    _ensure_exp_table()
    _ensure_audit_table()
    rows = db.session.execute(db.text(
        "SELECT e.id, e.task_prompt, e.recipe, e.tool_seq, e.hits, e.created_at, "
        "a.verdict, a.score, a.reason, a.action, a.audited_at "
        "FROM agent_experience e LEFT JOIN experience_audit a ON a.id = "
        "(SELECT id FROM experience_audit WHERE exp_id = e.id ORDER BY id DESC LIMIT 1) "
        "ORDER BY e.id DESC")).fetchall()
    exps = []
    for r in rows:
        (eid, task_prompt, recipe, tool_seq, hits, created_at,
         verdict, score, reason, action, audited_at) = r
        audit = None
        if action:
            audit = {"verdict": verdict, "score": round(score or 0, 1),
                     "reason": reason, "action": action, "at": audited_at}
        exps.append({"id": eid, "task_prompt": task_prompt or "",
                     "recipe": recipe or "", "tool_seq": tool_seq or "",
                     "hits": hits or 0, "created_at": created_at or "",
                     "audit": audit})
    return jsonify({"ok": True, "running": _audit_state["running"],
                    "last": _audit_state["last"],
                    "last_summary": _audit_state["last_summary"],
                    "experiences": exps})


@bp.route("/api/agent/experience/delete", methods=["POST"])
@admin_required
def agent_experience_delete():
    """人工确认删除经验（真删；前端有二次确认，巡检绝不自动调它）。"""
    data = request.json or {}
    exp_id = int(data.get("id") or 0)
    if exp_id <= 0:
        return jsonify({"ok": False, "error": "缺少 id"}), 400
    db.session.execute(db.text(
        "DELETE FROM agent_experience WHERE id=:i"), {"i": exp_id})
    db.session.execute(db.text(
        "UPDATE experience_audit SET action='deleted' "
        "WHERE exp_id=:i AND action='pending'"), {"i": exp_id})
    db.session.commit()
    _log.info("人工确认删除经验 #%d", exp_id)
    return jsonify({"ok": True, "msg": f"经验 #{exp_id} 已删除"})


@bp.route("/api/agent/experience/audit", methods=["POST"])
@admin_required
def agent_experience_audit_now():
    """手动触发一轮经验巡检（后台线程；进行中返回 409）。"""
    if _audit_state["running"]:
        return jsonify({"ok": False, "error": "巡检正在进行中"}), 409
    threading.Thread(target=run_experience_audit, daemon=True).start()
    return jsonify({"ok": True, "msg": "巡检已启动，完成后刷新列表查看建议"})


@bp.route("/api/agent/experience/keep", methods=["POST"])
@admin_required
def agent_experience_keep():
    """人工保留：撤销建议删除（该条后续巡检不再重复建议）。"""
    data = request.json or {}
    exp_id = int(data.get("id") or 0)
    if exp_id <= 0:
        return jsonify({"ok": False, "error": "缺少 id"}), 400
    db.session.execute(db.text(
        "UPDATE experience_audit SET action='kept' "
        "WHERE exp_id=:i AND action='pending'"), {"i": exp_id})
    db.session.commit()
    return jsonify({"ok": True, "msg": f"经验 #{exp_id} 已保留"})

# ================== 历史会话（DeepSeek 式：多会话持久化） ==================
# 会话 = 消息序列（JSON），一轮 run 的 user/assistant 文本完成后追加落库。
# 单表 JSON 存储（每会话几十 KB 内，无需独立消息表）。
_CONV_TABLE = """
CREATE TABLE IF NOT EXISTS agent_conversation (
  id VARCHAR(20) PRIMARY KEY,
  title VARCHAR(100) DEFAULT '',
  messages TEXT DEFAULT '[]',
  created_at VARCHAR(20) DEFAULT '',
  updated_at VARCHAR(20) DEFAULT '')"""


def _ensure_conv_table():
    try:
        db.session.execute(db.text(_CONV_TABLE))
        db.session.commit()
    except Exception:
        pass


def _conv_msgs(conv_id):
    """读会话消息列表 [{role, content}]。会话不存在返回 None。"""
    try:
        row = db.session.execute(db.text(
            "SELECT messages FROM agent_conversation WHERE id=:i"),
            {"i": conv_id}).fetchone()
    except Exception:
        return None
    if not row:
        return None
    try:
        return json.loads(row[0] or "[]")
    except Exception:
        return []


def _conv_save_messages(conv_id, messages, title=None):
    """整段写回会话消息 + 时间戳（尽力而为，失败不影响主流程）。"""
    try:
        from datetime import datetime
        now = datetime.now().strftime("%Y-%m-%d %H:%M")
        msgs_json = json.dumps(messages[-60:], ensure_ascii=False)
        if title:
            db.session.execute(db.text(
                "UPDATE agent_conversation SET messages=:m, title=:t, "
                "updated_at=:u WHERE id=:i"),
                {"m": msgs_json, "t": title[:100], "u": now, "i": conv_id})
        else:
            db.session.execute(db.text(
                "UPDATE agent_conversation SET messages=:m, updated_at=:u "
                "WHERE id=:i"),
                {"m": msgs_json, "u": now, "i": conv_id})
        db.session.commit()
        return True
    except Exception as e:
        _log.warning(f"会话保存失败: {e}")
        return False


@bp.route("/api/agent/conversations", methods=["GET"])
@admin_required
def agent_conversations_list():
    """会话列表（按最近更新倒序）：{id, title, updated_at, count}。"""
    _ensure_conv_table()
    rows = db.session.execute(db.text(
        "SELECT id, title, messages, updated_at FROM agent_conversation "
        "ORDER BY updated_at DESC, created_at DESC")).fetchall()
    out = []
    for cid, title, messages, updated_at in rows:
        try:
            count = len(json.loads(messages or "[]")) // 2
        except Exception:
            count = 0
        out.append({"id": cid, "title": title or "新会话",
                    "updated_at": updated_at or "", "count": count})
    return jsonify({"ok": True, "conversations": out})


@bp.route("/api/agent/conversations", methods=["POST"])
@admin_required
def agent_conversations_create():
    """新建会话（空消息）。返回 {id}。"""
    _ensure_conv_table()
    from datetime import datetime
    cid = uuid.uuid4().hex[:10]
    now = datetime.now().strftime("%Y-%m-%d %H:%M")
    db.session.execute(db.text(
        "INSERT INTO agent_conversation(id, title, messages, created_at, updated_at) "
        "VALUES(:i, '', '[]', :c, :c)"), {"i": cid, "c": now})
    db.session.commit()
    return jsonify({"ok": True, "id": cid, "title": "新会话"})


@bp.route("/api/agent/conversations/<conv_id>", methods=["GET"])
@admin_required
def agent_conversations_detail(conv_id):
    """会话详情（全部消息文本）。"""
    _ensure_conv_table()
    msgs = _conv_msgs(conv_id)
    if msgs is None:
        return jsonify({"ok": False, "error": "会话不存在"}), 404
    row = db.session.execute(db.text(
        "SELECT title, created_at, updated_at FROM agent_conversation "
        "WHERE id=:i"), {"i": conv_id}).fetchone()
    return jsonify({"ok": True, "id": conv_id,
                    "title": (row[0] if row else "") or "新会话",
                    "messages": msgs,
                    "created_at": row[1] if row else "",
                    "updated_at": row[2] if row else ""})


@bp.route("/api/agent/conversations/<conv_id>", methods=["DELETE"])
@admin_required
def agent_conversations_delete(conv_id):
    """删除会话（消息一并删除，不可恢复）。"""
    _ensure_conv_table()
    db.session.execute(db.text(
        "DELETE FROM agent_conversation WHERE id=:i"), {"i": conv_id})
    db.session.commit()
    _log.info(f"删除会话 {conv_id}")
    return jsonify({"ok": True, "msg": "会话已删除"})


@bp.route("/api/agent/conversations/<conv_id>/rename", methods=["POST"])
@admin_required
def agent_conversations_rename(conv_id):
    """重命名会话。"""
    data = request.json or {}
    title = (data.get("title") or "").strip()[:100]
    if not title:
        return jsonify({"ok": False, "error": "标题不能为空"}), 400
    _ensure_conv_table()
    db.session.execute(db.text(
        "UPDATE agent_conversation SET title=:t WHERE id=:i"),
        {"t": title, "i": conv_id})
    db.session.commit()
    return jsonify({"ok": True, "msg": "已重命名"})


# ================== 运行 ==================
@bp.route("/api/agent/run", methods=["POST"])
@admin_required
def agent_run():
    """启动 Agent：{prompt, serial?, conversation_id?}。运行中返回 409。"""
    data = request.json or {}
    prompt = (data.get("prompt") or "").strip()
    if not prompt:
        return jsonify({"ok": False, "error": "请输入指令"}), 400
    serial = (data.get("serial") or "").strip()
    conv_id = (data.get("conversation_id") or "").strip()
    cfg = _read_cfg()
    if not cfg.get("api_key"):
        return jsonify({"ok": False, "error": "请先在配置区填写 API Key"}), 400
    if not cfg.get("model"):
        return jsonify({"ok": False, "error": "请先填写模型名"}), 400
    if not serial and not (cfg.get("default_serial") or "").strip():
        return jsonify({"ok": False, "error": "请先选择目标设备（AI 只操作你指定的设备）"}), 400
    # 目标设备校验：必须在池、在线、且无任务运行（AI 不与任务抢设备；
    # MCP 层另有 busy 锁兜底外部客户端）
    if serial:
        try:
            from web import context
            devs, err = context.mgr.get_status()
            dev = next((x for x in (devs or []) if x.get("serial") == serial), None)
            if err or dev is None:
                return jsonify({"ok": False, "error": f"设备 {serial} 不在设备池"}), 400
            if not dev.get("present"):
                return jsonify({"ok": False, "error": f"设备 {serial} 当前离线，请稍后再试"}), 400
            if dev.get("worker_status") in ("running", "connecting"):
                return jsonify({"ok": False, "error":
                    f"设备 {serial} 正在执行任务「{dev.get('task_job') or ''}」——"
                    f"AI 不与任务抢设备，任务结束后才能操作（或在任务页先停止）"}), 409
        except Exception:
            pass  # 状态服务异常不阻塞（MCP busy 锁兜底）
    # 会话校验：不存在则自动新建（标题=首条消息截断）
    if conv_id:
        _ensure_conv_table()
        if _conv_msgs(conv_id) is None:
            try:
                from datetime import datetime as _dt2
                now = _dt2.now().strftime("%Y-%m-%d %H:%M")
                db.session.execute(db.text(
                    "INSERT OR IGNORE INTO agent_conversation(id, messages, "
                    "created_at, updated_at) VALUES(:i, '[]', :c, :c)"),
                    {"i": conv_id, "c": now})
                db.session.commit()
            except Exception:
                pass
    with _lock:
        if _run["state"] == "running":
            return jsonify({"ok": False, "error":
                f"已有 Agent 运行中（{_run.get('prompt', '')[:40]}…），"
                f"请等待完成或先停止"}), 409
        run_id = uuid.uuid4().hex[:8]
        from datetime import datetime as _dt
        _run.update(id=run_id, state="running", prompt=prompt,
                    serial=serial, conv_id=conv_id,
                    started=_dt.now().strftime("%H:%M:%S"),
                    answer="", error="")
        # history 保留（多轮上下文），由会话/「新建会话」管理
        _queues[run_id] = queue.Queue()
        _stop_events[run_id] = threading.Event()
    _log.info(f"Agent 启动: {prompt[:60]} @ {serial or 'default'} conv={conv_id or '-'}")
    threading.Thread(target=_agent_thread,
                     args=(run_id, prompt, serial, cfg),
                     daemon=True).start()
    return jsonify({"ok": True, "run_id": run_id})
@bp.route("/api/agent/config", methods=["GET"])
@admin_required
def agent_config_get():
    """读 Agent 配置（key 打码返回）。"""
    cfg = _read_cfg()
    if cfg["api_key"]:
        k = cfg["api_key"]
        cfg["api_key_masked"] = k[:6] + "***" + k[-4:]
    return jsonify({"ok": True, **cfg})


@bp.route("/api/agent/config", methods=["POST"])
@admin_required
def agent_config_save():
    """保存 Agent 配置：{api_base?, model?, api_key?, default_serial?} 部分更新。"""
    data = request.json or {}
    for key, meta_key in _CFG_KEYS.items():
        if key in data and data[key] is not None:
            _meta_put(meta_key, str(data[key]).strip())
    db.session.commit()
    return jsonify({"ok": True, "msg": "已保存"})


@bp.route("/api/agent/run", methods=["GET"])
@admin_required
def agent_run_status():
    """当前 Agent 运行状态（多窗口/页面刷新恢复用）：
    idle/running/done + run_id + 最终答案 + 会话历史。

    前端刷新后据此：渲染历史轮次、running 时重新订阅事件流（服务端队列
    保留积压事件，重连后补发）、done 时展示最终答案。
    """
    with _lock:
        hist = [{"role": h.get("role"), "content": (h.get("content") or "")[:4000]}
                for h in (_run.get("history") or [])]
        return jsonify({"ok": True,
                        "state": _run.get("state", "idle"),
                        "run_id": _run.get("id") or "",
                        "prompt": _run.get("prompt", ""),
                        "serial": _run.get("serial", ""),
                        "started": _run.get("started", ""),
                        "answer": (_run.get("answer") or "")[:4000],
                        "error": (_run.get("error") or "")[:400],
                        "history": hist[-16:]})


@bp.route("/api/agent/devices")
@admin_required
def agent_devices():
    """AI 可用设备列表：在线状态 + 是否有任务运行（前端选择器 busy 设备禁选）。

    worker 状态实时（内存）；busy = 任务 running/connecting。
    """
    try:
        from web import context
        devs, err = context.mgr.get_status()
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)[:120]}), 503
    out = []
    for d in (devs or []):
        busy = d.get("worker_status") in ("running", "connecting")
        out.append({"serial": d.get("serial"),
                    "model": d.get("model") or "",
                    "online": bool(d.get("present")),
                    "busy": busy,
                    "worker_status": d.get("worker_status") or "idle",
                    "task_job": d.get("task_job") or ""})
    return jsonify({"ok": True, "devices": out})


@bp.route("/api/agent/stream")
@admin_required
def agent_stream():
    """SSE 事件流（EventSource）：delta/step/done/error。"""
    run_id = request.args.get("run_id", "")
    with _lock:
        if run_id != _run["id"]:
            return jsonify({"ok": False, "error": "run_id 不存在"}), 404
        q = _queues.get(run_id)

    def gen():
        while True:
            try:
                evt = q.get(timeout=15)
            except queue.Empty:
                yield ": keepalive\n\n"   # 心跳防超时
                continue
            if evt is None:
                break
            kind, payload = evt
            yield f"event: {kind}\ndata: {json.dumps(payload, ensure_ascii=False)}\n\n"
            if kind in ("done", "error"):
                break

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


@bp.route("/api/agent/stop", methods=["POST"])
@admin_required
def agent_stop():
    """中断当前运行的 Agent（下一个检查点生效，通常在数秒内）。"""
    with _lock:
        if _run["state"] != "running":
            return jsonify({"ok": False, "error": "当前没有运行中的任务"}), 400
        evt = _stop_events.get(_run["id"])
    if evt:
        evt.set()
        _log.info("用户请求中断 Agent")
    return jsonify({"ok": True, "msg": "已请求停止"})


@bp.route("/api/agent/clear", methods=["POST"])
@admin_required
def agent_clear():
    """清空对话历史。"""
    with _lock:
        _run["history"] = []
    return jsonify({"ok": True, "msg": "已清空"})


def _shrink_image(b64, width=220, quality=50):
    """截图降采样（SSE step 事件用，控制传输体积）。失败原样返回。"""
    try:
        from PIL import Image
        img = Image.open(io.BytesIO(base64.b64decode(b64)))
        if img.width > width:
            img = img.resize((width, int(img.height * width / img.width)))
        buf = io.BytesIO()
        img.convert("RGB").save(buf, "JPEG", quality=quality)
        return base64.b64encode(buf.getvalue()).decode()
    except Exception:
        return b64


def _agent_thread(run_id, prompt, serial, cfg):
    """后台线程：Agent 流式执行，事件推入队列供 SSE 消费。"""
    q = _queues.get(run_id)
    try:
        # 平台进程 cwd=/app（含 mcp_agent 包），进程内 import
        sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
        from mcp_agent.agent import Agent

        def on_delta(text, kind):
            q.put(("delta", {"text": text, "kind": kind}))

        tool_seq = []   # 本轮工具序列（经验提炼用）

        def on_tool(step):
            # args 保留对象（json.dumps 序列化）——前端要解析 serial 做画面跟随；
            # 之前 str() 成 Python repr（单引号）导致前端 JSON.parse 失败、跟随失效
            rec = {"tool": step.get("tool"),
                   "args": step.get("args") or {}}
            if step.get("image"):
                rec["image"] = _shrink_image(step["image"])
            q.put(("step", rec))
            # 记录精简工具序列
            try:
                args = step.get("args") or {}
                brief = {k: v for k, v in args.items() if k != "serial"}
                tool_seq.append(f"{step.get('tool')}({str(brief)[:60]})")
            except Exception:
                pass

        agent = Agent()
        agent.s.api_base = cfg.get("api_base") or agent.s.api_base
        agent.s.model = cfg.get("model") or agent.s.model
        agent.s.api_key = cfg.get("api_key") or agent.s.api_key
        agent.s.default_serial = cfg.get("default_serial") or agent.s.default_serial
        try:
            agent.s.max_steps = max(1, min(200, int(cfg.get("max_steps") or 40)))
        except (TypeError, ValueError):
            pass  # 非法值用默认 40

        with _lock:
            conv_id = _run.get("conv_id") or ""
        # 历史来源：绑定了会话 → 从会话读（多轮上下文延续，DeepSeek 式）；
        # 无会话（CLI/兼容）→ 用运行态 history
        if conv_id:
            # 后台线程需自行推 app context（与 _find_experiences 同思路）
            with _flask_app.app_context():
                msgs = _conv_msgs(conv_id)
            history = [{"role": m.get("role"), "content": m.get("content", "")[:4000]}
                       for m in (msgs or []) if m.get("role") in ("user", "assistant")]
            history = history[-24:]
            with _lock:
                _run["history"] = list(history)
        else:
            with _lock:
                history = list(_run.get("history") or [])
        target = serial or cfg.get("default_serial") or ""
        stop_evt = _stop_events.get(run_id)

        # 经验检索：相似历史任务的操作配方注入 system（自进化记忆）
        exp_ctx = _find_experiences(prompt)
        if exp_ctx:
            _log.info("命中历史经验，注入参考配方")
            q.put(("step", {"tool": "🧠 经验记忆",
                            "args": f"命中 {exp_ctx.count(chr(10) + '- ')} 条同类历史经验，已注入参考",
                            "image": None}))

        mcp_url = getattr(getattr(agent, "s", None), "mcp_url", "")

        def _mcp_unreachable(err):
            """MCP 客户端在工具服务不可达/返回非 MCP 响应时只抛含糊字样，
            识别出来以便给用户明确指引（默认端口 8033）。"""
            m = str(err).lower()
            return any(k in m for k in (
                "server returned an error response", "connecterror",
                "connection refused", "all connection attempts failed",
                "connection error", "server disconnected",
                "network is unreachable", "connect timeout"))

        async def _execute():
            try:
                await agent._load_tools()
            except Exception as e:
                if _mcp_unreachable(e):
                    raise RuntimeError(
                        f"MCP server(8033) 不可达：无法加载设备工具（{mcp_url}）。"
                        f"请确认 MCP server 已启动（本机可运行 "
                        f"python -m mcp_server.mcp_server，默认端口 8033）") from e
                raise
            try:
                return await agent.run_stream(prompt, target,
                                              history=history,
                                              on_delta=on_delta, on_tool=on_tool,
                                              should_stop=lambda: bool(
                                                  stop_evt and stop_evt.is_set()),
                                              extra_context=exp_ctx)
            except Exception as e:
                if _mcp_unreachable(e):
                    raise RuntimeError(
                        f"MCP server(8033) 连接中断：AI 无法继续操作设备（{mcp_url}）") from e
                raise

        # 整体超时保护：卡死时结束，释放单实例
        answer = asyncio.run(asyncio.wait_for(_execute(), timeout=900))
        with _lock:
            _run["state"] = "done"
            _run["answer"] = answer
            # 追加本轮进历史（多轮连续性；上限 12 轮防 token 膨胀）
            hist = _run.setdefault("history", [])
            hist.append({"role": "user", "content": prompt[:2000]})
            hist.append({"role": "assistant", "content": (answer or "")[:4000]})
            _run["history"] = hist[-24:]
        # 会话落库：本轮追加写回（新会话自动用首条消息作标题）。
        # 后台线程 db 访问需 app context。
        if conv_id and _flask_app is not None:
            try:
                with _flask_app.app_context():
                    saved = list(_run["history"])
                    title = None
                    if not db.session.execute(db.text(
                            "SELECT title FROM agent_conversation WHERE id=:i"),
                            {"i": conv_id}).scalar():
                        title = "".join(prompt.split())[:30] or "新会话"
                    _conv_save_messages(conv_id, saved, title=title)
                _log.info(f"会话已保存: {conv_id}（{len(saved)} 条消息）")
            except Exception as e:
                _log.warning(f"会话落库失败: {e}")

        # 自进化：成功执行过工具则提炼配方写入经验。必须在 done 之前完成——
        # done 发出后 SSE 关流，用户就看不到「已写入经验」的提示了。
        # 提炼/保存失败静默（不阻塞、不影响结果），只在成功时推送 🧠 卡片。
        if tool_seq:
            try:
                recipe = _distill_experience(cfg, prompt, " -> ".join(tool_seq))
                if recipe and "配方" not in recipe[:50]:
                    if _qualify_experience(prompt, recipe):
                        if _save_experience(prompt, recipe, " -> ".join(tool_seq)):
                            _log.info("经验已写入记忆库，随事件流提示")
                            q.put(("step", {"tool": "🧠 经验记忆",
                                            "args": "本轮操作已提炼为经验并写入记忆库"
                                                    "（下次相似任务会自动参考）",
                                            "image": None}))
            except Exception as e:
                _log.warning(f"经验保存异常: {e}")
        q.put(("done", {"answer": answer}))
    except Exception as e:
        _log.warning(f"Agent 运行异常: {e}")
        # 诊断：打印消息结构（tool_calls 与 tool 消息配对检查）
        try:
            roles = [m.get("role", "?") for m in agent.messages]
            tcs = sum(1 for m in agent.messages
                      if m.get("tool_calls") and isinstance(m.get("tool_calls"), list))
            tools_msg = sum(1 for m in agent.messages if m.get("role") == "tool")
            _log.warning(f"诊断 messages: {len(agent.messages)} 条 roles={roles[-8:]} "
                         f"tool_calls消息={tcs} tool回应={tools_msg}")
        except Exception:
            pass
        with _lock:
            _run["state"] = "error"
            _run["error"] = f"{type(e).__name__}: {str(e)[:200]}"
        q.put(("error", {"message": str(e)[:200]}))
    finally:
        q.put(None)   # 关闭 SSE
        _queues.pop(run_id, None)
        _stop_events.pop(run_id, None)