Please address the comments from this code review:

## Individual Comments

### Comment 1
<location path="astrbot/core/agent/runners/tool_loop_agent_runner.py" line_range="1008" />
<code_context>
+            data=AgentResponseData(chain=MessageChain(type="aborted")),
+        )
+
+    async def _cancel_tool_iteration(
+        self,
+        executor: T.Any,
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the new tool-executor abort logic by replacing the polling-based `_iter_tool_executor_results`/`_cancel_tool_iteration` pair with a single helper that races `anext(executor)` directly against the abort signal.

The main extra complexity comes from `_iter_tool_executor_results` + `_cancel_tool_iteration` and the time-based polling. You can keep all the new abort behavior while simplifying the control flow by:

1. Removing the polling interval and loop around `asyncio.wait`.
2. Racing a single `anext(executor)` task against `self._abort_signal.wait()` directly.
3. Inlining the cancellation into `_iter_tool_executor_results` so `_cancel_tool_iteration` is no longer needed.

That keeps explicit cancellation of a long-running `anext(executor)` without the nested polling loops or the `tool_executor_poll_interval` knob.

### Suggested refactor

First, drop the poll-interval configuration and `_cancel_tool_iteration`:

```python
# __init__ / __post_init__:
self._abort_signal = asyncio.Event()
# Remove:
# self._tool_executor_poll_interval = ...
```

Remove `_cancel_tool_iteration` entirely.

Then rewrite `_iter_tool_executor_results` to use a single race between the next result and the abort signal:

```python
async def _iter_tool_executor_results(
    self,
    executor: T.Any,
) -> T.AsyncGenerator[T.Any, None]:
    while True:
        if self._is_stop_requested():
            # Ensure executor is closed if we haven't started a next_result_task yet
            close_executor = getattr(executor, "aclose", None)
            if close_executor is not None:
                with suppress(asyncio.CancelledError, RuntimeError, StopAsyncIteration):
                    await close_executor()
            raise _ToolExecutionInterrupted(
                "Tool execution interrupted before reading the next tool result."
            )

        next_result_task = asyncio.create_task(anext(executor))
        abort_task = asyncio.create_task(self._abort_signal.wait())
        try:
            done, _ = await asyncio.wait(
                {next_result_task, abort_task},
                return_when=asyncio.FIRST_COMPLETED,
            )

            if abort_task in done:
                # Stop requested: cancel the in-flight anext and close executor
                if not next_result_task.done():
                    next_result_task.cancel()
                    with suppress(asyncio.CancelledError, StopAsyncIteration):
                        await next_result_task

                close_executor = getattr(executor, "aclose", None)
                if close_executor is not None:
                    with suppress(asyncio.CancelledError, RuntimeError, StopAsyncIteration):
                        await close_executor()

                raise _ToolExecutionInterrupted("Tool execution interrupted by a stop request.")

            # Normal path: next_result_task finished first
            try:
                yield next_result_task.result()
            except StopAsyncIteration:
                return
        finally:
            # Best-effort cleanup of the abort task
            if not abort_task.done():
                abort_task.cancel()
                with suppress(asyncio.CancelledError):
                    await abort_task
```

This preserves:

- Pre-tool-start interruption (still handled in the `for func_tool_name, ...` loop via `_ToolExecutionInterrupted`).
- Mid-tool-call interruption (explicitly cancels `anext(executor)` and calls `aclose()`).
- Centralized stop handling via `_finalize_aborted_step` and `_ToolExecutionInterrupted`.

But it:

- Removes the extra `tool_executor_poll_interval` configuration and associated logic.
- Avoids nested polling loops and multiple calls to `_cancel_tool_iteration`.
- Keeps cancellation logic local to a single, linear helper that’s easier to reason about.
</issue_to_address>

Please address the comments from this code review:

## Overall Comments
- In `_handle_function_tools`, the `except _ToolExecutionInterrupted: raise` block is redundant since the generic code path already lets this exception propagate; you can simplify by removing that explicit catch-and-rethrow.
- The executor shutdown logic in `_iter_tool_executor_results` is duplicated in both the early-stop branch at the top of the loop and the `abort_task`-completed branch; consider extracting a small helper (e.g. `_close_executor(executor)`) to keep this behavior consistent and easier to maintain.

## Individual Comments

### Comment 1
<location path="astrbot/core/agent/runners/tool_loop_agent_runner.py" line_range="996" />
<code_context>
+            data=AgentResponseData(chain=MessageChain(type="aborted")),
+        )
+
+    async def _iter_tool_executor_results(
+        self,
+        executor: T.Any,
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring the new stop-handling logic to centralize executor cleanup and rely on a single interruption mechanism to make the control flow easier to follow and maintain.

You can simplify the new stop-handling without changing behavior by:

1. **Centralizing executor cleanup in a small helper**

Right now `_iter_tool_executor_results` duplicates the `aclose` + `suppress(...)` logic in two places. A small helper makes the control flow clearer and easier to maintain:

```python
async def _close_executor(self, executor: T.Any) -> None:
    close_executor = getattr(executor, "aclose", None)
    if close_executor is None:
        return
    with suppress(asyncio.CancelledError, RuntimeError, StopAsyncIteration):
        await close_executor()
```

Then `_iter_tool_executor_results` can reuse this:

```python
async def _iter_tool_executor_results(
    self,
    executor: T.Any,
) -> T.AsyncGenerator[T.Any, None]:
    while True:
        if self._is_stop_requested():
            await self._close_executor(executor)
            raise _ToolExecutionInterrupted(
                "Tool execution interrupted before reading the next tool result."
            )

        next_result_task = asyncio.create_task(anext(executor))
        abort_task = asyncio.create_task(self._abort_signal.wait())
        try:
            done, _ = await asyncio.wait(
                {next_result_task, abort_task},
                return_when=asyncio.FIRST_COMPLETED,
            )

            if abort_task in done:
                if not next_result_task.done():
                    next_result_task.cancel()
                    with suppress(asyncio.CancelledError, StopAsyncIteration):
                        await next_result_task

                await self._close_executor(executor)
                raise _ToolExecutionInterrupted(
                    "Tool execution interrupted by a stop request."
                )

            try:
                yield next_result_task.result()
            except StopAsyncIteration:
                return
        finally:
            if not abort_task.done():
                abort_task.cancel()
                with suppress(asyncio.CancelledError):
                    await abort_task
```

This keeps the same semantics but removes duplicated cleanup code and narrows the suppression to one place.

2. **Use a single control-flow path for tool interruption**

You currently mix:

- explicit `_is_stop_requested()` check before the tool loop, and
- `_ToolExecutionInterrupted` raised inside `_iter_tool_executor_results`, and
- a separate `_is_stop_requested()` check *after* the `async for` in `step`.

If you want `_ToolExecutionInterrupted` to be the primitive for tool interruption, you can remove the pre/post checks and rely on the exception only. For example:

```python
async def _handle_function_tools(...):
    for func_tool_name, func_tool_args, func_tool_id in zip(...):
        # Remove this outer check:
        # if self._is_stop_requested():
        #     raise _ToolExecutionInterrupted(...)

        ...
        try:
            async for resp in self._iter_tool_executor_results(executor):
                ...
        except _ToolExecutionInterrupted:
            raise
        except Exception as e:
            ...
```

And in `step` you only handle the exception, without an extra `_is_stop_requested()` guard:

```python
if llm_resp.tools_call_name:
    if self.tool_schema_mode == "skills_like":
        llm_resp, _ = await self._resolve_tool_exec(llm_resp)

    tool_call_result_blocks = []
    cached_images = []
    try:
        async for result in self._handle_function_tools(self.req, llm_resp):
            ...
    except _ToolExecutionInterrupted:
        yield await self._finalize_aborted_step(llm_resp)
        return

    # Remove this:
    # if self._is_stop_requested():
    #     yield await self._finalize_aborted_step(llm_resp)
    #     return
```

This keeps the stop feature intact but reduces the number of ways interruption can affect control flow, making it easier to reason about when/where tool execution exits.
</issue_to_address>

Please address the comments from this code review:

## Overall Comments
- Consider extracting the repeated interruption message string ("[SYSTEM: User actively interrupted the response generation. Partial output before interruption is preserved.]") into a module-level constant so the behavior and wording remain consistent if it needs to be updated later.
- The `_iter_tool_executor_results` helper currently takes `executor: Any`; you could tighten this to an `AsyncIterator`/`AsyncGenerator` type (and possibly make the helper generic over the yielded type) to improve type safety and clarity around the expected executor interface.