Hacktoberfest 2026: le issue che i maintainer hanno segnato per ottobre, aperte e adatte ai principianti. Sfoglia le issue Hacktoberfest

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

Aperta
  • Qualcosa non va?
#1,313 0 commenti 0 reazioni 2 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 2 giorni

@rohityan ci sta già lavorando.

Dal 5/10/2026.

Valutazione

Questa issue non è ancora stata valutata.

Descrizione

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
Lingua principale
Python
Stelle
2.2k
Fork
509
Merge medio
3g 11h
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.