fix: reorder fake tool call messages at tail to correct timing

[Rail1bc]

Motivation / 动机

通过向 req.contexts 尾部注入 assistant(tool_calls) → tool(result) 消息对，可以强制 LLM 认为自己已经调用了某个工具并拿到了结果，从而：

• 跳过 LLM 自身的工具调用决策，节省 token
• 降低延迟（不需要等 LLM 生成工具调用）
• 增加确定性（固定或规则触发，不受模型输出波动影响，增强确定性）

这是一种通用手法，已在 LivingMemory 插件中长期记忆注入中使用，未来可能被更多插件用于各种确定性工具调用场景。

但在 OpenAI_Provider._prepare_chat_payload() 中，处理顺序是：

1. deepcopy(contexts) —— 此时尾部已含伪造工具调用对
2. append(new_record) —— 追加当前用户消息

最终发给 LLM API 的 messages 序列为：

... assistant(tool_calls=...), tool(result), user("...")

LLM 视角下，assistant 在用户提问之前就"预知"了查询内容。正确顺序应为：

... user("..."), assistant(tool_calls=...), tool(result)

该问题影响所有使用该手法的插件，以及所有走 OpenAI_Provider 的 LLM 路径

Refs: #9450

Modifications / 改动点

astrbot/core/provider/sources/openai_source.py

在 _sanitize_assistant_messages() 末尾新增检测与重排逻辑：

1. 从尾部弹出 user 消息
2. 用 while 循环收集所有 assistant(tool_calls) + tool 成对消息（验证 tool_call_id 匹配）
3. 重排为 user → assistant₁, tool₁ → ... → assistant_N, tool_N

支持多插件各自注入多轮伪造对的场景。

• 不依赖具体插件实现，通用模式检测
• 不改动 req 对象，不破坏其他 on_llm_request handler
• 覆盖 _query / _query_streaming 全部路径
• 真实工具调用场景下 tool 后不会直接跟 user，不会误杀

此为轻量妥协修复。未来若实现专用的上下文操作钩子或伪造工具调用钩子，可考虑移除此方法。

• This is NOT a breaking change. / 这不是一个破坏性变更。

Test Results / 测试结果

逻辑验证（手动）：

• 无伪造对：不触发，不变
• 单对：..., asst(tc), tool, user → ..., user, asst(tc), tool
• 多对：..., asst₁, tool₁, asst₂, tool₂, user → ..., user, asst₁, tool₁, asst₂, tool₂
• 真实工具调用：..., asst(tc), tool, asst(content), user → 不变，不误杀

---

Checklist / 检查清单

• 😊 新功能已通过 Issue 与作者讨论
• 👀 我的更改经过了充分测试，测试结果已在上方提供
• 🤓 无新依赖引入
• 😮 无恶意代码

Summary by Sourcery

Ensure fake tool call assistant/tool/user message sequences are reordered so user messages precede injected tool call pairs, while preserving real tool call ordering and markers across all compatible providers and serialization paths.

Bug Fixes:

• Correct message ordering when plugins inject fake assistant(tool_calls) and tool messages at the tail so the LLM sees user queries before tool call pairs.
• Prevent crashes and misordering when message lists contain non-dict entries or incomplete/mismatched tool call IDs.
• Avoid reordering and protect timing of real tool calls by marking them and excluding marked pairs from fake-tool-call reordering.
• Ensure internal real-tool-call markers are stripped from outbound payloads so provider APIs never receive internal flags.

Enhancements:

• Share a reusable fake tool call fixture across provider tests to keep scenarios consistent.
• Propagate and persist real-tool-call markers through Message serialization, ToolCallsResult conversion, modalities sanitization, and checkpoint roundtrips to support robust reordering logic.
• Apply the fake-tool-call reorder logic to OpenAI, Anthropic, Gemini, and OpenAI Responses providers, including both streaming and non-streaming code paths.
• Add comprehensive tests across providers to validate reordering behavior, marker preservation, and mixed real/fake tool call scenarios.

Tests:

• Extend unit tests for OpenAI, Anthropic, Gemini, and OpenAI Responses providers to cover fake tool call reordering, real tool call protection, marker stripping, and edge cases such as mixed pairs and malformed messages.

[dosubot]

📄 Knowledge review

Dosu skipped reviewing this PR because your organization has used its 200 included credits for the month. Your usage will reset on 2026-08-01. To have Dosu review this PR before then, ask your organization admin to upgrade to a pro account.

---

[sourcery-ai]

Hey - I've found 1 issue

Prompt for AI Agents

Please address the comments from this code review:

## Individual Comments

### Comment 1
<location path="astrbot/core/provider/sources/openai_source.py" line_range="534" />
<code_context>
+        ProviderOpenAIOfficial._reorder_tailing_tool_call_user(payloads)
+
+    @staticmethod
+    def _reorder_tailing_tool_call_user(payloads: dict) -> None:
+        """重排因伪造工具调用导致尾部 assistant(tc) → tool → user 乱序的消息。
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring `_reorder_tailing_tool_call_user` to use an index-based scan with a separate pair-validation helper instead of pop-based mutation to clarify its control flow and data handling.

You can simplify `_reorder_tailing_tool_call_user` by making it index-based and separating pair validation from mutation. This reduces control-flow nesting, avoids repeated `pop()`/reverse operations, and makes the behavior easier to reason about.

### 1. Extract a small validator helper

```python
@staticmethod
def _is_valid_tool_pair(asst_msg: dict, tool_msg: dict) -> bool:
    if asst_msg.get("role") != "assistant" or tool_msg.get("role") != "tool":
        return False

    tool_calls = asst_msg.get("tool_calls")
    if not tool_calls:
        return False

    tc_ids = {tc.get("id") for tc in tool_calls if isinstance(tc, dict)}
    return tool_msg.get("tool_call_id") in tc_ids
```

### 2. Use indices + slicing instead of pops + reverse

```python
@staticmethod
def _reorder_tailing_tool_call_user(payloads: dict) -> None:
    messages = payloads.get("messages")
    if not isinstance(messages, list) or len(messages) < 2:
        return

    # 必须以 user 结尾
    if messages[-1].get("role") != "user":
        return

    # 从尾部向前扫描匹配的 assistant/tool 成对消息
    end = len(messages) - 1  # index of tailing user
    i = end - 1
    while i >= 1:
        asst_msg = messages[i - 1]
        tool_msg = messages[i]
        if not ProviderOpenAIOfficial._is_valid_tool_pair(asst_msg, tool_msg):
            break
        i -= 2  # 每次向前跳过一对

    # 没有成对工具调用，保持原样
    if i == end - 1:
        return

    # 现在 messages 结构为: [ ... prefix ..., pair_start..pair_end, user ]
    prefix = messages[:i + 1]
    pairs = messages[i + 1:end]  # contiguous assistant/tool pairs
    user_msg = messages[end]

    # 重排：prefix + [user] + pairs
    payloads["messages"] = prefix + [user_msg] + pairs
```

This keeps all behavior but removes the mutable stack-like operations and nested breaks, making the tail reordering logic more straightforward and testable.
</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.}

[xiaoyuyu6420]

Review

Clean fix for #9450. The approach of popping the trailing user message and re-inserting it before the assistant(tool_calls) → tool pairs is correct and minimal.

Strengths

1. Validates tool_call_id matching — the code checks tool_msg.get("tool_call_id") not in tc_ids before treating a pair as a tool-call pair. This prevents misidentifying unrelated assistant → tool sequences.
2. Reverses pairs before re-insertion — correctly handles the LIFO collection order.
3. Handles the no-op case — when no fake pairs are found, the user message is put back in place.

Issues

1. No tests

This is a non-trivial message reordering algorithm with several branches (no pairs, one pair, multiple pairs, mismatched tool_call_id). There should be at least one test per branch:

• Trailing user after a single fake tool call pair
• Trailing user after multiple fake tool call pairs
• No fake pairs (message order unchanged)
• assistant → tool without tool_calls (should stop collecting)

2. Only handles the OpenAI official provider

This fix is in ProviderOpenAIOfficial, but the issue (#9450) says the problem occurs in _prepare_chat_payload. If other providers (e.g. OpenAI-compatible providers) have the same _prepare_chat_payload pattern, they'll have the same bug. Worth checking if this needs to be applied more broadly.

3. Edge case: what if the user message itself contains tool-call-related content?

If a plugin injects a fake tool call AND the user message has role "user" but also contains function-related content, the reordering still treats it as a plain user message. This is probably correct (OpenAI expects user messages after tool results), but worth a comment.

4. Performance: O(n) pop from the end in a while loop

Each messages.pop() is O(1) from the end, and the while loop processes at most len(messages)/2 iterations, so this is fine. Just noting that the list is mutated in place.

Summary

Solid algorithm, correct logic. Main gap is missing tests — a message reordering function with 4+ branches should have test coverage before merging.

[Rail1bc]

修复伪造工具调用（fake tool call）消息对在 OpenAI / Anthropic / Gemini 格式 provider 中的时序错乱问题。

修复从openai扩展到OpenAI / Anthropic / Gemini 格式
覆盖了测试

Modifications / 改动点

• 在 astrbot/core/provider/provider.py 新增共享函数 reorder_tailing_tool_call_user(messages: list)：从尾部弹出 user 消息，持续收集 assistant(tool_calls) + tool 成对消息（校验 tool_call_id 匹配），重排为 user → assistant → tool。

• 在 astrbot/core/provider/__init__.py 中 re-export 该函数。

• openai_source.py：删除原 static method _reorder_tailing_tool_call_user，改用共享函数（_query 与 _query_streaming 两条路径均覆盖）。

• anthropic_source.py：在 text_chat / text_chat_stream 调用 _prepare_payload 之前，对 OpenAI-format 的 context_query 应用同一重排函数。

• gemini_source.py：在 text_chat / text_chat_stream 组装 context_query 之后（_prepare_conversation 转换之前）应用同一重排函数。Gemini 的 functionCall 回合位置约束使其在重排前直接报 400，修复后合法且时序正确。

• 测试：tests/test_openai_source.py 更新为重排函数的直接调用（16 个用例）；tests/test_anthropic_kimi_code_provider.py 新增 2 个用例验证 Anthropic 格式转换后的时序；tests/test_gemini_source.py 新增 2 个用例验证 Gemini 的 messages 重排与转换后 user → model(functionCall) → user(functionResponse) 结构。

• This is NOT a breaking change. / 这不是一个破坏性变更。

Screenshots or Test Results / 运行截图或测试结果

$ python -m pytest tests/test_openai_source.py -k reorder -q
16 passed

$ python -m pytest tests/test_anthropic_kimi_code_provider.py -q
29 passed

$ python -m pytest tests/test_gemini_source.py -q
5 passed

$ python -m ruff check astrbot/core/provider/
All checks passed!

[Rail1bc]

@sourcery-ai review

[sourcery-ai]

Hey - I've reviewed your changes and they look great!

---

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]

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

• The helper _FAKE_TOOL_CALL_CONTEXTS is duplicated across the Anthropic and Gemini tests; consider extracting it (and related expected structures) into a shared test utility to avoid drift and keep the scenarios consistent.
• You might want to tighten the type hints on reorder_tailing_tool_call_user (e.g., list[dict[str, Any]] instead of plain list) to better document the expected shape of the messages parameter and help static analysis.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- The helper `_FAKE_TOOL_CALL_CONTEXTS` is duplicated across the Anthropic and Gemini tests; consider extracting it (and related expected structures) into a shared test utility to avoid drift and keep the scenarios consistent.
- You might want to tighten the type hints on `reorder_tailing_tool_call_user` (e.g., `list[dict[str, Any]]` instead of plain `list`) to better document the expected shape of the `messages` parameter and help static analysis.

## Individual Comments

### Comment 1
<location path="astrbot/core/provider/provider.py" line_range="214" />
<code_context>
         )


+def reorder_tailing_tool_call_user(messages: list) -> None:
+    """重排因伪造工具调用导致尾部 assistant(tc) → tool → user 乱序的消息。
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring `reorder_tailing_tool_call_user` to avoid in-place popping and instead scan with indices and rebuild the tail to make the control flow clearer.

The current implementation works but the in-place mutation plus nested loop makes it harder to reason about. You can keep the behavior while simplifying control flow by:

- Walking backwards with indices instead of popping inside the loop.
- Reconstructing the tail in one shot.
- (Optionally) splitting the tool-call pairing check into a small helper.

For example, you can refactor `reorder_tailing_tool_call_user` like this:

```python
def reorder_tailing_tool_call_user(messages: list) -> None:
    if not isinstance(messages, list) or len(messages) < 2:
        return

    # 尾部必须是 user
    last = messages[-1]
    if last.get("role") != "user":
        return

    pairs: list[tuple[dict, dict]] = []
    i = len(messages) - 2  # 从最后一个非 user 开始向前扫

    while i >= 1:
        asst_msg = messages[i - 1]
        tool_msg = messages[i]

        if asst_msg.get("role") != "assistant" or tool_msg.get("role") != "tool":
            break

        tool_calls = asst_msg.get("tool_calls")
        if not tool_calls:
            break

        tc_ids = {tc.get("id") for tc in tool_calls if isinstance(tc, dict)}
        if tool_msg.get("tool_call_id") not in tc_ids:
            break

        pairs.append((asst_msg, tool_msg))
        i -= 2  # 每次向前跳过一对 assistant + tool

    if not pairs:
        return  # 没有伪造对，无需重排

    # prefix: 未参与重排的前半段
    prefix = messages[:i + 1]
    user_msg = last

    new_tail: list[dict] = [user_msg]
    for asst_msg, tool_msg in reversed(pairs):
        new_tail.append(asst_msg)
        new_tail.append(tool_msg)

    # 一次性写回，保持原列表对象
    messages[:] = prefix + new_tail
```

If you want to further clarify the intent, you can factor out the pairing check into a helper without changing behavior:

```python
def _is_valid_tool_pair(asst_msg: dict, tool_msg: dict) -> bool:
    if asst_msg.get("role") != "assistant" or tool_msg.get("role") != "tool":
        return False
    tool_calls = asst_msg.get("tool_calls")
    if not tool_calls:
        return False
    tc_ids = {tc.get("id") for tc in tool_calls if isinstance(tc, dict)}
    return tool_msg.get("tool_call_id") in tc_ids
```

Then use it inside the loop:

```python
while i >= 1:
    asst_msg = messages[i - 1]
    tool_msg = messages[i]
    if not _is_valid_tool_pair(asst_msg, tool_msg):
        break
    pairs.append((asst_msg, tool_msg))
    i -= 2
```

This keeps all current functionality but reduces mutation during pattern matching, makes the index invariants explicit, and isolates the tool-call validation logic.
</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.}

[Rail1bc]

本次改动在框架层修复了伪造工具调用（fake tool call）消息对的时序错乱问题，覆盖 OpenAI / Anthropic / Gemini /
Responses 四条 provider 路径：
▎ - 新增共享函数 reorder_tailing_tool_call_user /
strip_internal_markers（astrbot/core/provider/provider.py），将插件注入的 assistant(tool_calls) → tool
对重排到用户消息之后，与真实工具调用时序对齐；
▎ - 误伤防护：真实工具执行产生的对通过 _from_real_tool_call
内部标记识别（经序列化/持久化/重试全链路保留），重排扫描遇到即停止；标记在发送边界对副本剥离，绝不进入发往 API
的请求；
▎ - Gemini 路径修复后不再触发 400 INVALID_ARGUMENT 的 functionCall 位置约束报错；

▎ 测试：4 个相关测试文件 139 passed、3 failed（3 个失败为 Windows 平台预存在的 file_uri_to_path
问题，与本次改动无关，上游基线同样失败）；ruff check 全部通过。
@xiaoyuyu6420 希望能抽空跟进此pr，谢谢！

[sourcery-ai]

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

• The _from_real_tool_call propagation logic is now duplicated in several places (ToolCallsResult.to_openai_messages[_model], _ensure_message_to_dicts, _message_to_dict, modalities), consider centralizing this into a single helper to reduce maintenance overhead and risk of inconsistent behavior across providers.
• reorder_tailing_tool_call_user silently returns on many edge cases and mutates the input list in place; it could be helpful to document its side effects and constraints more explicitly at the call sites (e.g., that messages must be OpenAI-style dicts and that only a trailing user is considered) to avoid subtle misuse in future extensions.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- The `_from_real_tool_call` propagation logic is now duplicated in several places (`ToolCallsResult.to_openai_messages[_model]`, `_ensure_message_to_dicts`, `_message_to_dict`, modalities), consider centralizing this into a single helper to reduce maintenance overhead and risk of inconsistent behavior across providers.
- `reorder_tailing_tool_call_user` silently returns on many edge cases and mutates the input list in place; it could be helpful to document its side effects and constraints more explicitly at the call sites (e.g., that messages must be OpenAI-style dicts and that only a trailing `user` is considered) to avoid subtle misuse in future extensions.

## Individual Comments

### Comment 1
<location path="astrbot/core/provider/provider.py" line_range="252-253" />
<code_context>
+    if not isinstance(messages, list) or len(messages) < 2:
+        return
+
+    last = messages[-1]
+    if last.get("role") != "user":
+        return
+
</code_context>
<issue_to_address>
**issue (bug_risk):** Guard against non-dict messages before calling `.get` in `reorder_tailing_tool_call_user`.

`last = messages[-1]` assumes the last element is always a dict, so `last.get("role")` will raise `AttributeError` if a non-dict message is ever appended (e.g., by a plugin). Please add a type check like `isinstance(last, dict)` before calling `.get` to keep this helper robust to mixed message types.
</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.}

[Rail1bc]

@sourcery-ai review

[sourcery-ai]

Hey - I've found 1 issue

Prompt for AI Agents

Please address the comments from this code review:

## Individual Comments

### Comment 1
<location path="astrbot/core/provider/provider.py" line_range="219-229" />
<code_context>
         )


+def _is_valid_tool_pair(asst_msg: dict[str, Any], tool_msg: dict[str, Any]) -> bool:
+    """判断 assistant(tool_calls) 与 tool 是否为 tool_call_id 匹配的一对。"""
+    if not isinstance(asst_msg, dict) or not isinstance(tool_msg, dict):
+        return False
+    if asst_msg.get("role") != "assistant" or tool_msg.get("role") != "tool":
+        return False
+    tool_calls = asst_msg.get("tool_calls")
+    if not tool_calls:
+        return False
+    tc_ids = {tc.get("id") for tc in tool_calls if isinstance(tc, dict)}
+    return tool_msg.get("tool_call_id") in tc_ids
+
</code_context>
<issue_to_address>
**suggestion:** Tool call ID collection may treat missing IDs as valid matches.

In `_is_valid_tool_pair`, `tc_ids` collects `tc.get("id")` for all dict `tool_calls`, so `None` values are included. A `tool_msg` with `tool_call_id=None` will then match any `tool_call` missing an `id`. To avoid treating missing IDs as valid, filter out `None` when building `tc_ids`, e.g. `{tc_id for tc in tool_calls if isinstance(tc, dict) and (tc_id := tc.get("id")) is not None}`.

```suggestion
def _is_valid_tool_pair(asst_msg: dict[str, Any], tool_msg: dict[str, Any]) -> bool:
    """判断 assistant(tool_calls) 与 tool 是否为 tool_call_id 匹配的一对。"""
    if not isinstance(asst_msg, dict) or not isinstance(tool_msg, dict):
        return False
    if asst_msg.get("role") != "assistant" or tool_msg.get("role") != "tool":
        return False
    tool_calls = asst_msg.get("tool_calls")
    if not tool_calls:
        return False
    # 仅收集非 None 的 tool_call_id，避免将缺失 ID 视为有效匹配
    tc_ids = {
        tc_id
        for tc in tool_calls
        if isinstance(tc, dict) and (tc_id := tc.get("id")) is not None
    }
    return tool_msg.get("tool_call_id") in tc_ids
```
</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.}

[Rail1bc]

@sourcery-ai review

[sourcery-ai]

Hey - I've found 3 issues, and left some high level feedback:

• ToolCallsResult.to_openai_messages_model mutates its AssistantMessageSegment/ToolCallMessageSegment instances by setting _from_real_tool_call in-place; if these objects are reused elsewhere this hidden state could be surprising, so consider cloning or documenting the mutation clearly at the API boundary.
• The internal marker key "_from_real_tool_call" is now referenced from multiple modules (provider, message, entities, fixtures); centralizing this in a shared constant would reduce the risk of typos and make future refactors of the marker semantics easier.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- ToolCallsResult.to_openai_messages_model mutates its AssistantMessageSegment/ToolCallMessageSegment instances by setting _from_real_tool_call in-place; if these objects are reused elsewhere this hidden state could be surprising, so consider cloning or documenting the mutation clearly at the API boundary.
- The internal marker key "_from_real_tool_call" is now referenced from multiple modules (provider, message, entities, fixtures); centralizing this in a shared constant would reduce the risk of typos and make future refactors of the marker semantics easier.

## Individual Comments

### Comment 1
<location path="tests/test_anthropic_kimi_code_provider.py" line_range="898-907" />
<code_context>
+@pytest.mark.asyncio
</code_context>
<issue_to_address>
**suggestion (testing):** Add Anthropic tests that cover real ToolCallsResult-driven tool calls to ensure they are not reordered and that markers never leak into the outbound payload.

Current tests cover fake tool call contexts using `FAKE_TOOL_CALL_CONTEXTS` and the reordering of the tail `assistant(tool_use) → user(tool_result)` pair. To also validate behavior for real tool calls (`tool_calls_result`), consider adding tests that:

- Construct a `ToolCallsResult` and pass it to `text_chat` / `text_chat_stream` via `tool_calls_result`.
- Assert that `messages` preserve the original tool-call order for real pairs (no reordering that puts the user before the tool).
- Verify that any internal `_from_real_tool_call` markers used for misfire protection are stripped from the payload sent to Anthropic (e.g., by inspecting the kwargs of the underlying client call).

This would align Anthropic coverage with the existing OpenAI/Responses tests and confirm the real/fake tool-call separation guaranteed by the shared utilities.

Suggested implementation:

```python
@pytest.mark.asyncio
async def test_text_chat_reorders_fake_tool_call_pair(monkeypatch):
    """伪造工具调用对应重排到用户消息之后，与真实工具调用时序对齐。"""
    provider = _setup_provider_with_mock_client(monkeypatch)

    await provider.text_chat(prompt="我的名字是？", contexts=FAKE_TOOL_CALL_CONTEXTS)

    assert _capture_payloads_create.last_kwargs["messages"] == (
        _EXPECTED_REORDERED_MESSAGES
    )


@pytest.mark.asyncio
async def test_text_chat_preserves_real_tool_call_order(monkeypatch):
    """真实工具调用对应的 tool_calls_result 保持原始时序，不发生用户在工具之前的重排。"""
    provider = _setup_provider_with_mock_client(monkeypatch)

    # 构造一个真实的工具调用结果，包含一个 assistant(tool_use) → user(tool_result) 对
    real_tool_calls_result = ToolCallsResult(
        tool_calls=[
            {
                "id": "real_tool_use_1",
                "type": "tool_use",
                "name": "recall",
                "input": {"key": "abc"},
                # 内部标记（如果存在的话）仅用于测试工具调用分离逻辑
                "_from_real_tool_call": True,
            }
        ],
        tool_results=[
            {
                "tool_use_id": "real_tool_use_1",
                "type": "tool_result",
                "content": "memory json",
                "_from_real_tool_call": True,
            }
        ],
    )

    await provider.text_chat(
        prompt="我的名字是？",
        tool_calls_result=real_tool_calls_result,
    )

    messages = _capture_payloads_create.last_kwargs["messages"]

    # 确认真实工具调用的顺序保持为 assistant(tool_use) → user(tool_result)，不会被重排为 user 在前
    tool_use_index = next(
        i for i, m in enumerate(messages) if m.get("role") == "assistant" and m.get("type") == "tool_use"
    )
    tool_result_index = next(
        i for i, m in enumerate(messages) if m.get("role") == "user" and m.get("type") == "tool_result"
    )

    assert tool_use_index < tool_result_index


@pytest.mark.asyncio
async def test_text_chat_strips_internal_markers_for_real_tool_calls(monkeypatch):
    """真实工具调用对应的内部标记（如 _from_real_tool_call）不会泄漏到发往 Anthropic 的 payload 中。"""
    provider = _setup_provider_with_mock_client(monkeypatch)

    real_tool_calls_result = ToolCallsResult(
        tool_calls=[
            {
                "id": "real_tool_use_2",
                "type": "tool_use",
                "name": "recall",
                "input": {"key": "xyz"},
                "_from_real_tool_call": True,
            }
        ],
        tool_results=[
            {
                "tool_use_id": "real_tool_use_2",
                "type": "tool_result",
                "content": "memory json xyz",
                "_from_real_tool_call": True,
            }
        ],
    )

    await provider.text_chat(
        prompt="请回忆 xyz 对应的内容？",
        tool_calls_result=real_tool_calls_result,
    )

    payload_messages = _capture_payloads_create.last_kwargs["messages"]

    # 检查消息级别是否存在内部标记字段
    for msg in payload_messages:
        assert "_from_real_tool_call" not in msg
        # 某些实现可能将内容项展开为列表；在这种情况下也需要检查子项
        content = msg.get("content")
        if isinstance(content, list):
            for item in content:
                assert "_from_real_tool_call" not in item

```

1. Ensure `ToolCallsResult` is imported at the top of `tests/test_anthropic_kimi_code_provider.py` from the shared utilities module that defines it, for example:
   `from shared_utils.tool_calls import ToolCallsResult` (adjust the path/name to match your actual codebase).
2. If your Anthropic client uses a different schema for tool call messages (e.g. `"role": "assistant", "content": [{"type": "tool_use", ...}]`), adapt the `tool_use_index` / `tool_result_index` detection logic to match that shape (e.g. by inspecting `msg["content"]` entries instead of top-level `type`).
3. If `text_chat` accepts `tool_calls_result` in a different shape than this example (e.g. dataclass fields `calls` / `results` or nested objects), adjust the construction of `ToolCallsResult` to follow the existing tests/utilities for OpenAI/Responses so that a real tool-use → tool-result pair is created.
4. To mirror OpenAI/Responses streaming coverage, consider adding analogous `text_chat_stream` tests that:
   - Pass a `ToolCallsResult` via `tool_calls_result` when invoking `text_chat_stream`.
   - Assert `messages` in the payload passed to the underlying streamed Anthropic client preserve order and strip `_from_real_tool_call`, using the same assertions as above on the captured payload(s) for the streaming path.
</issue_to_address>

### Comment 2
<location path="docs/PR.md" line_range="39" />
<code_context>
+- 测试：`tests/fixtures/fake_tool_call.py` 新增共享 `make_fake_pair` helper；`tests/test_openai_source.py` 覆盖重排、误伤防护、标记往返与三项回归（重试不误伤 / modalities 标记保留 / 非 dict 不崩溃——中部与尾部，共 32 个新用例）；`tests/test_openai_responses_source.py` 验证 Responses 路径重排与无标记泄漏（新增 6 个用例，文件共 14 个）；`tests/test_anthropic_kimi_code_provider.py` 新增 2 个用例验证 Anthropic 格式转换后的时序；`tests/test_gemini_source.py` 新增 2 个用例验证 Gemini 的 messages 重排与转换后 `user → model(functionCall) → user(functionResponse)` 结构。
+
+- [x] This is NOT a breaking change. / 这不是一个破坏性变更。
+<!-- If your changes is a breaking change, please uncheck the checkbox above -->
+
+### Screenshots or Test Results / 运行截图或测试结果
</code_context>
<issue_to_address>
**nitpick (typo):** Fix grammar in the commented template line ("changes is" → "changes are" or "change is").

In the HTML comment, "If your changes is a breaking change" is grammatically incorrect. Please change it to either "If your changes are a breaking change" or "If your change is a breaking change" to keep the template wording correct.

```suggestion
<!-- If your change is a breaking change, please uncheck the checkbox above -->
```
</issue_to_address>

### Comment 3
<location path="astrbot/core/provider/provider.py" line_range="219" />
<code_context>
         )


+def _is_valid_tool_pair(asst_msg: dict[str, Any], tool_msg: dict[str, Any]) -> bool:
+    """判断 assistant(tool_calls) 与 tool 是否为 tool_call_id 匹配的一对。"""
+    if not isinstance(asst_msg, dict) or not isinstance(tool_msg, dict):
</code_context>
<issue_to_address>
**issue (complexity):** Consider inlining the tool-pair validation into `_is_fake_tool_pair` and extracting tail fake-pair collection into a helper to simplify the control flow and indexing logic.

A couple of targeted tweaks can reduce the index/indirection complexity without changing behavior.

### 1. Merge `_is_valid_tool_pair` into `_is_fake_tool_pair`

You can avoid the extra hop and keep all logic in a single helper:

```python
def _is_fake_tool_pair(asst_msg: dict[str, Any], tool_msg: dict[str, Any]) -> bool:
    """判断一对 assistant(tool_calls) / tool 是否为需要重排的伪造对。"""
    if not isinstance(asst_msg, dict) or not isinstance(tool_msg, dict):
        return False
    if asst_msg.get("role") != "assistant" or tool_msg.get("role") != "tool":
        return False

    tool_calls = asst_msg.get("tool_calls")
    if not tool_calls:
        return False

    tc_ids = {
        tc_id
        for tc in tool_calls
        if isinstance(tc, dict) and (tc_id := tc.get("id")) is not None
    }
    if tool_msg.get("tool_call_id") not in tc_ids:
        return False

    # 真实工具执行产生的消息带 `_from_real_tool_call` 标记；伪造对是裸 dict
    return not bool(
        asst_msg.get("_from_real_tool_call") or tool_msg.get("_from_real_tool_call")
    )
```

This keeps the semantics but removes one helper and a conceptual layer for readers.

### 2. Extract tail fake-pair detection into a small helper

Separating “where is the fake block” from “how do we reorder” makes `reorder_tailing_tool_call_user` much easier to follow and avoids having to mentally trace `i` / `i + 1`:

```python
def _collect_tailing_fake_pairs(
    messages: list[dict[str, Any]],
) -> tuple[int, list[tuple[dict[str, Any], dict[str, Any]]]]:
    """返回尾部伪造对块的起始索引及按外层到内层顺序排列的对。"""
    if not isinstance(messages, list) or len(messages) < 2:
        return len(messages), []

    last = messages[-1]
    if not isinstance(last, dict) or last.get("role") != "user":
        return len(messages), []

    pairs: list[tuple[dict[str, Any], dict[str, Any]]] = []
    i = len(messages) - 2
    while i >= 1 and _is_fake_tool_pair(messages[i - 1], messages[i]):
        pairs.append((messages[i - 1], messages[i]))
        i -= 2

    # i+1 为伪造对块的起点；对按外层→内层顺序返回
    return i + 1, list(reversed(pairs))
```

Then `reorder_tailing_tool_call_user` becomes more declarative:

```python
def reorder_tailing_tool_call_user(messages: list[dict[str, Any]]) -> None:
    start_idx, pairs = _collect_tailing_fake_pairs(messages)
    if not pairs:
        return

    last_user = messages[-1]
    prefix = messages[:start_idx]
    fake_block = [m for pair in pairs for m in pair]

    # 重排：prefix → user → fake_block
    messages[:] = prefix + [last_user] + fake_block
```

This keeps all current constraints (in-place modification, marker-based protection of real tool calls, inner/outer ordering) but removes the non-obvious `i`, `i-1`, `i+1` coupling in the main function.
</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.}

[Rail1bc]

@sourcery-ai review

[sourcery-ai]

Hey - I've found 1 issue

Prompt for AI Agents

Please address the comments from this code review:

## Individual Comments

### Comment 1
<location path="astrbot/core/provider/provider.py" line_range="220" />
<code_context>
         )


+def _is_fake_tool_pair(asst_msg: dict[str, Any], tool_msg: dict[str, Any]) -> bool:
+    """判断一对 assistant(tool_calls) / tool 是否为需要重排的伪造对。
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the fake tool-call helpers by restructuring `_is_fake_tool_pair`, inlining the tail-pair collection into `reorder_tailing_tool_call_user`, and consolidating outgoing sanitization into a single helper function.

The fake-tool handling helpers can be simplified without losing behavior or safety, which should reduce the mental overhead for future readers.

### 1. Make `_is_fake_tool_pair` a straight-line sequence of checks

You already have good comments; the function can be made more readable by removing the walrus/comprehension and structuring as named steps:

```python
def _is_fake_tool_pair(asst_msg: dict[str, Any], tool_msg: dict[str, Any]) -> bool:
    if not isinstance(asst_msg, dict) or not isinstance(tool_msg, dict):
        return False

    if asst_msg.get("role") != "assistant" or tool_msg.get("role") != "tool":
        return False

    tool_calls = asst_msg.get("tool_calls")
    if not tool_calls:
        return False

    tc_ids: set[str] = set()
    for tc in tool_calls:
        if not isinstance(tc, dict):
            continue
        tc_id = tc.get("id")
        if tc_id is not None:
            tc_ids.add(tc_id)

    if tool_msg.get("tool_call_id") not in tc_ids:
        return False

    # exclude real tool-call pairs
    if asst_msg.get(FROM_REAL_TOOL_CALL_KEY):
        return False
    if tool_msg.get(FROM_REAL_TOOL_CALL_KEY):
        return False

    return True
```

This keeps all semantics but reduces nested logic and inline assignment.

### 2. Inline `_collect_tailing_fake_pairs` logic into `reorder_tailing_tool_call_user`

Since `_collect_tailing_fake_pairs` is only used by `reorder_tailing_tool_call_user`, you can inline it and avoid the tuple `(start_idx, pairs)` and reversal, which simplifies the control flow:

```python
def reorder_tailing_tool_call_user(messages: list[dict[str, Any]]) -> None:
    if not isinstance(messages, list) or len(messages) < 3:
        return

    last_user = messages[-1]
    if not isinstance(last_user, dict) or last_user.get("role") != "user":
        return

    pairs: list[tuple[dict[str, Any], dict[str, Any]]] = []
    i = len(messages) - 2

    # collect inner-most fake pairs from the tail
    while i >= 1 and _is_fake_tool_pair(messages[i - 1], messages[i]):
        pairs.append((messages[i - 1], messages[i]))
        i -= 2

    if not pairs:
        return

    start_idx = i + 1
    fake_block = [m for pair in reversed(pairs) for m in pair]

    # prefix → user → assistant_1, tool_1 → ... → assistant_N, tool_N
    messages[:] = messages[:start_idx] + [last_user] + fake_block
```

This removes one helper and the extra indirection while preserving the same behavior (same stop condition, same ordering, same in-place update).

### 3. Fold `strip_internal_markers` into a single sanitization helper

To hide the internal `_from_real_tool_call` mechanics from call sites, consider a single “outgoing sanitization” function that both reorders and strips markers. For example:

```python
def sanitize_outgoing_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
    # work on a shallow copy to preserve payload markers for retries
    working = [msg.copy() if isinstance(msg, dict) else msg for msg in messages]

    reorder_tailing_tool_call_user(working)

    return [
        {k: v for k, v in msg.items() if k != FROM_REAL_TOOL_CALL_KEY}
        if isinstance(msg, dict)
        else msg
        for msg in working
    ]
```

Then call `sanitize_outgoing_messages(...)` at the send boundary instead of calling `reorder_tailing_tool_call_user` and `strip_internal_markers` separately. This consolidates the feature into one abstraction and reduces the number of functions readers need to understand.
</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.}

[Rail1bc]

@sourcery-ai review

[sourcery-ai]

Hey - I've found 2 issues, and left some high level feedback:

• ToolCallsResult.to_openai_messages_model mutates the underlying Message segments by setting _from_real_tool_call; consider making this side-effect more explicit in the method name or docstring to avoid surprises for callers that reuse these segments elsewhere.
• reorder_tailing_tool_call_user operates in-place on the messages list; given it is now wired into multiple providers, it may be safer to make the copying behavior explicit (either always copy or clearly document that callers must not reuse the original order).

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- ToolCallsResult.to_openai_messages_model mutates the underlying Message segments by setting _from_real_tool_call; consider making this side-effect more explicit in the method name or docstring to avoid surprises for callers that reuse these segments elsewhere.
- reorder_tailing_tool_call_user operates in-place on the messages list; given it is now wired into multiple providers, it may be safer to make the copying behavior explicit (either always copy or clearly document that callers must not reuse the original order).

## Individual Comments

### Comment 1
<location path="astrbot/core/provider/provider.py" line_range="306-316" />
<code_context>
+    messages[:] = messages[:start_idx] + [last_user] + fake_block
+
+
+def strip_internal_markers(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """返回剥离 ``_from_real_tool_call`` 标记的消息副本，不改动原列表。
+
+    标记只在发送前对副本剥离，payload 中的标记保留供重试时重排继续做误伤防护。
+    """
+    return [
+        {k: v for k, v in msg.items() if k != FROM_REAL_TOOL_CALL_KEY}
+        if isinstance(msg, dict)
+        else msg
+        for msg in messages
+    ]
</code_context>
<issue_to_address>
**suggestion:** strip_internal_markers’ type hints don’t match its behavior for non-dict entries.

The function is typed as `list[dict[str, Any]]`, but the comprehension preserves non-dict elements (`else msg`), so callers may receive mixed element types. Either constrain inputs to dicts (e.g., with validation) or adjust the annotation (e.g., `list[Any]` or `Sequence[Mapping[str, Any] | Any]`) to reflect the actual behavior.

```suggestion
def strip_internal_markers(messages: list[Any]) -> list[Any]:
    """返回剥离 ``_from_real_tool_call`` 标记的消息副本，不改动原列表。

    仅对字典消息剥离标记，非字典元素原样保留，从而可能返回包含多种元素类型的列表。
    标记只在发送前对副本剥离，payload 中的标记保留供重试时重排继续做误伤防护。
    """
    return [
        {k: v for k, v in msg.items() if k != FROM_REAL_TOOL_CALL_KEY}
        if isinstance(msg, dict)
        else msg
        for msg in messages
    ]
```
</issue_to_address>

### Comment 2
<location path="astrbot/core/provider/provider.py" line_range="220" />
<code_context>
         )


+def _is_fake_tool_pair(asst_msg: dict[str, Any], tool_msg: dict[str, Any]) -> bool:
+    """判断一对 assistant(tool_calls) / tool 是否为需要重排的伪造对。
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring the fake tool-call handling helpers and tail reordering logic to separate structural checks from semantics and use slice-based block manipulation, while grouping marker injection/stripping in a single serialization path for clearer control flow.

The fake tool-call handling is functionally solid, but there are a couple of spots where you can reduce cognitive complexity without changing behavior:

---

### 1) Separate structure vs. “fake” semantics in `_is_fake_tool_pair`

Right now `_is_fake_tool_pair` mixes:

- structural validation (roles, `tool_calls` presence, id matching),
- and fake/real semantics (`FROM_REAL_TOOL_CALL_KEY`).

You can make this easier to reason about by extracting a structural predicate and leaving `_is_fake_tool_pair` to focus on “fake vs real”.

```python
def _is_tool_pair_candidate(
    asst_msg: dict[str, Any],
    tool_msg: dict[str, Any],
) -> bool:
    return (
        isinstance(asst_msg, dict)
        and isinstance(tool_msg, dict)
        and asst_msg.get("role") == "assistant"
        and tool_msg.get("role") == "tool"
        and asst_msg.get("tool_calls")
    )


def _tool_pair_ids(asst_msg: dict[str, Any]) -> set[str]:
    ids: set[str] = set()
    for tc in asst_msg.get("tool_calls", []):
        if isinstance(tc, dict):
            tc_id = tc.get("id")
            if tc_id is not None:
                ids.add(tc_id)
    return ids


def _is_fake_tool_pair(asst_msg: dict[str, Any], tool_msg: dict[str, Any]) -> bool:
    if not _is_tool_pair_candidate(asst_msg, tool_msg):
        return False

    if tool_msg.get("tool_call_id") not in _tool_pair_ids(asst_msg):
        return False

    # only “fake” semantics here
    if asst_msg.get(FROM_REAL_TOOL_CALL_KEY) or tool_msg.get(FROM_REAL_TOOL_CALL_KEY):
        return False

    return True
```

This keeps the same behavior but makes the contract clearer: one helper for “is it structurally a tool pair?”, one for “does it look fake?”.

---

### 2) Simplify tail collection and reordering with explicit slice indices

You can avoid index arithmetic + reversed pairs + list-flattening by returning the start/end indices of the tail block and relying on slices. This keeps the contiguous-block assumption explicit and makes `reorder_tailing_tool_call_user` easier to read.

```python
def _collect_tailing_fake_block(
    messages: list[dict[str, Any]],
) -> tuple[int, int]:
    """Return (start_idx, end_idx_exclusive) of the tailing fake tool-call block."""
    if not isinstance(messages, list) or len(messages) < 2:
        return 0, 0

    last = messages[-1]
    if not isinstance(last, dict) or last.get("role") != "user":
        return len(messages), len(messages)

    # Walk backwards over assistant/tool pairs, but only compute the start index;
    # the tail block itself comes from a slice.
    i = len(messages) - 2
    while i >= 1 and _is_fake_tool_pair(messages[i - 1], messages[i]):
        i -= 2

    # i + 1 is the index of the first assistant in the block; end is the last user.
    start_idx = i + 1
    end_idx = len(messages) - 1
    return start_idx, end_idx
```

Then `reorder_tailing_tool_call_user` can be expressed in terms of three named pieces rather than flattening `pairs`:

```python
def reorder_tailing_tool_call_user(messages: list[dict[str, Any]]) -> None:
    start_idx, end_idx = _collect_tailing_fake_block(messages)
    if start_idx == end_idx:
        return

    prefix = messages[:start_idx]
    fake_block = messages[start_idx:end_idx]  # already in outer→inner order
    last_user = messages[-1]

    # prefix → user → fake_block
    messages[:] = prefix + [last_user] + fake_block
```

This keeps all current semantics (including protection of real tool calls via `_is_fake_tool_pair`) but makes the control flow and data flow much easier to track:

- `_collect_tailing_fake_block` only answers “where is the block?”
- `reorder_tailing_tool_call_user` only arranges `prefix`, `user`, and `fake_block` via slices.

---

### 3) Keep marker injection/stripping close together

You now have:

- marker injection in `message_to_dict_with_marker`,
- marker stripping in `strip_internal_markers`.

To avoid spreading marker semantics further, you could lean on a single “serialization path” and keep the strip logic adjacent to the existing conversion helper:

```python
def messages_to_payload(messages: list[Message | dict]) -> list[dict]:
    # wrapper used at the send boundary
    return strip_internal_markers(
        [
            message_to_dict_with_marker(m) if isinstance(m, Message) else m
            for m in messages
            if not is_checkpoint_message(m)
        ]
    )
```

This kind of wrapper keeps marker inject/strip logically grouped and makes it clear that markers live only in internal state and are removed only at the payload boundary, without changing any of the current behavior.
</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.}

[whatevertogo]

我结合 AstrBot 当前的上下文组装链路做了一个可运行原型，确认更适合复用现有的 ProviderRequest.tool_calls_result，而不是在 provider 发送前扫描消息并推断 fake/real。

具体做法是：

• 插件在 on_llm_request 中调用 req.append_tool_calls_result(...)
• ToolLoopAgentRunner.reset() 先组装当前 user 消息
• 然后追加 request.tool_calls_result

最终顺序天然为：

history
→ current user
→ assistant(tool_calls)
→ tool(result)

验证结果：

• 测试覆盖了完整的 ToolLoopAgentRunner → OpenAI provider → SDK create() 路径
• 测试中主动禁用了 reorder_tailing_tool_call_user
• 最终消息顺序仍然正确
• Runner 测试 40 个通过
• Checkpoint/集成测试 17 个通过
• Ruff 和格式检查通过
• 使用 deepseek-v4-flash 实网请求返回 200，并正确读取预计算工具结果

因此可以确认，这条路径不依赖启发式 reorder，也不依赖 _from_real_tool_call marker。

建议当前 PR 改为：

1. 由 runner 消费 ProviderRequest.tool_calls_result
2. 从 astrbot.api.provider 导出 ToolCallsResult、AssistantMessageSegment 和 ToolCallMessageSegment
3. 将相关插件迁移到 req.append_tool_calls_result()
4. 删除尾部 reorder、_from_real_tool_call marker，以及 marker 在序列化、持久化、modalities、retry 和各 provider 中的传播
5. 补充临时注入是否持久化的明确语义

这个方案让消息生产者显式表达注入位置，由 runner 统一负责顺序，provider 只负责格式转换，职责边界更清晰，也不会误伤旧历史或外部传入的真实工具调用。

[Rail1bc]

已经在新的pr里使用更好的修复方案

[xiaoyuyu6420]

感谢 @Rail1bc @whatevertogo 的深入分析和可运行原型！看了 #9673 的方案——由 ToolLoopAgentRunner 统一消费 tool_calls_result 通道，插件显式声明注入位置，顺序天然正确，而且 _from_real_tool_call 标记穿透序列化/持久化/重试的那整套复杂度都消失了。认同这个方向明显更优，本 PR 关闭没有异议。这次学到的重要一课：让消息生产者显式表达意图，优于在 provider 发送边界做启发式推断。