diff --git a/CHANGELOG.md b/CHANGELOG.md index c22567854..ca5e0fefe 100644 --- a/CHANGELOG.md +++ b/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 ### 新增 diff --git a/src/octop/api/routers/chat/ws.py b/src/octop/api/routers/chat/ws.py index 910c3fda7..b16294a34 100644 --- a/src/octop/api/routers/chat/ws.py +++ b/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"}) diff --git a/src/octop/infra/gateway/gateway.py b/src/octop/infra/gateway/gateway.py index 16a2067f8..9c7bb5985 100644 --- a/src/octop/infra/gateway/gateway.py +++ b/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 diff --git a/src/octop/infra/gateway/process/processor.py b/src/octop/infra/gateway/process/processor.py index 9f4488bfa..016fa2c39 100644 --- a/src/octop/infra/gateway/process/processor.py +++ b/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. diff --git a/src/octop/infra/gateway/ws/turn_watchdog.py b/src/octop/infra/gateway/ws/turn_watchdog.py new file mode 100644 index 000000000..7f99d1f8d --- /dev/null +++ b/src/octop/infra/gateway/ws/turn_watchdog.py @@ -0,0 +1,289 @@ +"""TurnWatchdog — detect stalled Dashboard turns and recover them. + +Octop 的流式链路（WS user_turn → gateway worker → iter_turn_chunks → +agent.stream → LLM provider）本身没有超时：LLM 上游请求挂起（连接黑洞、 +200 无字节、慢流）、或工具执行静默挂起（bash 卡死、网络 IO 悬挂）时， +turn 会永久卡住——前端无回帧、session lock 被永久持有、同一会话后续 +消息全部排队等待一把永远不释放的锁。 + +看门狗周期性扫描 WebSocketHub 登记的 active turns： +- 无进展超时（``OCTOP_TURN_STALL_SECONDS``，默认 300）：模型阶段自最后 + 一个 chunk 起无任何产出 → 判定卡死； +- 工具阶段无进展超时（``OCTOP_TURN_TOOL_STALL_SECONDS``，默认 900）： + 工具执行中不产 chunk 是正常态（bash 跑 10 分钟无中间帧），阈值放宽； +- 模型阶段总时长超时（``OCTOP_TURN_MAX_SECONDS``，默认 1800）：只累计 + 模型阶段时间（工具段不计入），防模型死循环，不误杀长工具链。 + +触发后执行恢复（顺序很重要）： +1. 标记 notified + 从 hub 活登记表摘除（后续 chunk 不再更新进展，重复 + 触发被 mark_turn_active 的 notified 分支挡住）； +2. 向 thread 订阅者推送 error + done 帧——客户端立刻回到可交互态，不再 + 无限等待； +3. cancel_stream —— harness 的 cancel 是可靠的（``_iter_until_cancelled`` + 能打断被阻塞的 ``__anext__``，即 LLM 请求挂起也能打断），取消后正常 + 链路收尾、session lock 释放，排队的后续消息继续处理。 + +notified 记录由 harness 收尾 finally 的 mark_turn_idle 删除；若极端情况 +下 harness 收尾不了（进程内卡死），``OCTOP_TURN_NOTIFIED_GRACE_SECONDS`` +（默认 120）后强制清理，避免该 thread 永久无法登记新 turn。 + +参数全部环境变量可配（对齐 OCTOP_BROWSER_IDLE_TIMEOUT_MINUTES 惯例）： +- ``OCTOP_TURN_STALL_SECONDS`` 模型阶段无进展超时（秒，默认 300） +- ``OCTOP_TURN_TOOL_STALL_SECONDS`` 工具阶段无进展超时（秒，默认 900） +- ``OCTOP_TURN_MAX_SECONDS`` 模型阶段总时长上限（秒，默认 1800） +- ``OCTOP_TURN_NOTIFIED_GRACE_SECONDS`` 收尾宽限（秒，默认 120） +- ``OCTOP_TURN_WATCHDOG_INTERVAL_SECONDS`` 扫描间隔（秒，默认 15） +- ``OCTOP_TURN_WATCHDOG_DISABLED=1`` 逃生阀，彻底关闭 +""" + +from __future__ import annotations + +import asyncio +import contextlib +import logging +import os +import time +from typing import Any + +from octop.infra.gateway.ws.ws_hub import WebSocketHub + +logger = logging.getLogger(__name__) + +DEFAULT_STALL_SECONDS = 300 +DEFAULT_TOOL_STALL_SECONDS = 900 +DEFAULT_MAX_SECONDS = 1800 +DEFAULT_NOTIFIED_GRACE_SECONDS = 120 +DEFAULT_INTERVAL_SECONDS = 15 + + +def _env_int(name: str, default: int) -> int: + raw = os.environ.get(name, "").strip() + if not raw: + return default + try: + return max(1, int(raw)) + except ValueError: + logger.warning("invalid %s=%r, falling back to %d", name, raw, default) + return default + + +def watchdog_disabled() -> bool: + return os.environ.get("OCTOP_TURN_WATCHDOG_DISABLED", "").strip() == "1" + + +class TurnWatchdog: + """Background task that scans active dashboard turns for stalls.""" + + def __init__( + self, + *, + hub: WebSocketHub, + agent_manager: Any, + audit_repo: Any | None = None, + gateway: Any | None = None, + stall_seconds: int | None = None, + tool_stall_seconds: int | None = None, + max_seconds: int | None = None, + notified_grace_seconds: int | None = None, + interval_seconds: int | None = None, + ) -> None: + self._hub = hub + self._agent_manager = agent_manager + self._audit_repo = audit_repo + self._gateway = gateway + self._stall_seconds = ( + stall_seconds + if stall_seconds is not None + else _env_int("OCTOP_TURN_STALL_SECONDS", DEFAULT_STALL_SECONDS) + ) + self._tool_stall_seconds = ( + tool_stall_seconds + if tool_stall_seconds is not None + else _env_int("OCTOP_TURN_TOOL_STALL_SECONDS", DEFAULT_TOOL_STALL_SECONDS) + ) + self._max_seconds = ( + max_seconds + if max_seconds is not None + else _env_int("OCTOP_TURN_MAX_SECONDS", DEFAULT_MAX_SECONDS) + ) + self._notified_grace = ( + notified_grace_seconds + if notified_grace_seconds is not None + else _env_int("OCTOP_TURN_NOTIFIED_GRACE_SECONDS", DEFAULT_NOTIFIED_GRACE_SECONDS) + ) + self._interval = ( + interval_seconds + if interval_seconds is not None + else _env_int("OCTOP_TURN_WATCHDOG_INTERVAL_SECONDS", DEFAULT_INTERVAL_SECONDS) + ) + self._task: asyncio.Task[None] | None = None + self._started_at: float | None = None + + def start(self) -> None: + """Launch the scan loop (idempotent).""" + if self._task is not None and not self._task.done(): + return + self._started_at = time.monotonic() + self._task = asyncio.create_task(self._scan_loop(), name="turn-watchdog") + logger.info( + "TurnWatchdog started: stall=%ss tool_stall=%ss model_max=%ss " + "notified_grace=%ss interval=%ss", + self._stall_seconds, + self._tool_stall_seconds, + self._max_seconds, + self._notified_grace, + self._interval, + ) + + async def stop(self) -> None: + """Cancel the scan loop (idempotent).""" + task = self._task + self._task = None + if task is None or task.done(): + return + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + async def _scan_loop(self) -> None: + try: + while True: + try: + self._scan_once() + except Exception: # noqa: BLE001 - watchdog must never die + logger.exception("TurnWatchdog scan failed") + await asyncio.sleep(self._interval) + except asyncio.CancelledError: + logger.info("TurnWatchdog stopped") + raise + + def _scan_once(self) -> None: + now = time.monotonic() + self._sweep_notified(now) + for thread_id, record in self._hub.snapshot_active_turns().items(): + if record.notified: + continue + stalled_for = now - record.last_progress_at + stall_limit = self._tool_stall_seconds if record.in_tool else self._stall_seconds + model_time = record.model_time(now) + if stalled_for < stall_limit and model_time < self._max_seconds: + continue + if stalled_for >= stall_limit: + reason = ( + f"no output for {stalled_for:.0f}s " + f"({'tool' if record.in_tool else 'model'} threshold {stall_limit}s)" + ) + else: + reason = ( + f"model phase exceeded {model_time:.0f}s " + f"(limit {self._max_seconds}s)" + ) + self._recover(thread_id, record, reason) + + def _sweep_notified(self, now: float) -> None: + """Force-remove notified records whose harness never wound down. + + Normally the cancelled stream's finally-path calls mark_turn_idle and + deletes the record within a second. If the harness is itself wedged + (no finally, no cancellation response), the notified record would + otherwise block this thread from registering a new turn forever. + """ + for thread_id, record in self._hub.snapshot_notified_turns().items(): + if record.notified_at is None: + record.notified_at = now + continue + if now - record.notified_at <= self._notified_grace: + continue + logger.warning( + "TurnWatchdog force-cleared zombie turn record: agent=%s thread=%s " + "(harness wind-down exceeded %ss grace)", + record.agent_id or "?", + thread_id, + self._notified_grace, + ) + self._hub.mark_turn_idle(thread_id) + + def _recover(self, thread_id: str, record: Any, reason: str) -> None: + """Declare a turn dead: notify client, cancel stream, release slot. + + The record stays in the hub table with ``notified=True`` so a racing + ``mark_turn_active`` (user retry) cannot resurrect the zombie; the + harness finally-path removes it via ``mark_turn_idle`` once the + cancelled stream winds down (or _sweep_notified force-clears it). + """ + record.notified = True + record.notified_at = time.monotonic() + agent_id = record.agent_id or "?" + logger.warning( + "TurnWatchdog recovered stalled turn: agent=%s thread=%s %s", + agent_id, + thread_id, + reason, + ) + self._audit_stall(agent_id, thread_id, reason) + asyncio.ensure_future( + self._notify_and_cancel(thread_id, agent_id, reason, record.session_key or "") + ) + + async def _notify_and_cancel( + self, + thread_id: str, + agent_id: str, + reason: str, + session_key: str = "", + ) -> None: + message = ( + f"Turn watchdog aborted a stalled turn ({reason}). " + "The request may still be running server-side; please retry." + ) + try: + await self._hub.push_to_thread(thread_id, {"type": "error", "message": message}) + await self._hub.push_to_thread(thread_id, {"type": "done"}) + except Exception: # noqa: BLE001 - notify failure must not block cancel + logger.exception("TurnWatchdog notify failed thread=%s", thread_id) + # IM turns have no WS subscribers; push a plain-text notice to the + # channel the message came from so the user is not left hanging. + if session_key and self._gateway is not None: + try: + session = self._gateway.thread_registry.get_session(session_key) + if session is not None and session.channel_id: + await self._gateway.push_text( + session.channel_type, + session.channel_id, + session.to_channel_subject(), + "[回复超时] 已自动取消，请重试。", + ) + except Exception: # noqa: BLE001 + logger.exception("TurnWatchdog IM notify failed session=%s", session_key) + try: + if self._agent_manager is not None and agent_id != "?": + self._agent_manager.cancel_stream(agent_id, thread_id) + except Exception: # noqa: BLE001 + logger.exception("TurnWatchdog cancel_stream failed agent=%s", agent_id) + + def _audit_stall(self, agent_id: str, thread_id: str, reason: str) -> None: + if self._audit_repo is None: + return + try: + self._audit_repo.write( + actor="turn-watchdog", + action="turn.stall.recovered", + target=thread_id, + payload=f"agent={agent_id} {reason}", + ) + except Exception: # noqa: BLE001 - audit must never crash the watchdog + logger.warning("TurnWatchdog audit write failed", exc_info=True) + + @property + def started_at(self) -> float | None: + return self._started_at + + +__all__ = [ + "DEFAULT_INTERVAL_SECONDS", + "DEFAULT_MAX_SECONDS", + "DEFAULT_NOTIFIED_GRACE_SECONDS", + "DEFAULT_STALL_SECONDS", + "DEFAULT_TOOL_STALL_SECONDS", + "TurnWatchdog", +] diff --git a/src/octop/infra/gateway/ws/ws_channel.py b/src/octop/infra/gateway/ws/ws_channel.py index 50ed02cf2..242a75d23 100644 --- a/src/octop/infra/gateway/ws/ws_channel.py +++ b/src/octop/infra/gateway/ws/ws_channel.py @@ -143,9 +143,21 @@ async def handle_inbound(self, raw_payload: Any) -> None: await self._hub.push_to_thread(thread_id, {"type": "done"}) return - self._hub.mark_turn_active(thread_id) + self._hub.mark_turn_active(thread_id, agent_id=str(message.tenant_id or "")) try: async for chunk in processor.iter_turn_chunks(message): + # Phase tracking: a tool that runs for many minutes emits no + # chunks, so tell the watchdog we are inside a tool call and + # let it use the longer tool stall threshold. + ctype = chunk.get("type") + if ctype == "tool_call_chunk": + self._hub.mark_tool_state(thread_id, True) + elif ctype == "tool_result": + self._hub.mark_tool_state(thread_id, False) + # Every produced chunk counts as progress so the TURN watchdog + # can tell "model still thinking / tool still running" from + # "turn stuck with no output at all". + self._hub.mark_turn_progress(thread_id) # Never let delivery failures abort the harness turn — otherwise a # flaky client disconnect can stop generation mid-node and leave # an incomplete checkpoint. diff --git a/src/octop/infra/gateway/ws/ws_hub.py b/src/octop/infra/gateway/ws/ws_hub.py index bc573f03f..83b5adcdc 100644 --- a/src/octop/infra/gateway/ws/ws_hub.py +++ b/src/octop/infra/gateway/ws/ws_hub.py @@ -5,11 +5,55 @@ import asyncio import json import logging +import time from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field from typing import Any logger = logging.getLogger(__name__) + +@dataclass +class ActiveTurn: + """One in-flight dashboard turn, tracked for the TURN watchdog. + + ``last_progress_at`` advances on every chunk the turn produces, so the + watchdog can distinguish "model still thinking/tool still running" from + "turn is stuck with no output at all". + + Tool phases are tracked separately so a long-running tool (bash fetching + pages for 10 minutes emits no chunk) is not treated as a stall: the + watchdog uses ``tool_stall_seconds`` while ``in_tool`` is True, and the + total-turn cap only counts model-phase time (tool time is excluded). + """ + + agent_id: str + thread_id: str + session_key: str = "" + started_at: float = field(default_factory=time.monotonic) + last_progress_at: float = field(default_factory=time.monotonic) + notified: bool = False + notified_at: float | None = None + in_tool: bool = False + tool_segment_started_at: float | None = None + tool_budget_spent: float = 0.0 + # 2026-09-07 修复：同一 thread 多并发 WS 连接（多标签页）各自开启 turn 时， + # 先结束的一方不能把登记删掉（另一 turn 仍在跑）——引用计数，最后一个结束才置 idle。 + refcount: int = 1 + + def __post_init__(self) -> None: + if self.started_at == self.last_progress_at: + self.last_progress_at = self.started_at + + def model_time(self, now: float | None = None) -> float: + """Accumulated model-phase time (total minus tool segments).""" + now = now if now is not None else time.monotonic() + total = now - self.started_at + spent = self.tool_budget_spent + if self.in_tool and self.tool_segment_started_at is not None: + spent += now - self.tool_segment_started_at + return max(0.0, total - spent) + SendFn = Callable[[dict[str, Any]], Awaitable[None]] @@ -38,7 +82,7 @@ def __init__(self) -> None: self._conn_thread: dict[str, str] = {} self._user_conns: dict[int, set[str]] = {} self._conn_user: dict[str, int] = {} - self._active_turns: set[str] = set() + self._active_turns: dict[str, ActiveTurn] = {} def register( self, @@ -100,16 +144,85 @@ def _drop_subscriber(self, thread_id: str, connection_id: str) -> None: if not conns: self._thread_subscribers.pop(thread_id, None) - def mark_turn_active(self, thread_id: str) -> None: + def mark_turn_active(self, thread_id: str, agent_id: str = "", session_key: str = "") -> None: + tid = thread_id.strip() + if not tid: + return + existing = self._active_turns.get(tid) + if existing is not None and existing.notified: + # Watchdog already declared this turn dead and the harness is + # winding down; refuse to resurrect the zombie registration. + # The old turn's finally-path will mark_turn_idle soon, after + # which a retry registers fresh. + return + if existing is not None: + # 并发 turn（同 thread 多连接）：递增引用计数，不覆盖既有登记。 + existing.refcount += 1 + return + self._active_turns[tid] = ActiveTurn( + agent_id=agent_id, + thread_id=tid, + session_key=session_key, + ) + + def mark_turn_progress(self, thread_id: str) -> None: + """Advance the last-progress timestamp for an active turn.""" tid = thread_id.strip() if tid: - self._active_turns.add(tid) + rec = self._active_turns.get(tid) + if rec is not None: + rec.last_progress_at = time.monotonic() + + def mark_tool_state(self, thread_id: str, in_tool: bool) -> None: + """Track model vs tool phase so long tools are not treated as stalls.""" + tid = thread_id.strip() + if not tid: + return + rec = self._active_turns.get(tid) + if rec is None: + return + now = time.monotonic() + if in_tool and not rec.in_tool: + rec.in_tool = True + rec.tool_segment_started_at = now + rec.last_progress_at = now + elif not in_tool and rec.in_tool: + if rec.tool_segment_started_at is not None: + rec.tool_budget_spent += now - rec.tool_segment_started_at + rec.in_tool = False + rec.tool_segment_started_at = None + rec.last_progress_at = now def mark_turn_idle(self, thread_id: str) -> None: - self._active_turns.discard(thread_id.strip()) + tid = thread_id.strip() + rec = self._active_turns.get(tid) + if rec is None: + return + rec.refcount -= 1 + if rec.refcount <= 0: + self._active_turns.pop(tid, None) def is_turn_active(self, thread_id: str) -> bool: - return thread_id.strip() in self._active_turns + rec = self._active_turns.get(thread_id.strip()) + # A notified (watchdog-declared-dead) turn is no longer "active" for + # subscribers, even though its record lingers until the harness winds + # down. + return rec is not None and not rec.notified + + def get_active_turn(self, thread_id: str) -> ActiveTurn | None: + return self._active_turns.get(thread_id.strip()) + + def snapshot_active_turns(self) -> dict[str, ActiveTurn]: + """Copy of live (non-notified) turn records for the TURN watchdog.""" + return { + tid: rec + for tid, rec in self._active_turns.items() + if not rec.notified + } + + def snapshot_notified_turns(self) -> dict[str, ActiveTurn]: + """Copy of watchdog-declared-dead records still awaiting wind-down.""" + return {tid: rec for tid, rec in self._active_turns.items() if rec.notified} async def push(self, connection_id: str, frame: dict[str, Any]) -> None: send_fn = self._connections.get(connection_id) diff --git a/tests/unit/gateway/test_dashboard_ws.py b/tests/unit/gateway/test_dashboard_ws.py index 9c973895f..1772a0255 100644 --- a/tests/unit/gateway/test_dashboard_ws.py +++ b/tests/unit/gateway/test_dashboard_ws.py @@ -165,6 +165,20 @@ def test_ws_hub_turn_active_flags() -> None: assert hub.is_turn_active("t1") is False +def test_ws_hub_concurrent_turns_refcount() -> None: + """Two overlapping turns on the same thread: the first to finish must not + idle the registration while the second is still streaming (multi-tab).""" + hub = WebSocketHub() + hub.mark_turn_active("t1") + hub.mark_turn_active("t1") # second connection starts a turn on same thread + hub.mark_turn_idle("t1") # first connection finishes + assert hub.is_turn_active("t1") is True, "second turn still streaming" + hub.mark_turn_idle("t1") # second finishes + assert hub.is_turn_active("t1") is False + hub.mark_turn_idle("t1") # redundant idle is a no-op + assert hub.is_turn_active("t1") is False + + @pytest.mark.asyncio async def test_ws_channel_streams_chunks() -> None: hub = WebSocketHub() diff --git a/tests/unit/gateway/test_gateway.py b/tests/unit/gateway/test_gateway.py index 54d913984..860454483 100644 --- a/tests/unit/gateway/test_gateway.py +++ b/tests/unit/gateway/test_gateway.py @@ -51,12 +51,35 @@ async def test_gateway_boot_and_shutdown(tmp_path: Path) -> None: await gw.boot() assert gw._channel_manager is not None assert gw._processor is not None + assert gw._turn_watchdog is not None # enabled by default await gw.shutdown() assert gw._channel_manager is None await registry.shutdown() +@pytest.mark.asyncio +async def test_gateway_boot_respects_watchdog_disabled(tmp_path: Path) -> None: + services = _make_services(tmp_path) + with patch("octop.infra.agents.manager.HarnessAgentManager") as mock_hm_cls: + mock_hm_cls.return_value = MagicMock() + + registry = AgentManager( + repos=services.repos, + paths=services.paths, + config=services.config, + ) + await registry.boot() + + gw = Gateway(agent_manager=registry, repos=services.repos) + with patch.dict("os.environ", {"OCTOP_TURN_WATCHDOG_DISABLED": "1"}): + await gw.boot() + assert gw._turn_watchdog is None # escape hatch honored + + await gw.shutdown() + await registry.shutdown() + + @pytest.mark.asyncio async def test_create_channel_updates_existing_same_kind(tmp_path: Path) -> None: services = _make_services(tmp_path) diff --git a/tests/unit/gateway/test_turn_watchdog.py b/tests/unit/gateway/test_turn_watchdog.py new file mode 100644 index 000000000..95868c2b6 --- /dev/null +++ b/tests/unit/gateway/test_turn_watchdog.py @@ -0,0 +1,206 @@ +"""tests/unit/gateway/test_turn_watchdog.py — TURN watchdog recovery logic.""" + +from __future__ import annotations + +import asyncio +import time +from typing import Any + +import pytest + +from octop.infra.gateway.ws.turn_watchdog import TurnWatchdog, watchdog_disabled +from octop.infra.gateway.ws.ws_hub import WebSocketHub + + +class _FakeAgentManager: + def __init__(self) -> None: + self.cancelled: list[tuple[str, str]] = [] + + def cancel_stream(self, agent_id: str, thread_id: str) -> None: + self.cancelled.append((agent_id, thread_id)) + + +class _FakeSession: + channel_id = "wx123" + channel_type = "weixin" + + def to_channel_subject(self) -> str: + return "subject-object" + + +class _FakeGateway: + thread_registry: Any + + def __init__(self) -> None: + self.pushed: list[tuple[str, str, str]] = [] + + async def push_text(self, channel_type: str, channel_id: str, subject: str, text: str) -> None: + self.pushed.append((channel_type, channel_id, text)) + + +async def _scan(wd: TurnWatchdog) -> None: + wd._scan_once() + await asyncio.sleep(0.15) + + +@pytest.mark.asyncio +async def test_stall_recovery_pushes_error_done_and_cancels() -> None: + hub = WebSocketHub() + sent: list[dict[str, Any]] = [] + + async def capture(frame: dict[str, Any]) -> None: + sent.append(frame) + + hub.register("c1", capture, user_id=1) + hub.subscribe("t1", "c1") + hub.mark_turn_active("t1", agent_id="agentA") + hub.get_active_turn("t1").last_progress_at -= 400 + + am = _FakeAgentManager() + wd = TurnWatchdog(hub=hub, agent_manager=am, stall_seconds=300, interval_seconds=60) + await _scan(wd) + + assert [f.get("type") for f in sent] == ["error", "done"] + assert am.cancelled == [("agentA", "t1")] + assert not hub.is_turn_active("t1") + + +@pytest.mark.asyncio +async def test_healthy_turn_untouched() -> None: + hub = WebSocketHub() + hub.mark_turn_active("t3", agent_id="agentC") + am = _FakeAgentManager() + wd = TurnWatchdog(hub=hub, agent_manager=am, stall_seconds=300, interval_seconds=60) + await _scan(wd) + assert hub.is_turn_active("t3") + assert am.cancelled == [] + + +@pytest.mark.asyncio +async def test_tool_phase_uses_longer_stall_threshold() -> None: + hub = WebSocketHub() + hub.mark_turn_active("t1", agent_id="agentA") + hub.mark_tool_state("t1", True) + hub.get_active_turn("t1").last_progress_at -= 400 # > model stall, < tool stall + + am = _FakeAgentManager() + wd = TurnWatchdog( + hub=hub, + agent_manager=am, + stall_seconds=300, + tool_stall_seconds=900, + interval_seconds=60, + ) + await _scan(wd) + assert am.cancelled == [] + assert hub.is_turn_active("t1") + + +@pytest.mark.asyncio +async def test_model_phase_cap_excludes_tool_time() -> None: + hub = WebSocketHub() + hub.mark_turn_active("t2", agent_id="agentB") + rec = hub.get_active_turn("t2") + rec.started_at -= 7200 + rec.last_progress_at = time.monotonic() - 1 + rec.tool_budget_spent = 5400 # 1.5h in tools of 2h total + + am = _FakeAgentManager() + wd = TurnWatchdog( + hub=hub, + agent_manager=am, + stall_seconds=300, + max_seconds=1800, + interval_seconds=60, + ) + await _scan(wd) + assert am.cancelled == [("agentB", "t2")] + + +@pytest.mark.asyncio +async def test_notified_zombie_not_resurrected_and_force_cleared() -> None: + hub = WebSocketHub() + hub.mark_turn_active("t4", agent_id="agentD") + rec = hub.get_active_turn("t4") + rec.notified = True + rec.notified_at = time.monotonic() - 200 # past grace + + am = _FakeAgentManager() + wd = TurnWatchdog( + hub=hub, + agent_manager=am, + stall_seconds=300, + notified_grace_seconds=120, + interval_seconds=60, + ) + await _scan(wd) + assert hub.get_active_turn("t4") is None, "zombie record must be force-cleared" + + # After cleanup, a fresh turn registers normally. + hub.mark_turn_active("t4", agent_id="agentD") + assert hub.is_turn_active("t4") + + +@pytest.mark.asyncio +async def test_im_turn_recovery_pushes_notice_and_cancels() -> None: + hub = WebSocketHub() + hub.mark_turn_active("thrX", agent_id="agentA", session_key="sk:im") + hub.get_active_turn("thrX").last_progress_at -= 400 + + am = _FakeAgentManager() + + class _Registry: + def get_session(self, session_key: str) -> _FakeSession | None: + return _FakeSession() if session_key == "sk:im" else None + + gw = _FakeGateway() + gw.thread_registry = _Registry() + wd = TurnWatchdog( + hub=hub, + agent_manager=am, + stall_seconds=300, + gateway=gw, + interval_seconds=60, + ) + await _scan(wd) + + assert am.cancelled == [("agentA", "thrX")] + assert gw.pushed and gw.pushed[0][0] == "weixin" + assert "超时" in gw.pushed[0][2] + + +@pytest.mark.asyncio +async def test_ws_turn_recovery_no_im_notice() -> None: + hub = WebSocketHub() + hub.mark_turn_active("thrY", agent_id="agentB") # no session_key (dashboard WS) + hub.get_active_turn("thrY").last_progress_at -= 400 + + am = _FakeAgentManager() + gw = _FakeGateway() + gw.thread_registry = type("R", (), {"get_session": lambda self, sk: None})() + wd = TurnWatchdog( + hub=hub, + agent_manager=am, + stall_seconds=300, + gateway=gw, + interval_seconds=60, + ) + await _scan(wd) + + assert am.cancelled == [("agentB", "thrY")] + assert gw.pushed == [] + + +def test_watchdog_disabled_env(monkeypatch) -> None: + """OCTOP_TURN_WATCHDOG_DISABLED=1 disables the watchdog; anything else enables it.""" + monkeypatch.setenv("OCTOP_TURN_WATCHDOG_DISABLED", "1") + assert watchdog_disabled() is True + + monkeypatch.setenv("OCTOP_TURN_WATCHDOG_DISABLED", "0") + assert watchdog_disabled() is False + + monkeypatch.setenv("OCTOP_TURN_WATCHDOG_DISABLED", "") + assert watchdog_disabled() is False + + monkeypatch.delenv("OCTOP_TURN_WATCHDOG_DISABLED") + assert watchdog_disabled() is False