Repository navigation
[BUG][P1] A2A 流取消应中止阻塞的 HTTP 读取并释放连接与工作线程 #277
Description
已确认问题
A2A streaming 在 boundedElastic 工作线程中使用同步 HttpClient 和 BufferedReader.readLine。代码只在每次读取前检查 sink.isCancelled，没有将取消绑定到在途 HTTP 请求/响应关闭。远端已经建立 SSE 连接但停止发送数据时，消费者取消无法唤醒阻塞读取，try-with-resources 的释放也无法及时执行。
本地 HTTP/SSE 复现
- 在 127.0.0.1 随机端口启动测试 HTTP 服务，消费请求，返回
Content-Type: text/event-stream。 - 发送并 flush 一个 SSE 注释行，然后由受控 latch 保持响应打开，不继续发送数据。
- 通过 A2aNodeActionWithConfig 的公开 apply 获取 Flux 并订阅，确认客户端工作线程已进入 BufferedReader.readLine。
- dispose 订阅，观察线程与资源；最后由测试服务关闭响应，以保证夹具清理。
实测：取消前、取消 250ms 后均有 boundedElastic-1 停留在该 HTTP 读取调用栈；只有测试服务关闭远端响应后读取线程才退出。本次验证是本地可控 HTTP/SSE，不是生产远端服务验收，也没有将“取消外层订阅”误当作传输已释放。
交付与验收
- 将订阅取消绑定到当前请求中止/响应与输入流关闭，及时解除阻塞读取并释放本次调用持有的连接和 worker。
- 处理连接建立、响应头到达、响应读取和正常完成各阶段的取消竞态，关闭操作幂等。
- 区分主动取消与传输失败，不把取消转成成功 done 或普通错误文本；错误终态与 [BUG][OPT-22] A2A 传输与解析失败应传播结构化错误而非正常文本完成 #213 协调。
- 用静默 SSE、响应头前阻塞、持续输出、远端正常结束与并发多调用验证；不通过全局关闭共享线程池终止单个运行。
- 必要的等待/读取时限作为补充边界，不能仅依靠远端最终断开或很长的超时释放资源；取消仍须主动生效。
与既有任务的边界
关联 #212、#213。#212 聚焦队列/背压满载与可靠终态，旧问题依据是 LinkedBlockingQueue；本单针对当前 Flux 实现，即使没有队列满载、远端没有任何业务事件也能复现的传输取消缺陷。它不是 Extensions 的持久化 A2A 取消或 SDK 升级任务。
验证基线与变更约束
Core b34835354b27f1f320ea8438edf2e9d170e69c84；Extensions cb2635ab2fc338e8aade28a132a15f48b4e655fc。本轮 Core 19 模块 clean install 通过，2517 个测试中 173 个跳过；此处缺陷由额外的独立 Java 探针确认，说明已有测试尚未覆盖对应行为，不代表真实模型或生产环境验收。
优先修复现有契约的实现，不强制拆模块或重命名，不让独立 Agent/Workflow 反向依赖 Graph/Loop，不向 Core 引入供应商 SDK。若必须改变公开 API、事件、配置或持久格式，说明较小替代方案、消费者影响、收益、迁移成本与目标版本；不得通过删测试、吞错误或弱化取消保证获得通过。
Activity
konlue commentedon Oct 28, 2026
你好，我想认领 #277。
基于 main
04c5686fe确认：createStreamingFlux在Schedulers.boundedElastic()上同步执行 HttpClient5 的httpClient.execute(post, response -> ...)（L253/L264），SSE 用BufferedReader.readLine()逐行读（L290）。循环条件!sink.isCancelled() && (line = reader.readLine()) != null只能在两次读取之间看到取消——一旦线程已经阻塞在readLine()，取消不会唤醒它。post.cancel()，CloseableHttpClient的close()要等execute返回后由 try-with-resources 才执行 → 远端只发一个注释行后保持静默时，worker 和连接一起被钉住，正是 issue 里boundedElastic-1停在该调用栈的现象。sendSpan.error(e)/receiveSpan.error(ex)，把主动取消记成传输失败（错误文本与 done 已被emit/complete的isCancelled挡住，但 span 归因是错的）。方案（单 PR，不动公开 API / 事件 / 配置 / 持久格式）
sink.onCancel(...)绑定传输侧释放：post.cancel()（请求已进入执行时中止本次 exchange）+close()本次调用自己创建的CloseableHttpClient（未进入执行/正在等响应头时由连接管理器关闭兜底）。两者都幂等，且该 client 是每次调用新建的，不触碰任何共享线程池或共享 client。execute前补一次sink.isCancelled()检查，避免取消后仍发起请求。!sink.isCancelled()守卫），非取消路径的错误文本 + done 终态保持原样（与 [BUG][OPT-22] A2A 传输与解析失败应传播结构化错误而非正常文本完成 #213 的终态契约不冲突，只是取消不再被当成失败）。fix(a2a): abort the in-flight HTTP exchange when a streaming call is …
Merge pull request #336 from konlue/fix/a2a-stream-cancel