[Bug]: Blocking continuation returns stale INPUT_REQUIRED when executor emits current task

Problem

Reproduced with a2a-sdk==1.3.0 on Python 3.13. A blocking DefaultRequestHandler.on_message_send() continuation returns the old TASK_STATE_INPUT_REQUIRED snapshot when the executor enqueues context.current_task before emitting this turn's WORKING/artifact/COMPLETED events. The same task later reads as COMPLETED with its artifact. return_immediately is unset/false.

Both task_id and context_id are supplied, so this is separate from #1206. Related streaming/history behavior is tracked in #1285.

Reproduction

Install a2a-sdk==1.3.0, then run this self-contained script:

import asyncio
from uuid import uuid4
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 import AgentCard, AgentCapabilities, AgentInterface, AgentSkill, Message, Part, SendMessageRequest, Task, TaskState, TaskStatus, GetTaskRequest

class PausingAgent(AgentExecutor):
    async def execute(self, context, queue):
        await queue.enqueue_event(context.current_task or Task(id=context.task_id,
            context_id=context.context_id, status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED),
            history=[context.message]))
        updater = TaskUpdater(queue, context.task_id, context.context_id)
        if context.current_task is None:
            await updater.requires_input(message=updater.new_agent_message([Part(text='Which region?')]))
            return
        await updater.start_work()
        await asyncio.sleep(.05)
        await updater.add_artifact([Part(text='EU result')], name='result', last_chunk=True)
        await updater.complete()

    async def cancel(self, context, queue):
        await TaskUpdater(queue, context.task_id, context.context_id).cancel()

async def main():
    card = AgentCard(name='Repro', description='Blocking continuation', version='1',
        capabilities=AgentCapabilities(streaming=True),
        default_input_modes=['text/plain'], default_output_modes=['text/plain'],
        supported_interfaces=[AgentInterface(url='http://localhost/a2a', protocol_binding='JSONRPC', protocol_version='1.0')],
        skills=[AgentSkill(id='repro', name='Repro', description='Pause and resume', tags=['repro'])])
    handler = DefaultRequestHandler(agent_executor=PausingAgent(), task_store=InMemoryTaskStore(), agent_card=card)
    context = ServerCallContext()
    def request(text, **ids):
        return SendMessageRequest(message=Message(message_id=str(uuid4()), role=1,
            parts=[Part(text=text)], **ids))
    try:
        first = await handler.on_message_send(request('Analyze'), context)
        print('first:', TaskState.Name(first.status.state))
        second = await handler.on_message_send(request('EU', task_id=first.id, context_id=first.context_id), context)
        print('blocking continuation:', TaskState.Name(second.status.state))
        await asyncio.sleep(.15)
        canonical = await handler.on_get_task(GetTaskRequest(id=first.id), context)
        print('canonical after .15s:', TaskState.Name(canonical.status.state), 'artifacts:', len(canonical.artifacts))
        assert second.status.state == TaskState.TASK_STATE_INPUT_REQUIRED
        assert canonical.status.state == TaskState.TASK_STATE_COMPLETED
    finally:
        await handler.aclose()

asyncio.run(main())

Observed:

first: TASK_STATE_INPUT_REQUIRED
blocking continuation: TASK_STATE_INPUT_REQUIRED
canonical after .15s: TASK_STATE_COMPLETED artifacts: 1

Expected behavior

The blocking continuation should observe this execution through its terminal/interrupted state and return COMPLETED with the result, rather than satisfying the wait from a replay of the previous paused snapshot. Publishing the current task before status updates is useful for an initial stream snapshot; it should not make blocking mode return a stale answer.

Relevant code and workaround

In default_request_handler_v2.py:on_message_send, the first Task in a terminal/interrupted state immediately satisfies the blocking wait. The reproducer's initial current-task event is already INPUT_REQUIRED. Our application workaround drains super().on_message_send_stream() for the continuation and then reads the canonical task with history projection. A regression should cover INPUT_REQUIRED (and AUTH_REQUIRED where applicable), initial-current-task emission, blocking mode, and explicit task/context IDs.

I searched for the exact blocking behavior before filing; #1206 and #1285 cover related but distinct cases. This report claims reproduction on 1.3.0, not validation against unreleased main.