修复测试用例interruptrecovery093

[openjiuwen-sync-bot]

Paired: GitHub #69 ↔ GitCode !212

InterruptRecovery093 重复执行修复方案

一、结论

之前把：

assertThat(loopNode1.getTimes()).isEqualTo(3);
assertThat(loopNode4.getTimes()).isEqualTo(3);

放宽成：

isBetween(3, 6)

这个处理不正确。

这不是“没有执行但计数增加”的幽灵计数。times++ 就在节点的 invoke() 中，次数超过 3 说明节点确实被再次执行了。真正的问题是：

节点已经执行并产生副作用，但成功结果被 Java 的取消机制丢弃；恢复时框架误认为该节点失败，于是重新执行了一次。

因此不能通过放宽断言接受重复执行，应该修复 Core 的任务取消与恢复语义，并恢复精确次数断言。

二、093 的正确执行次数

InterruptRecovery093 有 3 个循环元素：

• loopNode1、loopNode3、loopNode4：每个元素成功执行一次，因此都是 3 次。
• loopNode2：每个元素第一次异常、恢复后第二次成功，因此是 3 × 2 = 6 次。

Python 原测试也是严格断言 3 / 6 / 3 / 3，见 [test_interrupt_recovery_093.py]。

如果 Java 中 loopNode1/4 出现 4～6 次，说明失败恢复时重复执行了已经运行过的节点。由于测试节点只是“加十”，最终结果可能仍然正确；但真实业务节点可能已经发送消息、写数据库或调用外部接口，重复执行会产生真实副作用。

三、根因分析

Python 使用单线程 asyncio 协作调度。AddTenNode.invoke() 从 times += 1 到正常返回之间没有 await，这一段不会被兄弟任务抢占，见 [workflow_aw.py]。

Java 翻译后使用虚拟线程并行执行兄弟节点。旧实现发生了以下竞态：

loopNode1/4 已经进入 invoke，times 已增加
                  ↓
兄弟节点 loopNode2 抛出异常
                  ↓
waitAll 对所有未完成 Future 调用 cancel(true)
                  ↓
FutureTask 被不可逆地标记为 cancelled
                  ↓
loopNode1/4 即使随后正常返回，结果也无法再变成 success
                  ↓
节点被写入 PendingNode
                  ↓
恢复时重新 invoke，times 再增加一次

旧实现的问题点在于 FutureTask.cancel(true)：

• 它不仅发送线程中断；
• 还会立即把 Future 置为终态 cancelled；
• 后续正常返回无法覆盖这个状态；
• TaskFuture.done() 会进一步取消外层结果 Future；
• 最终该节点被当作失败节点保存并恢复执行。

这和 Python asyncio.Task.cancel() 的协作式取消并不等价。Python 是在后续 await 注入取消异常，协程可以响应、清理，甚至捕获取消后正常返回；Java FutureTask.cancel(true) 则先永久丢弃结果。

四、修复原则

核心规则是：

兄弟节点失败，可以尝试终止仍在执行的节点；但已经正常返回的节点必须完成路由并保留结果，不能因为 Future 尚未整体结束就把它标记为失败。

为此需要同时区分“节点执行阶段”和“取消来源”。

节点阶段	兄弟节点失败	显式 cancelAll()
NOT_STARTED	取消，节点未执行，可恢复重试	强制取消
INVOKING	只发送协作式 interrupt，不终态取消 Future	cancel(true)
ROUTING	不再中断，等待路由完成并保留结果	强制中断并取消
COMPLETED	使用已经产生的结果	无需处理

兄弟失败和显式取消必须分开：

• SIBLING_FAILURE：目标是停止不必要的工作，但不能丢弃正常执行结果。
• FORCED：调用方明确要求停止整个执行，可以强制取消并丢弃结果。

五、具体代码修改

1. 增加 invocation 完成边界

在 [NodeTask.java]中，节点函数正常返回后、路由开始前通知执行器：

invokeFunc(func, kwargs);
invocationCompletedHandler.run();
throwIfInterrupted();

只有 NodeTask 知道“业务调用已经正常返回”的准确位置，所以不能只根据整个 Future 是否完成来判断。

原有公开构造方法保留，默认使用 no-op callback，不改变独立使用 NodeTask 的行为。

2. 引入任务阶段

在 [TaskExecutorPool.java] 中增加：

NOT_STARTED → INVOKING → ROUTING → COMPLETED

原来的 hasStarted 布尔值只能判断“开始/未开始”，无法区分：

• 正在调用业务节点；
• 业务调用已经成功，只剩路由和结果发布。

这正是旧实现误判的根源。

3. 兄弟失败改为协作式取消

在 [TaskExecutorPool.java]中：

• NOT_STARTED：仍然取消 Future，确保节点不会进入。
• INVOKING：只 interrupt 执行线程，不调用 FutureTask.cancel()。
• ROUTING：不 interrupt，等待正常结果。
• 如果 invocation 响应中断并抛出异常，仍进入 pending。
• 如果 invocation 捕获中断后正常返回，则保留为成功，不再恢复执行。

4. 用锁封闭竞态窗口

phase、cancellation、executionThread 都由每个任务自己的 lifecycleLock 保护。

这样无论哪个动作先发生，结果都确定：

• 兄弟失败先发生：登记协作式取消并 interrupt；节点若最终正常返回，则清理这次 sibling interrupt 并继续路由。
• invocation 先返回：状态先进入 ROUTING；后到的兄弟失败不再中断它。
• 强制取消先发生：取消类型为 FORCED，不会误清强制中断信号。

没有增加全局锁，不影响不同节点之间的并行执行。

5. 等待真实执行和结果发布都完成

[settlePendingTasksAfterFailure()]同时等待：

底层执行真正退出
        AND
结果 Future 完成发布

不能只看到 Future 被标记为 cancelled 就立即写 checkpoint，否则旧任务可能还在执行，而恢复任务已经启动，仍会产生并行重复执行。

六、为什么不能采用其他简单方案

• 不能使用 isBetween(3, 6)：这会把重复副作用合法化。
• 不能把 cancel(true) 改成 cancel(false) 用于运行中任务：运行中的 FutureTask.cancel(false) 仍可能把 Future 永久标记为 cancelled，只是不发送 interrupt，结果仍会丢失。
• 不能完全不取消、直接等待全部节点：会改变 FIRST_EXCEPTION 语义，慢任务可能一直运行。
• 不能要求所有业务组件自行幂等：幂等是额外保护，不能替代框架正确保存已经成功的节点结果。
• 不能移动测试中的 times++：真实业务副作用可能发生在节点执行任意位置，修改计数器只是掩盖问题。

七、测试修改及验证

测试断言已恢复为精确次数，见 [InterruptRecovery093Test.java]。

新增 Core 回归测试见 [TaskExecutorPoolTest.java]，确定性制造：

A 已进入 invocation
→ B 抛异常
→ A 收到 interrupt，但正常返回
→ 模拟恢复

旧实现结果为：

expected invocationCount = 1
actual invocationCount   = 2

新实现保证：

• invocation 只执行一次；
• interrupt 不泄漏到 router；
• 路由消息得到保留；
• 正常节点不进入 failed；
• 显式 cancelAll() 仍能强制中断 router。

验证结果：

• TaskExecutorPoolTest：通过。
• ReActAgentInterruptRegressionTest：通过。
• 093 次数路径临时绕过独立的 index 断言后重复运行 10 次，均稳定为 3 / 6 / 3 / 3；临时改动已撤销，index 未纳入本次修改。
• Core 标准全量：3981 个测试，2 failures + 2 errors。
• 这 4 个异常均为 macOS /var 与 /private/var 路径问题，并已在未修改的 HEAD 上原样复现，确认不是本次修改引入。
• 标准全量不包含 system-test 标签。

八、影响范围与边界

本次只修改“同一 super-step 中普通节点异常后，兄弟任务如何收敛”：

• 不改变节点与 router 的公开接口。
• 不改变 GraphInterrupt 的交互恢复语义。
• 不改变 checkpoint 数据结构。
• 不改变普通异常优先级。
• cancelAll() 仍保持强制取消。
• index 问题保持不动。

这不是全局 exactly-once 保证。如果节点产生外部副作用后主动抛出异常、JVM 崩溃或 router 自身失败，框架仍可能重试；这类场景需要幂等键、事务或补偿机制。

当前单元测试已经覆盖 INVOKING + sibling failure 窗口。合入前建议再增加一个确定性的 ROUTING + sibling failure 测试：让成功节点进入阻塞 router 后再触发兄弟异常，断言 router 不被 sibling failure 中断且节点不进入 pending。这是测试覆盖增强，不改变当前修复方向。

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[openjiuwen-collaboration-bot]

head_sha: 7bd16060b6449db7ef5a445494940c2fa7c8fac9

变更摘要

本 PR 修复测试用例 interruptrecovery093：此前将节点执行次数断言从精确值放宽为 isBetween(3, 6) 的做法不正确——times++ 位于节点的 invoke() 内，次数超过预期说明节点确实被重复执行。根因是 Java 虚拟线程并行调度下，兄弟节点失败时取消机制会丢弃已成功执行（已产生副作用）的节点结果，恢复时框架误判其失败而重新执行。本 PR 通过在 NodeTask、TaskExecutorPool 中引入"调用完成"生命周期回调与协作式取消语义，区分节点调用阶段与路由阶段，使已正常返回的调用不再被当作失败重复执行，并恢复精确次数断言。

主要改动

• NodeTask 增加调用生命周期回调: 新增 invocationCompletedHandler 字段（默认 NO_OP_INVOCATION_COMPLETED_HANDLER）及带该参数的新构造器，在 invokeFunc 正常返回后、throwIfInterrupted() 之前回调 invocationCompletedHandler.run()，用于标记节点调用已完成。
• RunningTask 状态机重构: TaskExecutorPool.RunningTask 用枚举 TaskPhase（NOT_STARTED/INVOKING/ROUTING/COMPLETED）和 TaskCancellation（NONE/SIBLING_FAILURE/FORCED）取代原 hasStarted、isCancellationRequested 布尔标志，并新增 cancelForSiblingFailure、markInvocationCompleted 方法区分处理调用阶段与路由阶段。
• 兄弟失败后的协作式取消: 新增 settlePendingTasksAfterFailure，在兄弟失败后对未完成 future 请求协作式取消并等待全部 settle；处于 INVOKING 阶段的任务仅中断执行线程，节点正常返回后由 markInvocationCompleted 清除协作式中断信号，避免成功结果被丢弃导致重复执行。
• 中断与取消语义修正: 捕获 InterruptedException 时恢复 Thread.currentThread().interrupt()；cancelAll（FORCED 取消）在节点调用返回后仍可中断 ROUTING 阶段，未启动任务取消时补调 complete() 完成状态。
• TaskExecutorPoolTest 新增回归用例: 新增 testSiblingFailurePreservesNormallyReturnedInvocation（验证正常返回的调用不被重复执行、兄弟失败中断不泄漏进路由）和 testCancelAllInterruptsRouting（验证 cancelAll 可中断路由阶段），并在既有用例中补充兄弟失败后节点保持 pending 以待恢复的断言。

[openjiuwen-collaboration-bot]

head_sha: 7bd16060b6449db7ef5a445494940c2fa7c8fac9

代码审查

✅ 未发现问题

[openjiuwen-collaboration-bot]

head_sha: 7bd16060b6449db7ef5a445494940c2fa7c8fac9

TASK	STATUS	DETAILS
CodeCheck	❌FAILED	Click here
AntiPoison	✅SUCCESS	Click here
Software Composition Analysis	❌FAILED	Click here
Maven Build	✅SUCCESS	See CHECK tab