feat(AI 建任务): 直接创建 + 草稿沉淀 + MCP de_snapshot；修「建任务页收不到 done」

用户报的"探索完无法点击创建任务"真因：一个 run 的事件原先只有**一条** queue.Queue，聊天页与建任务页同时开着时两个 EventSource 会**瓜分**它——建任务页的回放卡在中间、 `done` 被聊天页取走 → 永远等不到草稿，页面上自然没有可点的"创建"。一、修（根因 + 表现） - `web/agent_api.py` 新增 `_Fanout`：**每个订阅者一个专属队列**，多开页面各看各的，还带单轮事件缓冲（晚订阅/刷新重连也能补齐回放，终止事件一定送达）。实测两路订阅者收到完全一致的 1039 条事件（含 done）。 - `static/admin/agent.js`：断线重连的兜底订阅也按 `mode` 让开（此前漏了这一处）。二、补齐上一批的三项 - **「直接创建」**：`POST /api/agent/task_draft/create`（草稿体只在服务端、创建前再校验一次、成功后清草稿避免重复建）+ 草稿预览里的「✓ 直接创建任务」按钮 + 「探索完直接创建任务」勾选框。 - **草稿沉淀经验/动作**：designer 轮次也走 `_distill_experience/_distill_actions`，但**只在草稿通过校验时**（没走通的试错不入库，免得把误点当经验）。 - **MCP `de_snapshot`**（第 20 个工具）：截图+元素树一次取齐（省一次来回、不会因界面在动而错位），附带 `screen_state`/`unstable`；两套提示词都改为优先用它。平台侧 `/api/uiauto/snapshot` 随之多返回 `screen_state`。三、文档 - AI_TASK_GEN §10：§10.3 记两个 bug 的真因与修法、§10.4 三项标完成、§10.5 剩余项。 - AI_CONSOLE（扇出语义、多页面同时看一轮）、API（task_draft/create、snapshot 字段）、 MCP/MCP_DESIGN/staffdeck/README/ARCHITECTURE：工具数 19→20 + de_snapshot 条目。 - backlog：记一条新发现的缺陷——`mcp_server/platform_client._login()` 会把"登录页 200" 当成登录成功（现场进程缺 `MCP_PLATFORM_PASS` 时表现为含糊的 platform_unavailable）。自测：真机浏览器端到端（勾上"探索完直接创建"）→ 探索 12 步 → 草稿 → 自动建任务成功； `de_snapshot` 直连真机校验；校验器 21 条用例、扇出单元用例、本地工具契约用例全绿。自测产生的任务/草稿已全部清理（未碰用户既有数据）。
2026-09-14 08:14:11 +08:00
parent 46e6ea1f37
commit ed9e8bacb1
18 changed files with 384 additions and 67 deletions
@@ -58,7 +58,7 @@ _run = {"id": None, "state": "idle", "prompt": "", "serial": "",
        "draft_error": "",    # designer：草稿被拦下的原因（errors 拼接）
        "warnings": [],       # designer：草稿的提醒项（不拦，前端醒目展示）
        "history": []}   # 多轮对话历史 [{role: user|assistant, content}]
-_queues = {}            # run_id -> queue.Queue（SSE 消费者读取）
+_queues = {}            # run_id -> _Fanout（SSE 消费者各自订阅，见下）
 _stop_events = {}       # run_id -> threading.Event（用户中断）
 _lock = threading.Lock()
 _flask_app = None       # web_server 注册时注入（后台线程 db 操作需 app context）
@@ -69,6 +69,62 @@ def set_app(app):
    _flask_app = app


+class _Fanout:
+    """一个 Agent 轮次的事件，扇出给**所有**订阅者（每个订阅者一个独立队列）。
+
+    历史实现是"一个 run 一个 queue.Queue"：第二个页面一订阅，两个 EventSource 就
+    开始**瓜分**同一条队列——谁先取到算谁的。2026-09-14 实测踩到：同一个浏览器里
+    聊天页与建任务页同时开着，建任务页的回放卡到一半就不动了、`done` 事件被另一个
+    页面取走 → 永远等不到草稿（表现为"探索完没法创建任务"）。
+    扇出之后，多开几个页面都各看各的，谁都不丢事件。
+    """
+
+    # 单轮事件缓冲上限：新订阅者从这里补齐"订阅之前"的回放（刷新/重连不丢进度）。
+    # 只留最近 _HISTORY_KEEP 条，避免长任务把内存拖大；终止事件（done/error）永远保留。
+    _HISTORY_KEEP = 400
+
+    def __init__(self):
+        self._subs = []
+        self._history = []
+        self._closed = False
+        self._lock = threading.Lock()
+
+    def subscribe(self):
+        """新订阅者拿一个专属队列（**预填已发生的事件**）；轮次已结束则返回 None。"""
+        with self._lock:
+            if self._closed:
+                return None
+            q = queue.Queue()
+            for item in self._history:
+                q.put(item)
+            self._subs.append(q)
+            return q
+
+    def unsubscribe(self, q):
+        with self._lock:
+            if q in self._subs:
+                self._subs.remove(q)
+
+    def put(self, item):
+        with self._lock:
+            subs = list(self._subs)
+            self._history.append(item)
+            # 只留最近 N 条。终止事件（done/error 及其后的哨兵）永远在尾部，
+            # 按这个规则裁剪不会把它们丢掉。
+            if len(self._history) > self._HISTORY_KEEP:
+                del self._history[:len(self._history) - self._HISTORY_KEEP]
+        for q in subs:
+            q.put(item)
+
+    def close(self):
+        """结束：给每个订阅者发终止哨兵（None）。"""
+        with self._lock:
+            self._closed = True
+            subs, self._subs = self._subs, []
+        for q in subs:
+            q.put(None)
+
+
 def _meta_get(key):
    # 走方言中立助手（app_meta 的列名 key 在 MySQL 里是保留字，裸 SQL 会语法错）
    from core.db_config import meta_get
@@ -1184,7 +1240,7 @@ def agent_run():
                    answer="", error="", usage={},
                    draft=None, draft_error="", warnings=[])
        # history 保留（多轮上下文），由会话/「新建会话」管理
-        _queues[run_id] = queue.Queue()
+        _queues[run_id] = _Fanout()
        _stop_events[run_id] = threading.Event()
    _log.info(f"Agent 启动[{mode}]: {prompt[:60]} @ {serial or 'default'} "
              f"conv={conv_id or '-'}")
@@ -1276,21 +1332,29 @@ def agent_stream():
    with _lock:
        if run_id != _run["id"]:
            return jsonify({"ok": False, "error": "run_id 不存在"}), 404
-        q = _queues.get(run_id)
+        fan = _queues.get(run_id)
+    # 每个订阅者一个专属队列（多开页面互不抢事件）；轮次已结束 → 404
+    q = fan.subscribe() if fan is not None else None
+    if q is None:
+        return jsonify({"ok": False, "error": "本轮已结束"}), 404

    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
+        try:
+            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
+        finally:
+            # 页面关掉/断线时摘掉自己的队列，别让扇出往里堆没人读的事件
+            fan.unsubscribe(q)

    return Response(gen(), mimetype="text/event-stream",
                    headers={"Cache-Control": "no-cache",
@@ -1408,6 +1472,61 @@ def agent_task_draft_save():
    return jsonify({"ok": True, "msg": "草稿已保存", "saved": saved})


+@bp.route("/api/agent/task_draft/create", methods=["POST"])
+@admin_required
+def agent_task_draft_create():
+    """把暂存的草稿**直接创建**成任务（AI 建任务页的「直接创建任务」）。
+
+    这是"AI 直接建任务"的落点，两条设计约束：
+      - **草稿体只在服务端**（app_meta）：客户端只能传 overrides，不能把任务内容
+        再提交一遍——少一个可被篡改的入口；
+      - 创建前**再校验一次**（草稿可能是人工改过的），通过后走与「新建任务」
+        **完全相同**的落库路径 `context.mgr.add_job`。
+    创建成功后清掉草稿（同一份草稿不该被重复建成多个任务）。
+    """
+    from core.task_draft import validate_draft
+    from web import context
+    saved = _load_draft_meta()
+    if not saved or not saved.get("draft"):
+        return jsonify({"ok": False, "error": "没有可创建的草稿，请先探索一次"}), 400
+    data = request.json or {}
+    overrides = data.get("overrides") if isinstance(data.get("overrides"), dict) else {}
+    groups, pool = _draft_env()
+    res = validate_draft(saved["draft"], groups=groups, pool=pool,
+                         default_serial=(data.get("serial") or "").strip(),
+                         overrides=overrides)
+    if not res["ok"]:
+        return jsonify({"ok": False, "error": "草稿校验未通过（请在编辑器里核对）",
+                        "errors": res["errors"]}), 400
+    task = res["draft"]["task"]
+    params = task["params"]
+    params.setdefault("skip_offline", True)   # 与编辑器保存时一致
+    try:
+        job = context.mgr.add_job(name=task["name"], task_type=task["task_type"],
+                                  target=task["target"], params=params,
+                                  schedule=task["schedule"], retry=task["retry"],
+                                  enabled=bool(task.get("enabled", True)))
+    except Exception as e:
+        _log.exception("AI 草稿创建任务失败")
+        return jsonify({"ok": False, "error": f"创建失败: {e}"}), 500
+    # 用掉就清掉，避免重复创建
+    _meta_put(_DRAFT_META_KEY, "")
+    try:
+        db.session.commit()
+    except Exception:
+        db.session.rollback()
+    with _lock:
+        _run["draft"] = None
+        _run["warnings"] = []
+    d = job.to_dict()
+    nr = context.mgr.next_run_of(job)
+    d["next_run"] = nr.strftime("%Y-%m-%d %H:%M") if nr else None
+    _log.info(f"AI 草稿已创建任务: {job.name}（{job.id}）"
+              f"{'，' + str(len(res['warnings'])) + ' 条提醒' if res['warnings'] else ''}")
+    return jsonify({"ok": True, "msg": f"任务「{job.name}」已创建",
+                    "job": d, "warnings": res["warnings"]})
+
+
@bp.route("/api/agent/task_draft/clear", methods=["POST"])
@admin_required
 def agent_task_draft_clear():
@@ -1713,13 +1832,14 @@ def _agent_thread(run_id, prompt, serial, cfg, mode="chat", settings=None):
            except Exception as e:
                _log.warning(f"会话落库失败: {e}")

-        # 自进化：AI 建任务（designer）**不沉淀**——探索轨迹是为"写出一条任务"服务的，
-        # 不是一次成功操作套路；沉淀它会把探索期的误点/试错当成经验，污染记忆库。
-        # （把"草稿→经验/动作"作为独立里程碑，见 doc/AI_TASK_GEN.md §6 P1）
-        # 必须在 done 之前完成——
+        # 自进化：成功执行过工具则提炼配方写入经验。必须在 done 之前完成——
        # done 发出后 SSE 关流，用户就看不到「已写入经验」的提示了。
        # 提炼/保存失败静默（不阻塞、不影响结果），只在成功时推送 🧠 卡片。
-        if tool_seq and not designer:
+        # 建任务（designer）模式也沉淀，但**只在草稿通过校验时**：那说明这轮探索
+        # 确实走通了一条路。没有草稿的探索（试错、半途而废）不入库，免得把误点当经验。
+        with _lock:
+            draft_ok = bool(_run.get("draft"))
+        if tool_seq and (not designer or draft_ok):
            try:
                recipe = _clean_recipe(_distill_experience(cfg, prompt, " -> ".join(tool_seq)))
                if recipe:
@@ -1774,6 +1894,10 @@ def _agent_thread(run_id, prompt, serial, cfg, mode="chat", settings=None):
                pass
        q.put(("error", {"message": str(e)[:200]}))
    finally:
-        q.put(None)   # 关闭 SSE
+        # 关闭 SSE：扇出把终止哨兵发给**每个**订阅者
+        if isinstance(q, _Fanout):
+            q.close()
+        else:
+            q.put(None)
        _queues.pop(run_id, None)
        _stop_events.pop(run_id, None)