Hacktoberfest 2026：维护者为十月标记出来的 issue，仍然开放、适合新手。 浏览 Hacktoberfest issue

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

未关闭
  • 哪里不对？
#1,313 1 条评论 0 个 reaction 已指派 1 人 在 GitHub 查看

维护者通常 2 天内回复

@rohityan 已经在做这个了。

开始于 2026年10月5日。

评估

这个 Issue 还没有评估数据。

描述

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
主要语言
Python
星标
2.2k
派生
509
平均合并
3 天 18 小时
30 天内合并 PR
44

环境准备

从这里开始

  1. 先读完整个 Issue，再读项目的贡献指南。
  2. 在 Issue 下留言说明你要接手 —— 这能避免两个人做同样的事。
  3. Fork 仓库，在一个分支上完成修改。
  4. 提交 Pull Request，并在描述里引用这个 Issue 编号。

a2aproject/a2a-python 的其他 Issue

查看 a2aproject/a2a-python 的全部 Issue

相似的 Issue

更多 Python Issue

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。