feat: add event loop diagnostics

[Soulter]

Summary

• add a low-overhead event loop lag monitor that logs when scheduling lag exceeds the threshold
• add a faulthandler watchdog that dumps Python thread stacks when the event loop is blocked too long
• write watchdog stack dumps to a rotating log file under the AstrBot data logs directory
• start diagnostics with the core lifecycle and cover fixed diagnostic behavior with unit tests

Defaults

• lag monitor interval: 5s
• lag warning threshold: 15s
• watchdog refresh interval: 5s
• watchdog dump timeout: 30s
• watchdog log path: data/logs/event_loop_watchdog.log
• watchdog log rotation: current file rotates to .1 at 1048576 bytes

Tests

• uv run pytest tests/unit/test_event_loop_diagnostics.py tests/unit/test_core_lifecycle.py
• ruff format .
• ruff check .

[sourcery-ai]

Hey - I've found 2 issues, and left some high level feedback:

• create_event_loop_diagnostic_tasks() calls asyncio.create_task immediately, which assumes a running event loop; consider either accepting an explicit loop or guarding its use so it can’t be invoked from non-async contexts without failing.
• faulthandler_event_loop_watchdog() cancels dump_traceback_later() on each iteration and in the finally block, which will also cancel any dumps scheduled elsewhere; consider whether this should be isolated or clearly constrained to avoid interfering with other faulthandler usage.
• In core_lifecycle._load(), diagnostic_tasks is always created even when all diagnostics are disabled; you could skip the call or return early from create_event_loop_diagnostic_tasks() to avoid constructing an unused list.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- create_event_loop_diagnostic_tasks() calls asyncio.create_task immediately, which assumes a running event loop; consider either accepting an explicit loop or guarding its use so it can’t be invoked from non-async contexts without failing.
- faulthandler_event_loop_watchdog() cancels dump_traceback_later() on each iteration and in the finally block, which will also cancel any dumps scheduled elsewhere; consider whether this should be isolated or clearly constrained to avoid interfering with other faulthandler usage.
- In core_lifecycle._load(), diagnostic_tasks is always created even when all diagnostics are disabled; you could skip the call or return early from create_event_loop_diagnostic_tasks() to avoid constructing an unused list.

## Individual Comments

### Comment 1
<location path="astrbot/core/utils/event_loop_diagnostics.py" line_range="154-163" />
<code_context>
+        expected = now + interval
+
+
+async def faulthandler_event_loop_watchdog(
+    *,
+    timeout: float = DEFAULT_WATCHDOG_TIMEOUT,
+    interval: float = DEFAULT_WATCHDOG_INTERVAL,
+    dump_file: TextIO | None = None,
+) -> 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.
+    """
+    output = dump_file or sys.stderr
+    if not faulthandler.is_enabled():
+        faulthandler.enable(file=output)
+
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Avoid leaving faulthandler globally enabled after the watchdog finishes.

This function enables faulthandler when it’s not already enabled, but in the `finally` block it only cancels the scheduled dump and never disables faulthandler. In environments where faulthandler isn’t expected to be globally enabled or is managed elsewhere, this can cause unexpected stack dumps on unrelated crashes. Please track whether this function enabled faulthandler and, if so, disable it in `finally`, leaving any prior configuration unchanged.

Suggested implementation:

```python
async def faulthandler_event_loop_watchdog(
    *,
    timeout: float = DEFAULT_WATCHDOG_TIMEOUT,
    interval: float = DEFAULT_WATCHDOG_INTERVAL,
    dump_file: TextIO | None = None,
) -> 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.
    """
    output = dump_file or sys.stderr
    # Track whether this function enabled faulthandler so we can restore the
    # previous global state when the watchdog finishes.
    faulthandler_was_enabled = faulthandler.is_enabled()
    if not faulthandler_was_enabled:
        faulthandler.enable(file=output)

```

The `finally` block of `faulthandler_event_loop_watchdog` (not shown in the snippet) should be updated to restore the original faulthandler state. Specifically:
1. Ensure the function body is wrapped in a `try`/`finally` (if it isn’t already).
2. At the start of the function (after the new `faulthandler_was_enabled` assignment), declare `faulthandler_was_enabled` as nonlocal within the scope that the `finally` block can access, or structure the code so the `finally` block can read that variable.
3. In the `finally` block, after cancelling any scheduled dumps, add:
   ```python
   if not faulthandler_was_enabled and faulthandler.is_enabled():
       faulthandler.disable()
   ```
   This disables faulthandler only if this function enabled it and leaves any prior configuration unchanged.
</issue_to_address>

### Comment 2
<location path="tests/unit/test_event_loop_diagnostics.py" line_range="51-60" />
<code_context>
+    assert settings.watchdog_timeout == 30
+
+
+@pytest.mark.asyncio
+async def test_create_event_loop_diagnostic_tasks_respects_env(monkeypatch):
+    """Disabled lag monitor and watchdog should create no tasks."""
+    _clear_diagnostic_env(monkeypatch)
+    monkeypatch.setenv(diagnostics.LAG_MONITOR_ENABLED_ENV, "false")
+    monkeypatch.setenv(diagnostics.WATCHDOG_ENABLED_ENV, "false")
+
+    tasks = diagnostics.create_event_loop_diagnostic_tasks()
+
+    assert tasks == []
+
+
</code_context>
<issue_to_address>
**suggestion (testing):** Also test the enabled path of `create_event_loop_diagnostic_tasks` to prove tasks are created with the expected configuration.

Current tests only cover fully disabled behavior and the default lag-monitor-only case. Please add a test that explicitly enables the watchdog via environment variables (including custom timeout/interval), then verifies that both `event_loop_lag_monitor` and `event_loop_faulthandler_watchdog` tasks are created, the enabling warning is logged, and the watchdog is invoked with the configured timeout/interval. This will validate that environment configuration is correctly propagated into task creation.

Suggested implementation:

```python
    assert settings.watchdog_timeout == 30



import pytest

from astrbot.core.utils import event_loop_diagnostics as diagnostics


@pytest.mark.asyncio
async def test_create_event_loop_diagnostic_tasks_enabled_watchdog(monkeypatch, caplog):
    """Explicitly enable lag monitor and watchdog and validate task creation/config propagation."""
    _clear_diagnostic_env(monkeypatch)

    # Explicitly enable both diagnostics via environment, with custom watchdog settings.
    monkeypatch.setenv(diagnostics.LAG_MONITOR_ENABLED_ENV, "true")
    monkeypatch.setenv(diagnostics.WATCHDOG_ENABLED_ENV, "true")
    monkeypatch.setenv(diagnostics.WATCHDOG_TIMEOUT_ENV, "45")
    monkeypatch.setenv(diagnostics.WATCHDOG_INTERVAL_ENV, "0.5")

    lag_monitor_calls = {}
    watchdog_calls = {}

    async def fake_lag_monitor(*args, **kwargs):
        lag_monitor_calls["args"] = args
        lag_monitor_calls["kwargs"] = kwargs

    async def fake_watchdog(*args, **kwargs):
        watchdog_calls["args"] = args
        watchdog_calls["kwargs"] = kwargs

    # Patch the underlying diagnostic coroutines so we can inspect how they are invoked.
    monkeypatch.setattr(diagnostics, "event_loop_lag_monitor", fake_lag_monitor)
    monkeypatch.setattr(diagnostics, "event_loop_faulthandler_watchdog", fake_watchdog)

    with caplog.at_level("WARNING"):
        tasks = diagnostics.create_event_loop_diagnostic_tasks()

    # Both diagnostics should be enabled and result in task creation.
    assert isinstance(tasks, list)
    assert len(tasks) == 2

    # Validate that the lag monitor and watchdog were invoked.
    assert "args" in lag_monitor_calls
    assert "args" in watchdog_calls

    # Validate watchdog configuration propagation (timeout/interval from environment).
    watchdog_kwargs = watchdog_calls.get("kwargs", {})
    assert watchdog_kwargs.get("timeout") == 45
    assert watchdog_kwargs.get("interval") == 0.5

    # Validate that an enabling warning was logged when the watchdog was turned on.
    assert any(
        "watchdog" in record.message.lower()
        and "enabled" in record.message.lower()
        for record in caplog.records
    )

```

1. If `create_event_loop_diagnostic_tasks` uses different parameter names or types for the watchdog (e.g. `timeout_seconds` instead of `timeout`, or expects floats vs ints), adjust the `watchdog_kwargs` assertions to match the actual signature.
2. If the logger used to emit the enabling warning is not the root logger, make sure the test's `caplog.at_level` target (e.g. `logger=diagnostics.logger`) is configured appropriately.
3. If the warning message text differs (e.g. does not contain both “watchdog” and “enabled”), update the `any(...)` assertion to match the actual wording so the test remains robust but still proves the warning is emitted.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[sourcery-ai]

suggestion (testing): Also test the enabled path of create_event_loop_diagnostic_tasks to prove tasks are created with the expected configuration.

Current tests only cover fully disabled behavior and the default lag-monitor-only case. Please add a test that explicitly enables the watchdog via environment variables (including custom timeout/interval), then verifies that both event_loop_lag_monitor and event_loop_faulthandler_watchdog tasks are created, the enabling warning is logged, and the watchdog is invoked with the configured timeout/interval. This will validate that environment configuration is correctly propagated into task creation.

Suggested implementation:

    assert settings.watchdog_timeout == 30



import pytest

from astrbot.core.utils import event_loop_diagnostics as diagnostics


@pytest.mark.asyncio
async def test_create_event_loop_diagnostic_tasks_enabled_watchdog(monkeypatch, caplog):
    """Explicitly enable lag monitor and watchdog and validate task creation/config propagation."""
    _clear_diagnostic_env(monkeypatch)

    # Explicitly enable both diagnostics via environment, with custom watchdog settings.
    monkeypatch.setenv(diagnostics.LAG_MONITOR_ENABLED_ENV, "true")
    monkeypatch.setenv(diagnostics.WATCHDOG_ENABLED_ENV, "true")
    monkeypatch.setenv(diagnostics.WATCHDOG_TIMEOUT_ENV, "45")
    monkeypatch.setenv(diagnostics.WATCHDOG_INTERVAL_ENV, "0.5")

    lag_monitor_calls = {}
    watchdog_calls = {}

    async def fake_lag_monitor(*args, **kwargs):
        lag_monitor_calls["args"] = args
        lag_monitor_calls["kwargs"] = kwargs

    async def fake_watchdog(*args, **kwargs):
        watchdog_calls["args"] = args
        watchdog_calls["kwargs"] = kwargs

    # Patch the underlying diagnostic coroutines so we can inspect how they are invoked.
    monkeypatch.setattr(diagnostics, "event_loop_lag_monitor", fake_lag_monitor)
    monkeypatch.setattr(diagnostics, "event_loop_faulthandler_watchdog", fake_watchdog)

    with caplog.at_level("WARNING"):
        tasks = diagnostics.create_event_loop_diagnostic_tasks()

    # Both diagnostics should be enabled and result in task creation.
    assert isinstance(tasks, list)
    assert len(tasks) == 2

    # Validate that the lag monitor and watchdog were invoked.
    assert "args" in lag_monitor_calls
    assert "args" in watchdog_calls

    # Validate watchdog configuration propagation (timeout/interval from environment).
    watchdog_kwargs = watchdog_calls.get("kwargs", {})
    assert watchdog_kwargs.get("timeout") == 45
    assert watchdog_kwargs.get("interval") == 0.5

    # Validate that an enabling warning was logged when the watchdog was turned on.
    assert any(
        "watchdog" in record.message.lower()
        and "enabled" in record.message.lower()
        for record in caplog.records
    )

1. If create_event_loop_diagnostic_tasks uses different parameter names or types for the watchdog (e.g. timeout_seconds instead of timeout, or expects floats vs ints), adjust the watchdog_kwargs assertions to match the actual signature.
2. If the logger used to emit the enabling warning is not the root logger, make sure the test's caplog.at_level target (e.g. logger=diagnostics.logger) is configured appropriately.
3. If the warning message text differs (e.g. does not contain both “watchdog” and “enabled”), update the any(...) assertion to match the actual wording so the test remains robust but still proves the warning is emitted.

[gemini-code-assist]

Code Review

This pull request introduces event loop diagnostics to monitor scheduling lag and blockages, integrating a lag monitor and a faulthandler-based watchdog into the core lifecycle. The review feedback highlights a potential issue where a watchdog timeout configured to be less than or equal to the check interval would trigger false-positive stack dumps. It recommends validating and adjusting the timeout dynamically and adding a corresponding unit test to verify this behavior.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

If watchdog_timeout is less than or equal to watchdog_interval, the watchdog thread will trigger false positive stack dumps under normal operation (since the loop sleeps for interval seconds, which exceeds the timeout threshold). We should validate and adjust the timeout to be at least greater than the interval (e.g., interval * 2.0) to prevent this configuration hazard.

    lag_monitor_enabled = _env_flag(LAG_MONITOR_ENABLED_ENV, True)
    lag_monitor_interval = _env_float(
        LAG_MONITOR_INTERVAL_ENV,
        DEFAULT_LAG_MONITOR_INTERVAL,
        0.1,
    )
    lag_monitor_threshold = _env_float(
        LAG_MONITOR_THRESHOLD_ENV,
        DEFAULT_LAG_MONITOR_THRESHOLD,
        0.1,
    )
    watchdog_enabled = _env_flag(WATCHDOG_ENABLED_ENV, False)
    watchdog_interval = _env_float(
        WATCHDOG_INTERVAL_ENV,
        DEFAULT_WATCHDOG_INTERVAL,
        0.1,
    )
    watchdog_timeout = _env_float(
        WATCHDOG_TIMEOUT_ENV,
        DEFAULT_WATCHDOG_TIMEOUT,
        1.0,
    )

    if watchdog_enabled and watchdog_timeout <= watchdog_interval:
        logger.warning(
            "Watchdog timeout (%.3fs) must be greater than interval (%.3fs) to prevent false positives. "
            "Adjusting timeout to %.3fs.",
            watchdog_timeout,
            watchdog_interval,
            watchdog_interval * 2.0,
        )
        watchdog_timeout = watchdog_interval * 2.0

    return EventLoopDiagnosticSettings(
        lag_monitor_enabled=lag_monitor_enabled,
        lag_monitor_interval=lag_monitor_interval,
        lag_monitor_threshold=lag_monitor_threshold,
        watchdog_enabled=watchdog_enabled,
        watchdog_interval=watchdog_interval,
        watchdog_timeout=watchdog_timeout,
    )

[gemini-code-assist]

Add a unit test to verify that the watchdog timeout is correctly adjusted when it is configured to be less than or equal to the watchdog interval.

Suggested change

	assert "is_enabled" in calls
	assert any(isinstance(call, tuple) and call[0] == "dump" for call in calls)
	assert calls[-1] == "cancel"
	assert "is_enabled" in calls
	assert any(isinstance(call, tuple) and call[0] == "dump" for call in calls)
	assert calls[-1] == "cancel"
	def test_load_event_loop_diagnostic_settings_invalid_watchdog_relation(monkeypatch):
	"""Watchdog timeout must be greater than interval; otherwise it should be adjusted."""
	_clear_diagnostic_env(monkeypatch)
	monkeypatch.setenv(diagnostics.WATCHDOG_ENABLED_ENV, "true")
	monkeypatch.setenv(diagnostics.WATCHDOG_INTERVAL_ENV, "5.0")
	monkeypatch.setenv(diagnostics.WATCHDOG_TIMEOUT_ENV, "2.0")
	settings = diagnostics.load_event_loop_diagnostic_settings()
	assert settings.watchdog_enabled is True
	assert settings.watchdog_interval == 5.0
	assert settings.watchdog_timeout == 10.0 # adjusted to interval * 2

[cloudflare-workers-and-pages]

Deploying with    Cloudflare Workers

The latest updates on your project. Learn more about integrating Git with Workers.

Status	Name	Latest Commit	Preview URL	Updated (UTC)
✅ Deployment successful! View logs	astrbot-docs	a18054c	Commit Preview URL Branch Preview URL	Jul 06 2026, 04:16 PM