Close P11 runtime watchdog loop

SESSION.md

@@ -2,7 +2,7 @@
 
 ## Current Goal
 
-Automation execution-layer P10 is closed locally. P10 adds visual-capture retention cleanup as an auditable operations hardening slice so stale capture artifacts can be cleaned through service, script, API, runtime automation, and workbench entry points while keeping the system analysis/restricted decision-support only.
+Automation execution-layer P11 is closed locally. P11 adds runtime stuck-task detection as an auditable operations hardening slice so stalled runtime tasks can be surfaced through status, API, script, and workbench views while keeping the system analysis/restricted decision-support only.
 
 ## Current Branch
 
@@ -14,14 +14,15 @@ Automation execution-layer P10 is closed locally. P10 adds visual-capture retent
 
 ## Automation Status
 
-- state: automation_execution_layer_p10_visual_capture_retention_cleanup_closed
-- current_phase: P10 Visual Capture Retention Cleanup
+- state: automation_execution_layer_p11_runtime_task_watchdog_closed
+- current_phase: P11 Runtime Task Watchdog
 - latest_p7_selector_choice: continue_periodic_analysis_decision_monitoring_with_maturity_rollup
-- latest_p10_slice: visual_capture_retention_cleanup
+- latest_p11_slice: runtime_task_watchdog
 - latest_visual_review_script_status: idle
 - latest_runtime_visual_watch_task_status: ok
 - latest_visual_capture_cleanup_task_status: ok
-- latest_runtime_visual_watch_state: idle
+ - latest_runtime_watchdog_task_status: implemented
+ - latest_runtime_visual_watch_state: idle
 - latest_visual_snapshot_capture_status: implemented
 - latest_visual_reviews_panel_status: visible
 - latest_visual_feedback_status: implemented
@@ -31,6 +32,8 @@ Automation execution-layer P10 is closed locally. P10 adds visual-capture retent
 - latest_alert_quality_route_status: ok
 - latest_visual_capture_cleanup_status: implemented
 - latest_visual_capture_cleanup_route_status: ok
+- latest_runtime_watchdog_status: implemented
+- latest_runtime_watchdog_route_status: ok
 - latest_runtime_blockers: none
 - guards_unchanged: execution_allowed, real_live_ready, risk_governor, trust_bucket, operator_confirmation_protocol
 
@@ -54,6 +57,9 @@ Automation execution-layer P10 is closed locally. P10 adds visual-capture retent
 - Added `VisualCaptureRetentionService` to remove stale `.claude/visual_watch/captures` HTML/JSON artifacts by retention window.
 - Added `scripts/run_visual_capture_cleanup.py` and `POST /trading/visual-watch/cleanup-captures`.
 - Added runtime task `visual_capture_cleanup` and a workbench `Cleanup Old Captures` button in the Visual Reviews panel.
+- Added `RuntimeTaskWatchdogService` to detect runtime tasks that are overdue or running too long.
+- Added `scripts/run_runtime_task_watchdog.py` and `GET /trading/runtime-watchdog`.
+- Extended `runtime_status` and the runtime workbench panel with `runtime_watchdog`, stalled task names, and per-task stall details.
 - Added focused unit tests for the service, script, API routing, and runtime task.
 
 ## Current Working State
@@ -61,16 +67,18 @@ Automation execution-layer P10 is closed locally. P10 adds visual-capture retent
 - P7 is closed and remains a monitor-maturity routing layer, not a Research Council expansion.
 - P8 review/capture/feedback surfaces remain implemented and unchanged in their execution guards.
 - P9 quality rollup remains implemented and visible in both the dashboard and review packet.
-- P10 retention cleanup is implemented: stale visual capture HTML/JSON artifacts can now be deleted by retention policy.
-- P10 runtime visibility is implemented: `runtime_status.tasks.visual_capture_cleanup` now reports root, retention window, eligible count, deleted count, deleted bytes, and idle reasons.
-- P10 workbench/manual visibility is implemented: the Visual Reviews panel exposes a `Cleanup Old Captures` action that hits the cleanup route and refreshes the dashboard.
+- P10 retention cleanup remains implemented and visible in runtime plus workbench flows.
+- P11 runtime watchdog is implemented: `runtime_status.runtime_watchdog` now reports `checked_ts`, stalled task names, warning codes, and detailed stall reasons.
+- P11 runtime panel visibility is implemented: the workbench runtime panel now shows stalled task count and per-task stalled/threshold detail.
+- P11 script/API visibility is implemented: `scripts/run_runtime_task_watchdog.py` and `/trading/runtime-watchdog` both expose the watchdog summary directly.
 - Latest `scripts/run_visual_review.py` smoke found no pending reviews: attempted=0, reviewed=0, status=idle.
 - Latest capture smoke returned `status=ignored` when no watch run was supplied, which is the expected safe no-op.
 - Latest cleanup smoke deleted one expired temporary artifact through `scripts/run_visual_capture_cleanup.py`.
+- Latest watchdog smoke returned `stalled_task_count=0` through `scripts/run_runtime_task_watchdog.py`.
 - Latest `scripts/run_trading_runtime.py` smoke returned blockers=none and includes a healthy `visual_watch` runtime task.
-- Latest local API smoke confirmed `/trading/visual-watch/cleanup-captures`, dashboard HTML, and dashboard JS all expose the P10 cleanup path successfully.
+- Latest local API smoke confirmed `/trading/runtime-watchdog`, `/trading/runtime-status`, and dashboard JS all expose the P11 watchdog path successfully.
 
-## P10 Blockers
+## P11 Blockers
 
 - none
 
@@ -80,11 +88,12 @@ Automation execution-layer P10 is closed locally. P10 adds visual-capture retent
 - Visual review remains advisory and must not change execution permission, live readiness, trust bucket, risk governor, or operator confirmation.
 - Alert-quality statistics remain advisory and must not change execution permission, live readiness, trust bucket, risk governor, or operator confirmation.
 - Capture cleanup remains operational only and must not change execution permission, live readiness, trust bucket, risk governor, or operator confirmation.
+- Runtime watchdog remains operational only and must not change execution permission, live readiness, trust bucket, risk governor, or operator confirmation.
 - Real-live order submission remains intentionally disabled because the OKX order adapter is still not a real-live submitter.
 
 ## Exact Next Step
 
-P10 is locally closed. Default next action is to continue periodic analysis-decision monitoring with maturity rollup and, when opening the next bounded product slice, move to the next P11-style automation-operations hardening candidate such as silent windows / noise control or stuck-task detection.
+P11 is locally closed. Default next action is to continue periodic analysis-decision monitoring with maturity rollup and, when opening the next bounded product slice, move to the next P12-style automation-operations hardening candidate such as silent windows / noise control.
 
 ## Validation Commands Used Recently
 
@@ -95,6 +104,7 @@ D:\ai-exchange\.venv\Scripts\python.exe -m unittest tests.unit.test_run_until_bl
 D:\ai-exchange\.venv\Scripts\python.exe scripts\run_visual_review.py
 D:\ai-exchange\.venv\Scripts\python.exe scripts\run_visual_snapshot_capture.py
 D:\ai-exchange\.venv\Scripts\python.exe scripts\run_visual_capture_cleanup.py
+D:\ai-exchange\.venv\Scripts\python.exe scripts\run_runtime_task_watchdog.py
 D:\ai-exchange\.venv\Scripts\python.exe scripts\run_trading_runtime.py
 D:\ai-exchange\.venv\Scripts\python.exe -m unittest tests.unit.test_visual_snapshot_capture_service tests.unit.test_run_visual_snapshot_capture tests.unit.test_api_static_server
 ```

TASKS.md

@@ -19,19 +19,21 @@
 - [x] P8 local closure complete
 - [x] P9 local closure complete
 - [x] P10 local closure complete
+- [x] P11 local closure complete
 - [ ] Continue periodic analysis-decision monitoring with maturity rollup in parallel
-- [ ] Open the next bounded P11-style automation-operations hardening slice
+- [ ] Open the next bounded P12-style automation-operations hardening slice
 - [ ] Treat browser screenshot capture and additional external media adapters as future enhancements
 
 ## Automation Status
 
-- [x] state: automation_execution_layer_p10_visual_capture_retention_cleanup_closed
-- [x] current_phase: P10 Visual Capture Retention Cleanup
+- [x] state: automation_execution_layer_p11_runtime_task_watchdog_closed
+- [x] current_phase: P11 Runtime Task Watchdog
 - [x] latest_p7_selector_choice: continue_periodic_analysis_decision_monitoring_with_maturity_rollup
-- [x] latest_p10_slice: visual_capture_retention_cleanup
+- [x] latest_p11_slice: runtime_task_watchdog
 - [x] latest_visual_review_script_status: idle
 - [x] latest_runtime_visual_watch_task_status: ok
 - [x] latest_visual_capture_cleanup_task_status: ok
+- [x] latest_runtime_watchdog_task_status: implemented
 - [x] latest_visual_snapshot_capture_status: implemented
 - [x] latest_visual_reviews_panel_status: visible
 - [x] latest_visual_feedback_status: implemented
@@ -41,6 +43,8 @@
 - [x] latest_alert_quality_route_status: ok
 - [x] latest_visual_capture_cleanup_status: implemented
 - [x] latest_visual_capture_cleanup_route_status: ok
+- [x] latest_runtime_watchdog_status: implemented
+- [x] latest_runtime_watchdog_route_status: ok
 - [x] latest_runtime_blockers: none
 - [x] no_new_research_council_phase_created: true
 
@@ -66,11 +70,17 @@
 - [x] Added `POST /trading/visual-watch/cleanup-captures`
 - [x] Added runtime `visual_capture_cleanup` task
 - [x] Added workbench `Cleanup Old Captures` action
+- [x] Added `RuntimeTaskWatchdogService`
+- [x] Added `scripts/run_runtime_task_watchdog.py`
+- [x] Added `GET /trading/runtime-watchdog`
+- [x] Added `runtime_status.runtime_watchdog`
+- [x] Added runtime panel stalled-task visibility
 - [x] Added runtime `visual_watch` task
 - [x] Added focused service, script, API, and runtime tests
 - [x] Smoked the visual review script with no pending reviews
 - [x] Smoked the capture script with no watch run provided
 - [x] Smoked the cleanup script with expired temporary artifacts
+- [x] Smoked the runtime watchdog script with `stalled_task_count=0`
 - [x] Smoked the trading runtime with blockers=none
 
 ## Blocked
@@ -113,6 +123,7 @@
 - [x] `automation_execution_layer_p8_visual_review_workbench_panel`
 - [x] `automation_execution_layer_p9_alert_quality_rollup`
 - [x] `automation_execution_layer_p10_visual_capture_retention_cleanup`
+- [x] `automation_execution_layer_p11_runtime_task_watchdog`
 
 ## Backlog
 

docs/AI_ICT_盯盘告警开发任务清单_v1.md

@@ -395,6 +395,8 @@ candidate -> decision alert -> dispatcher -> 外部消息 -> delivery record
 ### 当前进展
 - 2026-05-12 P10 已完成 `screenshot` 存储清理闭环：新增 `VisualCaptureRetentionService`、`scripts/run_visual_capture_cleanup.py`、`POST /trading/visual-watch/cleanup-captures`、runtime `visual_capture_cleanup` 任务，以及 workbench 手动触发入口。
 - 2026-05-12 P10 本地闭环已验证：cleanup 脚本可删除过期临时文件，runtime smoke 无 blocker，本地 API/workbench smoke 可见 cleanup 路径。
+- 2026-05-12 P11 已完成 `stuck task` 检测闭环：新增 `RuntimeTaskWatchdogService`、`scripts/run_runtime_task_watchdog.py`、`GET /trading/runtime-watchdog`、`runtime_status.runtime_watchdog`，以及 runtime 面板 stalled-task 呈现。
+- 2026-05-12 P11 本地闭环已验证：watchdog 脚本返回 `stalled_task_count=0`，runtime/API/workbench smoke 可见 watchdog 路径与状态输出。
 
 ### 验收
 - 非重大错误不打断主循环

docs/AI_ICT_盯盘告警技术设计稿_v1.md

@@ -647,6 +647,8 @@
 - `VISUAL_CAPTURE_RETENTION_HOURS`
 - `VISUAL_CAPTURE_CLEANUP_INTERVAL_SECONDS`
 - `VISUAL_CAPTURE_CLEANUP_MAX_DELETE_FILES`
+- `RUNTIME_TASK_STALE_MULTIPLIER`
+- `RUNTIME_TASK_STUCK_MIN_SECONDS`
 
 ---
 

scripts/run_runtime_task_watchdog.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from pathlib import Path
+import json
+import os
+import sys
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from src.api.trading_runtime import get_trading_runtime_watchdog
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    output_json = _get_bool(env, "TRADING_RUNTIME_WATCHDOG_OUTPUT_JSON", False)
+    payload = get_trading_runtime_watchdog(run_once=_get_bool(env, "TRADING_RUNTIME_WATCHDOG_RUN_ONCE", True))
+    if output_json:
+        print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str))
+    else:
+        print(f"Runtime watchdog checked: {payload.get('checked_ts')}")
+        print(f"Stalled tasks: {', '.join(payload.get('stalled_task_names') or []) or 'none'}")
+    return 0 if not payload.get("stalled_task_count") else 2
+
+
+def _get_bool(env, key: str, default: bool) -> bool:
+    value = env.get(key)
+    if value is None or value == "":
+        return default
+    return str(value).strip().lower() in {"1", "true", "yes", "on"}
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/api/__init__.py

@@ -32,6 +32,7 @@ from src.api.trading_runtime import (
     get_trading_plan_by_id,
     get_trading_review_packet,
     get_trading_runtime_status,
+    get_trading_runtime_watchdog,
     get_trading_session_view,
     get_trading_ticket_by_id,
     get_trading_trust_score,
@@ -90,6 +91,7 @@ __all__ = [
     "get_trading_launch_readiness",
     "get_trading_plan_by_id",
     "get_trading_runtime_status",
+    "get_trading_runtime_watchdog",
     "get_trading_trust_score",
     "get_trading_ticket_by_id",
     "get_trading_executions",

src/api/server.py

@@ -60,6 +60,7 @@ from src.api.trading_runtime import (
     get_trading_live_readiness,
     get_trading_launch_readiness,
     get_trading_runtime_status,
+    get_trading_runtime_watchdog,
     get_trading_session_view,
     get_trading_ticket_by_id,
     get_trading_trust_score,
@@ -114,6 +115,7 @@ def build_api_dependencies() -> dict:
         "trading_alert_quality_summary": get_alert_quality_summary,
         "trading_reconciliation_run": run_trading_reconciliation,
         "trading_runtime_status": get_trading_runtime_status,
+        "trading_runtime_watchdog": get_trading_runtime_watchdog,
         "trading_runtime_run_once": run_trading_runtime_once,
         "trading_trust_score": get_trading_trust_score,
         "trading_session_view": get_trading_session_view,
@@ -243,6 +245,9 @@ def route_request(
         return 200, deps["trading_market_state"](instrument_id=instrument_id, timeframe=timeframe)
     if method == "GET" and path == "/trading/runtime-status":
         return 200, deps["trading_runtime_status"]()
+    if method == "GET" and path == "/trading/runtime-watchdog":
+        run_once = _get_single(query, "run_once", "0")
+        return 200, deps["trading_runtime_watchdog"](run_once=str(run_once).strip().lower() in {"1", "true", "yes", "on"})
     if method == "POST" and path == "/trading/runtime/run-once":
         return 200, deps["trading_runtime_run_once"]()
     if method == "GET" and path == "/trading/decision-packet":

src/api/trading_runtime.py

@@ -375,6 +375,12 @@ def run_trading_runtime_once() -> dict:
     return get_trading_runtime_service().run_once()
 
 
+def get_trading_runtime_watchdog(run_once: bool = False) -> dict:
+    service = get_trading_runtime_service()
+    payload = service.run_once() if run_once else service.status()
+    return dict(payload.get("runtime_watchdog") or {})
+
+
 def get_trading_decision_packet(instrument_id: int) -> dict:
     return DecisionPacketService().build_preview(instrument_id=instrument_id)
 

src/config/settings.py

@@ -89,6 +89,9 @@ class Settings(BaseSettings):
     trading_runtime_research_council_interval_seconds: int = 60
     trading_runtime_decision_packet_interval_seconds: int = 20
     trading_runtime_plan_interval_seconds: int = 900
+    runtime_task_watchdog_enabled: bool = True
+    runtime_task_stale_multiplier: int = 3
+    runtime_task_stuck_min_seconds: int = 120
     visual_capture_cleanup_enabled: bool = True
     visual_capture_cleanup_interval_seconds: int = 3600
     visual_capture_retention_hours: int = 168

src/services/runtime/__init__.py

@@ -1,3 +1,4 @@
+from src.services.runtime.runtime_task_watchdog_service import RuntimeTaskWatchdogService
 from src.services.runtime.trading_runtime_service import (
     TradingRuntimeScheduler,
     TradingRuntimeSupervisor,
@@ -6,6 +7,7 @@ from src.services.runtime.trading_runtime_service import (
 )
 
 __all__ = [
+    "RuntimeTaskWatchdogService",
     "TradingRuntimeScheduler",
     "TradingRuntimeSupervisor",
     "get_trading_runtime_service",

src/services/runtime/runtime_task_watchdog_service.py

@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+
+class RuntimeTaskWatchdogService:
+    def __init__(
+        self,
+        *,
+        stale_multiplier: int = 3,
+        min_stuck_seconds: int = 120,
+    ) -> None:
+        self.stale_multiplier = max(int(stale_multiplier), 1)
+        self.min_stuck_seconds = max(int(min_stuck_seconds), 1)
+
+    def build_watchdog(self, tasks: dict[str, object], *, checked_ts: datetime | None = None) -> dict:
+        now = checked_ts.astimezone(timezone.utc) if isinstance(checked_ts, datetime) else datetime.now(timezone.utc)
+        stalled_tasks = []
+        for name, task in (tasks or {}).items():
+            normalized = _normalize_task(task)
+            stalled = self._evaluate_task(name=name, task=normalized, now=now)
+            if stalled is not None:
+                stalled_tasks.append(stalled)
+        return {
+            "checked_ts": now.isoformat(),
+            "stalled_task_count": len(stalled_tasks),
+            "stalled_task_names": [item["name"] for item in stalled_tasks],
+            "stalled_tasks": stalled_tasks,
+            "warning_codes": [f"runtime_task_stalled:{item['name']}:{item['reason']}" for item in stalled_tasks],
+        }
+
+    def _evaluate_task(self, *, name: str, task: dict, now: datetime) -> dict | None:
+        interval_seconds = max(int(task.get("interval_seconds") or 0), 1)
+        threshold_seconds = max(interval_seconds * self.stale_multiplier, self.min_stuck_seconds)
+        last_status = str(task.get("last_status") or "idle")
+        if last_status == "running":
+            last_started = _parse_iso(task.get("last_started_ts"))
+            if last_started is None:
+                return None
+            age_seconds = max((now - last_started).total_seconds(), 0.0)
+            if age_seconds <= threshold_seconds:
+                return None
+            return _stalled_payload(
+                name=name,
+                reason="running_too_long",
+                task=task,
+                age_seconds=age_seconds,
+                threshold_seconds=threshold_seconds,
+            )
+        if task.get("run_count") in (None, 0):
+            return None
+        reference_ts = _parse_iso(task.get("last_completed_ts")) or _parse_iso(task.get("last_success_ts"))
+        if reference_ts is None:
+            return None
+        age_seconds = max((now - reference_ts).total_seconds(), 0.0)
+        if age_seconds <= threshold_seconds:
+            return None
+        return _stalled_payload(
+            name=name,
+            reason="overdue",
+            task=task,
+            age_seconds=age_seconds,
+            threshold_seconds=threshold_seconds,
+        )
+
+
+def _normalize_task(task: object) -> dict:
+    if isinstance(task, dict):
+        return dict(task)
+    if hasattr(task, "to_dict"):
+        return dict(task.to_dict())
+    return {
+        "interval_seconds": getattr(task, "interval_seconds", 0),
+        "last_started_ts": getattr(task, "last_started_ts", None),
+        "last_completed_ts": getattr(task, "last_completed_ts", None),
+        "last_success_ts": getattr(task, "last_success_ts", None),
+        "last_status": getattr(task, "last_status", "idle"),
+        "run_count": getattr(task, "run_count", 0),
+    }
+
+
+def _parse_iso(value) -> datetime | None:
+    if value in (None, ""):
+        return None
+    if isinstance(value, datetime):
+        return value.astimezone(timezone.utc) if value.tzinfo else value.replace(tzinfo=timezone.utc)
+    try:
+        parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    return parsed.astimezone(timezone.utc) if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
+
+
+def _stalled_payload(*, name: str, reason: str, task: dict, age_seconds: float, threshold_seconds: int) -> dict:
+    return {
+        "name": name,
+        "reason": reason,
+        "age_seconds": age_seconds,
+        "threshold_seconds": threshold_seconds,
+        "interval_seconds": int(task.get("interval_seconds") or 0),
+        "last_status": task.get("last_status"),
+        "last_started_ts": task.get("last_started_ts"),
+        "last_completed_ts": task.get("last_completed_ts"),
+        "last_success_ts": task.get("last_success_ts"),
+    }

src/services/runtime/trading_runtime_service.py

@@ -13,6 +13,7 @@ from src.services.decision import DecisionPacketService
 from src.services.market_state import MarketStateService, get_market_truth_supervisor
 from src.services.reconciliation import BrokerExecutionReconciliationService
 from src.services.research_council import ResearchCouncilService
+from src.services.runtime.runtime_task_watchdog_service import RuntimeTaskWatchdogService
 from src.services.trade_plan_candidates import SetupWatcherService
 from src.services.trading_plans import TradingPlanService
 from src.services.visual_watch import VisualCaptureRetentionService, VisualReviewService
@@ -64,6 +65,7 @@ class TradingRuntimeScheduler:
         research_council_service: Optional[ResearchCouncilService] = None,
         visual_review_service: Optional[VisualReviewService] = None,
         visual_capture_retention_service: Optional[VisualCaptureRetentionService] = None,
+        runtime_task_watchdog_service: Optional[RuntimeTaskWatchdogService] = None,
     ) -> None:
         self.market_state_service = market_state_service or MarketStateService(market_truth_supervisor=get_market_truth_supervisor())
         self.trading_plan_service = trading_plan_service or TradingPlanService()
@@ -77,6 +79,10 @@ class TradingRuntimeScheduler:
         self.research_council_service = research_council_service or ResearchCouncilService()
         self.visual_review_service = visual_review_service or VisualReviewService()
         self.visual_capture_retention_service = visual_capture_retention_service or VisualCaptureRetentionService()
+        self.runtime_task_watchdog_service = runtime_task_watchdog_service or RuntimeTaskWatchdogService(
+            stale_multiplier=settings.runtime_task_stale_multiplier,
+            min_stuck_seconds=settings.runtime_task_stuck_min_seconds,
+        )
         self.symbols = [item.strip() for item in settings.trading_runtime_symbols.split(",") if item.strip()]
         self.bias_tf = settings.trading_runtime_bias_timeframe
         self.setup_tf = settings.trading_runtime_setup_timeframe
@@ -101,18 +107,18 @@ class TradingRuntimeScheduler:
         self.latest_market_state_by_symbol: dict[str, dict] = {}
         self.latest_broker_state_by_symbol: dict[str, dict] = {}
         self.runtime_health: dict = {}
+        self.runtime_watchdog: dict = {}
 
     def run_cycle(self) -> dict:
         if not settings.trading_runtime_enabled:
             self.current_blockers = ["trading_runtime_disabled"]
             self.current_warnings = []
             self.current_noisy_incidents = []
-            self.runtime_health = _build_runtime_health(self.current_blockers, self.current_warnings, self.current_noisy_incidents)
-            self.degraded_mode_state = {
-                "is_degraded": True,
-                "codes": list(self.current_blockers),
-            }
-            return self.status_snapshot()
+            snapshot = self.status_snapshot()
+            self.runtime_health = dict(snapshot.get("runtime_health") or {})
+            self.degraded_mode_state = dict(snapshot.get("degraded_mode_state") or {})
+            self.runtime_watchdog = dict(snapshot.get("runtime_watchdog") or {})
+            return snapshot
 
         blockers: list[str] = []
         warnings: list[str] = []
@@ -170,14 +176,24 @@ class TradingRuntimeScheduler:
         self.current_blockers = _dedupe(blockers)
         self.current_warnings = _dedupe(warnings)
         self.current_noisy_incidents = _dedupe(noisy_incidents)
-        self.runtime_health = _build_runtime_health(self.current_blockers, self.current_warnings, self.current_noisy_incidents)
-        self.degraded_mode_state = {
-            "is_degraded": bool(self.current_blockers or self.current_warnings),
-            "codes": [*self.current_blockers, *self.current_warnings],
-        }
-        return self.status_snapshot()
+        snapshot = self.status_snapshot()
+        self.runtime_health = dict(snapshot.get("runtime_health") or {})
+        self.degraded_mode_state = dict(snapshot.get("degraded_mode_state") or {})
+        self.runtime_watchdog = dict(snapshot.get("runtime_watchdog") or {})
+        return snapshot
 
     def status_snapshot(self) -> dict:
+        runtime_watchdog = (
+            self.runtime_task_watchdog_service.build_watchdog(self.tasks)
+            if settings.runtime_task_watchdog_enabled
+            else {"checked_ts": datetime.now(timezone.utc).isoformat(), "stalled_task_count": 0, "stalled_task_names": [], "stalled_tasks": [], "warning_codes": []}
+        )
+        current_warnings = _dedupe([*self.current_warnings, *list(runtime_watchdog.get("warning_codes") or [])])
+        runtime_health = _build_runtime_health(self.current_blockers, current_warnings, self.current_noisy_incidents)
+        degraded_mode_state = {
+            "is_degraded": bool(self.current_blockers or current_warnings),
+            "codes": [*self.current_blockers, *current_warnings],
+        }
         return {
             "enabled": bool(settings.trading_runtime_enabled),
             "scheduler_state": "running" if settings.trading_runtime_enabled else "disabled",
@@ -189,10 +205,11 @@ class TradingRuntimeScheduler:
             },
             "tasks": {name: task.to_dict() for name, task in self.tasks.items()},
             "current_blockers": list(self.current_blockers),
-            "current_warnings": list(self.current_warnings),
+            "current_warnings": list(current_warnings),
             "current_noisy_incidents": list(self.current_noisy_incidents),
-            "runtime_health": dict(self.runtime_health),
-            "degraded_mode_state": dict(self.degraded_mode_state),
+            "runtime_health": runtime_health,
+            "degraded_mode_state": degraded_mode_state,
+            "runtime_watchdog": runtime_watchdog,
             "incidents": list(self.incidents[-50:]),
             "latest_market_state_by_symbol": dict(self.latest_market_state_by_symbol),
             "latest_broker_state_by_symbol": dict(self.latest_broker_state_by_symbol),
@@ -589,7 +606,8 @@ class TradingRuntimeSupervisor:
 
     def status(self) -> dict:
         with self._lock:
-            status = dict(self._last_status)
+            status = self.scheduler.status_snapshot()
+            self._last_status = status
             status["supervisor_state"] = "running" if self._thread and self._thread.is_alive() else "stopped"
             status["started_ts"] = self._started_ts
             return status

src/web/app.js

@@ -289,6 +289,9 @@ function renderRuntimeStatus(runtimeStatus) {
   const tasks = runtimeStatus?.tasks || {};
   const blockers = runtimeStatus?.current_blockers || [];
   const incidents = runtimeStatus?.incidents || [];
+  const watchdog = runtimeStatus?.runtime_watchdog || {};
+  const stalledTasks = watchdog?.stalled_tasks || [];
+  const stalledMap = new Map(stalledTasks.map((item) => [item.name, item]));
   runtimeStatusPanel.appendChild(
     buildInfoGrid([
       ['scheduler state', runtimeStatus?.scheduler_state || 'unknown'],
@@ -297,25 +300,29 @@ function renderRuntimeStatus(runtimeStatus) {
       ['degraded', String(Boolean(runtimeStatus?.degraded_mode_state?.is_degraded))],
       ['symbols', formatList(runtimeStatus?.symbols || [])],
       ['blockers', formatList(blockers)],
+      ['stalled tasks', formatList(watchdog?.stalled_task_names || [])],
+      ['watchdog checked', formatTimestamp(watchdog?.checked_ts)],
       ['last incident', incidents.length ? `${incidents[incidents.length - 1].code} @ ${formatTimestamp(incidents[incidents.length - 1].observed_ts)}` : 'none'],
     ])
   );
   const grid = createElement('div', 'cards');
   Object.entries(tasks).forEach(([name, task]) => {
-    grid.appendChild(
-      buildItemCard(
-        name,
-        task.last_status || 'idle',
-        [
-          ['last success', formatTimestamp(task.last_success_ts)],
-          ['next run', formatTimestamp(task.next_run_ts)],
-          ['run count', String(task.run_count ?? 0)],
-          ['success / failure', `${task.success_count ?? 0} / ${task.failure_count ?? 0}`],
-          ['last error', task.last_error || 'none'],
-          ['details', formatJsonInline(task.details || {})],
-        ]
-      )
-    );
+    const stalled = stalledMap.get(name) || null;
+      grid.appendChild(
+        buildItemCard(
+          name,
+          stalled ? 'warning' : (task.last_status || 'idle'),
+          [
+            ['last success', formatTimestamp(task.last_success_ts)],
+            ['next run', formatTimestamp(task.next_run_ts)],
+            ['run count', String(task.run_count ?? 0)],
+            ['success / failure', `${task.success_count ?? 0} / ${task.failure_count ?? 0}`],
+            ['stalled', stalled ? `${stalled.reason} (${formatNumber(stalled.age_seconds, 1)}s > ${stalled.threshold_seconds}s)` : 'no'],
+            ['last error', task.last_error || 'none'],
+            ['details', formatJsonInline(task.details || {})],
+          ]
+        )
+      );
   });
   runtimeStatusPanel.appendChild(grid);
 }

tests/unit/test_run_runtime_task_watchdog.py

@@ -0,0 +1,32 @@
+import contextlib
+import io
+import json
+import unittest
+from unittest.mock import patch
+
+from scripts import run_runtime_task_watchdog
+
+
+class RunRuntimeTaskWatchdogScriptTests(unittest.TestCase):
+    def test_main_outputs_json(self) -> None:
+        with patch("scripts.run_runtime_task_watchdog.get_trading_runtime_watchdog") as watchdog_mock:
+            watchdog_mock.return_value = {
+                "checked_ts": "2026-05-12T00:00:00+00:00",
+                "stalled_task_count": 1,
+                "stalled_task_names": ["watch"],
+                "stalled_tasks": [{"name": "watch", "reason": "overdue"}],
+            }
+            output = io.StringIO()
+            with contextlib.redirect_stdout(output):
+                exit_code = run_runtime_task_watchdog.main(
+                    env={"TRADING_RUNTIME_WATCHDOG_OUTPUT_JSON": "1", "TRADING_RUNTIME_WATCHDOG_RUN_ONCE": "0"}
+                )
+
+        payload = json.loads(output.getvalue())
+        self.assertEqual(exit_code, 2)
+        self.assertEqual(payload["stalled_task_count"], 1)
+        watchdog_mock.assert_called_once_with(run_once=False)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_runtime_task_watchdog_service.py

@@ -0,0 +1,65 @@
+import unittest
+from datetime import datetime, timedelta, timezone
+
+from src.services.runtime.runtime_task_watchdog_service import RuntimeTaskWatchdogService
+
+
+class RuntimeTaskWatchdogServiceTests(unittest.TestCase):
+    def test_build_watchdog_returns_empty_when_tasks_are_fresh(self) -> None:
+        now = datetime(2026, 5, 12, 0, 0, tzinfo=timezone.utc)
+        service = RuntimeTaskWatchdogService(stale_multiplier=3, min_stuck_seconds=120)
+        tasks = {
+            "watch": {
+                "interval_seconds": 20,
+                "last_status": "ok",
+                "run_count": 1,
+                "last_completed_ts": (now - timedelta(seconds=10)).isoformat(),
+                "last_success_ts": (now - timedelta(seconds=10)).isoformat(),
+            }
+        }
+
+        payload = service.build_watchdog(tasks, checked_ts=now)
+
+        self.assertEqual(payload["stalled_task_count"], 0)
+        self.assertEqual(payload["warning_codes"], [])
+
+    def test_build_watchdog_flags_running_task_that_exceeds_threshold(self) -> None:
+        now = datetime(2026, 5, 12, 0, 0, tzinfo=timezone.utc)
+        service = RuntimeTaskWatchdogService(stale_multiplier=3, min_stuck_seconds=120)
+        tasks = {
+            "watch": {
+                "interval_seconds": 20,
+                "last_status": "running",
+                "run_count": 1,
+                "last_started_ts": (now - timedelta(seconds=180)).isoformat(),
+            }
+        }
+
+        payload = service.build_watchdog(tasks, checked_ts=now)
+
+        self.assertEqual(payload["stalled_task_count"], 1)
+        self.assertEqual(payload["stalled_tasks"][0]["reason"], "running_too_long")
+        self.assertIn("runtime_task_stalled:watch:running_too_long", payload["warning_codes"])
+
+    def test_build_watchdog_flags_overdue_task(self) -> None:
+        now = datetime(2026, 5, 12, 0, 0, tzinfo=timezone.utc)
+        service = RuntimeTaskWatchdogService(stale_multiplier=3, min_stuck_seconds=120)
+        tasks = {
+            "alert_dispatch": {
+                "interval_seconds": 20,
+                "last_status": "ok",
+                "run_count": 2,
+                "last_completed_ts": (now - timedelta(seconds=180)).isoformat(),
+                "last_success_ts": (now - timedelta(seconds=180)).isoformat(),
+            }
+        }
+
+        payload = service.build_watchdog(tasks, checked_ts=now)
+
+        self.assertEqual(payload["stalled_task_count"], 1)
+        self.assertEqual(payload["stalled_tasks"][0]["reason"], "overdue")
+        self.assertIn("runtime_task_stalled:alert_dispatch:overdue", payload["warning_codes"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_trading_runtime_service.py

@@ -1,4 +1,4 @@
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 import contextlib
 import io
 import json
@@ -11,6 +11,7 @@ os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:
 from scripts import run_trading_runtime
 from src.api.server import route_request
 from src.services.runtime import TradingRuntimeScheduler, TradingRuntimeSupervisor
+from src.services.runtime.runtime_task_watchdog_service import RuntimeTaskWatchdogService
 
 
 class FakeInstrumentRepository:
@@ -214,6 +215,7 @@ class TradingRuntimeSchedulerTests(unittest.TestCase):
         self.assertEqual(payload["tasks"]["visual_watch"]["details"]["visual_reviews_created"], 1)
         self.assertEqual(payload["tasks"]["visual_watch"]["details"]["last_visual_state"], "confirm")
         self.assertEqual(payload["tasks"]["visual_capture_cleanup"]["details"]["deleted_file_count"], 2)
+        self.assertEqual(payload["runtime_watchdog"]["stalled_task_count"], 0)
         self.assertEqual(payload["latest_broker_state_by_symbol"]["BTC-USDT"]["account"]["equity"], 1000.0)
         self.assertTrue(payload["tasks"]["watch"]["next_run_ts"])
 
@@ -260,6 +262,30 @@ class TradingRuntimeSchedulerTests(unittest.TestCase):
         self.assertEqual(payload["supervisor_state"], "stopped")
         self.assertEqual(payload["current_blockers"], [])
 
+    def test_status_snapshot_surfaces_watchdog_warnings_for_stalled_tasks(self) -> None:
+        scheduler = TradingRuntimeScheduler(
+            market_state_service=FakeMarketStateService(),
+            trading_plan_service=FakeTradingPlanService(),
+            setup_watcher_service=FakeSetupWatcherService(),
+            broker_read_model_service=FakeBrokerReadModelService(),
+            reconciliation_service=FakeReconciliationService(),
+            decision_packet_service=FakeDecisionPacketService(),
+            alert_dispatcher_service=FakeAlertDispatcherService(),
+            visual_review_service=FakeVisualReviewService(),
+            visual_capture_retention_service=FakeVisualCaptureRetentionService(),
+            research_council_service=FakeResearchCouncilService(),
+            runtime_task_watchdog_service=RuntimeTaskWatchdogService(stale_multiplier=1, min_stuck_seconds=1),
+        )
+        stale_ts = (datetime.now(timezone.utc) - timedelta(seconds=30)).isoformat()
+        scheduler.tasks["watch"].last_status = "running"
+        scheduler.tasks["watch"].run_count = 1
+        scheduler.tasks["watch"].last_started_ts = stale_ts
+
+        payload = scheduler.status_snapshot()
+
+        self.assertEqual(payload["runtime_watchdog"]["stalled_task_count"], 1)
+        self.assertIn("runtime_task_stalled:watch:running_too_long", payload["current_warnings"])
+
 
 class TradingRuntimeApiTests(unittest.TestCase):
     def test_routes_runtime_status_and_run_once(self) -> None:
@@ -293,6 +319,16 @@ class TradingRuntimeApiTests(unittest.TestCase):
         self.assertEqual(status, 200)
         self.assertEqual(payload["matched_execution_ids"], [1])
 
+    def test_routes_runtime_watchdog(self) -> None:
+        status, payload = route_request(
+            "/trading/runtime-watchdog",
+            {"run_once": ["1"]},
+            dependencies={"trading_runtime_watchdog": lambda run_once=False: {"stalled_task_count": 0, "run_once": run_once}},
+        )
+
+        self.assertEqual(status, 200)
+        self.assertTrue(payload["run_once"])
+
 
 class TradingRuntimeScriptTests(unittest.TestCase):
     def test_script_outputs_json(self) -> None: