Просмотр исходного кода

feat(opencode): 长任务进度上报与无限等待,bot 接入进度回调

Sisyphus Agent 2 месяцев назад
Родитель
Сommit
b848e2b095
2 измененных файлов с 141 добавлено и 27 удалено
  1. 130 26
      运营文案/opencode_client.py
  2. 11 1
      运营文案/xhs_bot_listener.py

+ 130 - 26
运营文案/opencode_client.py

@@ -8,10 +8,10 @@ opencode_client.py — 通过本机常驻 opencode 服务的 HTTP API 进行会
     3. 确保某 chat 的会话存在(build agent)
     3. 确保某 chat 的会话存在(build agent)
     4. 发消息 + 轮询取 assistant 最终文本回复
     4. 发消息 + 轮询取 assistant 最终文本回复
 
 
-权限策略(最安全):
-    - 只为会话启用「只读」类工具(读文件、浏览、文本生成)
-    - 禁用运行命令 / 写文件等需要授权的操作
-    - 因此 opencode 只能做无副作用的分析、生成回复,无法改动本机文件
+权限策略:
+    - 允许 bash/shell(可执行 git 等只读命令,满足「检查同步」类需求)
+    - 仍禁用 write/edit/apply_patch 等写文件操作与任务派发
+    - 因此 opencode 可以分析、执行只读命令、生成回复,但无法改动本机文件
 """
 """
 import io
 import io
 import json
 import json
@@ -40,13 +40,16 @@ WORK_DIR = CFC_ROOT
 # build agent 默认工作于该 project 的模型取服务端默认
 # build agent 默认工作于该 project 的模型取服务端默认
 AGENT = "build"
 AGENT = "build"
 
 
+# 默认模型。opencode 内置 big-pickle 免费额度有限流,显式指定可用的 axon/deepseek。
+MODEL = {"providerID": "axon", "modelID": "deepseek"}
+
 # 只读工具白名单:其余工具(bash 执行、写文件等)一律禁用
 # 只读工具白名单:其余工具(bash 执行、写文件等)一律禁用
 # 此处列出 opencode 内置常用只读工具;未列出的默认不可用(服务端按 tools 参数过滤)
 # 此处列出 opencode 内置常用只读工具;未列出的默认不可用(服务端按 tools 参数过滤)
 READONLY_TOOLS = []
 READONLY_TOOLS = []
 
 
 # 危险的需授权工具,显式禁用(防御性,即便服务端宽松也不给)
 # 危险的需授权工具,显式禁用(防御性,即便服务端宽松也不给)
 FORBIDDEN_TOOLS = [
 FORBIDDEN_TOOLS = [
-    "bash", "shell", "write", "edit", "apply_patch",
+    "write", "edit", "apply_patch",
     "task", "dispatch", "webfetch_post", "chrome_launch",
     "task", "dispatch", "webfetch_post", "chrome_launch",
 ]
 ]
 
 
@@ -122,7 +125,7 @@ def send_prompt(session_id: str, text: str, timeout: float = 30):
     parts = [{"type": "text", "text": text}]
     parts = [{"type": "text", "text": text}]
     body = {
     body = {
         "parts": parts,
         "parts": parts,
-        # 只读约束:明确禁用需授权的执行/写入类工具
+        "model": MODEL,
         "tools": {t: False for t in FORBIDDEN_TOOLS},
         "tools": {t: False for t in FORBIDDEN_TOOLS},
     }
     }
     st, raw = _api("POST", f"/session/{session_id}/prompt_async",
     st, raw = _api("POST", f"/session/{session_id}/prompt_async",
@@ -142,23 +145,80 @@ def _extract_assistant_text(msgs) -> str:
     return texts[-1] if texts else ""
     return texts[-1] if texts else ""
 
 
 
 
+def _extract_progress(msgs, sent_ts: float = 0.0):
+    """从消息列表提取当前进展摘要(含思考 reasoning、工具调用、输出文本)。
+
+    sent_ts:仅提取该时刻之后(本次 prompt 之后)新增的 assistant 消息,
+    防止复用会话时把旧回复当成新进展上报。
+    """
+    items = []
+    for m in reversed(msgs):
+        info = m.get("info", {})
+        if info.get("role") != "assistant":
+            continue
+        m_created = (info.get("time") or {}).get("created", 0)
+        if m_created < sent_ts:
+            continue
+        for p in reversed(m.get("parts", [])):
+            t = p.get("type")
+            if t == "reasoning":
+                txt = (p.get("text") or "").strip()
+                if txt:
+                    items.append(("思考", txt))
+            elif t == "text":
+                txt = (p.get("text") or "").strip()
+                if txt:
+                    items.append(("输出", txt))
+            elif t == "tool":
+                tool = p.get("tool") or {}
+                name = tool.get("name") if isinstance(tool, dict) else str(tool)
+                if name:
+                    items.append(("工具", str(name)))
+            if len(items) >= 3:
+                break
+        if len(items) >= 3:
+            break
+    if not items:
+        return "(处理中...)"
+    return " | ".join(f"[{k}] {v[:120]}" for k, v in items)
+
+
 def wait_for_reply(session_id: str, timeout: float = 180,
 def wait_for_reply(session_id: str, timeout: float = 180,
-                   poll_interval: float = 5.0) -> str:
+                   poll_interval: float = 5.0,
+                   progress_callback=None, progress_interval: float = 120.0,
+                   sent_ts: float = None) -> str:
     """等待会话最新 assistant 回复完成,返回最终文本。
     """等待会话最新 assistant 回复完成,返回最终文本。
 
 
     处理两种情形:
     处理两种情形:
       1) assistant 回复极快,进入本函数时已生成 → 直接看最后一条 assistant 文本
       1) assistant 回复极快,进入本函数时已生成 → 直接看最后一条 assistant 文本
       2) 正常流式:等待「发消息后新增」的 assistant 带文本出现
       2) 正常流式:等待「发消息后新增」的 assistant 带文本出现
+
+    超长任务支持(progress_callback 非 None 时启用):
+      - 每 progress_interval 秒提取进展(含 reasoning)回调给调用方
+      - 若两次回调间无任何进展,向 opencode 提交「在干什么」推动执行
+
+    sent_ts:发送 prompt 时的毫秒时间戳。所有返回路径只接受此时刻之后的
+    assistant 消息,防止在复用会话中错误返回旧消息。
+
+    timeout 为 None 时表示无限等待,直到 assistant 产生最终回复才返回。
     """
     """
-    # 进入时先记录当前消息条数,作为"已有"基线
+
+
+    _sent_ts = sent_ts or 0.0
+
     try:
     try:
         _, raw0 = _api("GET", f"/session/{session_id}/message", timeout=30)
         _, raw0 = _api("GET", f"/session/{session_id}/message", timeout=30)
         before = len(json.loads(raw0)) if raw0 else 0
         before = len(json.loads(raw0)) if raw0 else 0
     except Exception:
     except Exception:
         before = 0
         before = 0
 
 
-    deadline = time.time() + timeout
-    while time.time() < deadline:
+    deadline = (time.time() + timeout) if timeout is not None else None
+    last_report_at = time.time()
+    # 跟踪最后一条 assistant 消息的 created 时间戳,只有它不更新才算真卡住
+    last_asst_ts = 0
+    # nudge 冷却:同一会话 300s 内只允许触发一次
+    last_nudge_at = 0
+    while deadline is None or time.time() < deadline:
         time.sleep(poll_interval)
         time.sleep(poll_interval)
         try:
         try:
             st, raw = _api("GET", f"/session/{session_id}/message", timeout=30)
             st, raw = _api("GET", f"/session/{session_id}/message", timeout=30)
@@ -170,33 +230,77 @@ def wait_for_reply(session_id: str, timeout: float = 180,
         if not msgs:
         if not msgs:
             continue
             continue
 
 
-        # 情况1:最后一条是 assistant 且有文本(覆盖超快回复:before 已含该条)
+        # 快速路径:最后一条消息是新 assistant 且有文本 → 直接返回
         last = msgs[-1]
         last = msgs[-1]
         if last.get("info", {}).get("role") == "assistant":
         if last.get("info", {}).get("role") == "assistant":
-            lt = [p.get("text", "") for p in last.get("parts", [])
-                  if p.get("type") == "text" and (p.get("text") or "").strip()]
-            if lt:
-                return lt[-1]
-
-        # 情况2:出现新增的 assistant 消息(条数 > 基线)
+            last_created = (last.get("info", {}).get("time") or {}).get("created", 0)
+            if last_created >= _sent_ts:
+                lt = [p.get("text", "") for p in last.get("parts", [])
+                      if p.get("type") == "text" and (p.get("text") or "").strip()]
+                if lt:
+                    return lt[-1]
+
+        # 常规路径:遍历消息找第一个(最新)满足条件的 assistant 文本
         if len(msgs) > before:
         if len(msgs) > before:
             for m in reversed(msgs):
             for m in reversed(msgs):
-                role = m.get("info", {}).get("role")
-                if role == "assistant":
-                    txts = [p.get("text", "") for p in m.get("parts", [])
-                            if p.get("type") == "text" and (p.get("text") or "").strip()]
-                    if txts:
-                        return txts[-1]
+                info = m.get("info", {})
+                if info.get("role") != "assistant":
+                    continue
+                m_created = (info.get("time") or {}).get("created", 0)
+                if m_created < _sent_ts:
+                    continue
+                txts = [p.get("text", "") for p in m.get("parts", [])
+                        if p.get("type") == "text" and (p.get("text") or "").strip()]
+                if txts:
+                    return txts[-1]
+
+        # 超长任务进展上报(含卡住检测)
+        if progress_callback and (time.time() - last_report_at) >= progress_interval:
+            last_report_at = time.time()
+            progress = _extract_progress(msgs, sent_ts=_sent_ts)
+
+            # 找到最新的 assistant 消息的 created 时间戳
+            latest_asst_ts = 0
+            for m in msgs:
+                info = m.get("info", {})
+                if info.get("role") == "assistant":
+                    ts = (info.get("time") or {}).get("created", 0)
+                    if ts > latest_asst_ts:
+                        latest_asst_ts = ts
+
+            if latest_asst_ts == last_asst_ts and latest_asst_ts > _sent_ts:
+                # 真正的卡住:assistant 消息时间戳没有变化
+                now = time.time()
+                if now - last_nudge_at >= 300:
+                    last_nudge_at = now
+                    progress_callback("⏳ 2 分钟无新进展,已向 opencode 询问「在干什么」…")
+                    try:
+                        send_prompt(session_id, "在干什么")
+                    except Exception:
+                        pass
+            else:
+                last_asst_ts = latest_asst_ts
+                progress_callback(f"⏳ 进展:{progress}")
+
     raise TimeoutError(f"等待 opencode 回复超时({timeout}s)")
     raise TimeoutError(f"等待 opencode 回复超时({timeout}s)")
 
 
 
 
-def ask(chat_id: str, text: str, timeout: float = 180) -> str:
-    """高层封装:确保会话 → 发消息 → 等回复 → 返回最终文本。"""
+def ask(chat_id: str, text: str, timeout: float = None,
+        progress_callback=None, progress_interval: float = 120.0) -> str:
+    """高层封装:确保会话 → 发消息 → 等回复 → 返回最终文本。
+
+    timeout 为 None(默认)时无限等待,直到任务结束。
+    """
     sid = ensure_session(chat_id)
     sid = ensure_session(chat_id)
+    # 记录发送时刻(先于发送,保证新消息 created >= sent_ts,旧消息被过滤)
+    sent_ts = time.time() * 1000
     st, raw = send_prompt(sid, text)
     st, raw = send_prompt(sid, text)
     if st != 204:
     if st != 204:
         raise RuntimeError(f"发送到 opencode 失败 status={st}: {raw[:300]}")
         raise RuntimeError(f"发送到 opencode 失败 status={st}: {raw[:300]}")
-    return wait_for_reply(sid, timeout=timeout)
+    return wait_for_reply(sid, timeout=timeout,
+                          progress_callback=progress_callback,
+                          progress_interval=progress_interval,
+                          sent_ts=sent_ts)
 
 
 
 
 if __name__ == "__main__":
 if __name__ == "__main__":

+ 11 - 1
运营文案/xhs_bot_listener.py

@@ -282,7 +282,17 @@ def on_message_receive(data):
                 # 其余消息 -> 转发到该群的 opencode 会话处理(只读分析 + 生成回复)
                 # 其余消息 -> 转发到该群的 opencode 会话处理(只读分析 + 生成回复)
                 user_text = _mention_text(msg)
                 user_text = _mention_text(msg)
                 _log(f"[opencode] chat={chat_id} 转发到 opencode 会话处理")
                 _log(f"[opencode] chat={chat_id} 转发到 opencode 会话处理")
-                reply_text = opencode_client.ask(chat_id, user_text)
+
+                def _progress(text):
+                    try:
+                        feishu.reply_text(msg.message_id, text)
+                    except Exception:
+                        pass
+
+                reply_text = opencode_client.ask(
+                    chat_id, user_text,
+                    timeout=None, progress_callback=_progress, progress_interval=120,
+                )
                 _log(f"[opencode] 得到回复: {reply_text[:120]!r}")
                 _log(f"[opencode] 得到回复: {reply_text[:120]!r}")
                 if reply_text and reply_text.strip():
                 if reply_text and reply_text.strip():
                     feishu.reply_text(msg.message_id, reply_text)
                     feishu.reply_text(msg.message_id, reply_text)