Skip to content

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

Open

Description

@sophgen

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.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Labels

No labels
No labels

Type

No type

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions