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

Aberta
  • Algo errado?
#1,409 1 comentário 0 reações 1 responsável Ver no GitHub

Mantenedores costumam responder em até 2 dias

@rohityan já está trabalhando nisso.

Desde 1/11/2026.

Avaliação

Esta issue ainda não foi avaliada.

Descrição

component: server status:awaiting response
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.

Linguagem predominante
Python
Estrelas
2.2k
Forks
532
Merge médio
3d 11h
PRs com merge (30d)
43

Preparar o ambiente

Primeiros passos

  1. Leia a issue inteira e depois o guia de contribuição do projeto.
  2. Comente na issue dizendo que vai assumir — evita que duas pessoas façam o mesmo trabalho.
  3. Faça um fork do repositório e trabalhe em uma branch.
  4. Abra um pull request que referencie o número da issue.

Mais de a2aproject/a2a-python

Todas as issues de a2aproject/a2a-python

Issues semelhantes

Mais issues de Python

Receba novas issues na sua caixa de entrada

Um resumo curto de issues do GitHub para quem está começando.