[Bug]: cancellation before concurrent toolkit entry can still execute the tool #2767

Prerequisites

• Existing issues and discussions have been searched for this problem.
• This is a bug report, not a usage question.

Background / Description

Cancelling a reply while a concurrency-safe tool is waiting in on_acting middleware can cause the tool to run after the cancellation. The reply then ends with finished_reason="completed".

The timing matters: the middleware has started, but has not yet forwarded the call to toolkit.call_tool. A middleware that waits for a rate-limit slot or performs asynchronous setup can reach this point during normal use.

The same reproduction with a sequential tool stops without executing the tool and ends as interrupted. The difference is the tool's is_concurrency_safe flag.

Error Messages

No exception reaches the default reply consumer. The observable result is:

The second middleware entry is useful evidence: this is not simply a tool that was already running and finished too late. The pending call is entered again after cancellation.

Steps to Reproduce

Run the following against the base commit below. The model is an offline double and the tool only appends to a local list. An asyncio.Event makes the cancellation point deterministic; there are no timing sleeps or provider calls.

import asyncio
from types import SimpleNamespace

from agentscope.agent import Agent, InjectionConfig, ReActConfig
from agentscope.event import ReplyEndEvent
from agentscope.message import TextBlock, ToolCallBlock, UserMsg
from agentscope.middleware import MiddlewareBase
from agentscope.model import ChatResponse
from agentscope.tool import FunctionTool, Toolkit, ToolChunk

class Model:
    model = "offline"
    context_size = 1000000
    formatter = SimpleNamespace(supported_input_media_types=[])

    def __init__(self):
        self.calls = 0

    async def count_tokens(self, *args, **kwargs):
        return 1

    async def __call__(self, **kwargs):
        self.calls += 1
        content = (
            [ToolCallBlock(id="call-work", name="work", input="{}")]
            if self.calls == 1 else [TextBlock(text="done")]
        )
        return ChatResponse(content=content, is_last=True)

class WaitBeforeTool(MiddlewareBase):
    def __init__(self):
        self.entered = asyncio.Event()
        self.calls = 0

    async def on_acting(self, agent, input_kwargs, next_handler):
        self.calls += 1
        if self.calls == 1:
            self.entered.set()
            await asyncio.Event().wait()
        async for event in next_handler(**input_kwargs):
            yield event

async def run(concurrent):
    executed, events = [], []

    async def work() -> ToolChunk:
        """Record a local action."""
        executed.append("ran")
        return ToolChunk(content=[TextBlock(text="done")])

    middleware = WaitBeforeTool()
    agent = Agent(
        name="worker", system_prompt="test", model=Model(),
        injection_config=InjectionConfig(inject_runtime_state=False),
        react_config=ReActConfig(max_iters=3),
        toolkit=Toolkit(tools=[FunctionTool(
            work, is_concurrency_safe=concurrent, is_read_only=True,
        )]),
        middlewares=[middleware],
    )

    async def consume():
        async for event in agent.reply_stream(UserMsg("user", "perform work")):
            events.append(event)

    task = asyncio.create_task(consume())
    await asyncio.wait_for(middleware.entered.wait(), 5)
    task.cancel()
    await asyncio.wait_for(task, 5)
    ends = [e.finished_reason for e in events if isinstance(e, ReplyEndEvent)]
    print(concurrent, middleware.calls, executed, ends)
    assert executed == []
    assert ends == ["interrupted"]

async def main():
    await run(False)
    await run(True)

asyncio.run(main())

The sequential case passes; the concurrent case fails the executed == [] assertion.

The behavior was also reproduced by POSTing to /sessions/{session_id}/interrupt through an in-process HTTP client. That check used the real session router, ChatService.interrupt, message bus, cancellation dispatcher and run registry; authentication, session lookup and model output were fixtures. Both requests returned 202. After the tasks finished, the concurrent reply had still executed the tool and ended as completed. The observation concerns the final state, not a requirement that HTTP 202 mean synchronous completion.

Cause and proposed fix

_execute_concurrent_tool_calls() catches CancelledError, cancels its workers and drains their queued events, then calls uncancel() and returns. The caller depends on an interrupted tool-result event to recognize the cancellation. That event cannot exist yet when the middleware was cancelled before entering the toolkit. The tool call remains eligible to run, so the reply loop starts it again.

The proposed change keeps the worker cleanup and event draining, then propagates cancellation to the reply loop so its existing interruption handling closes unfinished calls. Regression tests should cover both cancellation-propagation settings, cancellation before and during tool execution, and a mixed batch with a completed sibling.

Related work

#2650 handles cancellation propagation after an interrupted model/tool event has been received. This report covers an earlier window where no such event has been emitted, and it reproduces on a base commit that includes #2650. #2671 concerns the RealtimeAgent output reader; #1905 concerns orphaned tool-call recovery. Neither covers the same-run re-execution here.

Environment

• AgentScope: source checkout at 00d295b9d7232edf2034f25a2bd9c94016bb3880 (main, version field 2.0.8).
• Python: 3.12.14.
• OS: Windows.
• No external model or tool service is needed.

A focused fix and regression tests are being prepared for the linked draft PR. Feedback on the cancellation and event-cleanup contract is welcome before merging.