diff --git a/astrbot/core/core_lifecycle.py b/astrbot/core/core_lifecycle.py
index c325a2ea38..215b817544 100644
--- a/astrbot/core/core_lifecycle.py
+++ b/astrbot/core/core_lifecycle.py
@@ -35,6 +35,9 @@
 from astrbot.core.subagent_orchestrator import SubAgentOrchestrator
 from astrbot.core.umop_config_router import UmopConfigRouter
 from astrbot.core.updator import AstrBotUpdator
+from astrbot.core.utils.event_loop_diagnostics import (
+    create_event_loop_diagnostic_tasks,
+)
 from astrbot.core.utils.llm_metadata import update_llm_metadata
 from astrbot.core.utils.migra_helper import migra
 from astrbot.core.utils.temp_dir_cleaner import TempDirCleaner
@@ -296,13 +299,18 @@ def _load(self) -> None:
                 self.temp_dir_cleaner.run(),
                 name="temp_dir_cleaner",
             )
+        diagnostic_tasks = create_event_loop_diagnostic_tasks()
 
         # 把插件中注册的所有协程函数注册到事件总线中并执行
         extra_tasks = []
         for task in self.star_context._register_tasks:
             extra_tasks.append(asyncio.create_task(task, name=task.__name__))  # type: ignore
 
-        tasks_ = [event_bus_task, *(extra_tasks if extra_tasks else [])]
+        tasks_ = [
+            event_bus_task,
+            *diagnostic_tasks,
+            *(extra_tasks if extra_tasks else []),
+        ]
         if cron_task:
             tasks_.append(cron_task)
         if temp_dir_cleaner_task:
diff --git a/astrbot/core/utils/event_loop_diagnostics.py b/astrbot/core/utils/event_loop_diagnostics.py
new file mode 100644
index 0000000000..50eabea5df
--- /dev/null
+++ b/astrbot/core/utils/event_loop_diagnostics.py
@@ -0,0 +1,216 @@
+import asyncio
+import faulthandler
+from dataclasses import dataclass
+from pathlib import Path
+from typing import TextIO
+
+from astrbot import logger
+from astrbot.core.utils.astrbot_path import get_astrbot_data_path
+
+DEFAULT_LAG_MONITOR_ENABLED = True
+DEFAULT_LAG_MONITOR_INTERVAL = 5.0
+DEFAULT_LAG_MONITOR_THRESHOLD = 15.0
+DEFAULT_WATCHDOG_ENABLED = True
+DEFAULT_WATCHDOG_INTERVAL = 5.0
+DEFAULT_WATCHDOG_TIMEOUT = 30.0
+DEFAULT_WATCHDOG_LOG_RELATIVE_PATH = Path("logs") / "event_loop_watchdog.log"
+DEFAULT_WATCHDOG_LOG_MAX_BYTES = 1024 * 1024
+
+
+@dataclass(frozen=True)
+class EventLoopDiagnosticSettings:
+    """Settings for event loop lag and blockage diagnostics.
+
+    Args:
+        lag_monitor_enabled: Whether to log event loop scheduling lag.
+        lag_monitor_interval: Seconds between lag monitor wakeups.
+        lag_monitor_threshold: Minimum lag seconds before logging a warning.
+        watchdog_enabled: Whether to arm the faulthandler watchdog.
+        watchdog_interval: Seconds between faulthandler watchdog refreshes.
+        watchdog_timeout: Seconds without event loop progress before dumping stacks.
+        watchdog_log_path: File that receives faulthandler watchdog output.
+        watchdog_log_max_bytes: Maximum watchdog log bytes before rotation.
+    """
+
+    lag_monitor_enabled: bool
+    lag_monitor_interval: float
+    lag_monitor_threshold: float
+    watchdog_enabled: bool
+    watchdog_interval: float
+    watchdog_timeout: float
+    watchdog_log_path: Path
+    watchdog_log_max_bytes: int
+
+
+def _watchdog_log_path() -> Path:
+    """Resolve the watchdog stack dump log path.
+
+    Returns:
+        Absolute path for watchdog stack dump output.
+    """
+    return Path(get_astrbot_data_path()) / DEFAULT_WATCHDOG_LOG_RELATIVE_PATH
+
+
+def load_event_loop_diagnostic_settings() -> EventLoopDiagnosticSettings:
+    """Load fixed event loop diagnostic settings.
+
+    Returns:
+        Event loop diagnostic settings.
+    """
+    return EventLoopDiagnosticSettings(
+        lag_monitor_enabled=DEFAULT_LAG_MONITOR_ENABLED,
+        lag_monitor_interval=DEFAULT_LAG_MONITOR_INTERVAL,
+        lag_monitor_threshold=DEFAULT_LAG_MONITOR_THRESHOLD,
+        watchdog_enabled=DEFAULT_WATCHDOG_ENABLED,
+        watchdog_interval=DEFAULT_WATCHDOG_INTERVAL,
+        watchdog_timeout=DEFAULT_WATCHDOG_TIMEOUT,
+        watchdog_log_path=_watchdog_log_path(),
+        watchdog_log_max_bytes=DEFAULT_WATCHDOG_LOG_MAX_BYTES,
+    )
+
+
+async def monitor_event_loop_lag(
+    *,
+    interval: float = DEFAULT_LAG_MONITOR_INTERVAL,
+    warn_after: float = DEFAULT_LAG_MONITOR_THRESHOLD,
+) -> None:
+    """Log a warning when the event loop wakes significantly later than expected.
+
+    Args:
+        interval: Seconds between monitor wakeups.
+        warn_after: Minimum lag seconds before logging a warning.
+    """
+    loop = asyncio.get_running_loop()
+    expected = loop.time() + interval
+    while True:
+        await asyncio.sleep(interval)
+        now = loop.time()
+        lag = now - expected
+        if lag > warn_after:
+            logger.warning(
+                "Event loop lag detected: %.3fs (threshold %.3fs).",
+                lag,
+                warn_after,
+            )
+        expected = now + interval
+
+
+def _rotate_watchdog_log_file(log_path: Path, max_bytes: int) -> None:
+    """Rotate the watchdog log when it reaches the configured size limit.
+
+    Args:
+        log_path: Current watchdog log path.
+        max_bytes: Maximum current log size before rotation.
+    """
+    try:
+        if not log_path.exists() or log_path.stat().st_size < max_bytes:
+            return
+        rotated_path = log_path.with_name(f"{log_path.name}.1")
+        if rotated_path.exists():
+            rotated_path.unlink()
+        log_path.replace(rotated_path)
+    except OSError as e:
+        logger.warning("Failed to rotate event loop watchdog log %s: %s", log_path, e)
+
+
+def _open_watchdog_log_file(log_path: Path, max_bytes: int) -> TextIO:
+    """Open the watchdog log file after applying size-based rotation.
+
+    Args:
+        log_path: Current watchdog log path.
+        max_bytes: Maximum current log size before rotation.
+
+    Returns:
+        Writable text file object for faulthandler output.
+    """
+    log_path.parent.mkdir(parents=True, exist_ok=True)
+    _rotate_watchdog_log_file(log_path, max_bytes)
+    return log_path.open("a", encoding="utf-8")
+
+
+async def faulthandler_event_loop_watchdog(
+    *,
+    timeout: float = DEFAULT_WATCHDOG_TIMEOUT,
+    interval: float = DEFAULT_WATCHDOG_INTERVAL,
+    dump_file: TextIO | None = None,
+    dump_path: Path | None = None,
+    max_bytes: int = DEFAULT_WATCHDOG_LOG_MAX_BYTES,
+) -> None:
+    """Dump all thread stacks if the event loop is blocked for too long.
+
+    Args:
+        timeout: Seconds without watchdog refresh before faulthandler dumps stacks.
+        interval: Seconds between watchdog refreshes while the event loop is healthy.
+        dump_file: File object that receives faulthandler output.
+        dump_path: Path that receives faulthandler output when dump_file is unset.
+        max_bytes: Maximum current log size before rotation.
+    """
+    log_path = dump_path or _watchdog_log_path()
+    try:
+        while True:
+            faulthandler.cancel_dump_traceback_later()
+            output: TextIO | None = None
+            should_close = False
+            try:
+                output = dump_file or _open_watchdog_log_file(log_path, max_bytes)
+                should_close = dump_file is None
+                faulthandler.dump_traceback_later(
+                    timeout,
+                    repeat=False,
+                    file=output,
+                )
+                await asyncio.sleep(interval)
+            except Exception as e:
+                logger.warning("Event loop faulthandler watchdog failed: %s", e)
+                await asyncio.sleep(interval)
+            finally:
+                faulthandler.cancel_dump_traceback_later()
+                if should_close and output is not None:
+                    output.close()
+    finally:
+        faulthandler.cancel_dump_traceback_later()
+
+
+def create_event_loop_diagnostic_tasks() -> list[asyncio.Task]:
+    """Create enabled event loop diagnostic tasks for the current loop.
+
+    Returns:
+        A list of created asyncio tasks.
+    """
+    settings = load_event_loop_diagnostic_settings()
+    tasks: list[asyncio.Task] = []
+
+    if settings.lag_monitor_enabled:
+        tasks.append(
+            asyncio.create_task(
+                monitor_event_loop_lag(
+                    interval=settings.lag_monitor_interval,
+                    warn_after=settings.lag_monitor_threshold,
+                ),
+                name="event_loop_lag_monitor",
+            )
+        )
+
+    if settings.watchdog_enabled:
+        logger.warning(
+            "Event loop faulthandler watchdog enabled: timeout=%.3fs interval=%.3fs. "
+            "If the loop is blocked, Python thread stacks will be written to %s "
+            "(rotates at %d bytes).",
+            settings.watchdog_timeout,
+            settings.watchdog_interval,
+            settings.watchdog_log_path,
+            settings.watchdog_log_max_bytes,
+        )
+        tasks.append(
+            asyncio.create_task(
+                faulthandler_event_loop_watchdog(
+                    timeout=settings.watchdog_timeout,
+                    interval=settings.watchdog_interval,
+                    dump_path=settings.watchdog_log_path,
+                    max_bytes=settings.watchdog_log_max_bytes,
+                ),
+                name="event_loop_faulthandler_watchdog",
+            )
+        )
+
+    return tasks
diff --git a/tests/unit/test_event_loop_diagnostics.py b/tests/unit/test_event_loop_diagnostics.py
new file mode 100644
index 0000000000..0737478f9e
--- /dev/null
+++ b/tests/unit/test_event_loop_diagnostics.py
@@ -0,0 +1,134 @@
+import asyncio
+
+import pytest
+
+from astrbot.core.utils import event_loop_diagnostics as diagnostics
+
+
+def test_load_event_loop_diagnostic_settings_defaults():
+    """Default settings enable lag monitoring and stack dump watchdog."""
+    settings = diagnostics.load_event_loop_diagnostic_settings()
+
+    assert settings.lag_monitor_enabled is True
+    assert settings.lag_monitor_interval == diagnostics.DEFAULT_LAG_MONITOR_INTERVAL
+    assert settings.lag_monitor_threshold == diagnostics.DEFAULT_LAG_MONITOR_THRESHOLD
+    assert settings.watchdog_enabled is True
+    assert settings.watchdog_interval == diagnostics.DEFAULT_WATCHDOG_INTERVAL
+    assert settings.watchdog_timeout == diagnostics.DEFAULT_WATCHDOG_TIMEOUT
+    assert settings.watchdog_log_max_bytes == diagnostics.DEFAULT_WATCHDOG_LOG_MAX_BYTES
+
+
+@pytest.mark.asyncio
+async def test_create_event_loop_diagnostic_tasks_defaults():
+    """Default diagnostics should create both event loop diagnostic tasks."""
+    tasks = diagnostics.create_event_loop_diagnostic_tasks()
+
+    try:
+        assert [task.get_name() for task in tasks] == [
+            "event_loop_lag_monitor",
+            "event_loop_faulthandler_watchdog",
+        ]
+    finally:
+        for task in tasks:
+            task.cancel()
+        await asyncio.gather(*tasks, return_exceptions=True)
+
+
+@pytest.mark.asyncio
+async def test_faulthandler_watchdog_cancels_pending_dump(monkeypatch):
+    """The faulthandler watchdog should cancel its pending dump on shutdown."""
+    calls = []
+
+    class FakeFaultHandler:
+        def cancel_dump_traceback_later(self):
+            calls.append("cancel")
+
+        def dump_traceback_later(self, timeout, repeat, file):
+            calls.append(("dump", timeout, repeat, file))
+
+    fake_faulthandler = FakeFaultHandler()
+    monkeypatch.setattr(diagnostics, "faulthandler", fake_faulthandler)
+
+    task = asyncio.create_task(
+        diagnostics.faulthandler_event_loop_watchdog(timeout=10, interval=1)
+    )
+    await asyncio.sleep(0)
+    task.cancel()
+    await asyncio.gather(task, return_exceptions=True)
+
+    assert any(isinstance(call, tuple) and call[0] == "dump" for call in calls)
+    assert calls[-1] == "cancel"
+
+
+@pytest.mark.asyncio
+async def test_faulthandler_watchdog_writes_rotating_log(tmp_path, monkeypatch):
+    """The faulthandler watchdog should write to and rotate its log file."""
+    log_path = tmp_path / "logs" / "event_loop_watchdog.log"
+    log_path.parent.mkdir()
+    log_path.write_text("x" * 8, encoding="utf-8")
+    calls = []
+
+    class FakeFaultHandler:
+        def cancel_dump_traceback_later(self):
+            calls.append("cancel")
+
+        def dump_traceback_later(self, timeout, repeat, file):
+            calls.append(("dump", timeout, repeat, file.name))
+            file.write("watchdog dump\n")
+            file.flush()
+
+    fake_faulthandler = FakeFaultHandler()
+    monkeypatch.setattr(diagnostics, "faulthandler", fake_faulthandler)
+
+    task = asyncio.create_task(
+        diagnostics.faulthandler_event_loop_watchdog(
+            timeout=10,
+            interval=1,
+            dump_path=log_path,
+            max_bytes=4,
+        )
+    )
+    await asyncio.sleep(0)
+    task.cancel()
+    await asyncio.gather(task, return_exceptions=True)
+
+    assert log_path.read_text(encoding="utf-8") == "watchdog dump\n"
+    assert log_path.with_name("event_loop_watchdog.log.1").read_text(
+        encoding="utf-8"
+    ) == "x" * 8
+    assert any(isinstance(call, tuple) and call[0] == "dump" for call in calls)
+
+
+@pytest.mark.asyncio
+async def test_faulthandler_watchdog_survives_dump_failure(tmp_path, monkeypatch):
+    """The watchdog should keep running after faulthandler arm failures."""
+    log_path = tmp_path / "event_loop_watchdog.log"
+    armed_again = asyncio.Event()
+    calls = []
+
+    class FakeFaultHandler:
+        def cancel_dump_traceback_later(self):
+            calls.append("cancel")
+
+        def dump_traceback_later(self, timeout, repeat, file):
+            calls.append(("dump", timeout, repeat, file.name))
+            if len([call for call in calls if isinstance(call, tuple)]) == 1:
+                raise RuntimeError("boom")
+            armed_again.set()
+
+    fake_faulthandler = FakeFaultHandler()
+    monkeypatch.setattr(diagnostics, "faulthandler", fake_faulthandler)
+
+    task = asyncio.create_task(
+        diagnostics.faulthandler_event_loop_watchdog(
+            timeout=10,
+            interval=0.01,
+            dump_path=log_path,
+        )
+    )
+    await asyncio.wait_for(armed_again.wait(), timeout=1)
+    task.cancel()
+    await asyncio.gather(task, return_exceptions=True)
+
+    dump_calls = [call for call in calls if isinstance(call, tuple)]
+    assert len(dump_calls) >= 2