fix(repeater): 分片下主动发言与维护任务归属 worker

主动发言仅选本 worker 已连接牛；speak_up 未连接则跳过；sync/context 清理仅在 shard-0 执行。

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: TogetsuDo

src/plugins/repeater/__init__.py

@@ -390,18 +390,24 @@ async def speak_up():
 
     bot_id, group_id, messages, target_id = ret
 
+    try:
+        bot = get_bot(str(bot_id))
+    except (KeyError, ValueError):
+        logger.debug("speak_up skip bot [{}] not connected on this worker", bot_id)
+        return
+
     for msg in messages:
         logger.info(f"bot [{bot_id}] ready to speak [{msg}] to group [{group_id}]")
         try:
-            await get_bot(str(bot_id)).call_api(
+            await bot.call_api(
                 "send_group_msg",
                 **{
                     "message": msg,
                     "group_id": group_id,
                 },
             )
             if target_id:
-                await get_bot(str(bot_id)).call_api(
+                await bot.call_api(
                     "group_poke",
                     **{
                         "user_id": target_id,
@@ -421,5 +427,9 @@ async def speak_up():
 
 @scheduler.scheduled_job("cron", hour=4)
 async def update_data():
+    from .shard_opt import repeater_maintenance_runs_on_worker
+
+    if not repeater_maintenance_runs_on_worker():
+        return
     await Chat.sync()
     await Chat.clearup_context()

src/plugins/repeater/shard_opt.py

@@ -12,6 +12,15 @@ def repeater_worker_handles_message(bot_id: int) -> bool:
     return True
 
 
+def local_connected_bot_ids() -> frozenset[int]:
+    """本 worker 当前已连接的牛牛 QQ 集合。"""
+    try:
+        from nonebot import get_bots
+    except Exception:
+        return frozenset()
+    return frozenset(int(key) for key in get_bots() if str(key).isdigit())
+
+
 def repeater_scheduler_runs_on_worker() -> bool:
     """主动发言定时任务：分片时仅代表牛所在 worker 执行，减少重复扫描。"""
     from src.platform.shard.registry.config import is_sharding_active
@@ -21,3 +30,12 @@ def repeater_scheduler_runs_on_worker() -> bool:
     from src.platform.shard.local_representative import local_worker_representative_bot_id
 
     return local_worker_representative_bot_id() is not None
+
+
+def repeater_maintenance_runs_on_worker() -> bool:
+    """跨 worker 全局维护（sync / context 清理）：分片时仅 shard 0 执行。"""
+    from src.platform.shard.registry.config import get_shard_registry_settings, is_sharding_active
+
+    if not is_sharding_active():
+        return True
+    return int(get_shard_registry_settings().shard_id) == 0

src/plugins/repeater/speaker.py

@@ -99,7 +99,17 @@ def cmp(a: int | float, b: int | float) -> int:
                     "reply_keywords": Speaker.SPEAK_FLAG,
                 })
 
-            bot_id = random.choice([bid for bid in group_replies.keys() if bid])
+            from src.platform.shard.registry.config import is_sharding_active
+
+            from .shard_opt import local_connected_bot_ids
+
+            bot_ids = [bid for bid in group_replies.keys() if bid]
+            if is_sharding_active():
+                local_bots = local_connected_bot_ids()
+                bot_ids = [bid for bid in bot_ids if bid in local_bots]
+            if not bot_ids:
+                continue
+            bot_id = random.choice(bot_ids)
 
             ban_keywords = await BanManager.find_ban_keywords(context=None, group_id=group_id)
 

tests/plugins/repeater/test_speaker.py

@@ -97,6 +97,55 @@ async def test_speak_filters_banned_keywords(beanie_fixture):
         reply_dict.clear()
 
 
+@pytest.mark.asyncio
+async def test_speak_skips_remote_bot_when_sharded(beanie_fixture):
+    from src.plugins.repeater.message_store import MessageStore
+    from src.plugins.repeater.speaker import Speaker
+
+    MessageStore._message_dict = defaultdict(list)
+    Speaker._recent_speak = defaultdict(lambda: deque(maxlen=Speaker.DUPLICATE_REPLY))
+
+    reply_dict = defaultdict(lambda: defaultdict(list))
+    reply_lock = asyncio.Lock()
+    recent_topics = defaultdict(lambda: deque(maxlen=16))
+    topics_lock = asyncio.Lock()
+
+    group_id = 30004
+    local_bot_id = 10001
+    remote_bot_id = 10002
+    msg_list = [_build_message(group_id, 20001 + i, f"warmup-{i}", f"warmup-{i}", i + 1) for i in range(10)]
+    MessageStore._message_dict[group_id] = msg_list
+    reply_dict[group_id][remote_bot_id] = [{"time": 1, "reply": "x", "reply_keywords": "x"}]
+    reply_dict[group_id][local_bot_id] = [{"time": 1, "reply": "y", "reply_keywords": "y"}]
+
+    chosen_msg = msg_list[-1]
+    try:
+        with (
+            patch("src.platform.shard.registry.config.is_sharding_active", return_value=True),
+            patch("src.plugins.repeater.shard_opt.local_connected_bot_ids", return_value=frozenset({local_bot_id})),
+            patch("src.plugins.repeater.speaker.time.time", return_value=10000),
+            patch(
+                "src.plugins.repeater.speaker.random.choice",
+                side_effect=[local_bot_id, [chosen_msg], chosen_msg],
+            ),
+            patch("src.plugins.repeater.speaker.random.random", return_value=1.0),
+            patch(
+                "src.plugins.repeater.speaker.BanManager.find_ban_keywords",
+                new_callable=AsyncMock,
+                return_value=set(),
+            ),
+            patch("src.plugins.repeater.speaker.BotConfig.taken_name", new_callable=AsyncMock, return_value=-1),
+            patch("src.plugins.repeater.reply_record_sync.publish_reply_record"),
+        ):
+            result = await Speaker.speak(reply_dict, reply_lock, recent_topics, topics_lock)
+            assert result is not None
+            assert result[0] == local_bot_id
+    finally:
+        MessageStore._message_dict.clear()
+        Speaker._recent_speak.clear()
+        reply_dict.clear()
+
+
 @pytest.mark.asyncio
 async def test_speak_recent_dedup_avoids_same_message_twice(beanie_fixture):
     from src.plugins.repeater.message_store import MessageStore

tests/plugins/test_repeater_shard_opt.py

@@ -3,6 +3,8 @@
 import pytest
 
 from src.plugins.repeater.shard_opt import (
+    local_connected_bot_ids,
+    repeater_maintenance_runs_on_worker,
     repeater_scheduler_runs_on_worker,
     repeater_worker_handles_message,
 )
@@ -84,3 +86,37 @@ def test_scheduler_skips_worker_without_rep(monkeypatch):
         lambda: None,
     )
     assert repeater_scheduler_runs_on_worker() is False
+
+
+def test_maintenance_runs_only_on_shard_zero_when_sharded(monkeypatch):
+    monkeypatch.setattr(
+        "src.platform.shard.registry.config.is_sharding_active",
+        lambda: True,
+    )
+    monkeypatch.setattr(
+        "src.platform.shard.registry.config.get_shard_registry_settings",
+        lambda: type("S", (), {"shard_id": 0})(),
+    )
+    assert repeater_maintenance_runs_on_worker() is True
+
+    monkeypatch.setattr(
+        "src.platform.shard.registry.config.get_shard_registry_settings",
+        lambda: type("S", (), {"shard_id": 3})(),
+    )
+    assert repeater_maintenance_runs_on_worker() is False
+
+
+def test_maintenance_runs_when_not_sharded(monkeypatch):
+    monkeypatch.setattr(
+        "src.platform.shard.registry.config.is_sharding_active",
+        lambda: False,
+    )
+    assert repeater_maintenance_runs_on_worker() is True
+
+
+def test_local_connected_bot_ids_reads_nonebot(monkeypatch):
+    monkeypatch.setattr(
+        "nonebot.get_bots",
+        lambda: {"1001": object(), "2002": object(), "bad": object()},
+    )
+    assert local_connected_bot_ids() == frozenset({1001, 2002})