fix(a2a): cooperative cancellation checkpoints in execute_a2a_task (#16174)

autobot-backend/a2a/task_executor.py

@@ -28,6 +28,26 @@
 logger = get_logger(__name__)
 
 
+async def _is_cancelled(task_id: str, manager) -> bool:
+    """True once `task_id` has moved to CANCELLED underneath a running executor.
+
+    Checked at defined points inside `_execute_claimed` (#16174) -- never
+    inside a single call into the orchestrator, which this cannot interrupt
+    once it has started. Cancelling stops the NEXT checkpoint from doing more
+    work; it does not abort work already in flight.
+    """
+    task = manager.get_task(task_id)
+    return task is not None and task.status.state == TaskState.CANCELLED
+
+
+async def _abort_if_cancelled(task_id: str, manager, where: str) -> bool:
+    """A checkpoint: log and return True once cancellation should stop more work."""
+    if await _is_cancelled(task_id, manager):
+        logger.info("A2A task %s cancelled %s; no further work performed", task_id, where)
+        return True
+    return False
+
+
 def _extract_response_text(result: Dict[str, Any]) -> str:
     """Pull the human-readable response from an orchestrator result dict.
 
@@ -361,13 +381,24 @@ async def _execute_claimed(
     peer_id: str | None,
     manager,
 ) -> None:
-    """The original execution body, unchanged, now inside the task's claims."""
+    """The original execution body, now with cooperative-cancellation checkpoints (#16174).
+
+    A cancelled task's executor cannot be interrupted mid-call -- `cancel_task`
+    only flips a state in Redis -- so each checkpoint below only ever stops
+    the NEXT step from starting. `hold_scopes`'s `finally` releases the claim
+    as soon as this function returns, at whichever checkpoint that is.
+    """
     try:
+        if await _abort_if_cancelled(task_id, manager, "before scrub"):
+            return
         scrubbed = _scrub_inbound(task_id, input_text, peer_id, manager)
         if scrubbed is None:
             return
         input_text = scrubbed
 
+        if await _abort_if_cancelled(task_id, manager, "before orchestration"):
+            return
+
         # Late import to avoid circular deps at module load time
         from agents.agent_orchestration import get_distributed_agent_coordinator
 
@@ -377,6 +408,9 @@ async def _execute_claimed(
             context=context,
         )
 
+        if await _abort_if_cancelled(task_id, manager, "after orchestration"):
+            return
+
         artifacts = _store_response_artifacts(task_id, result, peer_id, manager)
         if artifacts is None:
             return

autobot-backend/a2a/task_executor_cancellation_test.py

@@ -0,0 +1,200 @@
+# Copyright 2025-2026 mrveiss
+# SPDX-License-Identifier: Apache-2.0
+# AutoBot - AI-Powered Automation Platform
+# Author: mrveiss
+"""Cooperative cancellation checkpoints in execute_a2a_task (#16174, #15950).
+
+`cancel_task` only flips a state in Redis -- it never interrupts a running
+executor. These tests drive `execute_a2a_task` against a real work-claim
+registry (fakeredis) and a fake `TaskManager` whose `get_task` reflects a
+state a test can flip mid-run, the same way a concurrent cancel request would.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+import pytest_asyncio
+
+from a2a.task_executor import execute_a2a_task
+from a2a.types import TaskState
+from autobot_shared.coordination.work_claims import list_claims
+
+
+class _FakeTaskManager:
+    """Just enough of TaskManager's surface for the executor to drive."""
+
+    def __init__(self, initial: TaskState = TaskState.SUBMITTED):
+        self.state = initial
+        self.artifacts: list = []
+        self.update_state_calls: list[tuple] = []
+
+    def get_task(self, task_id):
+        return SimpleNamespace(status=SimpleNamespace(state=self.state))
+
+    def update_state(self, task_id, state, message=None):
+        self.update_state_calls.append((state, message))
+        self.state = state
+
+    def publish_event(self, *_args, **_kwargs):
+        pass
+
+    def add_artifact(self, task_id, artifact):
+        self.artifacts.append(artifact)
+
+    def cancel(self):
+        self.state = TaskState.CANCELLED
+
+
+@pytest_asyncio.fixture
+async def claim_registry(monkeypatch):
+    """A real work-claim registry over an in-process Redis (see task_executor_trust_test.py)."""
+    fakeredis_async = pytest.importorskip("fakeredis.aioredis")
+    from autobot_shared.coordination import work_claims
+
+    client = fakeredis_async.FakeRedis(server=fakeredis_async.FakeServer(), decode_responses=True)
+    monkeypatch.setattr(work_claims, "get_async_redis_client", AsyncMock(return_value=client))
+    yield client
+    await client.flushall()
+
+
+def _scrub_ok(text: str = "ok"):
+    result = MagicMock()
+    result.text = text
+    result.redaction_count = 0
+    return result
+
+
+@pytest.mark.asyncio
+async def test_cancelled_before_orchestration_skips_the_orchestrator_call(claim_registry):
+    """AC1: a checkpoint before the expensive call stops it from ever starting."""
+    tm = _FakeTaskManager()
+    orchestrator = MagicMock()
+    orchestrator.process_request = AsyncMock(return_value={"response": "should never run"})
+
+    def _scrub_then_cancel(text, **_kwargs):
+        tm.cancel()
+        return _scrub_ok(text)
+
+    with (
+        patch("a2a.task_executor.get_task_manager", return_value=tm),
+        patch("a2a.task_executor.get_trust_manager", return_value=MagicMock()),
+        patch("a2a.task_executor.scrub_outbound", side_effect=_scrub_then_cancel),
+        patch("agents.agent_orchestration.get_distributed_agent_coordinator", return_value=orchestrator, create=True),
+    ):
+        await execute_a2a_task("task-cp1", "do something", context={"declared_scopes": ["path:a/b"]})
+
+    orchestrator.process_request.assert_not_called()
+    assert TaskState.COMPLETED not in [s for s, _m in tm.update_state_calls]
+    assert [c.scope for c in await list_claims()] == [], "scope must be released once the checkpoint aborts"
+
+
+@pytest.mark.asyncio
+async def test_already_cancelled_before_the_first_checkpoint_does_nothing(claim_registry):
+    """The "before scrub" checkpoint, not just the later ones (#16174 review).
+
+    Every other test in this file cancels mid-run, inside `scrub_outbound` or
+    `process_request`'s side effect -- so removing the *first* checkpoint
+    (the one before either runs) would leave them all green. `execute_a2a_task`
+    itself unconditionally moves the task to WORKING before anything else
+    (task_executor.py:98), so an initial CANCELLED state cannot survive to the
+    checkpoint -- cancelling from *inside* that update_state call is the only
+    way to have the task be CANCELLED by the time "before scrub" checks it.
+    """
+    tm = _FakeTaskManager()
+
+    def _cancel_once_working(task_id, state, message=None):
+        tm.update_state_calls.append((state, message))
+        tm.state = TaskState.CANCELLED if state == TaskState.WORKING else state
+
+    tm.update_state = _cancel_once_working
+    scrub = MagicMock(side_effect=AssertionError("scrub_outbound must not run once already cancelled"))
+    orchestrator = MagicMock()
+    orchestrator.process_request = AsyncMock(side_effect=AssertionError("process_request must not run"))
+
+    with (
+        patch("a2a.task_executor.get_task_manager", return_value=tm),
+        patch("a2a.task_executor.get_trust_manager", return_value=MagicMock()),
+        patch("a2a.task_executor.scrub_outbound", scrub),
+        patch("agents.agent_orchestration.get_distributed_agent_coordinator", return_value=orchestrator, create=True),
+    ):
+        await execute_a2a_task("task-cp0", "do something", context={"declared_scopes": ["path:a/b"]})
+
+    scrub.assert_not_called()
+    orchestrator.process_request.assert_not_called()
+    assert tm.artifacts == [], "no artifact must be stored once already cancelled"
+    assert [c.scope for c in await list_claims()] == [], "scope must be released, never having done any work"
+
+
+@pytest.mark.asyncio
+async def test_scope_stays_held_while_the_orchestrator_call_is_still_running(claim_registry):
+    """AC3: cancelling mid-call must not free the scope while work is still in flight."""
+    tm = _FakeTaskManager()
+    held_when_cancelled: list[str] = []
+
+    async def _process(*_args, **_kwargs):
+        tm.cancel()  # a cancel request arrives while this call is still running
+        held_when_cancelled.extend(c.scope for c in await list_claims())
+        return {"response": "still finishes this call"}
+
+    orchestrator = MagicMock()
+    orchestrator.process_request = AsyncMock(side_effect=_process)
+
+    with (
+        patch("a2a.task_executor.get_task_manager", return_value=tm),
+        patch("a2a.task_executor.get_trust_manager", return_value=MagicMock()),
+        patch("a2a.task_executor.scrub_outbound", return_value=_scrub_ok()),
+        patch("agents.agent_orchestration.get_distributed_agent_coordinator", return_value=orchestrator, create=True),
+    ):
+        await execute_a2a_task("task-cp2", "do something", context={"declared_scopes": ["path:a/b"]})
+
+    assert "path:a/b" in held_when_cancelled, "the scope must still be held while the orchestrator call is running"
+
+
+@pytest.mark.asyncio
+async def test_cancelled_during_orchestration_performs_no_further_work_and_releases(claim_registry):
+    """AC2: once the next checkpoint sees the cancellation, no more work runs and the scope frees."""
+    tm = _FakeTaskManager()
+
+    async def _process(*_args, **_kwargs):
+        tm.cancel()
+        return {"response": "produced but must be discarded"}
+
+    orchestrator = MagicMock()
+    orchestrator.process_request = AsyncMock(side_effect=_process)
+    eval_mock = AsyncMock()
+
+    with (
+        patch("a2a.task_executor.get_task_manager", return_value=tm),
+        patch("a2a.task_executor.get_trust_manager", return_value=MagicMock()),
+        patch("a2a.task_executor.scrub_outbound", return_value=_scrub_ok()),
+        patch("a2a.task_executor.evaluate_task_output", new=eval_mock),
+        patch("agents.agent_orchestration.get_distributed_agent_coordinator", return_value=orchestrator, create=True),
+    ):
+        await execute_a2a_task("task-cp3", "do something", context={"declared_scopes": ["path:a/b"]})
+
+    eval_mock.assert_not_called()
+    assert tm.artifacts == [], "no response/metadata artifact must be stored once cancelled"
+    assert TaskState.COMPLETED not in [s for s, _m in tm.update_state_calls]
+    assert [c.scope for c in await list_claims()] == [], "scope must be released once the checkpoint aborts"
+
+
+@pytest.mark.asyncio
+async def test_never_cancelled_still_completes_normally(claim_registry):
+    """Contrast: without a cancellation, the run reaches COMPLETED as before."""
+    tm = _FakeTaskManager()
+    orchestrator = MagicMock()
+    orchestrator.process_request = AsyncMock(return_value={"response": "hello"})
+    eval_pass = MagicMock(passed=True, confidence=0.9, eval_reason=None)
+
+    with (
+        patch("a2a.task_executor.get_task_manager", return_value=tm),
+        patch("a2a.task_executor.get_trust_manager", return_value=MagicMock()),
+        patch("a2a.task_executor.scrub_outbound", return_value=_scrub_ok("hello")),
+        patch("a2a.task_executor.evaluate_task_output", new=AsyncMock(return_value=eval_pass)),
+        patch("agents.agent_orchestration.get_distributed_agent_coordinator", return_value=orchestrator, create=True),
+    ):
+        await execute_a2a_task("task-cp4", "do something", context={"declared_scopes": ["path:a/b"]})
+
+    assert TaskState.COMPLETED in [s for s, _m in tm.update_state_calls]
+    assert [c.scope for c in await list_claims()] == [], "scope releases on normal completion too"

autobot-backend/agents/scope_enforcement.py

@@ -84,10 +84,11 @@ async def _renew_forever(scopes: Sequence[str], *, agent_id: str, task_id: str,
     TTL, which bounds a cancelled task's hold without ever unlocking a scope
     while work continues.
 
-    That is a bound, not a release, and #15950's cancellation criterion is left
-    unticked because of it. Cooperative cancellation in the executor is the real
-    fix (#16174); this path stays afterwards as the backstop for an executor
-    wedged before it reaches any checkpoint.
+    That is a bound, not a release -- which is why #16174 made `execute_a2a_task`
+    check for cancellation at defined checkpoints and stop there instead of
+    relying on this loop alone. This path remains as the backstop for an
+    executor wedged before it reaches any checkpoint: cooperative cancellation
+    covers the common case, this covers the one it structurally cannot.
     """
     from autobot_shared.coordination.work_claims import renew
 

changelog/unreleased/16174-a2a-cancel-cooperative.md

@@ -0,0 +1,7 @@
+---
+type: fix
+scope: a2a
+issue: 16174
+pr: 0
+---
+Cancelling an A2A task now stops its executor at defined checkpoints instead of only bounding an orphaned work-claim at its TTL: `execute_a2a_task` checks for cancellation before scrubbing, before calling the orchestrator, and after it returns, so a cancelled task performs no further work and releases its claim as soon as the next checkpoint sees the cancellation — without ever unlocking a scope while the work holding it is still writing.