"""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} 已保留"})

@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=["POST"])
@admin_required
def agent_run():
    """启动 Agent：{prompt, serial?}。运行中返回 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()
    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 锁兜底）
    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,
                    started=_dt.now().strftime("%H:%M:%S"),
                    answer="", error="")
        # history 保留（同会话多轮对话），由前端「清空对话」调用 clear 重置
        _queues[run_id] = queue.Queue()
        _stop_events[run_id] = threading.Event()
    _log.info(f"Agent 启动: {prompt[:60]} @ {serial or 'default'}")
    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/run", methods=["GET"])
@admin_required
def agent_run_status():
    """当前 Agent 运行状态（多窗口/多人可见）：idle / running + 任务摘要。

    前端轮询它同步「运行中」状态（换浏览器/他人启动的任务也能看到并停止）。
    """
    with _lock:
        return jsonify({"ok": True,
                        "state": _run.get("state", "idle"),
                        "prompt": _run.get("prompt", ""),
                        "serial": _run.get("serial", ""),
                        "started": _run.get("started", "")})


@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:
            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}))

        async def _execute():
            await agent._load_tools()
            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)

        # 整体超时保护：卡死时结束，释放单实例
        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:]

        # 自进化：成功执行过工具则提炼配方写入经验。必须在 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)