fix: prevent late LLM responses after stop

[Sisyphbaous-DT-Project]

此前 /stop 会请求当前 Agent 停止，但不会强制取消已经发出的 provider 请求。如果 provider 在 stop 请求之后才返回，原始 LLMResponse 仍可能被保存为最终响应，继续进入 OnLLMResponseEvent、结果装饰、buffer/live 输出等后续链路，导致用户已经看到 stop 成功提示后，旧的 LLM 回复仍然被发送出去。

本 PR 保持现有软停止语义和 provider 请求行为不变，只确保 stop 后返回的迟到模型输出会被净化，不再进入用户可见发送链或普通 LLM 响应插件链。

Modifications / 改动点

• 在 ToolLoop Agent aborted 后构造空的安全响应，避免暴露原始模型正文、reasoning、result chain、raw completion 和 tool call 字段。

• 当事件处于 agent_stop_requested、agent_user_aborted 或已停止状态时，跳过 OnLLMResponseEvent 和 _llm_reasoning_content 写入。

• 保留 OnAgentDoneEvent，但 aborted 场景下传入净化后的空响应，保证生命周期清理类插件仍可执行。

• 阻止 stop 后继续 flush buffered LLM 输出、streaming finish 正文和 Live/TTS 已入队音频。

• 在 skills_like re-query fallback/repair 和 provider fallback 路径补充 stop 检查，避免 stop 后继续发起多余的迟到 provider 请求。

• 统一将 agent_user_aborted 视为 stop 状态，覆盖 runner 直接调用和 run_agent() 主链路。

• 补充回归测试，覆盖 aborted response 净化、hook 抑制、buffer 清理、Live/TTS 清理、skills_like stop、fallback stop，以及正常未 stop 路径。

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

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

本地验证命令：

TMPDIR=/tmp/astrbot_pytest_tmp python -m pytest tests/test_tool_loop_agent_runner.py -q

结果：

49 passed, 1 warning in 2.80s

额外检查：

ruff check astrbot/core/agent/runners/tool_loop_agent_runner.py astrbot/core/astr_agent_hooks.py astrbot/core/astr_agent_run_util.py astrbot/core/pipeline/context_utils.py astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py tests/test_tool_loop_agent_runner.py
ruff format --check astrbot/core/agent/runners/tool_loop_agent_runner.py astrbot/core/astr_agent_hooks.py astrbot/core/astr_agent_run_util.py astrbot/core/pipeline/context_utils.py astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py tests/test_tool_loop_agent_runner.py
git diff --check

结果：

All checks passed!
6 files already formatted
git diff --check passed with no output

---

Checklist / 检查清单

• 😊 If there are new features added in the PR, I have discussed it with the authors through issues/emails, etc.
  / 如果 PR 中有新加入的功能，已经通过 Issue / 邮件等方式和作者讨论过。

• 👀 My changes have been well-tested, and "Verification Steps" and "Screenshots" have been provided above.
  / 我的更改经过了良好的测试，并已在上方提供了“验证步骤”和“运行截图”。

• 🤓 I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations in requirements.txt and pyproject.toml.
  / 我确保没有引入新依赖库，或者引入了新依赖库的同时将其添加到 requirements.txt 和 pyproject.toml 文件相应位置。

• 😮 My changes do not introduce malicious code.
  / 我的更改没有引入恶意代码。

Summary by Sourcery

Ensure agent stop and user abort prevent late LLM outputs and audio from being delivered or processed while preserving lifecycle hooks.

Bug Fixes:

• Prevent late LLM text chunks and final completions from being sent to users after a stop or abort is requested.
• Stop buffering and flushing delayed LLM results and streaming finishes when the agent has been stopped or aborted.
• Avoid issuing additional provider requests in skills_like requery, repair, and fallback flows once a stop or abort has been requested.
• Treat user-aborted events as stop signals across runner, run_agent, hooks, and pipeline hook propagation so residual outputs are discarded.
• Prevent fallback providers from being called when a primary provider returns an error response after a stop request.
• Ensure live agent and TTS streaming discard queued or remaining text/audio once a stop or abort is detected.

Enhancements:

• Sanitize aborted final LLM responses to strip model-generated content while still recording usage and preserving identifiers.
• Adjust agent hooks and pipeline hook propagation to skip OnLLMResponseEvent for stopped or aborted events while still invoking OnAgentDoneEvent.
• Refine token usage accounting to avoid double-counting while still tracking usage for aborted responses.
• Allow follow-up messages to be accepted only when the agent is actively running and not stopping.

Tests:

• Add comprehensive tests covering aborted response sanitization, hook suppression, buffer and live/TTS cleanup, skills_like stop handling, fallback stop behavior, and normal non-stop paths.

[sourcery-ai]

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

• The logic for detecting a stop/abort and building an "empty" aborted LLMResponse is duplicated in several places (e.g., _event_requests_agent_stop in astr_agent_hooks vs context_utils, and _build_aborted_llm_response in both the runner and hooks); consider centralizing these helpers to avoid drift between slightly different implementations.
• The stop/abort semantics are now implemented independently in _is_stop_requested (runner), _should_stop_agent (run_util), _event_requests_agent_stop (hooks/context_utils), and _should_stop_hook_propagation; it would be easier to reason about late-stop behavior if these funneled through a single shared helper or enum-like state instead of multiple overlapping checks.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- The logic for detecting a stop/abort and building an "empty" aborted LLMResponse is duplicated in several places (e.g., `_event_requests_agent_stop` in `astr_agent_hooks` vs `context_utils`, and `_build_aborted_llm_response` in both the runner and hooks); consider centralizing these helpers to avoid drift between slightly different implementations.
- The stop/abort semantics are now implemented independently in `_is_stop_requested` (runner), `_should_stop_agent` (run_util), `_event_requests_agent_stop` (hooks/context_utils), and `_should_stop_hook_propagation`; it would be easier to reason about late-stop behavior if these funneled through a single shared helper or enum-like state instead of multiple overlapping checks.

## Individual Comments

### Comment 1
<location path="astrbot/core/agent/runners/tool_loop_agent_runner.py" line_range="175" />
<code_context>
         return extract_persona_custom_error_message_from_event(event)

-    async def _complete_with_assistant_response(self, llm_resp: LLMResponse) -> None:
+    async def _complete_with_assistant_response(self, llm_resp: LLMResponse) -> bool:
         """Finalize the current step as a plain assistant response with no tool calls."""
+        if self._is_stop_requested():
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting helpers for streaming stop handling in `step()` and post-completion abort logic in `_complete_with_assistant_response()` to centralize stop behavior and remove duplicated control flow.

The repeated and scattered stop/abort handling can be simplified without changing behavior by:

1. Extracting a small helper for the streaming stop path in `step()` to remove the duplicated `if self._is_stop_requested()` branches.
2. Encapsulating the three stop checks inside `_complete_with_assistant_response()` into a single helper so that the method clearly expresses a “normal completion” flow.

### 1. Extract a streaming-stop helper in `step()`

Right now the streaming path has several identical patterns:

```py
if self._is_stop_requested():
    llm_resp_result = llm_response
    break
```

You can wrap this in a focused helper that both captures the last chunk and returns a boolean:

```py
def _handle_streaming_stop(
    self,
    llm_response: LLMResponse,
) -> bool:
    """Capture latest chunk and signal exit from streaming loop."""
    if not self._is_stop_requested():
        return False

    # last chunk becomes the result of the step
    self.final_llm_resp = llm_response
    return True
```

Then simplify the loop:

```py
async for llm_response in self._iter_llm_responses_with_fallback():
    if llm_response.is_chunk:
        if self.stats.time_to_first_token == 0:
            self.stats.time_to_first_token = time.time() - self.stats.start_time

        if llm_response.reasoning_content:
            yield AgentResponse(
                type="streaming_delta",
                data=AgentResponseData(
                    chain=MessageChain(type="reasoning").message(
                        llm_response.reasoning_content,
                    ),
                ),
            )
        if self._handle_streaming_stop(llm_response):
            llm_resp_result = llm_response
            break

        if llm_response.result_chain:
            yield AgentResponse(
                type="streaming_delta",
                data=AgentResponseData(chain=llm_response.result_chain),
            )
        if self._handle_streaming_stop(llm_response):
            llm_resp_result = llm_response
            break

        elif llm_response.completion_text:
            yield AgentResponse(
                type="streaming_delta",
                data=AgentResponseData(
                    chain=MessageChain().message(llm_response.completion_text),
                ),
            )
        if self._handle_streaming_stop(llm_response):
            llm_resp_result = llm_response
            break

        continue

    llm_resp_result = llm_response
    if self._handle_streaming_stop(llm_response):
        break
```

This keeps the behavior (“capture last chunk and break”), but removes the copy‑pasted control flow, making the streaming section easier to follow and modify.

### 2. Centralize stop logic inside `_complete_with_assistant_response`

Currently `_complete_with_assistant_response()` has three separate `self._is_stop_requested()` checks with different side effects. You can encapsulate the “after completion, if stopped, clean up and mark as aborted” behavior into one helper that `_complete_with_assistant_response()` calls once:

```py
def _handle_completion_abort(
    self,
    llm_resp: LLMResponse,
) -> bool:
    """Handle a stop request after we have an assistant response."""
    if not self._is_stop_requested():
        return False

    # Remove the just-appended assistant message if it came from this llm_resp
    if (
        self.run_context.messages
        and self._is_message_from_llm_response(self.run_context.messages[-1], llm_resp)
    ):
        self.run_context.messages.pop()

    self.discard_late_aborted_result()
    self._resolve_unconsumed_follow_ups()
    return True
```

Then `_complete_with_assistant_response()` becomes a straight-line “normal completion” with one abort hook:

```py
async def _complete_with_assistant_response(self, llm_resp: LLMResponse) -> bool:
    if self._is_stop_requested():
        await self._finalize_aborted_step(llm_resp)
        return False

    self.final_llm_resp = llm_resp
    self._transition_state(AgentState.DONE)
    self.stats.end_time = time.time()

    parts: list[ContentPart] = []
    if llm_resp.reasoning_content is not None or llm_resp.reasoning_signature:
        parts.append(
            ThinkPart(
                think=llm_resp.reasoning_content or "",
                encrypted=llm_resp.reasoning_signature,
            )
        )
    if llm_resp.completion_text:
        parts.append(TextPart(text=llm_resp.completion_text))
    if not parts:
        logger.warning("LLM returned empty assistant message with no tool calls.")

    self.run_context.messages.append(Message(role="assistant", content=parts))

    try:
        await self.agent_hooks.on_agent_done(self.run_context, llm_resp)
    except Exception as e:
        logger.error(f"Error in on_agent_done hook: {e}", exc_info=True)

    if self._handle_completion_abort(llm_resp):
        return False

    self._resolve_unconsumed_follow_ups()
    return True
```

This preserves:

- Early abort before writing messages/hook invocation.
- Safe discarding of late aborted results.
- Follow‑up resolution and usage recording through existing helpers.

But the control flow is now:

1. Early abort.
2. Normal completion.
3. Single post‑completion abort hook.

Which makes future changes less error‑prone.

If you want to go further, `_is_stop_requested` could be moved into a shared utility to avoid drift from other stop logic, but the two changes above already reduce the local complexity without touching behavior or other modules.
</issue_to_address>

### Comment 2
<location path="astrbot/core/astr_agent_run_util.py" line_range="27" />
<code_context>


 def _should_stop_agent(astr_event) -> bool:
-    return astr_event.is_stopped() or bool(astr_event.get_extra("agent_stop_requested"))
+    return (
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring the new stop/abort handling into shared helpers and early‑exit patterns to consolidate duplicated logic and scattered `_should_stop_agent` checks.

The new stop/abort handling is functionally useful but increases complexity through duplicated buffer/abort logic and scattered `_should_stop_agent` checks. You can keep all behavior while consolidating this into a few helpers and early‑exit patterns.

### 1. Centralize “discard buffered LLM result on abort”

Right now, buffer clearing and `discard_late_aborted_result` wiring is duplicated:

- In the `resp.type == "aborted"` branch
- In the final `can_buffer_llm_result and agent_runner.done()` block

You can encapsulate this in a small helper and call it from both places:

```python
def _discard_buffered_llm_output(
    astr_event,
    buffered_llm_chains,
    agent_runner,
) -> None:
    buffered_llm_chains.clear()
    astr_event.clear_result()

    discard_late_result = getattr(
        agent_runner,
        "discard_late_aborted_result",
        None,
    )
    if callable(discard_late_result):
        discard_late_result()
```

Then in `run_agent`:

```python
if resp.type == "aborted":
    _discard_buffered_llm_output(astr_event, buffered_llm_chains, agent_runner)
    if not stop_watcher.done():
        ...
    astr_event.set_extra("agent_user_aborted", True)
    astr_event.set_extra("agent_stop_requested", False)
    return
```

And in the final buffering block:

```python
if can_buffer_llm_result and agent_runner.done():
    if _should_stop_agent(astr_event):
        _discard_buffered_llm_output(astr_event, buffered_llm_chains, agent_runner)
    else:
        merged_chain = _merge_buffered_llm_chains(buffered_llm_chains)
        if merged_chain:
            astr_event.set_result(
                MessageEventResult(
                    chain=merged_chain.chain,
                    result_content_type=ResultContentType.LLM_RESULT,
                ),
            )
            yield merged_chain
            astr_event.clear_result()
```

This removes duplicated buffer clearing and keeps the “late aborted result discard” behavior in exactly one place.

### 2. Use early‑exit in the live audio consumer loop

In `run_live_agent`, stop checks are per queue item:

```python
while True:
    queue_item = await audio_queue.get()

    if queue_item is None:
        break

    if agent_runner.was_aborted() or _should_stop_agent(...):
        continue
    ...
```

You can simplify by checking stop state once per iteration and exiting the loop when stop/abort is detected, instead of skipping items:

```python
while True:
    if agent_runner.was_aborted() or _should_stop_agent(
        agent_runner.run_context.context.event
    ):
        # no more audio chunks should be emitted after stop/abort
        break

    queue_item = await audio_queue.get()
    if queue_item is None:
        break

    text = None
    if isinstance(queue_item, tuple):
        text, audio_data = queue_item
    else:
        audio_data = queue_item

    ...
```

This keeps the behavior of “no further audio after stop”, but removes repeated branching inside the loop.

### 3. Reduce repeated `_should_stop_agent` checks in `_simulated_stream_tts`

Currently `_simulated_stream_tts` checks `_should_stop_agent` three times in the inner loop (before TTS, after TTS, before enqueuing audio). You can use an early‑exit + single mid‑iteration check:

```python
while True:
    if _should_stop_agent(astr_event):
        # stop/abort: bail out of the TTS loop
        break

    text = await text_queue.get()
    if text is None:
        break

    try:
        audio_path = await tts_provider.get_audio(text)
        if not audio_path or _should_stop_agent(astr_event):
            # skip emitting audio when stopped or when TTS fails
            continue

        with open(audio_path, "rb") as f:
            audio_data = f.read()

        astr_event.track_temporary_local_file(audio_path)
        await audio_queue.put((text, audio_data))
    except Exception as e:
        logger.error(
            f"[Live TTS Simulated] Error processing text '{text[:20]}...': {e}"
        )
```

This preserves the behavior of not emitting TTS after stop but makes the streaming/TTS pipeline more linear and easier to read.

### 4. Normalize stop checks in `_run_agent_feeder` final buffer

The final buffer condition combines several checks inline:

```python
# 处理剩余 buffer。若 stop 已到达，未送出的文本不能再进入 TTS。
if (
    buffer.strip()
    and not agent_runner.was_aborted()
    and not _should_stop_agent(agent_runner.run_context.context.event)
):
    await text_queue.put(buffer)
```

You can factor the stop logic into a small helper to make the intent clearer and reuse it if needed elsewhere:

```python
def _can_enqueue_tts_text(agent_runner, astr_event) -> bool:
    return (
        not agent_runner.was_aborted()
        and not _should_stop_agent(astr_event)
    )

...

if buffer.strip() and _can_enqueue_tts_text(
    agent_runner,
    agent_runner.run_context.context.event,
):
    await text_queue.put(buffer)
```

This keeps the “no new TTS after stop/abort” rule while reducing inline conditional complexity.

These small helpers/early‑exit patterns keep all the new stop/abort behaviors but consolidate the logic into clearer phases, which should address the complexity concerns without reverting any features.
</issue_to_address>

### Comment 3
<location path="astrbot/core/astr_agent_hooks.py" line_range="33" />
<code_context>
+    )
+
+
 class MainAgentHooks(BaseAgentRunHooks[AstrAgentContext]):
     async def on_agent_begin(
         self, run_context: ContextWrapper[AstrAgentContext]
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring `on_agent_done` to use a shared stop-request helper and split response preparation from hook emission for clearer, less fragmented control flow.

The new logic in `MainAgentHooks.on_agent_done` can be made simpler and more explicit without changing behavior.

### 1. Reuse a single “stop requested” helper

Instead of defining another `_event_requests_agent_stop` in this module, import and reuse the existing shared helper (e.g. from `context_utils`), so there is a single canonical policy:

```python
from astrbot.core.pipeline.context_utils import (
    call_event_hook,
    event_requests_agent_stop,  # or whatever the existing helper is named
)
```

Then drop the local `_event_requests_agent_stop` definition and use the shared one:

```python
async def on_agent_done(self, run_context, llm_response) -> None:
    event = run_context.context.event
    stop_requested = event_requests_agent_stop(event)
    ...
```

This reduces fragmentation and makes it clear where “agent stop requested” is defined.

### 2. Separate phases in `on_agent_done` and avoid reassigning `llm_response` inline

You can keep the same behavior but make the flow more obvious by splitting into small helpers: (a) prepare `llm_response` for hooks (including aborted response construction and `_llm_reasoning_content` handling), and (b) emit hooks.

For example:

```python
class MainAgentHooks(BaseAgentRunHooks[AstrAgentContext]):
    async def on_agent_done(self, run_context, llm_response) -> None:
        event = run_context.context.event
        stop_requested = event_requests_agent_stop(event)

        prepared_llm_response = self._prepare_llm_response_for_hooks(
            event, llm_response, stop_requested
        )

        await self._emit_llm_response_hook(event, prepared_llm_response, stop_requested)
        await call_event_hook(
            event,
            EventType.OnAgentDoneEvent,
            run_context,
            prepared_llm_response,
        )

    def _prepare_llm_response_for_hooks(self, event, llm_response, stop_requested):
        set_extra = getattr(event, "set_extra", None)

        # reasoning content handling
        if not stop_requested and llm_response and llm_response.reasoning_content:
            if callable(set_extra):
                set_extra("_llm_reasoning_content", llm_response.reasoning_content)
        elif stop_requested and callable(set_extra):
            set_extra("_llm_reasoning_content", None)

        # aborted response
        if stop_requested:
            return _build_aborted_llm_response(llm_response)
        return llm_response

    async def _emit_llm_response_hook(self, event, llm_response, stop_requested):
        if not stop_requested:
            await call_event_hook(
                event,
                EventType.OnLLMResponseEvent,
                llm_response,
            )
```

Key points:

- `stop_requested` is computed once and passed around, rather than calling `_event_requests_agent_stop` twice.
- `llm_response` isn’t reassigned mid‑method; a new `prepared_llm_response` is created instead.
- The policy is explicit:
  - `OnLLMResponseEvent` only runs when `stop_requested` is `False` and sees the original response.
  - `OnAgentDoneEvent` always runs and sees `prepared_llm_response` (possibly aborted).
- `_prepare_llm_response_for_hooks` encapsulates all reasoning content storage and aborted response building in one place, making it easier to reason about what hooks receive.
</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.}

[gemini-code-assist]

Code Review

This pull request enhances the agent's interruption and stop-request handling across the runner, hooks, and streaming utilities, ensuring that late-aborted results are safely discarded and model outputs are sanitized without leaking. It also adds extensive unit tests to validate these abort behaviors. The review feedback highlights two important improvements: first, replacing the use of id() with a list of references (_recorded_usages) to avoid token usage tracking omissions caused by Python's memory address reuse; second, strictly returning False for unrecognized message parts in _is_message_from_llm_response to prevent false positives when matching and popping messages.

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.