fix: avoid duplicate send_message_to_user replies

[Soulter]

Summary

• record plain text sent by send_message_to_user when it targets the current session
• skip RespondStage only when the pending reply text exactly matches text already sent by the tool
• log before skipping so duplicate suppression is visible

Testing

• uv run pytest tests/unit/test_message_tools.py -q
• uv run ruff format .
• uv run ruff check .

Closes #6005.
Supersedes #6067; after this PR is merged, #6067 can be closed.

Summary by Sourcery

Prevent duplicate user-facing replies when both the send_message_to_user tool and the RespondStage would emit the same text in the current session.

Bug Fixes:

• Track plain-text content sent by send_message_to_user to the current session and have RespondStage skip sending an identical plain-text reply in that session.
• Ensure send_message_to_user calls targeting other sessions do not affect duplicate suppression for the active session.

Tests:

• Add unit tests covering current-session tracking for send_message_to_user, behavior for other sessions, and RespondStage duplicate-suppression logic.

[gemini-code-assist]

Code Review

This pull request introduces a deduplication mechanism to prevent duplicate replies when a message has already been sent to the user in the current session via tools. It records sent plain texts in the event's extra data and skips the respond stage if the same text is about to be sent again. Unit tests have been added to verify this behavior. I have no additional feedback to provide.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[sourcery-ai]

Hey - I've found 1 issue, and left some high level feedback:

• The key _send_message_to_user_current_session_plain_texts is hard-coded in multiple places (tool, respond stage, tests); consider centralizing it as a shared constant or helper to avoid drift and typos.
• Instead of mutating the event with the ad-hoc _has_send_oper attribute, prefer storing this flag in event extras (like the plain-text list) to keep event state consistently encapsulated.
• When appending to _send_message_to_user_current_session_plain_texts, you might want to guard against unbounded growth (e.g., deduplicate or limit to the last N entries) in long-running sessions.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- The key `_send_message_to_user_current_session_plain_texts` is hard-coded in multiple places (tool, respond stage, tests); consider centralizing it as a shared constant or helper to avoid drift and typos.
- Instead of mutating the event with the ad-hoc `_has_send_oper` attribute, prefer storing this flag in `event` extras (like the plain-text list) to keep event state consistently encapsulated.
- When appending to `_send_message_to_user_current_session_plain_texts`, you might want to guard against unbounded growth (e.g., deduplicate or limit to the last N entries) in long-running sessions.

## Individual Comments

### Comment 1
<location path="astrbot/core/pipeline/respond/stage.py" line_range="182" />
<code_context>
         if result.result_content_type == ResultContentType.STREAMING_FINISH:
             event.set_extra("_streaming_finished", True)
             return
+        sent_plain_texts = event.get_extra(
+            "_send_message_to_user_current_session_plain_texts",
+            [],
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the duplicate-response detection block into well-named helper functions to make `process` easier to read and test.

You can reduce the cognitive load of `process` by extracting the duplicate-detection logic into a small helper and giving names to the implicit contracts.

For example:

```python
def _is_duplicate_plain_text_response(event: AstrMessageEvent, result) -> bool:
    sent_plain_texts = event.get_extra(
        "_send_message_to_user_current_session_plain_texts",
        [],
    )
    if not isinstance(sent_plain_texts, list):
        return False

    result_plain_text = result.get_plain_text().strip()
    if not result_plain_text:
        return False

    if not _is_simple_text_chain(result):
        return False

    return result_plain_text in sent_plain_texts


def _is_simple_text_chain(result) -> bool:
    allowed_types = {
        ComponentType.Plain,
        ComponentType.Reply,
        ComponentType.At,
    }
    return all(comp.type in allowed_types for comp in result.chain)
```

Then `process` becomes easier to scan and unit test:

```python
if _is_duplicate_plain_text_response(event, result):
    logger.info(
        "send_message_to_user already delivered the same text in this session, "
        "skip respond stage to avoid duplicate reply.",
    )
    return

logger.info(
    f"Prepare to send - {event.get_sender_name()}/{event.get_sender_id()}: "
    f"{event._outline_chain(result.chain)}",
)
```

This keeps all behavior (including `strip`, type checks, and the chain component constraint) but moves the nuanced logic into named helpers with clear intent.
</issue_to_address>

---

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

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[sourcery-ai]

issue (complexity): Consider extracting the duplicate-response detection block into well-named helper functions to make process easier to read and test.

You can reduce the cognitive load of process by extracting the duplicate-detection logic into a small helper and giving names to the implicit contracts.

For example:

def _is_duplicate_plain_text_response(event: AstrMessageEvent, result) -> bool:
    sent_plain_texts = event.get_extra(
        "_send_message_to_user_current_session_plain_texts",
        [],
    )
    if not isinstance(sent_plain_texts, list):
        return False

    result_plain_text = result.get_plain_text().strip()
    if not result_plain_text:
        return False

    if not _is_simple_text_chain(result):
        return False

    return result_plain_text in sent_plain_texts


def _is_simple_text_chain(result) -> bool:
    allowed_types = {
        ComponentType.Plain,
        ComponentType.Reply,
        ComponentType.At,
    }
    return all(comp.type in allowed_types for comp in result.chain)

Then process becomes easier to scan and unit test:

if _is_duplicate_plain_text_response(event, result):
    logger.info(
        "send_message_to_user already delivered the same text in this session, "
        "skip respond stage to avoid duplicate reply.",
    )
    return

logger.info(
    f"Prepare to send - {event.get_sender_name()}/{event.get_sender_id()}: "
    f"{event._outline_chain(result.chain)}",
)

This keeps all behavior (including strip, type checks, and the chain component constraint) but moves the nuanced logic into named helpers with clear intent.

[cloudflare-workers-and-pages]

Deploying with    Cloudflare Workers

The latest updates on your project. Learn more about integrating Git with Workers.

Status	Name	Latest Commit	Preview URL	Updated (UTC)
✅ Deployment successful! View logs	astrbot-docs	8044f4f	Commit Preview URL Branch Preview URL	Jun 27 2026, 08:55 AM