[Bug]: ParallelToolExecutor starts a pending tool after cancellation while waiting for a resource lock

Is there an existing issue for the same bug?

• I have searched existing issues and this is not a duplicate.

Bug Description

ParallelToolExecutor.execute_batch can start a pending tool after its cancellation token becomes cancelled while the call is waiting for a declared-resource lock.

The executor checks cancel_token.is_cancelled before resource resolution. If the token is still active at that point, the call can then block while acquiring a resource lock. When cancellation occurs during that wait, the executor does not re-check the token after acquiring the lock and invokes tool_runner.

This conflicts with the documented behavior that pending tool calls are skipped after cancellation and that a call cancelled before the tool begins should return a synthetic cancellation error.

Consequently, a queued file, terminal, browser, or other resource-using operation can begin after the user has interrupted the run.

Expected Behavior

If the cancellation token becomes cancelled before tool_runner begins, the pending call should not execute after its resource lock becomes available.

The executor should return an AgentErrorEvent containing the existing cancellation diagnostic:

Tool call cancelled by interrupt.

The behavior of tools that began running before cancellation should remain unchanged.

Actual Behavior

The lock-waiting call invokes tool_runner after the lock is released, even though:

cancel_token.is_cancelled is True

It returns the normal tool output rather than the executor's synthetic cancellation result.

The behavior reproduces on both v1.44.1 and current main.

Steps to Reproduce

1. Check out release v1.44.1.

2. Set up the development environment:

make build

3. Create tests/sdk/agent/test_parallel_executor_cancel_wait.py:

import threading
import time
from unittest.mock import MagicMock

from openhands.sdk.agent.parallel_executor import ParallelToolExecutor
from openhands.sdk.conversation.cancellation import CancellationToken
from openhands.sdk.conversation.resource_lock_manager import (
    ResourceLockManager,
)
from openhands.sdk.event import ActionEvent, AgentErrorEvent
from openhands.sdk.llm import MessageToolCall, TextContent
from openhands.sdk.tool.schema import Action
from openhands.sdk.tool.tool import DeclaredResources


class ProbeAction(Action):
    pass


def test_cancelled_tool_does_not_start_after_resource_lock_wait():
    resource_key = "file:/tmp/cancelled-waiter"
    lock_manager = ResourceLockManager(
        timeouts={"file": 2.0},
    )
    executor = ParallelToolExecutor(
        max_workers=2,
        lock_manager=lock_manager,
    )
    cancel_token = CancellationToken()
    resources_resolved = threading.Event()
    runner_called = threading.Event()

    parsed_action = ProbeAction()
    action = ActionEvent(
        thought=[TextContent(text="test")],
        action=parsed_action,
        tool_name="editor",
        tool_call_id="call-1",
        tool_call=MessageToolCall(
            id="call-1",
            name="editor",
            arguments="{}",
            origin="completion",
        ),
        llm_response_id="response-1",
    )

    tool = MagicMock()
    tool.name = "editor"

    def declared_resources(received):
        assert received is parsed_action
        resources_resolved.set()
        return DeclaredResources(
            keys=(resource_key,),
            declared=True,
        )

    tool.declared_resources = declared_resources
    tool_output = MagicMock(name="tool_output")

    def tool_runner(received):
        assert received is action
        runner_called.set()
        return [tool_output]

    result_holder = {}

    def execute():
        result_holder["result"] = executor.execute_batch(
            [action],
            tool_runner,
            {"editor": tool},
            cancel_token,
        )

    worker = threading.Thread(target=execute)

    with lock_manager.lock(resource_key):
        worker.start()
        assert resources_resolved.wait(timeout=1)
  
        # Allow the worker to enter the FIFO lock wait.
        time.sleep(0.05)
        assert worker.is_alive()
  
        cancel_token.cancel()
  
    worker.join(timeout=1)
    assert not worker.is_alive()

    result = result_holder["result"][0]

    assert not runner_called.is_set(), (
        "tool_runner was invoked after cancellation"
    )
    assert len(result) == 1
    assert isinstance(result[0], AgentErrorEvent)

4. Run:

uv run pytest \
    tests/sdk/agent/test_parallel_executor_cancel_wait.py \
    -q

5. Observe that the test fails because tool_runner is invoked after cancellation.

Acceptance Criteria

• A call cancelled while waiting for a resource lock does not invoke tool_runner.
• The call returns the existing synthetic cancellation AgentErrorEvent.
• Result ordering remains unchanged.
• Calls without cancellation continue after acquiring their locks.
• Calls cancelled before entering the executor remain skipped.
• Synchronous and asynchronous batch execution follow the same cancellation behavior.
• Regression coverage uses a real CancellationToken and ResourceLockManager.

Installation Method

Source checkout using make build (uv sync --dev).

If you selected "Other", please specify

Not applicable.

SDK Version

1.44.1

Version Confirmation

• I have confirmed this bug exists on the LATEST version of OpenHands SDK

Python Version

3.13.2

Model Name (if applicable)

Not applicable; reproduced through deterministic local tool-executor code.

Operating System

MacOS

Logs and Error Messages

v1.44.1@9d143aac35c2dcec9cbb046ff9f35ac5eb072f6a:
cancelled=True
runner_called=True
returned_tool_output=True

main@704cbe6015e3d59cabe04632175d99df2d448999:
cancelled=True
runner_called=True
returned_tool_output=True

Representative failure:

AssertionError: tool_runner was invoked after cancellation
assert not True

The returned batch contains the normal tool_runner output instead of an AgentErrorEvent.

Minimal Code Sample

import threading
from unittest.mock import MagicMock

from openhands.sdk.agent.parallel_executor import ParallelToolExecutor
from openhands.sdk.conversation.cancellation import CancellationToken
from openhands.sdk.conversation.resource_lock_manager import (
    ResourceLockManager,
)
from openhands.sdk.event import ActionEvent, AgentErrorEvent
from openhands.sdk.llm import MessageToolCall, TextContent
from openhands.sdk.tool.schema import Action
from openhands.sdk.tool.tool import DeclaredResources


class ProbeAction(Action):
    pass


def test_cancelled_tool_does_not_start_after_resource_lock_wait():
    resource_key = "file:/tmp/cancelled-waiter"
    lock_manager = ResourceLockManager(
        timeouts={"file": 2.0},
    )
    executor = ParallelToolExecutor(
        max_workers=2,
        lock_manager=lock_manager,
    )
    cancel_token = CancellationToken()
    resources_resolved = threading.Event()
    runner_called = threading.Event()

    parsed_action = ProbeAction()
    action = ActionEvent(
        thought=[TextContent(text="test")],
        action=parsed_action,
        tool_name="editor",
        tool_call_id="call-1",
        tool_call=MessageToolCall(
            id="call-1",
            name="editor",
            arguments="{}",
            origin="completion",
        ),
        llm_response_id="response-1",
    )

    tool = MagicMock()
    tool.name = "editor"

    def declared_resources(received):
        assert received is parsed_action
        resources_resolved.set()
        return DeclaredResources(
            keys=(resource_key,),
            declared=True,
        )

    tool.declared_resources = declared_resources
    tool_output = MagicMock(name="tool_output")

    def tool_runner(received):
        assert received is action
        runner_called.set()
        return [tool_output]

    result_holder = {}

    def execute():
        result_holder["result"] = executor.execute_batch(
            [action],
            tool_runner,
            {"editor": tool},
            cancel_token,
        )

    worker = threading.Thread(target=execute)

    with lock_manager.lock(resource_key):
        worker.start()
        assert resources_resolved.wait(timeout=1)
        cancel_token.cancel()

    worker.join(timeout=1)
    assert not worker.is_alive()

    result = result_holder["result"][0]

    assert not runner_called.is_set(), (
        "tool_runner was invoked after cancellation"
    )
    assert len(result) == 1
    assert isinstance(result[0], AgentErrorEvent)

Screenshots and Additional Context

No screenshot is required; this is a deterministic concurrency reproduction using the SDK's actual cancellation token and resource lock manager.

The current implementation checks cancellation only before resource resolution:

if cancel_token is not None and cancel_token.is_cancelled:
    return self._cancelled_error(action, span_owner)

It later acquires the resource lock and calls the tool without another check:

resources = self._extract_declared_resources(action, tool)
lock_keys = self._resolve_lock_keys(resources, tool)

if not lock_keys:
    return tool_runner(action)

with self._lock_manager.lock(*lock_keys):
    return tool_runner(action)

The token can therefore change from active to cancelled while the thread is blocked in ResourceLockManager.lock, but that state change is ignored once the lock becomes available.

This also affects aexecute_batch, because its worker path delegates to the same _run_safe implementation.

A possible fix is to re-check the cancellation token immediately before each tool_runner invocation, especially after acquiring a blocking resource lock:

with self._lock_manager.lock(*lock_keys):
    if cancel_token is not None and cancel_token.is_cancelled:
        return self._cancelled_error(
            action,
            span_owner,
        )
    return tool_runner(action)

Regression coverage should include:

• cancellation already set before execution;
• cancellation while waiting for a declared resource;
• cancellation while waiting for the fallback tool-wide mutex;
• execution without cancellation;
• synchronous and asynchronous batch entry points;
• preservation of cancellation result ordering.

The implementation is unchanged through:

main@704cbe6015e3d59cabe04632175d99df2d448999

Start in openhands/sdk/agent/parallel_executor.py, especially the _run_safe path used by execute_batch and aexecute_batch, and review the existing cancellation result handling. Run the proposed regression test at tests/sdk/agent/test_parallel_executor_cancel_wait.py with uv run pytest. Done means lock-waiting calls return the existing AgentErrorEvent without invoking tool_runner, while ordering and non-cancelled behavior remain unchanged.