mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-21 16:18:55 +00:00
fix(discord): close reconnect recovery edge cases
Include allowed mention-gated channels in default recovery scope, keep the newest bounded history window, narrow outage-message detection, avoid disabled-path ledger I/O, and persist forum replies.
This commit is contained in:
Teknium
parent
2278f2cb7e
commit
80744bc2bc
2 changed files with 115 additions and 20 deletions
|
|
@ -1912,7 +1912,12 @@ class DiscordAdapter(BasePlatformAdapter):
|
|||
"""
|
||||
raw = os.getenv("DISCORD_MISSED_MESSAGE_BACKFILL_CHANNELS", "")
|
||||
if not raw.strip():
|
||||
return self._discord_free_response_channels()
|
||||
allowed = {
|
||||
item.strip()
|
||||
for item in os.getenv("DISCORD_ALLOWED_CHANNELS", "").split(",")
|
||||
if item.strip()
|
||||
}
|
||||
return allowed | self._discord_free_response_channels()
|
||||
return {item.strip() for item in raw.split(",") if item.strip()}
|
||||
|
||||
def _missed_message_backfill_window_seconds(self) -> float:
|
||||
|
|
@ -2012,9 +2017,7 @@ class DiscordAdapter(BasePlatformAdapter):
|
|||
async def _iter_missed_message_backfill_candidates(self, channel_ids: set[str]):
|
||||
if not self._client:
|
||||
return
|
||||
import datetime as _dt
|
||||
|
||||
after = _dt.datetime.now(_dt.timezone.utc) - _dt.timedelta(
|
||||
after = dt.datetime.now(dt.timezone.utc) - dt.timedelta(
|
||||
seconds=self._missed_message_backfill_window_seconds()
|
||||
)
|
||||
limit = self._missed_message_backfill_limit()
|
||||
|
|
@ -2057,10 +2060,18 @@ class DiscordAdapter(BasePlatformAdapter):
|
|||
history = getattr(channel, "history", None)
|
||||
if callable(history):
|
||||
try:
|
||||
# Fetch the latest N messages in the window, then restore
|
||||
# chronological dispatch order. With oldest_first=True the API
|
||||
# returns the earliest N and can permanently starve newer work.
|
||||
history_iter = history(
|
||||
limit=limit,
|
||||
after=after,
|
||||
oldest_first=False,
|
||||
)
|
||||
messages = []
|
||||
async for message in history(limit=limit, after=after, oldest_first=True):
|
||||
async for message in history_iter: # type: ignore[attr-defined]
|
||||
messages.append(message)
|
||||
for message in messages:
|
||||
for message in reversed(messages):
|
||||
yield message
|
||||
except Exception as exc:
|
||||
logger.debug("[%s] Cannot read history for %s: %s", self.name, channel_key, exc)
|
||||
|
|
@ -2117,19 +2128,12 @@ class DiscordAdapter(BasePlatformAdapter):
|
|||
return False
|
||||
|
||||
def _is_down_notice_content(self, content: str) -> bool:
|
||||
"""Recognize only explicit Hermes/gateway outage notices."""
|
||||
text = (content or "").lower()
|
||||
down_markers = (
|
||||
"agent is down",
|
||||
"agent was down",
|
||||
"gateway is down",
|
||||
"gateway was down",
|
||||
"bmo is down",
|
||||
"currently down",
|
||||
"offline",
|
||||
"unavailable",
|
||||
"not running",
|
||||
)
|
||||
return any(marker in text for marker in down_markers)
|
||||
subject = r"(?:hermes|the agent|agent|the gateway|gateway|bmo)"
|
||||
state = r"(?:is|was|appears to be|is currently|was currently)"
|
||||
condition = r"(?:down|offline|unavailable|not running)"
|
||||
return re.search(rf"\b{subject}\s+{state}\s+{condition}\b", text) is not None
|
||||
|
||||
async def _message_has_non_down_bot_response(self, message: Any) -> bool:
|
||||
"""Detect an already-addressed message without trusting down notices."""
|
||||
|
|
@ -2259,6 +2263,8 @@ class DiscordAdapter(BasePlatformAdapter):
|
|||
return channel_id, thread_id, parent_id
|
||||
|
||||
def _record_discord_message_seen(self, message: Any, *, status: str) -> None:
|
||||
if not self._missed_message_backfill_enabled():
|
||||
return
|
||||
message_id = str(getattr(message, "id", "") or "")
|
||||
if not message_id:
|
||||
return
|
||||
|
|
@ -2290,6 +2296,8 @@ class DiscordAdapter(BasePlatformAdapter):
|
|||
self._with_discord_recovery_db(_op)
|
||||
|
||||
def _record_recovery_attempt(self, message: Any, *, status: str, error: Optional[str] = None) -> None:
|
||||
if not self._missed_message_backfill_enabled():
|
||||
return
|
||||
self._record_discord_message_seen(message, status=status)
|
||||
message_id = str(getattr(message, "id", "") or "")
|
||||
if not message_id:
|
||||
|
|
@ -2309,6 +2317,8 @@ class DiscordAdapter(BasePlatformAdapter):
|
|||
self._with_discord_recovery_db(_op)
|
||||
|
||||
def _record_discord_processing_start(self, event: MessageEvent, *, emoji_ack: bool) -> None:
|
||||
if not self._missed_message_backfill_enabled():
|
||||
return
|
||||
message = event.raw_message
|
||||
self._record_discord_message_seen(message, status="processing")
|
||||
message_id = str(getattr(message, "id", "") or getattr(event, "message_id", "") or "")
|
||||
|
|
@ -2325,6 +2335,8 @@ class DiscordAdapter(BasePlatformAdapter):
|
|||
self._with_discord_recovery_db(_op)
|
||||
|
||||
def _record_discord_processing_complete(self, event: MessageEvent, outcome: ProcessingOutcome) -> None:
|
||||
if not self._missed_message_backfill_enabled():
|
||||
return
|
||||
message_id = str(getattr(getattr(event, "raw_message", None), "id", "") or getattr(event, "message_id", "") or "")
|
||||
if not message_id:
|
||||
return
|
||||
|
|
@ -2340,7 +2352,7 @@ class DiscordAdapter(BasePlatformAdapter):
|
|||
self._with_discord_recovery_db(_op)
|
||||
|
||||
def _record_discord_response(self, *, reply_to: Optional[str], result: SendResult, content: str) -> None:
|
||||
if not reply_to:
|
||||
if not self._missed_message_backfill_enabled() or not reply_to:
|
||||
return
|
||||
now = self._utc_now_iso()
|
||||
outage = self._is_down_notice_content(content)
|
||||
|
|
@ -2684,7 +2696,13 @@ class DiscordAdapter(BasePlatformAdapter):
|
|||
|
||||
# Forum channels reject channel.send() — create a thread post instead.
|
||||
if self._is_forum_parent(channel):
|
||||
return await self._send_to_forum(channel, content)
|
||||
result = await self._send_to_forum(channel, content)
|
||||
self._record_discord_response(
|
||||
reply_to=reply_to,
|
||||
result=result,
|
||||
content=content,
|
||||
)
|
||||
return result
|
||||
|
||||
# Format and split message if needed
|
||||
formatted = self.format_message(content)
|
||||
|
|
|
|||
|
|
@ -177,6 +177,21 @@ async def test_backfills_when_only_down_notice_exists(adapter):
|
|||
assert await adapter._should_backfill_discord_message(message) is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generic_unavailable_response_counts_as_completed(adapter):
|
||||
bot_reply = SimpleNamespace(
|
||||
id=2,
|
||||
content="That package is unavailable on this platform.",
|
||||
author=SimpleNamespace(id=999, bot=True),
|
||||
reference=SimpleNamespace(message_id=1),
|
||||
created_at=datetime.now(timezone.utc),
|
||||
)
|
||||
channel = FakeChannel(history_messages=[bot_reply])
|
||||
message = make_message(message_id=1, channel=channel)
|
||||
|
||||
assert await adapter._should_backfill_discord_message(message) is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_backfill_dispatches_unaddressed_messages(adapter, monkeypatch):
|
||||
message = make_message(message_id=1)
|
||||
|
|
@ -294,6 +309,14 @@ def test_default_config_exposes_missed_message_backfill_settings():
|
|||
}
|
||||
|
||||
|
||||
def test_default_recovery_scope_includes_allowed_and_free_response_channels(adapter, monkeypatch):
|
||||
monkeypatch.delenv("DISCORD_MISSED_MESSAGE_BACKFILL_CHANNELS", raising=False)
|
||||
monkeypatch.setenv("DISCORD_ALLOWED_CHANNELS", "100,200")
|
||||
monkeypatch.setenv("DISCORD_FREE_RESPONSE_CHANNELS", "200,300")
|
||||
|
||||
assert adapter._missed_message_backfill_channels() == {"100", "200", "300"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_persistent_responded_record_suppresses_backfill(adapter):
|
||||
message = make_message(message_id=77)
|
||||
|
|
@ -362,6 +385,28 @@ def test_empty_successful_turn_is_not_persistently_complete(adapter):
|
|||
assert adapter._discord_message_is_persistently_complete("89") is False
|
||||
|
||||
|
||||
def test_disabled_recovery_does_not_create_hot_path_ledger(adapter, monkeypatch):
|
||||
monkeypatch.setenv("DISCORD_MISSED_MESSAGE_BACKFILL", "false")
|
||||
message = make_message(message_id=90)
|
||||
event = MessageEvent(
|
||||
text=message.content,
|
||||
message_type=MessageType.TEXT,
|
||||
raw_message=message,
|
||||
message_id=str(message.id),
|
||||
)
|
||||
|
||||
adapter._record_discord_processing_start(event, emoji_ack=False)
|
||||
adapter._record_discord_processing_complete(event, ProcessingOutcome.SUCCESS)
|
||||
adapter._record_discord_response(
|
||||
reply_to="90",
|
||||
result=SimpleNamespace(success=True, message_id="9003"),
|
||||
content="Done",
|
||||
)
|
||||
|
||||
db_path = adapter._discord_recovery_db_path()
|
||||
assert not db_path.exists()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_iter_candidates_includes_active_and_archived_threads(adapter):
|
||||
active_msg = make_message(message_id=201, channel=FakeChannel(channel_id=2010))
|
||||
|
|
@ -405,3 +450,35 @@ async def test_iter_candidates_applies_one_global_scan_limit(adapter, monkeypatc
|
|||
got.append(msg.id)
|
||||
|
||||
assert got == [1, 2, 3]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_iter_candidates_keeps_latest_messages_when_window_exceeds_limit(adapter, monkeypatch):
|
||||
class RealisticChannel(FakeChannel):
|
||||
def history(self, **kwargs):
|
||||
async def _gen():
|
||||
messages = list(self._history_messages)
|
||||
if not kwargs["oldest_first"]:
|
||||
messages.reverse()
|
||||
for message in messages[:kwargs["limit"]]:
|
||||
yield message
|
||||
|
||||
return _gen()
|
||||
|
||||
channel = RealisticChannel(
|
||||
channel_id=123,
|
||||
history_messages=[
|
||||
make_message(message_id=1),
|
||||
make_message(message_id=2),
|
||||
make_message(message_id=3),
|
||||
make_message(message_id=4),
|
||||
],
|
||||
)
|
||||
adapter._client.get_channel = lambda _channel_id: channel
|
||||
monkeypatch.setattr(adapter, "_missed_message_backfill_limit", lambda: 3)
|
||||
|
||||
got = []
|
||||
async for msg in adapter._iter_missed_message_backfill_candidates({"123"}):
|
||||
got.append(msg.id)
|
||||
|
||||
assert got == [2, 3, 4]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue