Please address the comments from this code review:

## Overall Comments
- In `_get_sender_id` the priority is `id` → `user_openid` → `member_openid`, which contradicts the description that `user_openid` should be preferred and may lead to inconsistent sender IDs across callbacks; consider reordering the checks to `user_openid` first, then `member_openid`, then `id`.
- The short content key for deduplication is currently `f"{sender_id}:{content[:50]}"`; if different messages share the same 50‑char prefix this may incorrectly treat them as duplicates, so consider either using the full content or a hash of the full content instead of truncation.

## Individual Comments

### Comment 1
<location path="astrbot/core/platform/sources/qqofficial/qqofficial_platform_adapter.py" line_range="39-47" />
<code_context>
     def set_platform(self, platform: QQOfficialPlatformAdapter) -> None:
         self.platform = platform

+    def _get_sender_id(self, message) -> str:
+        """Extract sender ID from different message types."""
+        if hasattr(message, 'author') and hasattr(message.author, 'id'):
+            return str(message.author.id)
+        if hasattr(message, 'author') and hasattr(message.author, 'user_openid'):
+            return str(message.author.user_openid)
+        if hasattr(message, 'author') and hasattr(message.author, 'member_openid'):
+            return str(message.author.member_openid)
+        return ""
+
     # 收到群消息
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Align sender ID selection strategy with `_parse_from_qqofficial` to avoid inconsistent dedup keys.

Here `_get_sender_id` prefers `author.id` over `user_openid`, but `_parse_from_qqofficial` prefers `user_openid` and only falls back to `id`. This can cause the same user to appear under different IDs, weakening deduplication (`_is_duplicate_message`) and event_bus fingerprinting. Please align the precedence order with `_parse_from_qqofficial` or extract a shared helper so both paths use the same logic.

```suggestion
    def _get_sender_id(self, message) -> str:
        """Extract sender ID from different message types.

        The precedence order is aligned with `_parse_from_qqofficial` to ensure
        consistent deduplication and event fingerprinting:
        1. author.user_openid
        2. author.member_openid
        3. author.id
        """
        if hasattr(message, "author") and hasattr(message.author, "user_openid"):
            return str(message.author.user_openid)
        if hasattr(message, "author") and hasattr(message.author, "member_openid"):
            return str(message.author.member_openid)
        if hasattr(message, "author") and hasattr(message.author, "id"):
            return str(message.author.id)
        return ""
```
</issue_to_address>

### Comment 2
<location path="astrbot/core/event_bus.py" line_range="39" />
<code_context>
         self.pipeline_scheduler_mapping = pipeline_scheduler_mapping
         self.astrbot_config_mgr = astrbot_config_mgr
+        # 跨平台实例短窗去重（兜底）：处理同一消息在极短时间内重复入队。
+        self._recent_event_fingerprints: dict[str, float] = {}
+        self._dedup_ttl_seconds = 0.5
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the new deduplication logic by replacing the hash-based dict approach with a tuple-key + deque/set TTL structure and documenting the single-consumer assumption for clarity and safety.

You can simplify the dedup logic without changing behavior by:

1. Using a tuple as the fingerprint key (no hashing / string joins).
2. Using a deque + set for TTL-based dedup, with straightforward eviction.
3. Explicitly documenting the single-consumer assumption.

### 1. Replace SHA-1 fingerprint with a tuple

```python
from collections import deque
import time

class EventBus:
    def __init__(...):
        ...
        # 最近事件指纹，短窗去重（0.5s），单消费者线程内使用
        self._dedup_ttl_seconds = 0.5
        self._dedup_seen = set()              # Set[Fingerprint]
        self._dedup_queue = deque()           # deque[(timestamp, Fingerprint)]
```

```python
    def _build_event_fingerprint(self, event: AstrMessageEvent) -> tuple:
        # 简单元组键即可，避免拼接和哈希
        return (
            event.get_platform_id() or "",
            event.unified_msg_origin or "",
            event.get_sender_id() or "",
            (event.get_message_str() or "").strip(),
        )
```

### 2. Simplify TTL cleanup with deque + set

```python
    def _clean_expired_event_fingerprints(self) -> None:
        now = time.time()
        expire_before = now - self._dedup_ttl_seconds
        while self._dedup_queue and self._dedup_queue[0][0] < expire_before:
            _, fingerprint = self._dedup_queue.popleft()
            self._dedup_seen.discard(fingerprint)

    def _is_duplicate_event(self, event: AstrMessageEvent) -> bool:
        # dispatch 是单消费者循环，未加锁是有意为之
        self._clean_expired_event_fingerprints()
        fingerprint = self._build_event_fingerprint(event)
        if fingerprint in self._dedup_seen:
            return True
        ts = time.time()
        self._dedup_seen.add(fingerprint)
        self._dedup_queue.append((ts, fingerprint))
        return False
```

This:

- Removes the need to iterate and mutate a dict copy.
- Makes eviction logic linear and ordered (pop from left while expired).
- Removes hashing / encoding complexity, while collision risk remains negligible in a 0.5s window.

### 3. Clarify concurrency assumption

Add a short comment where dedup is used to prevent future accidental multi-consumer usage:

```python
    async def dispatch(self) -> None:
        # event_queue 由单一消费者处理；去重结构不是线程安全的，按设计仅在此循环中使用。
        while True:
            event: AstrMessageEvent = await self.event_queue.get()
            if self._is_duplicate_event(event):
                ...
```
</issue_to_address>

Please address the comments from this code review:

## Overall Comments
- The short-window dedup maps in both `EventBus` and `QQOfficialPlatformAdapter` are cleaned on every message by scanning the entire dict, which may become a hot path under load; consider either bounding the size, cleaning probabilistically, or only cleaning at a coarser interval to reduce per-message overhead.
- The new dedup logic in `QQOfficialPlatformAdapter._is_duplicate_message` logs every newly registered message at `info` level, which could spam logs in busy deployments; consider downgrading to `debug` or logging only on duplicate detection.

## Individual Comments

### Comment 1
<location path="astrbot/core/platform/sources/qqofficial/qqofficial_platform_adapter.py" line_range="39-48" />
<code_context>
     def set_platform(self, platform: QQOfficialPlatformAdapter) -> None:
         self.platform = platform

+    def _get_sender_id(self, message) -> str:
+        """Extract sender ID from different message types.
+
+        The precedence order is aligned with `_parse_from_qqofficial` to ensure
+        consistent deduplication and event fingerprinting:
+        1. author.user_openid
+        2. author.member_openid
+        3. author.id
+        """
+        if hasattr(message, "author") and hasattr(message.author, "user_openid"):
+            return str(message.author.user_openid)
+        if hasattr(message, "author") and hasattr(message.author, "member_openid"):
+            return str(message.author.member_openid)
+        if hasattr(message, "author") and hasattr(message.author, "id"):
+            return str(message.author.id)
+        return ""
</code_context>
<issue_to_address>
**issue (bug_risk):** Sender ID precedence in `_get_sender_id` is inconsistent with `_parse_from_qqofficial`.

The docstring claims `_get_sender_id` matches `_parse_from_qqofficial`, but the precedence differs: `_get_sender_id` uses `user_openid -> member_openid -> id`, while `_parse_from_qqofficial` only falls back to `user_openid or id`. This can cause event-bus dedup keys to use `member_openid` while `abm.sender.id` uses `user_openid/id`, leading to inconsistent fingerprints. Please either add `member_openid` to `_parse_from_qqofficial` (and other `get_sender_id()` callers) or remove it from `_get_sender_id` so the precedence is truly consistent.
</issue_to_address>

### Comment 2
<location path="astrbot/core/platform/sources/qqofficial/qqofficial_platform_adapter.py" line_range="62" />
<code_context>
     ) -> None:
+        sender_id = self._get_sender_id(message)
+        content = getattr(message, 'content', '') or ""
+        if await self.platform._is_duplicate_message(message.id, content, sender_id):
+            return
         abm = QQOfficialPlatformAdapter._parse_from_qqofficial(
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Deduplication assumes `message.id` is a `str`, which may not always be true.

`_is_duplicate_message` expects a `str` but receives `message.id` directly. If the `id` type changes (or differs across message classes), the deduplication keys could become inconsistent. Consider explicitly normalizing with `str(message.id)` either at the call site or within `_is_duplicate_message` to make the behavior robust to type changes.

Suggested implementation:

```python
        sender_id = self._get_sender_id(message)
        content = getattr(message, 'content', '') or ""
        if await self.platform._is_duplicate_message(str(message.id), content, sender_id):
            return
        abm = QQOfficialPlatformAdapter._parse_from_qqofficial(

```

```python
    async def on_at_message_create(self, message: botpy.message.Message) -> None:
        sender_id = self._get_sender_id(message)
        content = getattr(message, 'content', '') or ""
        if await self.platform._is_duplicate_message(str(message.id), content, sender_id):
            return
        abm = QQOfficialPlatformAdapter._parse_from_qqofficial(

```
</issue_to_address>

Please address the comments from this code review:

## Overall Comments
- The QQOfficial deduplication maps (`message_id_timestamps` and `content_key_timestamps`) are unbounded; consider adding a max size or periodic background cleanup to avoid unbounded memory growth under long-running high-traffic workloads.
- Using a single `_dedup_lock` around the entire `_is_duplicate_message` check and registration can become a bottleneck under high concurrency; consider using a more granular structure (e.g., sharded locks or lock-free cleanup with atomic updates) to reduce contention.
- The event bus and QQOfficial dedup logic both log every duplicate/new message at `info` level, which may be noisy in production; consider downgrading to `debug` or adding sampling to keep logs manageable.

## Individual Comments

### Comment 1
<location path="astrbot/core/platform/sources/qqofficial/qqofficial_platform_adapter.py" line_range="175" />
<code_context>
+
+    def _clean_expired_messages(self) -> None:
+        """Clean up expired dedup caches."""
+        current_time = time.time()
+        expired_message_ids = [
+            msg_id
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Consider using a monotonic clock for TTL-based deduplication instead of `time.time()`.

Both this cleanup and the event_bus dedup logic use `time.time()` for TTL expiry. If the system clock jumps (NTP sync, manual change, DST quirks), entries can expire too early or too late. For relative TTLs, `time.monotonic()` is safer because it isn’t affected by wall-clock adjustments and keeps dedup timing consistent.

Suggested implementation:

```python
    def _clean_expired_messages(self) -> None:
        """Clean up expired dedup caches."""
        # Use a monotonic clock for TTL-based expiry to avoid issues with system clock changes.
        current_time = time.monotonic()
        expired_message_ids = [

```

```python
        expired_content_keys = [
            content_key
            for content_key, timestamp in self.content_key_timestamps.items()
            if current_time - timestamp > self._content_key_ttl_seconds

```

To fully implement the switch to monotonic time for TTL-based deduplication, you should also:

1. Change any code that *writes* to `self.message_id_timestamps` and `self.content_key_timestamps` to use `time.monotonic()` instead of `time.time()`. For example:
   - `self.message_id_timestamps[msg_id] = time.monotonic()`
   - `self.content_key_timestamps[content_key] = time.monotonic()`
2. In the "event_bus dedup logic" that currently uses `time.time()` for TTL checks, update both:
   - The timestamp storage to use `time.monotonic()`.
   - The TTL comparison to use `time.monotonic()` as the current time source.
3. Verify that no code mixes wall-clock timestamps (e.g. from `time.time()` or datetimes) with these dedup timestamp values; all TTL comparisons must be monotonic-to-monotonic.

These additional changes keep all dedup-related timing consistent and robust against wall-clock adjustments.
</issue_to_address>

Please address the comments from this code review:

## Overall Comments
- Consider removing `problems.md` from the repo (or moving it to an appropriate developer docs location) since it describes a one-off local environment issue rather than project-level behavior and may quickly become stale.
- The new deduplication logging (e.g., in `_is_duplicate_message` and `_is_duplicate_event`) uses `info` level for every hit/miss; consider downgrading some of these to `debug` to avoid log noise under high message volume.

## Individual Comments

### Comment 1
<location path="astrbot/core/platform/sources/qqofficial/qqofficial_platform_adapter.py" line_range="144" />
<code_context>
         qq_group = platform_config["enable_group_c2c"]
         guild_dm = platform_config["enable_guild_direct_message"]

+        # Lock for thread-safe deduplication
+        self._dedup_lock = asyncio.Lock()
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the deduplication logic and key extraction into reusable helpers to keep the adapter and handlers simpler and less repetitive.

You can keep the dedup behavior but significantly reduce complexity and duplication by:

1. Encapsulating the dedup dictionaries/lock/cleanup into a small helper.
2. Centralizing `(sender_id, content)` extraction so each handler doesn’t repeat it.

### 1. Extract a `MessageDeduplicator` helper

Move the cache management, TTLs, and hashing out of `QQOfficialPlatformAdapter` into a small class:

```python
class MessageDeduplicator:
    def __init__(
        self,
        message_id_ttl_seconds: float = 30 * 60,
        content_key_ttl_seconds: float = 3.0,
    ) -> None:
        self._message_id_timestamps: dict[str, float] = {}
        self._content_key_timestamps: dict[str, float] = {}
        self._message_id_ttl_seconds = message_id_ttl_seconds
        self._content_key_ttl_seconds = content_key_ttl_seconds
        self._lock = asyncio.Lock()

    def _clean_expired(self, now: float) -> None:
        self._message_id_timestamps = {
            k: ts
            for k, ts in self._message_id_timestamps.items()
            if now - ts <= self._message_id_ttl_seconds
        }
        self._content_key_timestamps = {
            k: ts
            for k, ts in self._content_key_timestamps.items()
            if now - ts <= self._content_key_ttl_seconds
        }

    async def is_duplicate(
        self,
        message_id: str,
        content: str = "",
        sender_id: str = "",
    ) -> bool:
        async with self._lock:
            now = time.monotonic()
            self._clean_expired(now)

            if message_id in self._message_id_timestamps:
                logger.info(
                    f"[QQOfficial] Duplicate message detected (by ID): {message_id[:50]}..."
                )
                return True

            content_key = None
            if content and sender_id:
                content_hash = hashlib.sha1(content.encode("utf-8")).hexdigest()[:16]
                content_key = f"{sender_id}:{content_hash}"
                if content_key in self._content_key_timestamps:
                    logger.info(
                        f"[QQOfficial] Duplicate message detected (by content): {content_key}"
                    )
                    return True

            # register
            self._message_id_timestamps[message_id] = now
            if content_key is not None:
                self._content_key_timestamps[content_key] = now

            logger.info(f"[QQOfficial] New message registered: {message_id[:50]}...")
            return False
```

Then in `QQOfficialPlatformAdapter.__init__`:

```python
class QQOfficialPlatformAdapter(PlatformAdapter):
    def __init__(...):
        ...
        self._deduplicator = MessageDeduplicator(
            message_id_ttl_seconds=30 * 60,
            content_key_ttl_seconds=3.0,
        )
```

And replace `_is_duplicate_message` with a thin wrapper (or just call the helper directly):

```python
    async def _is_duplicate_message(
        self, message_id: str, content: str = "", sender_id: str = ""
    ) -> bool:
        return await self._deduplicator.is_duplicate(message_id, content, sender_id)
```

This keeps behavior identical but removes the low-level cache/hashing mechanics from the adapter itself.

### 2. Centralize sender/content extraction for handlers

You already added `_get_sender_id`. You can also centralize the content extraction and reuse in all handlers:

```python
class botClient(Client):
    ...

    def _extract_dedup_key(self, message) -> tuple[str, str]:
        sender_id = self._get_sender_id(message)
        content = getattr(message, "content", "") or ""
        return sender_id, content

    async def _should_drop_message(self, message) -> bool:
        sender_id, content = self._extract_dedup_key(message)
        return await self.platform._is_duplicate_message(message.id, content, sender_id)

    async def on_group_at_message_create(self, message: botpy.message.GroupMessage) -> None:
        if await self._should_drop_message(message):
            return
        abm = QQOfficialPlatformAdapter._parse_from_qqofficial(
            message,
            MessageType.GROUP_MESSAGE,
        )
        ...
```

Then apply the same `_should_drop_message` call to `on_at_message_create`, `on_direct_message_create`, and `on_c2c_message_create` to remove repeated sender/content retrieval and dedup calls while preserving the dedup checks.
</issue_to_address>

### Comment 2
<location path="astrbot/core/event_bus.py" line_range="38" />
<code_context>
         # abconf uuid -> scheduler
         self.pipeline_scheduler_mapping = pipeline_scheduler_mapping
         self.astrbot_config_mgr = astrbot_config_mgr
+        # 跨平台实例短窗去重（兜底）：处理同一消息在极短时间内重复入队。
+        # 最近事件指纹，短窗去重（0.5s），单消费者线程内使用
+        self._dedup_ttl_seconds = 0.5
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the deduplication TTL and fingerprint bookkeeping into a small `EventDeduplicator` helper so `EventBus` focuses only on dispatch intent.

You can keep the new deduplication behavior while reducing the cognitive load on `EventBus` by extracting a tiny helper that encapsulates the TTL + fingerprint logic.

### 1. Extract a reusable `EventDeduplicator`

Move the state and low-level bookkeeping into a small class (could live in the same module or a separate one):

```python
import time
from collections import deque
from typing import Deque, Set, Tuple

class EventDeduplicator:
    def __init__(self, ttl_seconds: float = 0.5) -> None:
        self._ttl_seconds = ttl_seconds
        self._seen: Set[tuple] = set()
        self._queue: Deque[Tuple[float, tuple]] = deque()

    def _clean_expired(self) -> None:
        now = time.monotonic()
        expire_before = now - self._ttl_seconds
        while self._queue and self._queue[0][0] < expire_before:
            _, fingerprint = self._queue.popleft()
            self._seen.discard(fingerprint)

    def _build_fingerprint(self, event: AstrMessageEvent) -> tuple:
        return (
            event.get_platform_id() or "",
            event.unified_msg_origin or "",
            event.get_sender_id() or "",
            (event.get_message_str() or "").strip(),
        )

    def is_duplicate(self, event: AstrMessageEvent) -> bool:
        self._clean_expired()
        fingerprint = self._build_fingerprint(event)
        if fingerprint in self._seen:
            return True
        ts = time.monotonic()
        self._seen.add(fingerprint)
        self._queue.append((ts, fingerprint))
        return False
```

### 2. Simplify `EventBus` to delegate dedup

Then `EventBus` only expresses the *intent* (“deduplicate”) and no longer manages cache internals:

```python
class EventBus:
    def __init__(self, event_queue, pipeline_scheduler_mapping, astrbot_config_mgr) -> None:
        self.event_queue = event_queue
        self.pipeline_scheduler_mapping = pipeline_scheduler_mapping
        self.astrbot_config_mgr = astrbot_config_mgr
        self._deduplicator = EventDeduplicator(ttl_seconds=0.5)

    async def dispatch(self) -> None:
        while True:
            event: AstrMessageEvent = await self.event_queue.get()
            if self._deduplicator.is_duplicate(event):
                logger.info(
                    "Skip duplicate event in event_bus, umo=%s, sender=%s",
                    event.unified_msg_origin,
                    event.get_sender_id(),
                )
                continue
            # ... existing logic ...
```

This keeps all behavior (monotonic TTL, fingerprint definition, short-window dedup) but:

- `EventBus` no longer mixes dispatch logic with low-level cache maintenance.
- The dedup logic is reusable and testable in isolation, and can be shared/extended if you decide to harmonize platform-level and bus-level dedup later.
</issue_to_address>