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

Ouverte
  • Quelque chose ne va pas ?
#1,409 1 commentaire 0 réactions 1 personne assignée Voir sur GitHub

Les mainteneurs répondent en général sous 2 jours

@rohityan y travaille déjà.

Depuis le 1/11/2026.

Évaluation

Cette issue n'a pas encore été évaluée.

Description

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.

Langage dominant
Python
Étoiles
2.2k
Forks
528
Merge moyen
3 j 17 h
PR mergées (30 j)
43

Préparer son environnement

Par où commencer

  1. Lisez l'issue en entier, puis le guide de contribution du projet.
  2. Signalez en commentaire que vous la prenez — cela évite que deux personnes fassent le même travail.
  3. Forkez le dépôt et travaillez sur une branche.
  4. Ouvrez une pull request qui référence le numéro de l'issue.

Autres issues de a2aproject/a2a-python

Toutes les issues de a2aproject/a2a-python

Issues similaires

Plus d'issues Python

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.