fix(restart): deliver completion after channel reconnects (#4931)

This commit is contained in:
chengyongru
2026-07-15 01:08:39 +08:00
committed by GitHub
parent 37165b0db0
commit 88c38e9b38
19 changed files with 284 additions and 55 deletions
+13 -1
View File
@@ -197,7 +197,19 @@ The agent receives the message and processes it. Replies arrive in your `send()`
|--------|-------------|
| `async start()` | **Must block forever.** Connect to platform, listen for messages, call `_handle_message()` on each. If this returns, the channel is dead. |
| `async stop()` | Set `self._running = False` and clean up. Called when gateway shuts down. |
| `async send(msg: OutboundMessage)` | Deliver an outbound message to the platform. |
| `async send(msg: OutboundMessage)` | Deliver an outbound message to the platform. Raise when the transport does not accept it. |
#### Outbound delivery contract
A normal return from `send()` means either the visible payload was accepted by the
platform transport/API, or the channel deliberately had nothing to deliver (for example,
an empty progress event). Do not log and return when the client is disconnected, still
starting, or the platform rejects the request. Raise an exception so `ChannelManager` can
apply the shared retry policy.
`send()` may run as soon as `is_running` becomes true. If a channel sets `_running` before
its transport is ready, it must keep raising until delivery can be attempted safely. Small
platform-specific retries are fine, but the final failure must still reach the manager.
### Interactive Login
+6 -4
View File
@@ -710,10 +710,11 @@ class DingTalkChannel(BaseChannel):
"""Send a message through DingTalk."""
token = await self._get_access_token()
if not token:
return
raise RuntimeError("DingTalk access token unavailable")
if msg.content and msg.content.strip():
await self._send_markdown_text(token, msg.chat_id, msg.content.strip())
if not await self._send_markdown_text(token, msg.chat_id, msg.content.strip()):
raise RuntimeError("DingTalk text message was not delivered")
for media_ref in msg.media or []:
ok = await self._send_media_ref(token, msg.chat_id, media_ref)
@@ -722,11 +723,12 @@ class DingTalkChannel(BaseChannel):
self.logger.error("media send failed for {}", media_ref)
# Send visible fallback so failures are observable by the user.
filename = self._guess_filename(media_ref, self._guess_upload_type(media_ref))
await self._send_markdown_text(
if not await self._send_markdown_text(
token,
msg.chat_id,
f"[Attachment send failed: {filename}]",
)
):
raise RuntimeError("DingTalk attachment fallback was not delivered")
async def _on_message(
self,
+2 -3
View File
@@ -264,7 +264,7 @@ if DISCORD_AVAILABLE:
channel = await self.fetch_channel(channel_id)
except Exception as e:
self._channel.logger.warning("channel {} unavailable: {}", msg.chat_id, e)
return
raise
reference, mention_settings = self._build_reply_context(channel, msg.reply_to)
sent_media = False
@@ -466,8 +466,7 @@ class DiscordChannel(BaseChannel):
"""Send a message through Discord using discord.py."""
client = self._client
if client is None or not client.is_ready():
self.logger.warning("client not ready; dropping outbound message")
return
raise RuntimeError("Discord client is not ready")
is_progress = isinstance(msg.event, ProgressEvent)
+8 -1
View File
@@ -2407,7 +2407,14 @@ class FeishuChannel(BaseChannel):
if ok:
return
# Fall back to regular send if reply fails
self._send_message_sync(receive_id_type, msg.chat_id, m_type, content)
message_id = self._send_message_sync(
receive_id_type,
msg.chat_id,
m_type,
content,
)
if not message_id:
raise RuntimeError(f"Feishu {m_type} message was not delivered")
for file_path in msg.media:
if not os.path.isfile(file_path):
+68 -14
View File
@@ -28,7 +28,11 @@ from nanobot.channels._feishu_instances import ChannelInstanceSpec, feishu_insta
from nanobot.channels.base import BaseChannel
from nanobot.channels.registry import DEFAULT_ENABLED_CHANNELS
from nanobot.config.schema import Config
from nanobot.utils.restart import consume_restart_notice_from_env, format_restart_completed_message
from nanobot.utils.restart import (
RestartNotice,
consume_restart_notice_from_env,
format_restart_completed_message,
)
if TYPE_CHECKING:
from nanobot.session.manager import SessionManager
@@ -46,6 +50,8 @@ def _default_webui_dist() -> Path | None:
# Retry delays for message sending (exponential backoff: 1s, 2s, 4s)
_SEND_RETRY_DELAYS = (1, 2, 4)
_RESTART_NOTICE_START_TIMEOUT_S = 30.0
_RESTART_NOTICE_START_POLL_S = 0.25
_BOOL_CAMEL_ALIASES: dict[str, str] = {
"send_progress": "sendProgress",
@@ -476,15 +482,40 @@ class ChannelManager:
# Wait for all to complete (they should run forever)
await asyncio.gather(*tasks, return_exceptions=True)
def _notify_restart_done_if_needed(self) -> None:
"""Send restart completion message when runtime env markers are present."""
def _notify_restart_done_if_needed(self) -> asyncio.Task[None] | None:
"""Schedule restart completion after the target channel starts."""
notice = consume_restart_notice_from_env()
if not notice:
return
return None
return asyncio.create_task(self._send_restart_notice_when_started(notice))
async def _send_restart_notice_when_started(
self,
notice: RestartNotice,
*,
timeout_s: float = _RESTART_NOTICE_START_TIMEOUT_S,
poll_s: float = _RESTART_NOTICE_START_POLL_S,
) -> None:
"""Deliver a restart notice after the target channel starts."""
loop = asyncio.get_running_loop()
deadline = loop.time() + timeout_s
target = self.channels.get(notice.channel)
if not target:
if target is None:
logger.warning("Restart notice target channel is not enabled: {}", notice.channel)
return
asyncio.create_task(self._send_with_retry(
while not target.is_running:
remaining = deadline - loop.time()
if remaining <= 0:
logger.warning(
"Restart notice target did not start: {}:{}",
notice.channel,
notice.chat_id,
)
return
await asyncio.sleep(min(poll_s, remaining))
await self._send_with_retry(
target,
OutboundMessage(
channel=notice.channel,
@@ -492,7 +523,8 @@ class ChannelManager:
content=format_restart_completed_message(notice.started_at_raw),
metadata=dict(notice.metadata or {}),
),
))
deadline=deadline,
)
async def stop_all(self) -> None:
"""Stop all channels and the dispatcher."""
@@ -791,30 +823,52 @@ class ChannelManager:
merged = replace_outbound_event(first_msg, final_event, content=combined_content)
return merged, non_matching
async def _send_with_retry(self, channel: BaseChannel, msg: OutboundMessage) -> None:
async def _send_with_retry(
self,
channel: BaseChannel,
msg: OutboundMessage,
*,
deadline: float | None = None,
) -> None:
"""Send a message with retry on failure using exponential backoff.
When deadline is provided, retry until that monotonic time instead of
stopping at the configured attempt limit.
Note: CancelledError is re-raised to allow graceful shutdown.
"""
max_attempts = max(self.config.channels.send_max_retries, 1)
attempt = 0
for attempt in range(max_attempts):
while True:
attempt += 1
try:
await self._send_once(channel, msg)
return # Send succeeded
except asyncio.CancelledError:
raise # Propagate cancellation for graceful shutdown
except Exception as e:
if attempt == max_attempts - 1:
loop = asyncio.get_running_loop()
exhausted = (
attempt >= max_attempts
if deadline is None
else loop.time() >= deadline
)
if exhausted:
logger.exception(
"Failed to send to {} after {} attempts",
msg.channel, max_attempts
msg.channel, attempt,
)
return
delay = _SEND_RETRY_DELAYS[min(attempt, len(_SEND_RETRY_DELAYS) - 1)]
delay = _SEND_RETRY_DELAYS[min(attempt - 1, len(_SEND_RETRY_DELAYS) - 1)]
if deadline is not None:
delay = min(delay, max(0.0, deadline - loop.time()))
attempt_label = str(attempt)
if deadline is None:
attempt_label = f"{attempt}/{max_attempts}"
logger.warning(
"Send to {} failed (attempt {}/{}): {}, retrying in {}s",
msg.channel, attempt + 1, max_attempts, type(e).__name__, delay
"Send to {} failed (attempt {}): {}, retrying in {}s",
msg.channel, attempt_label, type(e).__name__, delay,
)
try:
await asyncio.sleep(delay)
+4 -2
View File
@@ -559,7 +559,7 @@ class MatrixChannel(BaseChannel):
async def send(self, msg: OutboundMessage) -> None:
"""Send outbound content; clear typing for non-progress messages."""
if not self.client:
return
raise RuntimeError("Matrix client not initialized")
text = msg.content or ""
candidates = self._collect_outbound_media_candidates(msg.media)
relates_to = self._build_thread_relates_to(msg.metadata)
@@ -582,7 +582,9 @@ class MatrixChannel(BaseChannel):
content = _build_matrix_text_content(text)
if relates_to:
content["m.relates_to"] = relates_to
await self._send_room_content(msg.chat_id, content)
response = await self._send_room_content(msg.chat_id, content)
if isinstance(response, RoomSendError):
raise RuntimeError(f"Matrix message was not delivered: {response}")
finally:
if not is_progress:
await self._stop_typing_keepalive(msg.chat_id, clear_typing=True)
+1 -2
View File
@@ -430,8 +430,7 @@ class NapcatChannel(BaseChannel):
async def send(self, msg: OutboundMessage) -> None:
if self._ws is None:
logger.warning("napcat: not connected, dropping outbound message")
return
raise RuntimeError("napcat: not connected")
kind, _, target = msg.chat_id.partition(":")
if kind not in ("private", "group") or not target:
+2 -2
View File
@@ -243,8 +243,7 @@ class QQChannel(BaseChannel):
"""Send attachments first, then text."""
try:
if not self._client:
self.logger.warning("client not initialized")
return
raise RuntimeError("QQ client not initialized")
msg_id = msg.metadata.get("message_id")
chat_type = self._chat_type_cache.get(msg.chat_id, "c2c")
@@ -284,6 +283,7 @@ class QQChannel(BaseChannel):
raise
except Exception:
self.logger.exception("Error sending message to chat_id={}", msg.chat_id)
raise
async def _send_text_only(
self,
+2 -2
View File
@@ -493,8 +493,7 @@ class WecomChannel(BaseChannel):
async def send(self, msg: OutboundMessage) -> None:
"""Send a message through WeCom."""
if not self._client:
self.logger.warning("client not initialized")
return
raise RuntimeError("WeCom client not initialized")
try:
content = (msg.content or "").strip()
@@ -553,3 +552,4 @@ class WecomChannel(BaseChannel):
except Exception:
self.logger.exception("Error sending message to chat_id={}", msg.chat_id)
raise
+9 -2
View File
@@ -9,6 +9,8 @@ from contextlib import suppress
from dataclasses import dataclass, field
from typing import Any
from nanobot.webui.metadata import WEBUI_TURN_METADATA_KEY
RESTART_NOTIFY_CHANNEL_ENV = "NANOBOT_RESTART_NOTIFY_CHANNEL"
RESTART_NOTIFY_CHAT_ID_ENV = "NANOBOT_RESTART_NOTIFY_CHAT_ID"
RESTART_NOTIFY_METADATA_ENV = "NANOBOT_RESTART_NOTIFY_METADATA"
@@ -40,9 +42,14 @@ def set_restart_notice_to_env(
os.environ[RESTART_NOTIFY_CHANNEL_ENV] = channel
os.environ[RESTART_NOTIFY_CHAT_ID_ENV] = chat_id
os.environ[RESTART_STARTED_AT_ENV] = str(time.time())
if metadata:
persisted_metadata = dict(metadata or {})
persisted_metadata.pop(WEBUI_TURN_METADATA_KEY, None)
if persisted_metadata:
try:
os.environ[RESTART_NOTIFY_METADATA_ENV] = json.dumps(metadata, default=str)
os.environ[RESTART_NOTIFY_METADATA_ENV] = json.dumps(
persisted_metadata,
default=str,
)
except (TypeError, ValueError):
os.environ.pop(RESTART_NOTIFY_METADATA_ENV, None)
else:
+48 -5
View File
@@ -2651,8 +2651,8 @@ async def test_start_all_creates_dispatch_task():
@pytest.mark.asyncio
async def test_notify_restart_done_enqueues_outbound_message():
"""Restart notice should schedule send_with_retry for target channel."""
async def test_notify_restart_done_waits_until_channel_starts():
"""Restart notice should not be sent before the target channel starts."""
fake_config = SimpleNamespace(
channels=ChannelsConfig(),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
@@ -2661,18 +2661,61 @@ async def test_notify_restart_done_enqueues_outbound_message():
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
mgr.channels = {"feishu": _StartableChannel(fake_config, mgr.bus)}
channel = _StartableChannel(fake_config, mgr.bus)
mgr.channels = {"feishu": channel}
mgr._dispatch_task = None
mgr._send_with_retry = AsyncMock()
notice = RestartNotice(channel="feishu", chat_id="oc_123", started_at_raw="100.0")
with patch("nanobot.channels.manager.consume_restart_notice_from_env", return_value=notice):
mgr._notify_restart_done_if_needed()
task = mgr._notify_restart_done_if_needed()
await asyncio.sleep(0)
mgr._send_with_retry.assert_not_awaited()
channel._running = True
assert task is not None
await asyncio.wait_for(task, timeout=1.0)
mgr._send_with_retry.assert_awaited_once()
sent_channel, sent_msg = mgr._send_with_retry.await_args.args
assert sent_channel is mgr.channels["feishu"]
assert sent_channel is channel
assert sent_msg.channel == "feishu"
assert sent_msg.chat_id == "oc_123"
assert sent_msg.content.startswith("Restart completed")
@pytest.mark.asyncio
async def test_restart_notice_retries_until_running_channel_accepts_delivery():
"""A running flag must not make an early transport failure final."""
class _EventuallyDeliverableChannel(_StartableChannel):
def __init__(self, config, bus):
super().__init__(config, bus)
self.attempts = 0
self.sent: OutboundMessage | None = None
async def send(self, msg: OutboundMessage) -> None:
self.attempts += 1
if self.attempts == 1:
raise RuntimeError("transport not ready")
self.sent = msg
fake_config = SimpleNamespace(
channels=ChannelsConfig(send_max_retries=1),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
channel = _EventuallyDeliverableChannel(fake_config, mgr.bus)
channel._running = True
mgr.channels = {"discord": channel}
notice = RestartNotice(channel="discord", chat_id="123", started_at_raw="")
with patch("nanobot.channels.manager._SEND_RETRY_DELAYS", (0,)):
await mgr._send_restart_notice_when_started(notice, timeout_s=0.1, poll_s=0.01)
assert channel.attempts == 2
assert channel.sent is not None
assert channel.sent.content == "Restart completed."
+31
View File
@@ -2,6 +2,7 @@ import asyncio
import zipfile
from io import BytesIO
from types import SimpleNamespace
from unittest.mock import AsyncMock
import httpx
import pytest
@@ -17,6 +18,7 @@ if not DINGTALK_AVAILABLE:
pytest.skip("DingTalk dependencies not installed (dingtalk-stream)", allow_module_level=True)
import nanobot.channels.dingtalk as dingtalk_module
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.dingtalk import DingTalkChannel, DingTalkConfig, NanobotDingTalkHandler
@@ -864,6 +866,35 @@ async def test_send_batch_message_returns_false_on_api_error() -> None:
assert result is True
@pytest.mark.asyncio
async def test_send_raises_when_access_token_is_unavailable(monkeypatch) -> None:
channel = DingTalkChannel(
DingTalkConfig(client_id="app", client_secret="secret", allow_from=["*"]),
MessageBus(),
)
monkeypatch.setattr(channel, "_get_access_token", AsyncMock(return_value=None))
with pytest.raises(RuntimeError, match="access token unavailable"):
await channel.send(
OutboundMessage(channel="dingtalk", chat_id="user123", content="hello")
)
@pytest.mark.asyncio
async def test_send_raises_when_text_is_not_delivered(monkeypatch) -> None:
channel = DingTalkChannel(
DingTalkConfig(client_id="app", client_secret="secret", allow_from=["*"]),
MessageBus(),
)
monkeypatch.setattr(channel, "_get_access_token", AsyncMock(return_value="token"))
monkeypatch.setattr(channel, "_send_markdown_text", AsyncMock(return_value=False))
with pytest.raises(RuntimeError, match="text message was not delivered"):
await channel.send(
OutboundMessage(channel="dingtalk", chat_id="user123", content="hello")
)
@pytest.mark.asyncio
async def test_send_media_ref_short_circuits_on_transport_error() -> None:
"""When the first send fails with a transport error, _send_media_ref must
+10 -6
View File
@@ -659,18 +659,19 @@ async def test_on_message_marks_failed_attachment_download(tmp_path, monkeypatch
@pytest.mark.asyncio
async def test_send_warns_when_client_not_ready() -> None:
# Sending without a running/ready client should be a safe no-op.
async def test_send_raises_when_client_not_ready() -> None:
# The manager must be able to retry while Discord is still connecting.
channel = DiscordChannel(DiscordConfig(enabled=True, allow_from=["*"]), MessageBus())
await channel.send(OutboundMessage(channel="discord", chat_id="123", content="hello"))
with pytest.raises(RuntimeError, match="client is not ready"):
await channel.send(OutboundMessage(channel="discord", chat_id="123", content="hello"))
assert channel._typing_tasks == {}
@pytest.mark.asyncio
async def test_send_skips_when_channel_not_cached() -> None:
# Outbound sends should be skipped when the destination channel is not resolvable.
async def test_send_raises_when_channel_cannot_be_resolved() -> None:
# The manager must be able to retry transient channel-resolution failures.
owner = DiscordChannel(DiscordConfig(enabled=True, allow_from=["*"]), MessageBus())
client = DiscordBotClient(owner, intents=discord.Intents.none())
fetch_calls: list[int] = []
@@ -681,7 +682,10 @@ async def test_send_skips_when_channel_not_cached() -> None:
client.fetch_channel = fetch_channel # type: ignore[method-assign]
await client.send_outbound(OutboundMessage(channel="discord", chat_id="123", content="hello"))
with pytest.raises(RuntimeError, match="not found"):
await client.send_outbound(
OutboundMessage(channel="discord", chat_id="123", content="hello")
)
assert client.get_channel(123) is None
assert fetch_calls == [123]
+18 -1
View File
@@ -266,8 +266,9 @@ async def test_send_uses_expected_feishu_msg_type_for_uploaded_files(
send_calls: list[tuple[str, str, str, str]] = []
def _record_send(receive_id_type: str, receive_id: str, msg_type: str, content: str) -> None:
def _record_send(receive_id_type: str, receive_id: str, msg_type: str, content: str) -> str:
send_calls.append((receive_id_type, receive_id, msg_type, content))
return "om_test"
with patch.object(channel, "_upload_file_sync", return_value="file-key"), patch.object(
channel, "_send_message_sync", side_effect=_record_send
@@ -398,6 +399,22 @@ async def test_send_fallback_to_create_when_reply_fails() -> None:
channel._client.im.v1.message.create.assert_called_once()
@pytest.mark.asyncio
async def test_send_raises_when_create_api_does_not_deliver() -> None:
channel = _make_feishu_channel()
with patch.object(channel, "_send_message_sync", return_value=None):
with pytest.raises(RuntimeError, match="message was not delivered"):
await channel.send(
OutboundMessage(
channel="feishu",
chat_id="oc_abc",
content="hello",
metadata={},
)
)
def test_send_message_sync_falls_back_to_text_for_interactive_error() -> None:
channel = _make_feishu_channel()
+22
View File
@@ -1820,6 +1820,28 @@ async def test_send_room_content_returns_room_send_response():
assert result is client.room_send_response
@pytest.mark.asyncio
async def test_send_raises_when_room_send_returns_error(monkeypatch) -> None:
class _FakeRoomSendError:
def __str__(self) -> str:
return "temporary homeserver failure"
client = _FakeAsyncClient("", "", "", None)
client.room_send_response = _FakeRoomSendError()
channel = MatrixChannel(_make_config(), MessageBus())
channel.client = client
monkeypatch.setattr(matrix_module, "RoomSendError", _FakeRoomSendError)
with pytest.raises(RuntimeError, match="temporary homeserver failure"):
await channel.send(
OutboundMessage(
channel="matrix",
chat_id="!room:matrix.org",
content="hello",
)
)
@pytest.mark.asyncio
async def test_send_delta_creates_stream_buffer_and_sends_initial_message() -> None:
channel = MatrixChannel(_make_config(), MessageBus())
+11
View File
@@ -2,6 +2,7 @@ import asyncio
import pytest
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.napcat import NapcatChannel, NapcatConfig
@@ -52,6 +53,16 @@ def _channel(config: NapcatConfig | None = None) -> NapcatChannel:
return NapcatChannel(config or NapcatConfig(allow_from=["*"]), MessageBus())
@pytest.mark.asyncio
async def test_send_raises_while_websocket_is_not_connected() -> None:
channel = _channel()
with pytest.raises(RuntimeError, match="not connected"):
await channel.send(
OutboundMessage(channel="napcat", chat_id="private:123", content="hello")
)
@pytest.mark.asyncio
async def test_group_message_requires_mention_by_default() -> None:
channel = _channel(NapcatConfig(allow_from=["user1"], group_policy="mention"))
+6 -5
View File
@@ -113,17 +113,18 @@ def test_guess_send_file_type_by_mime() -> None:
@pytest.mark.asyncio
async def test_send_exception_caught_not_raised() -> None:
"""Exceptions inside send() must not propagate."""
async def test_send_exception_propagates_for_manager_retry() -> None:
"""Delivery failures must propagate to the channel manager."""
channel = QQChannel(QQConfig(app_id="app", secret="secret", allow_from=["*"]), MessageBus())
channel._client = _FakeClient()
with patch.object(
channel, "_send_text_only", new_callable=AsyncMock, side_effect=RuntimeError("boom")
) as send_text:
await channel.send(
OutboundMessage(channel="qq", chat_id="user1", content="hello")
)
with pytest.raises(RuntimeError, match="boom"):
await channel.send(
OutboundMessage(channel="qq", chat_id="user1", content="hello")
)
send_text.assert_awaited_once()
+6 -5
View File
@@ -412,8 +412,8 @@ async def test_send_media_file_not_found() -> None:
@pytest.mark.asyncio
async def test_send_exception_caught_not_raised() -> None:
"""Exceptions inside send() must not propagate."""
async def test_send_exception_propagates_for_manager_retry() -> None:
"""Delivery failures must propagate to the channel manager."""
channel = WecomChannel(WecomConfig(bot_id="b", secret="s", allow_from=["*"]), MessageBus())
client = _FakeWeComClient()
channel._client = client
@@ -423,9 +423,10 @@ async def test_send_exception_caught_not_raised() -> None:
# Make reply_stream raise
client.reply_stream.side_effect = RuntimeError("boom")
await channel.send(
OutboundMessage(channel="wecom", chat_id="chat1", content="fail test")
)
with pytest.raises(RuntimeError, match="boom"):
await channel.send(
OutboundMessage(channel="wecom", chat_id="chat1", content="fail test")
)
client.reply_stream.assert_called_once()
+17
View File
@@ -56,6 +56,23 @@ def test_restart_notice_preserves_metadata_across_env(monkeypatch):
assert "NANOBOT_RESTART_NOTIFY_METADATA" not in os.environ
def test_restart_notice_drops_process_local_webui_turn_metadata(monkeypatch):
monkeypatch.delenv("NANOBOT_RESTART_NOTIFY_METADATA", raising=False)
set_restart_notice_to_env(
channel="websocket",
chat_id="chat-1",
metadata={
"webui_turn_id": "turn-from-old-process",
"slack": {"thread_ts": "1700.42"},
},
)
notice = consume_restart_notice_from_env()
assert notice is not None
assert notice.metadata == {"slack": {"thread_ts": "1700.42"}}
def test_restart_notice_clears_stale_metadata(monkeypatch):
monkeypatch.setenv("NANOBOT_RESTART_NOTIFY_METADATA", '{"stale": true}')
set_restart_notice_to_env(channel="cli", chat_id="direct")