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

Aperta
  • Qualcosa non va?
#1,409 1 commento 0 reazioni 1 assegnatario Vedi su GitHub

I maintainer di solito rispondono entro 2 giorni

@rohityan ci sta già lavorando.

Dal 1/11/2026.

Valutazione

Questa issue non è ancora stata valutata.

Descrizione

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.

Lingua principale
Python
Stelle
2.2k
Fork
528
Merge medio
3g 17h
PR unite (30g)
43

Preparare l'ambiente

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di a2aproject/a2a-python

Tutte le issue di a2aproject/a2a-python

Issue simili

Altre issue su Python

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.