Bugfix: 修复AgentNode返回error Event时AGUI连接被过早断开的问题

[weimch]

问题

部署成AGUI服务时,如果Agent产生Event带error,则会立即终止,但有的Agent并不是终止(比如GraphAgent),它会接着执行下一个Node,之前产生Event带error时,将会返回RunErrorEvent事件,导致连接被断开

解决方案

AGUI服务端,在收到Event时,不立即返回RunErrorEvent,如果服务是最后一个Event包含错误(Agent此时已经结束),则再发送RunErrorEvent

验证

构造复现示例:GraphAgent包含两个AgentNode(error_node -> normal_node),通过AgUiAgent SSE stream消费事件流

修复前:RunErrorEvent在第6个event就发出,但Agent继续执行到第23个event,连接被过早断开

修复后:error改为CustomEvent(trpc_error),Agent继续执行,stream以RunFinishedEvent正常结束

[helloopenworld]

AI Code Review

这是 _llm_agent._create_error_event（第 411 行），它不会附加 function_response。但是 tools processor 的 _create_error_event（第 728 行）则会。tools processor 的错误事件确实同时带有 function_response 和 error_code。这就是可恢复的工具错误路径。已确认。

我现在已经有了所需的一切。正在撰写审查报告。

发现的问题

🚨 Critical

• trpc_agent_sdk/server/ag_ui/_core/_agui_agent.py:1255：last_event_is_error 用 is_error() 判定会误把“可恢复的工具错误”当作致命错误，导致 RunErrorEvent 被错误触发，恰好违背本 PR 的初衷。
  • _event_translator.py:213 区分致命/可恢复的依据是 is_error() and not function_responses：工具执行错误事件同时携带 error_code 与 function_response（见 _tools_processor.py:728-786 的 _create_error_event），被归类为可恢复、不发 trpc_error，agent 继续运行。但此处 last_event_is_error = trpc_event.is_error() 只看 error_code，对该类事件也置为 True。若该可恢复工具错误恰好是流中最后一个事件（如工具失败后 agent return 结束，_llm_agent.py:679），循环结束后会发出终止性的 RunErrorEvent，客户端随之断连——这正是本 PR 想避免的“过早断开”。应改为 last_event_is_error = trpc_event.is_error() and not trpc_event.get_function_responses()，与 translator 的判定保持一致。

⚠️ Warning

• trpc_agent_sdk/server/ag_ui/_core/_event_translator.py:222：致命错误改为 CustomEvent(name="trpc_error") 后不再 return，错误后会继续执行后续 if trpc_event.custom_metadata 分支，可能对同一错误事件额外产出 trpc_metadata CustomEvent，向客户端多发一条与错误无关的元数据事件。

  • 建议在 yield trpc_error 后 return（或显式跳过 metadata 分支），保持原“致命错误即终止翻译”的语义，避免客户端收到混淆的二次事件。

• trpc_agent_sdk/server/ag_ui/_core/_agui_agent.py:1296：循环体引用循环变量 trpc_event 来读取错误信息，当 runner.run_async() 一次也不 yield 时 trpc_event 未定义；当前因 last_event_is_error 保持 False 而被 if 短路保护，但代码注释断言“trpc_event 仍指向最后一个事件”并不成立，属脆弱的隐式依赖。

  • 建议显式缓存最后一个事件的错误信息（如 last_error_event），而不是复用循环变量，避免后续改动引入 NameError。

• 测试缺失：新增的 trpc_error CustomEvent 路径与“run 以错误结束才发 RunErrorEvent”的逻辑均无测试覆盖。test_event_translator.py 的 _make_trpc_event 默认 is_error=False，且无任何用例将 is_error 设为 True；test_agui_agent.py 中现有错误用例（如 test_handles_error_and_puts_error_event）走的是异常分支，未覆盖新的延迟发路径。

  • 建议补充：fatal error 事件流 → 末尾产出 RunErrorEvent 且不含 RunFinishedEvent；recoverable tool error（is_error=True 且有 function_responses）作为末事件 → 不发 RunErrorEvent。

总结

本 PR 的延迟发 RunErrorEvent 思路正确，但 _agui_agent.py:1255 用裸 is_error() 判定最后事件是否致命，与 translator 的“function_responses 即可恢复”语义不一致，会把可恢复的工具错误误判为终止性错误并触发客户端断连，属于必须修复的回归性 Critical 问题；其余为错误事件后续处理与测试覆盖的 Warning。

测试建议

• 补充 recoverable tool error（带 function_responses 的 error 事件）作为流的最后一个事件的用例，断言不产出 RunErrorEvent、仍产出 RunFinishedEvent。
• 补充 fatal error（is_error=True 且无 function_responses）作为末事件的用例，断言产出 RunErrorEvent 且其 code/message 来自该事件，并验证 translator 侧产出 trpc_error CustomEvent 且不再产出 trpc_metadata。

[helloopenworld]

裸 is_error() 误判可恢复工具错误为致命错误

last_event_is_error = trpc_event.is_error() 只看 error_code，会把携带 function_response 的可恢复工具错误也判为 True。若该错误恰为流末事件，循环结束会发出终止性 RunErrorEvent 导致客户端断连，违背本 PR 初衷。应改为 is_error() and not trpc_event.get_function_responses()，与 translator 判定一致。

[helloopenworld]

AI Code Review

我已经掌握了所需的全部上下文。现在让我来汇总一下审查结果。

关键分析点：

1. trpc_event 在空运行后未定义 — 如果 runner.run_async() 产生 0 个事件，循环体将永远不会执行，trpc_event 也将从未被绑定。但由于 last_event_is_error 会保持为 False，因此 if last_event_is_error: 的判断为短路逻辑，trpc_event 永远不会被求值。所以这是安全的。

2. 循环后的 RunErrorEvent 未能正确触发下游的 has_error — 实际上可以触发：它被放入了 event_queue，被 _stream_events 消费，产出，且 isinstance(event, RunErrorEvent) 将 has_error 设为 True，因此 RunFinishedEvent 会被跳过。这看起来是正确的。

3. 致命错误后流未终止 — 原代码在遇到 RunErrorEvent 时执行了 yield ...; return。现在 translate() 不再返回；它继续执行到 if trpc_event.custom_metadata: 代码块。如果错误事件包含 custom_metadata，就会同时产生 trpc_error CustomEvent 和 trpc_metadata CustomEvent。此外，对于带有 custom_metadata 且其中 error 键包含错误信息的错误事件，客户端现在会同时收到 trpc_error custom 事件和 trpc_metadata custom 事件。这是一个小问题。

4. 关键：以错误结尾的运行后发射的 RunErrorEvent 在顺序上是错误的 / 重复的 — 考虑流中间发生致命错误的情况。按照新逻辑：翻译器产生 trpc_error CustomEvent（而不是 RunErrorEvent），且代理不会提前终止。last_event_is_error 仅在最后一个事件是错误时才在最后触发 RunErrorEvent。但考虑常见的 GraphAgent 情况，即子代理产生错误但图形继续运行，并以成功的最终事件结束：此时 last_event_is_error 为 False，不会触发 RunErrorEvent，客户端只能看到 trpc_error CustomEvent。这很可能是预期的行为（这是本次修复的目的）。

5. 关键问题：以错误结束的运行触发 RunErrorEvent，但图形已通过正常事件恢复 — 其实没问题。

6. 真正的问题 — 流中间的致命错误不再关闭流 / 仍然通过 _translate_text_content 等传递后续事件 — 当事件既是错误且包含内容（例如中断的内容）时，翻译器现在会继续处理函数调用等。其实它原本也会这样做。删除 return 是核心变更。

7. trpc_error CustomEvent 没有被任何客户端/消费者处理 — grep 确认没有消费者。错误现在仅显示为 CustomEvent；如果运行没有以该错误结束，客户端将永远无法看到 RunErrorEvent。根据设计，这是为了在流中间保持连接活跃而做出的权衡。但这意味着对于在流中间发生致命错误但随后恢复并成功结束的运行，客户端将永远不知道发生了错误（除了非标准的 trpc_error CustomEvent 之外）。这是一个行为变更，但没有消费者将 trpc_error 识别为错误。值得提出警告。

8. last_event_is_error 使用 not trpc_event.get_function_responses() — 与翻译器的条件一致。但翻译器也会特别处理 AgentCancelledEvent（静默终止，无错误事件）。如果最后一个事件是 AgentCancelledEvent，is_error() 返回 False（没有 error_code），所以 last_event_is_error 为 False。没问题。

9. 测试覆盖率 — 没有针对新 last_event_is_error 路径或 trpc_error CustomEvent 路径的测试。翻译器测试模拟了 is_error 始终为 False。没有针对以错误事件结束的运行的测试。这是一个关于测试有效性的警告。

10. 空运行后的 trpc_event 引用 — 安全（短路逻辑）。

11. CustomEvent 缺少时间戳 — trpc_metadata CustomEvent 包含 timestamp=timestamp_ms，但新的 trpc_error CustomEvent 没有时间戳。一致性/轻微问题。

让我再检查一下：翻译器现在会在产生 trpc_error 后继续执行到 custom_metadata 代码块，这意味着如果错误事件带有 custom_metadata，它会同时产生两者。此外，trpc_error 的值使用了 trpc_event.custom_metadata.get("error") 作为回退，因此如果存在，元数据也包含错误。重复了。小问题。

让我再仔细核实一下 has_error / RunFinishedEvent 的交互 —— 带有 code="MODEL_ERROR" 的循环后 RunErrorEvent 被放入队列，然后是 None 哨兵值。_stream_events 产出 RunErrorEvent，设置 has_error=True，然后遇到 None，中断。然后 if has_error: skip RunFinished。很好。但等等 —— 有一个细微的问题：循环后的 RunErrorEvent 是在状态快照事件和 force_close_streaming_message 之后放入队列的。RunErrorEvent 现在在队列中的 RunFinishedEvent（如果触发了）之前。但 has_error 逻辑跳过了 RunFinishedEvent。很好。

等等 —— 实际上还有另一个担忧。_stream_events 消费队列。循环后的 RunErrorEvent 被放入队列后，_stream_events 产出它。但 AG-UI 协议规定 RunErrorEvent 是终止性的 —— 合规的客户端会关闭连接。但服务器仍在继续（None 哨兵值，清理工作）。这与代码库中其他 RunErrorEvent 的发射方式（例如 EXECUTION_TIMEOUT）一致。没问题。

现在让我最终确定。最具体、可验证的问题：

警告 1：流中间的致命错误对客户端而言可能不可见 / 无消费者处理。 trpc_error CustomEvent 没有消费者（通过 grep 验证）。对于在运行过程中发生致命错误但随后恢复并正常完成的运行（GraphAgent 继续到下一个节点并成功），不再发射 RunErrorEvent，客户端只能看到非标准的 trpc_error CustomEvent，标准 AG-UI 客户端可能无法将其识别为错误。这是一个行为变更 —— 需要客户端感知或文档说明。但这是该修复的既定目的，所以这更多是一个完整性警告。

警告 2：对“最后一个事件是错误”的判断对于先发生致命错误但运行未结束的情况较为脆弱。 实际上这是按设计行事的。但有一个具体的 Bug：last_event_is_error 仅跟踪最后一个事件。考虑运行产生 [错误事件，然后是一些非错误事件（例如图形恢复的状态增量或文本），但最终仍然因为错误而失败] 的情况 —— last_event_is_error 将为 False，不会发射 RunErrorEvent。反之，如果运行 [成功文本，然后是错误事件] 并且错误是终止性的 —— RunErrorEvent 会正确触发。所以逻辑是“仅当字面意义上的最后一个事件是非可恢复错误时才触发 RunErrorEvent”。如果错误事件之后有任何尾随的非错误事件（状态快照、文本关闭等 —— 这些确实可能发生在之后），RunErrorEvent 将被抑制。这是一个真正的正确性担忧。

让我检查一下：在致命的 LLM 错误之后，GraphAgent 会产生尾随事件吗？根据提交信息，是的 —— “它会接着执行下一个Node”。所以如果下一个节点成功了，最后一个事件不是错误，没有 RunErrorEvent（预期内）。如果下一个节点也因错误结束，最后一个事件是错误，RunErrorEvent 触发。但如果图形在节点错误后产生最终的正常状态增量事件，RunErrorEvent 就会被抑制。这种边缘情况使得“运行是否以错误终止”的判断不可靠。

警告 3：trpc_error CustomEvent 缺少时间戳（与 trpc_metadata 不一致）。小问题 —— 建议。

警告 4：缺乏新错误路径的测试。 翻译器测试将 is_error Mock 为始终为 False；没有针对 trpc_error CustomEvent 发射或 last_event_is_error → RunErrorEvent 路径的测试。测试有效性警告。

让我再检查一下 trpc_event.custom_metadata or {} 的 None 安全性 —— 没问题。

开始撰写。基于我对 diff 及周边代码的分析，以下是我的审查结果。

发现的问题

⚠️ Warning

• trpc_agent_sdk/server/ag_ui/_core/_agui_agent.py:1255：用“最后一个事件是否为 fatal error”来判断运行是否以错误终止，判定不够稳健

  • last_event_is_error 仅记录流中最后一个 trpc_event 是否为不可恢复错误。但 fatal error 事件之后，agent / runner 仍可能再 yield 非错误事件（例如 GraphAgent 在子节点 error 后继续执行下一节点并产出 state_delta、text 关闭或 normal 事件），这会使 last_event_is_error 被覆盖为 False，从而抑制 RunErrorEvent，客户端无法得知运行其实因错误终止。反之原 translate() 中 return 是在看到 fatal error 时即终止翻译。建议改为“运行期间出现过 fatal error 且未被后续成功响应覆盖”这类更明确的终止语义，或在 runner 层提供明确的 run-failed 信号，而不是依赖最后一个事件的形态。

• trpc_agent_sdk/server/ag_ui/_core/_event_translator.py:222：fatal error 改发 trpc_error CustomEvent 后无任何消费方，且与后续 trpc_metadata 重复

  • 仓库内 grep 确认没有任何客户端/下游消费 name="trpc_error" 的 CustomEvent（trpc_error 仅在此处产生）。对于“中途 fatal error 但运行最终未以 error 结束”的场景，标准 AG-UI 客户端将完全感知不到错误（既无 RunErrorEvent，也不识别 trpc_error）。这是相对原行为（立即发 RunErrorEvent）的明显兼容性回退，至少需要前端约定或文档说明。此外删除 return 后，若该 error 事件本身带 custom_metadata，下方 if trpc_event.custom_metadata: 分支会再 yield 一个 trpc_metadata CustomEvent，与 trpc_error 内容（同样取自 custom_metadata.get("error")）重复下发，建议在此处 return 或显式跳过后续 metadata 输出。

• trpc_agent_sdk/server/ag_ui/_core/_event_translator.py:222：新增的 trpc_error CustomEvent 未设置 timestamp

  • 同文件 trpc_metadata 的 CustomEvent 都带了 timestamp=timestamp_ms，而 trpc_error 没有，事件到达客户端时缺少时间戳，且与同模块其他 CustomEvent 不一致。建议补上 timestamp=int(trpc_event.timestamp * 1000)。

• tests/server/ag_ui/_core/test_event_translator.py:92 与 tests/server/ag_ui/_core/test_agui_agent.py：新增的 fatal error 翻译路径与 last_event_is_error→RunErrorEvent 路径均无测试覆盖

  • 翻译器测试里 event.is_error 统一被 Mock 为 return_value=False，没有任何用例覆盖 is_error() and not function_responses 分支产出 trpc_error CustomEvent 的新行为；_run_trpc_in_background 也没有覆盖“最后一个 trpc_event 为 fatal error 时应补发 RunErrorEvent”的核心修复路径。高风险错误处理逻辑缺少测试，回归时难以发现。

总结

本 PR 改为将 fatal error 延迟到运行末尾再决定是否发 RunErrorEvent，思路正确，但用“最后一个事件是否为 error”作为终止判定不够稳健，且中途 error 改发的 trpc_error CustomEvent 目前无任何消费方、与原行为存在兼容性回退；建议补全判断逻辑与对应测试后再合入。

测试建议

• 在 test_event_translator.py 增加：is_error=True 且无 function_responses 的事件，断言产出 name="trpc_error" 的 CustomEvent（含 code/message/timestamp），且不再产出 RunErrorEvent。
• 在 test_agui_agent.py 覆盖 _run_trpc_in_background 的两类终止场景：最后一个 trpc_event 为 fatal error 时队列中出现 RunErrorEvent；fatal error 后又跟一个正常事件时验证当前行为是否符合预期（即是否应仍发 RunErrorEvent）。

[helloopenworld]

AI Code Review

现在我已经有了完整的认识。测试辅助工具将 is_error 默认设置为 False，且没有针对致命错误路径的测试（新的 trpc_error CustomEvent，last_event_is_error 的延迟触发）。让我完成我的审查。

发现的问题

⚠️ Warning

• _agui_agent.py:1296-1306：仅依据"最后一个 trpc_event 是否为 fatal error"来判定是否补发 RunErrorEvent，存在误判与漏判风险。

  • 若 agent 在 fatal error 事件后还产出了任意一个非 error 事件（例如 GraphAgent 在子 agent 出错后继续产出 state_delta、空 content 事件或 trailing 事件），last_event_is_error 会被覆盖为 False，导致最终一次 fatal 错误被静默吞掉，客户端只会收到 trpc_error CustomEvent 而无终止信号。反之若最后一个事件恰好是可恢复的工具错误（带 function_response），not get_function_responses() 为 False，不会误报，逻辑正确；但 fatal 错误后继有其他事件的场景下错误终止信号会丢失。建议改为"流中曾出现 fatal error 且最终未正常完成"的判定（例如记录 saw_fatal_error 并结合 run 的最终状态），或至少在保留 fatal 终止语义上更保守。属于正确性/稳定性风险，不一定会立即失败故定为 Warning。

• _event_translator.py:206-229 与 _agui_agent.py:1301-1306：错误协议从"立即 RunErrorEvent 终止"改为"中途发 trpc_error CustomEvent + 流末按条件补发 RunErrorEvent"，是一次面向客户端的契约变更，且 trpc_error 这个 CustomEvent name 无文档、无版本约定。

  • 既有客户端若只监听 RunErrorEvent 来判定运行失败，在"fatal error 后 agent 仍继续产出事件、最终 last_event 非 error"的路径下（见上条）将既收不到 RunErrorEvent 也收不到 RUN_FINISHED 之外的失败信号，误判为成功。建议在 README/CHANGELOG 中显式声明 trpc_error CustomEvent 契约，并明确中途错误与最终错误的区分规则，避免下游解析不一致。

💡 Suggestion

• tests/server/ag_ui/_core/test_event_translator.py、tests/server/ag_ui/_core/test_agui_agent.py：本次变更引入了 fatal-error 转译为 trpc_error CustomEvent、以及"最后一事件为 fatal 时补发 RunErrorEvent"两条关键路径，但现有测试中 is_error 默认 Mock 为 False，且无任何用例覆盖 fatal（is_error=True 且无 function_response）分支，也未覆盖"流末 fatal 补发 RunErrorEvent""流中 fatal 后续有事件导致补发被跳过"等风险路径。建议补充对应单测，至少覆盖 fatal 中途转译为 CustomEvent、流末 fatal 补发 RunErrorEvent 两条路径。

总结

整体逻辑方向合理（避免 mid-stream RunErrorEvent 终止导致客户端与仍在运行的后端失同步），但用"最后一个事件是否 fatal"作为最终错误判定的启发式不够稳健，存在 fatal 错误被后续非 error 事件覆盖而漏报终止信号的风险；同时新增的 trpc_error CustomEvent 契约缺少文档与测试。无 Critical 阻塞问题，建议合并前补测试并明确契约。

测试建议

• 补充单测：fatal error（is_error=True 且 get_function_responses() 为空）应转译为 name="trpc_error" 的 CustomEvent，且不再立即发 RunErrorEvent。
• 补充单测：当 runner.run_async 产出的最后一个事件为 fatal error 时，_run_trpc_in_background 应在流末补发 RunErrorEvent 并以 None 收尾；当 fatal error 后还跟有非 error 事件时，验证当前行为是否符合预期（建议显式断言预期结果，以暴露上文的漏判风险）。