Hacktoberfest 2026: los issues que los mantenedores marcaron para octubre, abiertos y aptos para principiantes. Explorar issues de Hacktoberfest

[Bug]: Push notification store failure rewrites a completed task as FAILED (DefaultRequestHandlerV2)

Abierto
  • ¿Algo no está bien?
#1,313 1 comentario 0 reacciones 1 asignado Ver en GitHub

Los mantenedores suelen responder en 2 días

@rohityan ya está trabajando en esto.

Desde el 5/10/2026.

Evaluación

Este issue todavía no se ha evaluado.

Descripción

component: server status:awaiting response
What happened?

In DefaultRequestHandlerV2, a failure in the push-notification infrastructure (not in the agent) rewrites an otherwise successful task to TASK_STATE_FAILED and surfaces an error to the client instead of the task result.

In the reproduction below, the agent does everything right — it creates the task and completes it via TaskUpdater (which stamps status.timestamp). The only injected failure is a transient exception from PushNotificationConfigStore.get_info_for_dispatch (e.g. a database hiccup while reading webhook configs). Observed:

--- CONTROL: healthy push config store ---
  on_message_send -> Task (state=TASK_STATE_COMPLETED)
  persisted task state: TASK_STATE_COMPLETED, timestamp set: True
--- BUG CASE: config store read fails ---
  on_message_send RAISED RuntimeError: transient DB error while reading push configs
  persisted task state: TASK_STATE_FAILED, timestamp set: False

The task the agent completed is persisted as FAILED, and message/send raises instead of returning the Task.

Root cause

Two layers of missing error boundary:

  1. BasePushNotificationSender.send_notification (src/a2a/server/tasks/base_push_notification_sender.py): get_info_for_dispatch (line 72) and the push_url_validator call (lines 93–98) run outside the try: that starts at line 99. The dispatch path itself is properly guarded — _dispatch_notification catches everything and returns bool (lines 127–133) — so the design intent is clearly "push failures must not affect the task", but the config-read and validator escapes that boundary.
  2. EventConsumer._update_task_state (src/a2a/server/agent_execution/active_task.py:345-350) awaits send_notification inline without any guard. It is called from _process_event_inner (line 235); any exception propagates to the consumer's generic except Exception (lines 139–153), which calls _write_failed_status() — rewriting the task to TASK_STATE_FAILED and enqueueing the error to subscribers.

Secondary observation (visible in the repro output): the framework-written FAILED status carries no timestamp, which also nulls last_updated for DatabaseTaskStore rows.

Reproduction
Self-contained script (a2a-sdk @ main e649325, Python 3.10) 
import asyncio

import httpx

from a2a.auth.user import UnauthenticatedUser
from a2a.helpers.proto_helpers import new_task_from_user_message
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.context import ServerCallContext
from a2a.server.events import EventQueue
from a2a.server.request_handlers import DefaultRequestHandlerV2
from a2a.server.tasks import (
    InMemoryPushNotificationConfigStore,
    InMemoryTaskStore,
    TaskUpdater,
)
from a2a.server.tasks.base_push_notification_sender import (
    BasePushNotificationSender,
)
from a2a.types.a2a_pb2 import (
    AgentCapabilities,
    AgentCard,
    Message,
    Part,
    Role,
    SendMessageConfiguration,
    SendMessageRequest,
    TaskState,
)


class CompletingExecutor(AgentExecutor):
    # The agent does everything right: creates the task, then completes it
    # via TaskUpdater (which stamps status.timestamp).
    task_id = None

    async def execute(self, context, event_queue):
        task = new_task_from_user_message(context.message)
        self.task_id = task.id
        await event_queue.enqueue_event(task)
        updater = TaskUpdater(event_queue, task.id, task.context_id)
        await updater.update_status(TaskState.TASK_STATE_COMPLETED)

    async def cancel(self, context, event_queue):
        pass


class FlakyConfigStore(InMemoryPushNotificationConfigStore):
    # Simulates a transient infrastructure failure on the dispatch read path.
    async def get_info_for_dispatch(self, task_id):
        raise RuntimeError('transient DB error while reading push configs')


def build_handler(store):
    executor = CompletingExecutor()
    handler = DefaultRequestHandlerV2(
        agent_executor=executor,
        task_store=InMemoryTaskStore(),
        push_config_store=store,
        push_sender=BasePushNotificationSender(
            httpx_client=httpx.AsyncClient(), config_store=store
        ),
        agent_card=AgentCard(
            name='t',
            version='1.0',
            capabilities=AgentCapabilities(
                streaming=True, push_notifications=True
            ),
        ),
    )
    return handler, executor


async def run_case(label, store):
    handler, executor = build_handler(store)
    params = SendMessageRequest(
        message=Message(
            role=Role.ROLE_USER, message_id='m1', parts=[Part(text='Hi')]
        ),
        configuration=SendMessageConfiguration(accepted_output_modes=['text/plain']),
    )
    print(f'--- {label} ---')
    try:
        result = await handler.on_message_send(
            params, ServerCallContext(user=UnauthenticatedUser())
        )
        state = getattr(getattr(result, 'status', None), 'state', None)
        print(
            f'  on_message_send -> {type(result).__name__} '
            f'(state={TaskState.Name(state) if state is not None else None})'
        )
    except Exception as e:
        print(f'  on_message_send RAISED {type(e).__name__}: {e}')
    task = await handler.task_store.get(
        executor.task_id, ServerCallContext(user=UnauthenticatedUser())
    )
    print(
        f'  persisted task state: {TaskState.Name(task.status.state)}, '
        f'timestamp set: {task.status.HasField("timestamp")}'
    )


async def main():
    await run_case(
        'CONTROL: healthy push config store',
        InMemoryPushNotificationConfigStore(),
    )
    await run_case('BUG CASE: config store read fails', FlakyConfigStore())


asyncio.run(main())
Expected behavior

A failure to read push configs or to run the URL validator should degrade to "notification not sent" (logged), and must never mutate the task's state or fail the client request. This matches the intent already expressed by _dispatch_notification's error boundary.

Suggested fix

Two small, non-breaking changes:

  • In EventConsumer._update_task_state, wrap the send_notification call in try/except Exception with logger.exception(...) — push delivery is best-effort.
  • In BasePushNotificationSender.send_notification, move the get_info_for_dispatch call and the validator invocation inside the error boundary (log and return early on failure).

Happy to submit a PR with tests if the approach sounds right.

Relevant log output
RuntimeError: transient DB error while reading push configs
Dispatcher task was cancelled or finished. Events may be lost.
Code of Conduct
  • I agree to follow this project's Code of Conduct
Lenguaje dominante
Python
Estrellas
2.2k
Forks
509
Merge medio
3 d 18 h
PR fusionados (30 d)
44

Preparar el entorno

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de a2aproject/a2a-python

Todos los issues de a2aproject/a2a-python

Issues similares

Más issues de Python

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.