fix: coalesce queued private chat replies

[94yi]

Summary

• Coalesce eligible queued private text messages from the same session into one bounded reply, preserving receive order and isolation between contacts.
• Skip superseded or stale messages and prevent terminated sessions from entering another LLM request.
• Preserve command, attachment and provider-request handling separately from delivery-safety work.

Upstream conflict resolution (2026-10-18)

• Merge current upstream master (e5e21b71aec5642f110eb115eb8e9a0d8028b7e5) without rewriting branch history.
• Preserve upstream provider wake-prefix checks, file preparation before the session lock, and the provider/session enabled-state recheck after acquiring it.
• Register pending merge text only after the wake-prefix check, so rejected requests cannot enter a later batch.
• Add regressions for prefix rejection and attachment-before-lock ordering; update the upstream session-lock test event fixture to include termination and platform methods.

Review fixes (2026-10-18)

• Remove stopped/canceled pending input in final cleanup without discarding ordinary superseded input needed by the newest request.
• Keep the latest oversized input within the character budget and explicitly disclose truncation rather than answering only older messages.
• Reclaim idle state after the existing cooldown and cap retained sessions at 1024 without evicting active queues/cooldowns. New sessions beyond the cap retain normal routing. Global sequence IDs prevent old handlers from claiming a recreated session queue.
• Add six regressions covering cancel/stop cleanup, preserved superseded batches, oversized newest input, state capacity/cooldown and recycled-sequence isolation.

Validation

• 208 targeted tests passed: inbound reply merging, internal agent termination, session locks and main-agent construction.
• Repository-wide Ruff passed; formatting check passed for 519 Python files.
• The PR diff against upstream passes whitespace checks. API/UI definitions are unchanged.
• New upstream OpenTelemetry dependencies were installed only in an isolated test directory; no production dependencies, services or account state were changed.

The merge conflict is resolved; required maintainer review still applies.

Summary by Sourcery

Consolidate eligible private chat backlogs into safe, session-scoped replies without altering separate command, attachment, or provider-request handling.

New Features:

• Coalesce eligible queued private text messages from the same session into a single bounded reply while preserving order and contact isolation.

Bug Fixes:

• Prevent stale or superseded queued messages from triggering duplicate replies and stop terminated sessions from starting new LLM requests.

Enhancements:

• Keep commands, attachments, provider requests, and wake-prefix validation on their existing independent processing paths.

Tests:

• Add regression coverage for message merging, session isolation, stale and canceled input, termination handling, wake-prefix rejection, and attachment preparation ordering.

[sourcery-ai]

Hey - I've found 3 issues

Prompt for AI Agents

Please address the comments from this code review:

## Individual Comments

### Comment 1
<location path="astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py" line_range="380-392" />
<code_context>
+                    logger.warning("send_typing failed", exc_info=True)
             if await call_event_hook(event, EventType.OnWaitingLLMRequestEvent):
                 return
+            inbound_merge_registered = (
+                _inbound_reply_merger.register(event)
+                if inbound_merge_candidate
+                else False
+            )

             async with session_lock_manager.acquire_lock(event.unified_msg_origin):
+                if event.is_stopped() or event.get_extra("agent_stop_requested"):
+                    logger.info(
+                        "Session terminated while waiting for lock, skipping LLM request. umo=%s",
+                        event.unified_msg_origin,
+                    )
+                    return
+                if inbound_merge_registered:
+                    while True:
</code_context>
<issue_to_address>
**issue (bug_risk):** A registered message remains in `_InboundMergeState.pending` when the event is stopped while waiting for the session lock. The next message from the same UMO then merges the canceled message into its prompt and produces a reply for stale input.

**Triggers:** When an eligible private message is terminated after registration but before it acquires the session lock.

**Suggested fix:** Remove the event's pending message on every early return after registration, including termination and cancellation paths.
</issue_to_address>

### Comment 2
<location path="astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py" line_range="96" />
<code_context>
+    """Coalesce queued private texts without crossing conversation boundaries."""
+
+    def __init__(self) -> None:
+        self._states: dict[str, _InboundMergeState] = {}
+
+    def register(self, event: AstrMessageEvent) -> bool:
</code_context>
<issue_to_address>
**issue (bug_risk):** `_InboundReplyMerger._states` retains one `_InboundMergeState` for every UMO forever, even after its pending list is cleared. A long-running bot accumulates unbounded per-user state as distinct private contacts send eligible messages.

**Triggers:** When the bot receives eligible private messages from many distinct UMOs over its lifetime.

**Suggested fix:** Delete an empty state after its batch is finalized, while preserving any required sequence or timing state elsewhere with bounded cleanup.
</issue_to_address>

### Comment 3
<location path="astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py" line_range="196-205" />
<code_context>
+    selected: list[str] = []
+    used_chars = 0
+    omitted = 0
+    for message in reversed(messages):
+        text = message.text.strip()
+        projected = used_chars + len(text)
+        if (
+            len(selected) >= _INBOUND_MERGE_MAX_MESSAGES
+            or projected > _INBOUND_MERGE_MAX_CHARS
+        ):
+            omitted += 1
+            continue
+        selected.append(text)
+        used_chars = projected
+    selected.reverse()
</code_context>
<issue_to_address>
**issue (bug_risk):** The reverse selection loop drops the newest message when that message alone exceeds `_INBOUND_MERGE_MAX_CHARS`, then selects older messages instead. The resulting LLM prompt therefore omits the latest user input and answers stale content.

**Triggers:** When the newest queued message is longer than 6000 characters, or when the newest message cannot fit under the character limit.

**Suggested fix:** Always retain the newest message, truncating it or applying the limit after selecting the newest item rather than skipping it.

```suggestion
    for message in reversed(messages):
        text = message.text.strip()
        if not selected and len(text) > _INBOUND_MERGE_MAX_CHARS:
            text = text[:_INBOUND_MERGE_MAX_CHARS]
        projected = used_chars + len(text)
        if (
            len(selected) >= _INBOUND_MERGE_MAX_MESSAGES
            or projected > _INBOUND_MERGE_MAX_CHARS
        ):
            omitted += 1
            continue
        selected.append(text)
```
</issue_to_address>

Sourcery assessment

Needs a human reviewer. 3 findings to address first, and if the merge bookkeeping or eligibility checks are wrong, queued private messages can be superseded or omitted, and a single altered LLM reply may be sent to the user before reverting can undo it. Reverting restores the previous per-event processing behavior, but it cannot retract replies already delivered or recover messages dropped by the new coalescing path.

Blocking findings: astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py:392, astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py:96, astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py:205

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

[DFGHJ43]

你好！你做出的修改已经属于新功能而不是修复的范畴了，添加新功能之前最好在issue和其他开发者讨论一下再添加，而且其实已经有社区插件实现了合并临近消息的功能了，你可以看一看。