bug修复

task_receiver.py

@@ -12,18 +12,45 @@
 """
 import asyncio
 import logging
-import threading
+import os
 
 from fastapi import APIRouter, HTTPException, Request
 from fastapi.responses import JSONResponse
 
 import gateway_client
-from google.adk.runners import InMemoryRunner
+from google.adk.runners import Runner
+from google.adk.sessions.sqlite_session_service import SqliteSessionService
+from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService
+from google.adk.agents.run_config import RunConfig, StreamingMode
 from google.genai import types as genai_types
 
 logger = logging.getLogger(__name__)
 
 
+def _sessions_db_path() -> str:
+    """返回会话数据库路径（与 chat.py 一致，位于项目 data 目录）。"""
+    here = os.path.dirname(os.path.abspath(__file__))
+    data_dir = os.path.join(here, "data")
+    os.makedirs(data_dir, exist_ok=True)
+    return os.path.join(data_dir, "sessions.db")
+
+
+def _first_sentence(text: str) -> str:
+    """从文本中提取第一个完整句子（用于"Agent 接受任务回复"展示）。
+
+    ADK 流式事件会把回复拆成多个 text 片段，第一个片段常只有一两个字。这里把
+    累积文本按句子结束符切分，返回第一句完整内容；若没有句子结束符则回退为
+    完整文本。
+    """
+    if not text:
+        return ""
+    for sep in ("。", "！", "？", "!", "?", "\n", ";"):
+        idx = text.find(sep)
+        if idx != -1:
+            return text[: idx + 1].strip()
+    return text.strip()
+
+
 def _payload_to_prompt(payload: dict) -> str:
     """将网关任务 payload 转换为 agent 的用户指令。"""
     if not payload:
@@ -48,47 +75,53 @@ def create_task_router(app):
         pass
 
     async def _run_agent_once(prompt: str, request_id: str) -> tuple[str, str]:
-        """用 ADK InMemoryRunner 运行一次 agent，返回 (接受任务后的首条回复, 最终总结)。
+        """运行一次 agent，返回 (接受任务后的首条回复, 最终总结)。
 
-        app 是 App 容器（根 agent 非裸 LlmAgent），run_async 不会自动创建
-        session，需先用 runner.session_service 显式创建。
+        必须使用 Runner + SqliteSessionService + streaming_mode=SSE（与 chat.py
+        一致）：InMemoryRunner 无法驱动带 compaction 配置的 App 容器，会导致
+        LLM 不调用、回复为空（"Root node was cancelled"）。
         """
-        runner = InMemoryRunner(app=app)
-        session_id = f"task-{request_id}"
-        await runner.session_service.create_session(
-            app_name=runner.app_name,
-            user_id="gateway",
-            session_id=session_id,
+        runner = Runner(
+            app=app,
+            session_service=SqliteSessionService(db_path=_sessions_db_path()),
+            artifact_service=InMemoryArtifactService(),
+            auto_create_session=True,
         )
+        session_id = f"task-{request_id}"
+        message = genai_types.Content(parts=[genai_types.Part(text=prompt)])
         texts: list[str] = []
         final_text = ""
         async for event in runner.run_async(
             user_id="gateway",
             session_id=session_id,
-            new_message=genai_types.Content(role="user", parts=[genai_types.Part(text=prompt)]),
+            new_message=message,
+            run_config=RunConfig(streaming_mode=StreamingMode.SSE),
         ):
             # 可中断：每收到一个事件检查一次停止标志
             if gateway_client.is_stop_requested(request_id):
                 raise _TaskCancelled()
-            if event.is_final_response():
-                if event.content and event.content.parts:
-                    for part in event.content.parts:
-                        text = getattr(part, "text", None)
-                        if text:
-                            final_text += text
-                break
+            # 只收集非思考（thought）的用户可见文本：过滤掉 thought 片段
             if event.content and event.content.parts:
                 for part in event.content.parts:
                     text = getattr(part, "text", None)
-                    if text:
+                    is_thought = getattr(part, "thought", False)
+                    if not text or is_thought:
+                        continue
+                    if event.is_final_response():
+                        final_text += text
+                    else:
                         texts.append(text)
-        # 首条回复 = 接受任务后的第一条回应；最终总结 = final response
-        reply = (texts[0] if texts else final_text).strip()
-        summary = final_text.strip() or "\n".join(t for t in texts if t).strip() or "(无输出)"
+        # 最终总结 = final response 文本；首条回复 = 累积中间文本直到完整句子
+        summary = final_text.strip() or "".join(texts).strip() or "(无输出)"
+        reply = _first_sentence("".join(texts)) or summary
         return reply, summary
 
-    def _execute_and_report(request_id: str, payload: dict) -> None:
-        """后台线程：执行 agent，成功后回传 success，异常回传 failed，被取消时回传 cancelled。
+    async def _execute_and_report(request_id: str, payload: dict) -> None:
+        """后台执行：执行 agent，成功后回传 success，异常回传 failed，被取消时回传 cancelled。
+
+        此协程通过 asyncio.create_task 在主事件循环中调度，与 agent 的 MCP
+        session / opentelemetry 上下文保持同一事件循环，避免跨线程/跨 loop 导致的
+        "Root node was cancelled" / "Failed to detach context" 崩溃。
 
         执行过程中每步都检查停止标志（gateway_client.is_stop_requested），一旦收到
         取消指令（task_stop）即中断并回传失败（cancelled），网关 on_result 终态保护
@@ -97,7 +130,7 @@ def create_task_router(app):
         gateway_client.clear_stop_requested(request_id)
         try:
             prompt = _payload_to_prompt(payload)
-            reply, summary = asyncio.run(_run_agent_once(prompt, request_id))
+            reply, summary = await _run_agent_once(prompt, request_id)
             if gateway_client.is_stop_requested(request_id):
                 raise _TaskCancelled()
             gateway_client.report_result(
@@ -107,6 +140,15 @@ def create_task_router(app):
                 progress=100,
                 result={"reply": reply, "output": summary},
             )
+        except asyncio.CancelledError:
+            logger.info("agent task cancelled (loop) request=%s", request_id)
+            gateway_client.report_result(
+                request_id,
+                agent_id=app.name,
+                status="failed",
+                progress=100,
+                error_info="cancelled by user",
+            )
         except _TaskCancelled:
             logger.info("agent task cancelled request=%s", request_id)
             gateway_client.report_result(
@@ -132,6 +174,9 @@ def create_task_router(app):
 
         body 中若携带 cli_session_id，则同时启动该会话的 SSE 停止指令订阅线程，
         用于接收网关取消任务时下发的 task_stop。
+
+        执行在 asyncio.create_task 中调度（与 MCP session 同事件循环），
+        不再使用新线程 + asyncio.run，避免跨事件循环导致 agent 崩溃。
         """
         body = await request.json()
         if body.get("auth") != gateway_client.GATEWAY_AUTH:
@@ -140,12 +185,10 @@ def create_task_router(app):
         cli_session_id = body.get("cli_session_id")
         if cli_session_id:
             gateway_client.start_stop_listener(cli_session_id)
-        threading.Thread(
-            target=_execute_and_report,
-            args=(request_id, payload),
+        asyncio.create_task(
+            _execute_and_report(request_id, payload),
             name=f"task-{request_id[:8]}",
-            daemon=True,
-        ).start()
+        )
         logger.info("task received request=%s payload=%s", request_id, payload)
         return JSONResponse(status_code=202, content={"ok": True, "request_id": request_id, "status": "accepted"})