feat(channels): record runtime channel errors into status surface
Start, send and reload-stop failures are log-only today, so a channel that built fine but crashed on start shows running:false in the UI with no reason anywhere. _record_error merges a scrubbed one-line summary (URL userinfo stripped, capped at 200 chars), a frozen error context and a UTC timestamp into the per-channel status that /channels/status and /channels/config already expose; a clean rebuild clears all three fields. Precedent: AstrBot PlatformStatus + record_error surfaced in its WebUI. Refs #1625. Signed-off-by: shadowinlife <shadowinlife@gmail.com>
This commit is contained in:
1 parent
14cabaf195
commit
00a756bb06
3 files changed
+263
-3
No files matched your search
@@ -221,6 +221,8 @@ def _channel_entry(name: str, section: dict[str, Any], status_map: dict[str, Any
|
||||
"loaded": bool(status_map.get("loaded", False)),
|
||||
"install_hint": str(status_map.get("install_hint") or ""),
|
||||
"error": str(status_map.get("error") or ""),
|
||||
"error_context": str(status_map.get("error_context") or ""),
|
||||
"error_at": str(status_map.get("error_at") or ""),
|
||||
"supports_test": _supports_connection_test(name),
|
||||
"sdk_available": bool(status_map.get("sdk_available", available)),
|
||||
"fields": channel_field_hints(name),
|
||||
|
||||
@@ -5,9 +5,12 @@ from __future__ import annotations
|
||||
import asyncio
|
||||
import hashlib
|
||||
import logging
|
||||
import re
|
||||
from collections import defaultdict
|
||||
from contextlib import suppress
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
from urllib.parse import urlsplit, urlunsplit
|
||||
|
||||
from src.channels.base import BaseChannel
|
||||
from src.channels.bus.events import OutboundMessage
|
||||
@@ -38,6 +41,30 @@ _RELOAD_STOP_TIMEOUT_S = 10.0
|
||||
# Inbound messages whose reply fingerprint is kept for duplicate suppression.
|
||||
_MAX_REPLY_FINGERPRINTS = 4096
|
||||
|
||||
# Runtime-error bookkeeping for the status surface (Refs #1625).
|
||||
_ERROR_CONTEXTS = frozenset({"start", "send", "reload_stop"})
|
||||
_ERROR_TEXT_LIMIT = 200
|
||||
_URL_IN_TEXT_RE = re.compile(r"[a-zA-Z][a-zA-Z0-9+.\-]*://\S+")
|
||||
|
||||
|
||||
def _strip_url_userinfo(text: str) -> str:
|
||||
"""Return *text* with username/password stripped from every embedded URL."""
|
||||
|
||||
def _strip(match: re.Match[str]) -> str:
|
||||
url = match.group(0)
|
||||
try:
|
||||
parts = urlsplit(url)
|
||||
if parts.username is None and parts.password is None:
|
||||
return url
|
||||
host = parts.hostname or ""
|
||||
if parts.port is not None: # raises ValueError on a malformed port
|
||||
host = f"{host}:{parts.port}"
|
||||
return urlunsplit((parts.scheme, host, parts.path, parts.query, parts.fragment))
|
||||
except ValueError:
|
||||
return url
|
||||
|
||||
return _URL_IN_TEXT_RE.sub(_strip, text)
|
||||
|
||||
|
||||
class ChannelManager:
|
||||
"""Manages chat channels and coordinates message routing.
|
||||
@@ -140,6 +167,8 @@ class ChannelManager:
|
||||
running=channel.is_running,
|
||||
display_name=getattr(cls, "display_name", name),
|
||||
error="",
|
||||
error_context="",
|
||||
error_at="",
|
||||
)
|
||||
logger.info("%s channel enabled", cls.display_name)
|
||||
return channel
|
||||
@@ -224,8 +253,9 @@ class ChannelManager:
|
||||
"""Start a channel and log any exceptions."""
|
||||
try:
|
||||
await channel.start()
|
||||
except Exception:
|
||||
except Exception as exc:
|
||||
logger.exception("Failed to start channel %s", name)
|
||||
self._record_error(name, "start", exc)
|
||||
|
||||
def _spawn_channel_start(self, name: str, channel: BaseChannel) -> None:
|
||||
"""Start a channel in the background, keeping a strong task reference."""
|
||||
@@ -301,17 +331,22 @@ class ChannelManager:
|
||||
|
||||
async def _reload_channel_locked(self, name: str, section: dict | None) -> dict[str, Any]:
|
||||
old = self.channels.get(name)
|
||||
stop_error: BaseException | None = None
|
||||
if old is not None:
|
||||
try:
|
||||
await asyncio.wait_for(old.stop(), timeout=_RELOAD_STOP_TIMEOUT_S)
|
||||
except asyncio.TimeoutError:
|
||||
except asyncio.TimeoutError as exc:
|
||||
logger.warning("Stopping %s timed out; replacing", name)
|
||||
except Exception:
|
||||
stop_error = exc
|
||||
except Exception as exc:
|
||||
logger.warning("Stopping %s failed; replacing", name, exc_info=True)
|
||||
stop_error = exc
|
||||
self.channels.pop(name, None)
|
||||
self._store_channel_section(name, section)
|
||||
|
||||
if section is None or not self._section_enabled(section):
|
||||
if stop_error is not None:
|
||||
self._record_error(name, "reload_stop", stop_error)
|
||||
return self._record_status(
|
||||
name,
|
||||
configured=section is not None,
|
||||
@@ -323,6 +358,10 @@ class ChannelManager:
|
||||
channel = self._build_channel(name, section)
|
||||
if channel is None:
|
||||
return self._record_status(name)
|
||||
if stop_error is not None:
|
||||
# After the build's success-clear: a wedged stop means the old adapter
|
||||
# may still be alive; the status must explain the degraded swap.
|
||||
self._record_error(name, "reload_stop", stop_error)
|
||||
self.channels[name] = channel
|
||||
if self._is_manager_running():
|
||||
# start() can be long-running (reconnect loops), so it must not be
|
||||
@@ -339,6 +378,19 @@ class ChannelManager:
|
||||
self._status[name].update(fields)
|
||||
return self._status[name]
|
||||
|
||||
def _record_error(self, name: str, context: str, exc: BaseException) -> None:
|
||||
"""Merge a scrubbed one-line runtime-error summary into ``_status[name]``.
|
||||
|
||||
``context`` is one of ``_ERROR_CONTEXTS``; the traceback stays in the log
|
||||
because the status surface flows into HTTP responses (Refs #1625).
|
||||
"""
|
||||
self._record_status(
|
||||
name,
|
||||
error=_strip_url_userinfo(f"{type(exc).__name__}: {exc}")[:_ERROR_TEXT_LIMIT],
|
||||
error_context=context,
|
||||
error_at=datetime.now(timezone.utc).isoformat(),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _section_enabled(section: Any) -> bool:
|
||||
"""Return whether a channel section enables the adapter."""
|
||||
@@ -549,6 +601,7 @@ class ChannelManager:
|
||||
"Failed to send to %s after %d attempts",
|
||||
msg.channel, max_attempts,
|
||||
)
|
||||
self._record_error(msg.channel, "send", e)
|
||||
return
|
||||
delay = _SEND_RETRY_DELAYS[min(attempt, len(_SEND_RETRY_DELAYS) - 1)]
|
||||
logger.warning(
|
||||
|
||||
@@ -0,0 +1,205 @@
|
||||
"""Contracts for the channel runtime health surface (``_record_error``)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from collections.abc import Iterator
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
import src.channels.manager as manager_module
|
||||
from src.channels.bus.events import OutboundMessage
|
||||
from src.channels.bus.queue import MessageBus
|
||||
from src.channels.manager import ChannelManager
|
||||
|
||||
|
||||
class FakeChannel:
|
||||
"""Adapter stand-in whose failures are config-driven; no platform SDK is touched."""
|
||||
|
||||
display_name = "Fake"
|
||||
send_progress = True
|
||||
send_tool_hints = False
|
||||
show_reasoning = True
|
||||
|
||||
instances: list["FakeChannel"] = []
|
||||
|
||||
def __init__(self, config: dict[str, Any], bus: MessageBus, **kwargs: Any) -> None:
|
||||
del kwargs
|
||||
if config.get("bad_config"):
|
||||
raise ValueError("bad channel config")
|
||||
self.config = config
|
||||
self.bus = bus
|
||||
self.tag = str(config.get("tag", len(FakeChannel.instances)))
|
||||
self._running = False
|
||||
FakeChannel.instances.append(self)
|
||||
|
||||
@property
|
||||
def is_running(self) -> bool:
|
||||
return self._running
|
||||
|
||||
async def start(self) -> None:
|
||||
if self.config.get("fail_start"):
|
||||
raise RuntimeError("start failed")
|
||||
self._running = True
|
||||
|
||||
async def stop(self) -> None:
|
||||
if self.config.get("raise_stop"):
|
||||
raise TimeoutError("stop timed out")
|
||||
if self.config.get("hang_stop"):
|
||||
await asyncio.sleep(60)
|
||||
self._running = False
|
||||
|
||||
async def send(self, message: Any) -> None:
|
||||
del message
|
||||
if self.config.get("fail_send"):
|
||||
raise ConnectionError("send failed")
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_fake_channel() -> Iterator[None]:
|
||||
"""Give every test a clean instance ledger."""
|
||||
FakeChannel.instances = []
|
||||
yield
|
||||
|
||||
|
||||
def _fake_inspect_channels(config: Any) -> dict[str, dict[str, Any]]:
|
||||
"""``inspect_channels`` stand-in covering the fake names under test."""
|
||||
statuses: dict[str, dict[str, Any]] = {}
|
||||
if not isinstance(config, dict):
|
||||
return statuses
|
||||
for name, section in config.items():
|
||||
if not isinstance(section, dict):
|
||||
continue
|
||||
statuses[name] = {
|
||||
"name": name,
|
||||
"available": True,
|
||||
"display_name": name.title(),
|
||||
"install_hint": "",
|
||||
"error": "",
|
||||
"configured": True,
|
||||
"enabled": bool(section.get("enabled")),
|
||||
"loaded": False,
|
||||
"running": False,
|
||||
}
|
||||
return statuses
|
||||
|
||||
|
||||
def _build_manager(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
config: dict[str, Any],
|
||||
*,
|
||||
names: tuple[str, ...] = ("fakea",),
|
||||
) -> ChannelManager:
|
||||
"""Construct a manager whose adapters are :class:`FakeChannel`."""
|
||||
monkeypatch.setattr(manager_module, "discover_channel_names", lambda: list(names))
|
||||
monkeypatch.setattr(manager_module, "load_channel_class", lambda name: FakeChannel)
|
||||
monkeypatch.setattr(manager_module, "inspect_channels", _fake_inspect_channels)
|
||||
return ChannelManager(config, MessageBus())
|
||||
|
||||
|
||||
def test_start_failure_records_error_into_status(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
async def scenario() -> None:
|
||||
manager = _build_manager(
|
||||
monkeypatch, {"fakea": {"enabled": True, "fail_start": True}}
|
||||
)
|
||||
await manager.start_all()
|
||||
try:
|
||||
status = manager.get_status()["fakea"]
|
||||
assert status["running"] is False
|
||||
assert "RuntimeError" in status["error"]
|
||||
assert "start failed" in status["error"]
|
||||
assert status["error_context"] == "start"
|
||||
assert datetime.fromisoformat(status["error_at"]).tzinfo is not None
|
||||
finally:
|
||||
await manager.stop_all()
|
||||
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
def test_send_final_failure_records_send_context(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
async def scenario() -> None:
|
||||
manager = _build_manager(
|
||||
monkeypatch,
|
||||
{"send_max_retries": 1, "fakea": {"enabled": True, "fail_send": True}},
|
||||
)
|
||||
msg = OutboundMessage(channel="fakea", chat_id="chat-1", content="hello")
|
||||
|
||||
await manager._send_with_retry(manager.channels["fakea"], msg)
|
||||
|
||||
status = manager.get_status()["fakea"]
|
||||
assert status["error_context"] == "send"
|
||||
assert "ConnectionError" in status["error"]
|
||||
assert datetime.fromisoformat(status["error_at"]).tzinfo is not None
|
||||
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
@pytest.mark.parametrize("old_extra", [{"hang_stop": True}, {"raise_stop": True}])
|
||||
def test_reload_stop_failure_records_reload_stop_and_reload_completes(
|
||||
monkeypatch: pytest.MonkeyPatch, old_extra: dict[str, Any],
|
||||
) -> None:
|
||||
async def scenario() -> None:
|
||||
monkeypatch.setattr(manager_module, "_RELOAD_STOP_TIMEOUT_S", 0.05)
|
||||
manager = _build_manager(
|
||||
monkeypatch, {"fakea": {"enabled": True, "tag": "old", **old_extra}}
|
||||
)
|
||||
|
||||
status = dict(
|
||||
await manager.reload_channel("fakea", {"enabled": True, "tag": "new"})
|
||||
)
|
||||
|
||||
assert manager.channels["fakea"].tag == "new" # the reload still completed
|
||||
assert status["loaded"] is True
|
||||
assert status["error_context"] == "reload_stop"
|
||||
assert status["error"].startswith("TimeoutError")
|
||||
assert datetime.fromisoformat(status["error_at"]).tzinfo is not None
|
||||
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
def test_successful_rebuild_clears_stale_error(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
async def scenario() -> None:
|
||||
manager = _build_manager(monkeypatch, {"fakea": {"enabled": True, "tag": "old"}})
|
||||
manager._record_error("fakea", "start", RuntimeError("earlier crash"))
|
||||
assert manager.get_status()["fakea"]["error_context"] == "start"
|
||||
|
||||
status = dict(
|
||||
await manager.reload_channel("fakea", {"enabled": True, "tag": "new"})
|
||||
)
|
||||
|
||||
assert status["error"] == ""
|
||||
assert status["error_context"] == ""
|
||||
assert status["error_at"] == ""
|
||||
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
def test_record_error_scrubs_credentials_and_bounds_length(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
manager = _build_manager(monkeypatch, {"fakea": {"enabled": True}})
|
||||
|
||||
manager._record_error(
|
||||
"fakea",
|
||||
"start",
|
||||
ConnectionError("cannot reach https://bot:sup3rsecret@chat.example.com:443/api"),
|
||||
)
|
||||
error = manager.get_status()["fakea"]["error"]
|
||||
assert "sup3rsecret" not in error
|
||||
assert "bot" not in error
|
||||
assert "https://chat.example.com:443/api" in error
|
||||
|
||||
manager._record_error(
|
||||
"fakea",
|
||||
"start",
|
||||
ValueError("bad url https://user:pass@host:notaport/x"),
|
||||
)
|
||||
error = manager.get_status()["fakea"]["error"]
|
||||
assert "https://user:pass@host:notaport/x" in error # malformed URLs pass through
|
||||
|
||||
manager._record_error("fakea", "send", RuntimeError("x" * 500))
|
||||
error = manager.get_status()["fakea"]["error"]
|
||||
assert error.startswith("RuntimeError: ")
|
||||
assert len(error) <= 200
|
||||
Reference in new issue
Block a user