feat: TURN watchdog — recover stalled dashboard/IM chat turns

CHANGELOG.md

@@ -6,6 +6,11 @@
 
 ## [Unreleased]
 
+### 新增
+
+- TURN watchdog：Dashboard WS 与 IM 通道回合活性看门狗——模型/工具分阶段无进展超时与模型阶段总时长上限，超时自动推 error/done 帧（IM 推超时提示）、cancel_stream 释放会话锁；pre-enqueue 准备阶段超时兜底
+
+
 ## [0.9.32] - 2026-09-06
 
 ### 新增

src/octop/api/routers/chat/ws.py

@@ -2,9 +2,11 @@
 
 from __future__ import annotations
 
+import asyncio
 import contextlib
 import json
 import logging
+import os
 import uuid
 from typing import Any
 
@@ -26,6 +28,19 @@
 
 logger = logging.getLogger(__name__)
 
+_DEFAULT_PREPARE_TIMEOUT_SECONDS = 60
+
+
+def _prepare_timeout() -> float:
+    """Timeout for the pre-enqueue turn preparation phase (env-tunable)."""
+    raw = os.environ.get("OCTOP_TURN_PREPARE_TIMEOUT_SECONDS", "").strip()
+    if not raw:
+        return _DEFAULT_PREPARE_TIMEOUT_SECONDS
+    try:
+        return max(1.0, float(raw))
+    except ValueError:
+        return _DEFAULT_PREPARE_TIMEOUT_SECONDS
+
 router = APIRouter()
 
 
@@ -145,12 +160,26 @@ async def send_frame(frame: dict[str, Any]) -> None:
                 continue
 
             try:
-                prepared = await prepare_dashboard_turn(
-                    server,
-                    agent_id=agent_id,
-                    user=user,
-                    turn=turn,
+                # prepare 阶段含 MCP 加载等网络 IO，可能挂住（watchdog 看不到
+                # 未入队的 turn）——加超时让客户端立即拿到报错而不是无回帧。
+                prepared = await asyncio.wait_for(
+                    prepare_dashboard_turn(
+                        server,
+                        agent_id=agent_id,
+                        user=user,
+                        turn=turn,
+                    ),
+                    timeout=_prepare_timeout(),
                 )
+            except asyncio.TimeoutError:
+                await send_frame(
+                    {
+                        "type": "error",
+                        "message": "turn preparation timed out (MCP/skills load took too long)",
+                    }
+                )
+                await send_frame({"type": "done"})
+                continue
             except OctopError as exc:
                 await send_frame({"type": "error", "message": str(exc)})
                 await send_frame({"type": "done"})

src/octop/infra/gateway/gateway.py

@@ -34,6 +34,7 @@
     WebSocketChannel,
     WebSocketHub,
 )
+from octop.infra.gateway.ws.turn_watchdog import TurnWatchdog, watchdog_disabled
 from octop.infra.utils.locale import DEFAULT_LOCALE, Locale
 
 if TYPE_CHECKING:
@@ -120,6 +121,7 @@ def __init__(
         self._cli_channel: CliChannel | None = None
         self._runtime_status: dict[str, ChannelRuntimeStatus] = {}
         self._history_backfill = HistoryBackfillQueue()
+        self._turn_watchdog: TurnWatchdog | None = None
 
     def replace_repos(self, repos: RepoBundle) -> None:
         """Point channel/thread persistence at a rebound control-plane pool."""
@@ -225,6 +227,18 @@ async def boot(self) -> None:
         )
         await self._channel_manager.add_channel(self._ws_channel)
 
+        if not watchdog_disabled():
+            self._turn_watchdog = TurnWatchdog(
+                hub=self._ws_hub,
+                agent_manager=self._agent_manager,
+                audit_repo=self._repos.audit_repo,
+                gateway=self,
+            )
+            self._turn_watchdog.start()
+            logger.info("TurnWatchdog started (interval=%ds)", self._turn_watchdog._interval)
+        else:
+            logger.warning("TurnWatchdog disabled via OCTOP_TURN_WATCHDOG_DISABLED=1")
+
         self._cli_channel = CliChannel(
             self._processor,
             hub=self._cli_hub,
@@ -283,6 +297,9 @@ async def reload_channels_from_db(self) -> None:
 
     async def shutdown(self) -> None:
         await self._history_backfill.close()
+        if self._turn_watchdog is not None:
+            await self._turn_watchdog.stop()
+            self._turn_watchdog = None
         if self._channel_manager:
             await self._channel_manager.stop()
         self._channel_manager = None

src/octop/infra/gateway/process/processor.py

@@ -711,6 +711,7 @@ async def __call__(self, msg: InboundMessage) -> AsyncIterator[MessageEvent]:
         usage_tracker = UsageTracker()
         history_tracker = TurnHistoryTracker.from_request(request)
         projection_state = StreamProjectionState()
+        self._mark_turn_active(thread_id, agent_id=agent_id, session_key=session_key)
         try:
             async for ev in project_stream(
                 self._agent_manager,
@@ -730,6 +731,15 @@ async def __call__(self, msg: InboundMessage) -> AsyncIterator[MessageEvent]:
                     channel_type=channel_type,
                 ),
             ):
+                # IM turn progress: every event counts, and TOOL_START/TOOL_END
+                # flip the tool phase so the watchdog uses the longer tool
+                # stall threshold while a tool is running (same semantics as
+                # the dashboard WS path).
+                self._mark_turn_progress(thread_id)
+                if ev.type == MessageEventType.TOOL_START:
+                    self._mark_tool_state(thread_id, True)
+                elif ev.type == MessageEventType.TOOL_END:
+                    self._mark_tool_state(thread_id, False)
                 yield ev
             stream_ok = True
             hitl_paused = projection_state.hitl_paused
@@ -746,8 +756,40 @@ async def __call__(self, msg: InboundMessage) -> AsyncIterator[MessageEvent]:
                     usage=usage_tracker.usage,
                 )
                 self._record_turn_history(thread_id, history_tracker)
+        self._mark_turn_idle(thread_id)
         yield MessageEvent.completed()
 
+    # -- TURN watchdog registration for IM turns ------------------------------
+    # IM turns stream through __call__ (MessageEvent). They register into the
+    # same in-process hub table as dashboard WS turns so the single TurnWatchdog
+    # scan covers both transports. The hub is a plain dict — no WebSocket
+    # dependency — and watchdog recovery (cancel_stream) works identically.
+
+    def _hub(self) -> Any:
+        if self._gateway is None:
+            return None
+        return getattr(self._gateway, "ws_hub", None)
+
+    def _mark_turn_active(self, thread_id: str, *, agent_id: str, session_key: str) -> None:
+        hub = self._hub()
+        if hub is not None:
+            hub.mark_turn_active(thread_id, agent_id=agent_id, session_key=session_key)
+
+    def _mark_turn_progress(self, thread_id: str) -> None:
+        hub = self._hub()
+        if hub is not None:
+            hub.mark_turn_progress(thread_id)
+
+    def _mark_tool_state(self, thread_id: str, in_tool: bool) -> None:
+        hub = self._hub()
+        if hub is not None:
+            hub.mark_tool_state(thread_id, in_tool)
+
+    def _mark_turn_idle(self, thread_id: str) -> None:
+        hub = self._hub()
+        if hub is not None:
+            hub.mark_turn_idle(thread_id)
+
     # -- Raw harness-chunk stream (Dashboard WS, etc.) -------------------------
     # IM channels (DingTalk, Feishu, …) stream via __call__ → MessageEvent instead.