[Bug]: FAILED Task 的 Push Notification 在异常兜底与快速终态场景下可能丢失

[openjiuwen-collaboration-bot]

Checklist

• 我已经搜索过相关问题，但没有得到预期的帮助。
• 最新版本中该错误尚未修复。
• 请注意，如果您提交的Bug描述缺少相应的环境信息和最小可复现的demo，我们将很难复现和解决该问题，从而降低收到反馈的可能性，甚至该问题将被关闭。

🐞 问题详细描述

FAILED Task 的 Push Notification 在异常兜底与快速终态场景下可能丢失

结论摘要

Push Notification 契约要求 Task 进入 COMPLETED 或 FAILED 结果性状态时投递 callback。当前实现中：

• COMPLETED callback 已通过 Deep Research Agent 实跑验证。
• Runtime 将异常转换为 TaskStatusUpdateEvent(FAILED)，且 callback config 已完成绑定时，FAILED callback 正常。
• 普通 RuntimeException 逃出 Runtime A2AAgentExecutor 后，A2A SDK 将其转换为 InternalError。SDK 虽然把 Task 持久化为 FAILED，但 MainEventBusProcessor 因原始事件不是 StreamingEventKind 而不调用 PushNotificationSender，callback 稳定丢失。
• 对于执行启动后立即失败的 TaskStatusUpdateEvent(FAILED)，还存在 callback config 绑定竞态：SDK 先启动 Agent，后保存 inline push config。终态事件若先被 sender 消费，sender 查不到配置并直接返回，且没有终态重放。

这是 Runtime/A2A SDK 执行链问题，不是 Demo 业务代码问题，也不是 HttpPushNotificationSender 不支持 FAILED。

1. 场景

1.1 特性与需求来源

特性：服务入口与 Push Notification

相关事实要求：

• 能力范围设计 第 2 章“callback 触发范围”：COMPLETED 返回完成结果，FAILED 返回异常状态、错误码和失败原因。
• 能力范围设计 第 5.1.4 节：FAILED 必须通知异常状态、错误码和失败原因。
• 详细设计 第 3.4 节：SendMessage + pushNotificationConfig 必须先保证 Task 可查询、push config 已绑定、后台执行已可靠调度，再返回 accepted Task 表面。
• 详细设计 第 4.5 节：MainEventBusProcessor、PushNotificationConfigStore、PushNotificationSender 共同承载终态 callback 投递。

1.2 涉及模块

• agent-runtime-java/service/agent-service-app
  • A2AAgentExecutor
  • A2AAutoConfiguration
  • HttpPushNotificationSender
• A2A SDK a2a-java-sdk-server-common:1.0.0.Final
  • DefaultRequestHandler
  • AgentEmitter
  • TaskManager
  • MainEventBusProcessor
• 使用方场景：multi-deep-research-demo
  • Deep Research Runtime
  • Search Runtime
  • 调用方 Runtime 的固定 callback receiver

1.3 验证基线

验证日期：2026-08-07

agent-runtime-java: develop @ acd12ce7259f66e424a8ee3e9348a6a8fb081004
agent-core-java:    730     @ e0482cd0224b76c6422189a3eb1cf13fe7b52601
agent-solution:     common  @ 55226a1d8d4ecd9c76d22e28bf241b7e04c033d6
A2A SDK:            1.0.0.Final
Java:               26.0.1（Maven 编译 release 17）

1.4 测试场景

编号	场景	原始终态事件	结果
S1	Deep Research 正常完成，不调用 Search	TaskStatusUpdateEvent(COMPLETED)	callback 正常
S2	Search 模型地址不可连接，Core 抛出普通 RuntimeException	InternalError，随后 Task snapshot 变为 FAILED	callback 丢失，稳定复现
S3	Handler 延迟 500ms 后抛 IllegalStateException，Runtime 调用 emitter.fail(message)	TaskStatusUpdateEvent(FAILED)	callback 正常
S4	Handler 立即失败，并将 push config store 写入延迟 500ms	TaskStatusUpdateEvent(FAILED)	Task 进入 FAILED，但 callback 稳定丢失且不重放

2. 问题概述

2.1 问题一：InternalError 已形成 FAILED Task，但未触发 callback

Runtime 的 A2AAgentExecutor.execute() 只捕获：

IllegalArgumentException | IllegalStateException | NullPointerException

Core RunnerImpl 会把模型连接失败等异常包装成普通 RuntimeException("Failed to invoke agent")。该异常逃出 Runtime executor 后，由 A2A SDK DefaultRequestHandler 兜底转换为 InternalError 并发布到 EventBus。

SDK TaskManager 能识别 A2AError，因此 TaskStore 中的 Task 最终会变为 TASK_STATE_FAILED。但是 MainEventBusProcessor 只有在原始事件实现 StreamingEventKind 时才调用 PushNotificationSender。InternalError 只实现 Event，不实现 StreamingEventKind，所以 sender 完全没有被调用。

最终表现为：

GetTask -> TASK_STATE_FAILED
callback receiver -> 没有收到 POST

2.2 问题二：Agent 执行早于 inline push config 绑定

DefaultRequestHandler 当前时序是：

创建 Task/Queue
  -> 启动 Agent 异步执行
  -> non-blocking 聚合返回 accepted Task
  -> 保存 inline push notification config

如果 Agent 很快进入 FAILED，MainEventBusProcessor 可能在 push config 保存前调用 sender。HttpPushNotificationSender.firstConfig(taskId) 返回空后直接结束，本次终态不会在 config 保存后重放。

自然执行下该竞态取决于 Agent 线程、HTTP 请求线程和 MainEventBusProcessor 的调度顺序；零延迟探针分别观察到一次失败和一次成功。进一步通过仅延迟 PushNotificationConfigStore.setInfo()、不延迟 Agent 失败的方式，已将该竞态确定性复现。

2.3 影响概述

调用方 Runtime 已释放 SendMessage HTTP 请求并等待 callback，但被调用方 Task 已经失败且不再投递 callback。调用方只能等待超时或主动 GetTask，无法按异步完成契约及时恢复父 Task。

模型不可达、限流、连接超时、Core Runner 包装异常等属于常见生产故障，因此问题不局限于人工构造异常。

3. 问题分析

3.1 集成实跑结果

以下证据来自真实 Demo 调用和 Runtime 临时集成探针；建议将对应场景纳入正式自动化回归。

S1：COMPLETED 对照组

请求 Deep Research：

请你记住我的名字：cynthia xue

结果：

Task ID: 5dddd5fa-b34a-45b6-a3ee-8c32cf160821
GetTask: TASK_STATE_COMPLETED
Agent output: 我记住了，你的名字是 Cynthia Xue。
callback POST count: 1
callback state: TASK_STATE_COMPLETED
notificationId: d83f4db3dd2e6b5c0bd3e79be9692e8eea1d244ef9c3c8c3dbd928581b4f277a

说明 callback 配置绑定、HTTP sender、callback URL 和基本终态投递链可用。

S2：InternalError -> FAILED 问题组

将 Search Agent 的模型地址配置为不可连接地址后发起异步 SendMessage，连续两次复现：

Agent execution threw unexpected RuntimeException
java.lang.RuntimeException: Failed to invoke agent
Caused by: java.net.ConnectException: Failed to connect to /127.0.0.1:19999

Enqueued event org.a2aproject.sdk.spec.InternalError:
    Agent execution failed: Failed to invoke agent

MainEventBusProcessor: Processing event ...: InternalError
TaskManager: A2AError event detected, transitioning task ... to FAILED
TaskStore updated ... InternalError (final: true, replicated: false)
MainEventBusProcessor: Distributing InternalError ...

缺失日志：

Sending push notification for task <task-id>

callback receiver 未收到 POST，但 GetTask 返回 TASK_STATE_FAILED。

S3：StreamingEventKind(FAILED) 对照组

临时 Handler 延迟 500ms 后抛出 IllegalStateException。Runtime 捕获后调用 emitter.fail(message)，SDK 生成：

TaskStatusUpdateEvent[
  status=TaskStatus[
    state=TASK_STATE_FAILED,
    message=...probe failure translated to TaskStatusUpdateEvent...
  ]
]

关键日志：

Storing push notification config for new task 4e78e83c-...
TaskStore updated ... TaskStatusUpdateEvent (final: true, replicated: false)
Sending push notification for task 4e78e83c-...

测试结果：

Tests run: 1, Failures: 0, Errors: 0, Skipped: 0
BUILD SUCCESS

callback receiver 收到一次 POST，body 中 Task 状态为 TASK_STATE_FAILED，并保留失败 message。

这证明：

• HttpPushNotificationSender 支持 FAILED。
• TaskStatusUpdateEvent(FAILED) 可以正常触发 callback。
• S2 的主要差异是原始事件为 InternalError，不是 sender 或 callback receiver 故障。

S4：快速 FAILED 的绑定竞态

自然零延迟探针分别出现过一次未收到 callback 和一次正常收到 callback，说明问题受线程调度影响。为排除偶然因素，又构造了确定性对照：

• Handler 立即抛出 IllegalStateException，仍由 Runtime 转换为 TaskStatusUpdateEvent(FAILED)。
• 自定义 PushNotificationConfigStore 只把 setInfo() 延迟 500ms，getInfo() 保持实时读取。
• callback receiver 等待 1 秒，确认没有收到 POST。
• TaskStore 轮询确认同一 Task 已进入 TASK_STATE_FAILED。

关键时序：

17:48:39.107 Agent execution starting
17:48:39.119 Agent execution failed
17:48:39.126 Enqueued TaskStatusUpdateEvent(FAILED)
17:48:39.138 TaskStore updated ... TaskStatusUpdateEvent (final: true)
17:48:39.138 Sending push notification for task 9d123adc-...
17:48:39.639 delayed PushNotificationConfigStore.setInfo() 返回

sender 在 config 保存完成前执行，firstConfig(taskId) 读取为空并直接返回。config 后续保存成功，但没有终态事件重放，所以 receiver 始终没有 POST。

确定性探针结果：

Task state: TASK_STATE_FAILED
callback received within 1 second: false
Tests run: 1, Failures: 0, Errors: 0, Skipped: 0
BUILD SUCCESS

3.2 对照结论

COMPLETED TaskStatusUpdateEvent + 已绑定 config -> callback
FAILED TaskStatusUpdateEvent    + 已绑定 config -> callback
FAILED TaskStatusUpdateEvent    + config 绑定竞态 -> callback 可能丢失
InternalError -> FAILED Task snapshot           -> callback 稳定丢失

4. 复现条件

4.1 前置条件

1. Runtime 开启 A2A push notification 能力。
2. 调用方提供可访问的固定 callback receiver。
3. 使用 SendMessage 并携带 inline push notification config，使请求进入 returnImmediately 异步路径。
4. 被调用 Agent 在执行阶段产生普通 RuntimeException，例如 LLM 地址不可连接。
5. 开启以下 DEBUG 日志便于确认事件类型：

logging:
  level:
    org.a2aproject.sdk: DEBUG

4.2 典型启动方式

以下只使用无效地址和假 token，不需要真实密钥：

$env:LLM_PROVIDER = 'OpenAI'
$env:LLM_API_KEY = 'dummy'
$env:LLM_API_BASE = 'http://127.0.0.1:19999/v1'
$env:LLM_MODEL = 'dummy'
java -jar agent-solution/common/example/multi-deep-research-demo/agent-search/target/agent-search-0.1.0.jar `
  --logging.level.org.a2aproject.sdk=DEBUG

准备任意能记录 POST body 的 callback receiver，例如：

http://127.0.0.1:19092/a2a/push-notifications/callback

4.3 SendMessage 请求

当前 Runtime parser 同时兼容 inline params.pushNotificationConfig 和 configuration.taskPushNotificationConfig。按当前 Push Notification 请求契约可发送：

{
  "jsonrpc": "2.0",
  "id": "failed-callback-repro",
  "method": "SendMessage",
  "params": {
    "message": {
      "role": "ROLE_USER",
      "messageId": "msg-failed-callback-repro",
      "contextId": "ctx-failed-callback-repro",
      "parts": [
        {
          "kind": "text",
          "text": "callback repro"
        }
      ]
    },
    "pushNotificationConfig": {
      "id": "push-failed-callback-repro",
      "callbackUrl": "http://127.0.0.1:19092/a2a/push-notifications/callback"
    }
  }
}

4.4 观察步骤

1. SendMessage 返回 accepted/working Task，记录 taskId。
2. 轮询 GetTask，最终可观察到 TASK_STATE_FAILED。
3. callback receiver 没有收到对应 taskId 的 POST。
4. Runtime 日志出现 InternalError、transitioning task ... to FAILED 和 final: true。
5. 同一 taskId 后没有 Sending push notification。

4.5 临时对照探针命令

Runtime 内构造 IllegalStateException -> emitter.fail(message) -> TaskStatusUpdateEvent(FAILED) 后执行：

mvn -pl service/agent-service-app -Dtest=StreamingFailedCallbackProbeTest test

临时探针只用于本次定位，未保留在仓库；建议将该场景纳入正式自动化回归。

5. 问题代码分析

5.1 Runtime 只捕获三种 RuntimeException

文件：

service/agent-service-app/src/main/java/com/openjiuwen/service/app/controller/a2a/A2AAgentExecutor.java:99-113

当前代码：

try {
    if (req.isStream()) {
        executeStreaming(msgCtx, ctx, req, emitter);
    } else {
        executeQuery(msgCtx, ctx, req, emitter);
    }
} catch (IllegalArgumentException | IllegalStateException | NullPointerException ex) {
    log.error("Agent execution failed for contextId={}", ctx.getContextId(), ex);
    failAndDrain(emitter, msgCtx, ex);
}

Core RunnerImpl 包装出的普通 RuntimeException 不会进入 failAndDrain()。

5.2 Runtime 的正常失败转换本身可用

文件：

service/agent-service-app/src/main/java/com/openjiuwen/service/app/controller/a2a/A2AAgentExecutor.java:316-327

private void failAndDrain(AgentEmitter emitter, A2AMessageContext msgCtx, Throwable error) {
    String errorMessage = error.getMessage() == null ? "Agent execution failed" : error.getMessage();
    Message message = Message.builder()
        .role(Message.Role.ROLE_AGENT)
        .parts(List.of(new TextPart(errorMessage)))
        .build();
    emitter.fail(message);
    // drain omitted
}

AgentEmitter.fail(Message) 生成 TaskStatusUpdateEvent(FAILED)，S3 已证明其 callback 正常。

5.3 SDK 异常兜底选择了不同的 fail 重载

依赖：

org.a2aproject.sdk:a2a-java-sdk-server-common:1.0.0.Final
org.a2aproject.sdk.server.requesthandlers.DefaultRequestHandler

异常逃出 Agent executor 后，SDK 构造 InternalError，实际走的是：

AgentEmitter.fail(A2AError error)

而不是：

AgentEmitter.fail(Message message)

两个重载的行为不同：

fail(Message)
  -> TaskStatusUpdateEvent(FAILED)
  -> implements StreamingEventKind

fail(A2AError)
  -> 直接 enqueue InternalError
  -> implements Event only

5.4 TaskManager 生成 FAILED snapshot，但丢失错误详情

SDK TaskManager.process(Event, ...) 对 A2AError 的处理逻辑等价于：

if (event instanceof A2AError) {
    TaskStatusUpdateEvent failed = TaskStatusUpdateEvent.builder()
        .taskId(taskId)
        .contextId(contextId)
        .status(new TaskStatus(TaskState.TASK_STATE_FAILED))
        .build();
    return saveTaskEvent(failed, replicated, taskSnapshot);
}

因此 TaskStore 能看到 FAILED，但新建的 TaskStatus 没有携带原始 InternalError 的错误码和失败原因。这与异步 FAILED callback 契约对错误内容的要求也不一致。

5.5 MainEventBusProcessor 使用原始事件类型决定是否发送

SDK MainEventBusProcessor.processEvent() 的发送条件等价于：

UpdateResult update = updateTaskStore(taskId, originalEvent, replicated);

if (!replicated
        && processedEvent == originalEvent
        && originalEvent instanceof StreamingEventKind streamingEvent) {
    sendPushNotification(taskId, streamingEvent, update.taskSnapshot());
}

InternalError 不实现 StreamingEventKind，所以即使 update.taskSnapshot() 已是最终 FAILED Task，仍然跳过 sender。

5.6 HttpPushNotificationSender 已支持 FAILED

文件：

service/agent-service-app/src/main/java/com/openjiuwen/service/app/controller/a2a/HttpPushNotificationSender.java:69-98

private boolean isCallbackState(Task task) {
    TaskState state = task.status().state();
    return state == TaskState.TASK_STATE_COMPLETED
        || state == TaskState.TASK_STATE_FAILED;
}

因此不能通过修改该状态判断解决 S2；S2 中 sender 根本没有被调用。

5.7 callback config 绑定顺序存在竞态

SDK 日志显示 DefaultRequestHandler 先异步启动 Agent，再在 non-blocking 聚合分支保存 push config：

Agent execution starting
Agent execution failed
Enqueued TaskStatusUpdateEvent(FAILED)
Storing push notification config for new task

这与 详细设计 第 3.4 节“push config 已绑定、后台执行已可靠调度”的顺序承诺不一致。

6. 问题影响

1. runtime-to-runtime 调用方可能永远收不到远端失败通知，父 Task 停留在等待 callback 的状态直到超时。
2. 调用方只能通过 GetTask 补偿，与 SendMessage + push notification 释放长连接并异步恢复的设计目标不一致。
3. 模型网络故障、连接超时、SDK/handler 未预期异常等常见故障最容易进入问题路径。
4. Agent Card 仍可能声明 pushNotifications=true，但 FAILED 完成通知不是可靠能力，存在 capability 过度声明风险。
5. 调用方超时重试可能产生重复远程执行；当前 notification id 幂等不能补偿“通知从未发送”。
6. 即使只修复漏发，InternalError 转换后的 FAILED Task 仍缺少稳定错误码和失败原因，调用方无法程序化判断失败类别。

7. 建议修复方案

方案 A：Runtime 捕获所有 RuntimeException，作为近期止血

将：

catch (IllegalArgumentException | IllegalStateException | NullPointerException ex)

调整为：

catch (RuntimeException ex)

统一调用现有 failAndDrain()，使当前 Runtime 中的模型/Core/远程调用运行时异常转换为 TaskStatusUpdateEvent(FAILED)。

优点：改动小，可直接覆盖当前主要生产路径。

局限：不能修复 A2A SDK 对任意自定义 AgentExecutor、取消流程或其他 A2AError 的通用漏发问题。

方案 B：SDK 按持久化后的 terminal Task snapshot 决定 callback，推荐作为根修复

MainEventBusProcessor 完成 TaskManager.process() 后，应依据 UpdateResult.taskSnapshot() 判断是否已经进入允许 callback 的结果性状态，而不是只判断原始事件是否为 StreamingEventKind。

可选实现：

1. 对 A2AError 根据最终 Task snapshot 合成 TaskStatusUpdateEvent(FAILED) 后调用现有 sender。
2. 扩展 PushNotificationSender 接受通用 Event + Task，sender 只依据 Task 终态构造 callback；该方式会改变 SDK SPI，需要评估兼容性。
3. 在 onTaskFinalized(taskId) 回调中从 TaskStore 获取最终 Task 并投递，但必须避免与正常 TaskStatusUpdateEvent 重复发送。

推荐优先选择第 1 种，兼容当前 PushNotificationSender SPI，改动范围较小。

方案 C：在启动 Agent 前完成 inline push config 与新 Task 的原子绑定

DefaultRequestHandler 应调整为：

创建 taskId
  -> 校验并保存 push config(taskId)
  -> 创建/登记可查询 Task
  -> 启动 Agent 后台执行
  -> 返回 accepted Task

至少必须保证 Agent 产生任何终态事件前，sender 已能按 taskId 读取 callback config。

方案 D：增加终态重放，作为绑定顺序无法原子调整时的补偿

如果 SDK 架构暂时无法先绑定再启动，可在 push config 保存后检查 TaskStore：

• 如果 Task 已进入 COMPLETED / FAILED，立即按稳定 notification id 补投一次。
• 与正常事件投递共享同一幂等记录，避免重复 callback。
• 不能依赖当前仅进程内的偶然线程调度顺序。

方案 E：保留 FAILED 的错误码和失败原因

TaskManager.process(A2AError) 转换为 failed Task 时，应把以下信息映射到稳定的 Task status/message/metadata 契约：

• SDK/A2A error code
• 对外可返回的失败原因
• 可选的稳定业务错误分类

不得直接暴露内部堆栈、密钥或敏感连接信息。

方案 F：补充正式自动化回归覆盖

建议在 agent-runtime-验收测试 增加以下黑盒场景：

1. COMPLETED callback 一次且 payload 完整。
2. Handler 显式 FAILED callback 一次。
3. 模型/Core 抛普通 RuntimeException 后，GetTask=FAILED 且 callback 一次。
4. Agent 在 callback config 绑定窗口内立即完成/失败，callback 不丢失。
5. 同一 notification 重试保持 notification id 稳定且接收端幂等。
6. FAILED callback 包含稳定错误码和失败原因，不暴露内部异常细节。

建议拆分方式

如果 A2A SDK 与 Runtime 由不同团队维护，建议由本 Issue 作为 Push Notification 总问题跟踪，并拆成两个实现子 Issue：

1. Runtime/A2A SDK: InternalError-converted FAILED Task does not trigger push notification
2. A2A SDK: Agent execution starts before inline push notification config is bound

两者有相同的用户可见症状，但根因、修改位置和回归重点不同。

详细的环境信息描述

runtime Java develop分支

其他辅助信息

版本信息

感谢您的贡献 🎉!