fix: 修复 Fork.join() 并发 NPE 问题 (#404)

* test: 添加 NPE 复现测试用于 GitHub Actions 验证

问题背景:
- PatternTest.shouldOkWhenAiFlowWithExampleSelector 在 GitHub Actions 中偶发失败
- 错误: NullPointerException at Tip.merge(Tip.java:121)
- 失败率: 0.5% (5次/1000次运行, 平均每22天一次)
- 只在 GitHub Actions 环境中出现，本地无法稳定复现

测试策略（TDD 红色阶段）:
1. shouldReproduceNPEInRunnableParallel (50次重复)
   - 使用 200ms 延迟制造快慢分支，增大竞态窗口
   - 预期：在 GitHub Actions 中应该能偶发触发 NPE

2. shouldReproduceOriginalTestFailure (20次重复)
   - 使用原始测试配置重复运行
   - 预期：在 GitHub Actions 中应该能偶发触发 NPE

验证目标:
- 如果这些测试在 GitHub Actions 中失败（NPE），证明测试有效
- 如果全部通过，需要调整延迟时间或重复次数
- 为后续修复提供可靠的验证基准

下一步:
- 推送到 99.99.x 分支触发 GitHub Actions
- 观察测试结果，确认能够复现 NPE
- 然后添加修复代码，再次验证

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* test: 增加 NPE 复现测试重复次数到 1000 次

调整原因:
- 第一次运行 74 个测试未触发 NPE（符合 0.5% 失败率的概率分布）
- 根据历史数据（5/1000 次失败），需要更多测试次数才能稳定复现

新的测试配置:
- shouldReproduceNPEInRunnableParallel: 50 → 500 次
- shouldReproduceOriginalTestFailure: 20 → 500 次
- 总计: 1004 个测试（原 4 个 + 1000 次重复）

预期结果:
- 根据 0.5% 失败率，1000 次运行预期触发约 5 次 NPE
- 99% 概率至少触发 1 次 NPE
- 如果成功复现，将验证测试有效性，然后添加修复代码

TDD 原则:
- 必须先看到红色（NPE 失败）
- 然后才能看到绿色（修复后通过）
- 这样才能确信修复是真正有效的

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* test: 简化 NPE 复现策略 - 只重复运行原始失败测试

优化原因:
- 之前运行了太多测试（1004 个），包括人工延迟的测试
- 更高效的方式：只重复运行历史上真正失败过的测试

新策略:
- 只运行 shouldOkWhenAiFlowWithExampleSelector 1000 次
- 使用真实的 ExampleSelector 配置（不添加人工延迟）
- 移除了模拟测试，更接近真实失败场景

预期:
- 运行更快（无人工延迟）
- 只运行 1003 个测试（原 3 个 + 1000 次重复）
- 根据 0.5% 失败率，1000 次应该触发约 5 次 NPE

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* fix: 修复 runnableParallel 中的 NPE 竞态条件

问题验证（TDD 红色阶段）:
✅ Run ID: 20535263656
✅ 测试失败：1003 个测试中 1 个错误
✅ 错误：NullPointerException at Tip.merge(Tip.java:121)
✅ 调用链：AiStart.runnableParallel → Fork.process → acc.merge(null)

根本原因:
- Fork.join() 设计上会在前 N-1 个分支完成时返回 null
- AiStart.runnableParallel() 没有正确处理这个 null 值
- 导致偶发的 acc.merge(null) → NPE

修复方案:
1. AiStart.java:596-600 - 在 reducer 中添加 null 检查，过滤 Fork 返回的 null 值
2. Tip.java:121 - 添加防御性 null 验证

预期效果（TDD 绿色阶段）:
- 1003 个测试全部通过
- NPE 失败率从 0.5% 降至 0%
- CI/CD 稳定性提升

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* fix: 修复 Tip.merge() 的 null 处理逻辑

问题分析:
✅ Run 20535263656: 成功复现 NPE（第 15 次重复）
❌ Run 20535342469: 第一次修复失败（导致 IllegalStateException: Required parameters are missing）

第一次修复的问题:
- 在 AiStart 中添加了 if (data != null) 检查
- 这导致 null 分支的数据被跳过
- 最终导致模板渲染时参数丢失

新的修复策略:
- 只在 Tip.merge() 中处理 null
- 如果 other 为 null，返回 this（不改变当前 Tip）
- 这样既避免了 NPE，也不会丢失数据

理由:
1. 更简单：只在一个地方处理 null
2. 更安全：不会导致数据丢失
3. 更合理：null 分支本来就没有数据可merge

预期效果:
- 1003 个测试全部通过
- NPE 失败率从 0.5% 降至 0%

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* feat: 添加诊断日志以定位 Fork.join() NPE 根本原因

添加详细的日志记录以追踪：
1. Fork.join() reducer 接收的每个数据值
2. Pattern 执行的输入和输出
3. null 值出现的完整堆栈跟踪
4. 线程信息以分析并发行为

这些日志将帮助我们理解为什么 input.getData() 会返回 null，
从而找到 NPE 的真正根源。

相关 issue: #247

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* fix: 使用项目标准 Logger 替代 Lombok @slf4j

将日志框架从 lombok.extern.slf4j.Slf4j 改为
modelengine.fitframework.log.Logger 以符合项目规范。

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* fix: 将诊断日志级别提升为 WARN 以确保输出

将所有诊断日志从 DEBUG 改为 WARN 级别，
确保在测试运行时能够捕获到这些关键信息。

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* debug: 在 Tip.merge() 中添加 System.err 诊断输出

使用 System.err.println 直接输出诊断信息，
确保无论日志配置如何都能看到输出。

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* fix: 防御性处理 Fork.join 并发竞态导致的 null 数据

## 问题描述
在并发场景下，Fork.join() 的 reducer 可能接收到 null 作为
data 参数，导致 NPE 或参数丢失。

## 根本原因
waterflow 框架在某些竞态条件下，可能传递 data 为 null 的
FlowContext 给 Fork 的 wrapper。详见 issue #247。

## 修复方案
采用防御性编程，在两个层面处理 null：

1. **AiStart.runnableParallel()** (主要修复):
   - 在 reducer 中检查 data 是否为 null
   - 如果为 null，记录警告并保持累加器不变
   - 避免 NPE 并保留已有数据

2. **Tip.merge()** (次要防御):
   - 保留 null 检查作为最后防线
   - 清理诊断代码，只保留核心逻辑

## 测试验证
- 本地测试：1000 次运行全部通过
- GitHub Actions：待验证

## 相关 Issue
Fixes #247

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* fix: 改进 null 数据处理逻辑，始终调用 Tip.merge() 以避免逻辑不一致

- 移除 AiStart 中的提前返回，始终调用 acc.merge(data)
- 依赖 Tip.merge() 内部的 null 检查来处理 null 情况
- 保留警告日志以便追踪并发竞态条件
- 这样可以保证处理逻辑的一致性，避免数据丢失

相关 issue: #247

* chore: 添加完整的诊断输出以追踪 NPE 根本原因

在关键调用链的每个层级添加 System.err 诊断输出：

1. Fork.java:96 - processor 调用前后
   - 记录 input.getData(), acc, branchCount 等关键状态

2. AiStart.java:605 - merge 调用前后
   - 记录 acc, data 参数和 merge 结果

3. Tip.merge() - merge 内部
   - 记录 this, other 参数和 null 处理逻辑

这将帮助我们理解：
- 哪个分支的数据为 null
- null 是在哪个环节产生的
- Fork 的聚合逻辑是如何执行的
- 完整的数据流动路径

相关 issue: #247

* fix: 修复 Fork.java 的类型转换编译错误

将 inputData 的类型从 Object 改为泛型 O，
以匹配 processor.process() 的参数类型要求。

* fix: 修复 Fork.join() 并发场景下的 null 数据处理 (#247)

问题根因：在并发场景下，Fork.join() 的 reducer 接收到 input.getData() = null，
导致 NPE 或数据丢失（"Required parameters are missing"）。

修复方案（阶段1）：
- Fork.java: 添加智能 null 处理，跳过 null 分支避免崩溃
- 使用 Logger.warn() 记录异常情况，便于监控
- 清理所有 System.err 诊断代码
- Tip.merge(): 保留防御性 null 检查

技术细节：
- 当 inputData 为 null 时，记录警告日志并跳过此分支
- 如果是最后一个分支，返回已有数据（避免整个流程失败）
- 保留 Tip.merge() 的 null 检查作为额外防御层

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix: 改进 Fork.join() null 数据处理逻辑

修复问题：之前的修复在检测到 null 数据且是最后一个分支时会返回已有数据，
导致聚合提前完成，丢失分支数据。

改进策略：当 inputData 为 null 时，不更新分支计数，直接返回 null，
等待正确的数据到来后正常完成聚合。

这样可以避免因竞态条件导致的数据丢失问题。

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* refactor: 清理代码并优化测试结构

1. 清理代码：
   - Fork.java: 移除注释和日志，保留简洁的 null 检查
   - Tip.java: 移除注释，保留防御性 null 检查

2. 优化测试结构：
   - shouldOkWhenAiFlowWithExampleSelector: 恢复为单次测试
   - shouldStableWhenRunnableParallelUnderConcurrency: 新增专门的并发稳定性测试（1000次重复）

详细的问题分析和修复说明已同步到 Issue #247

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix: 修复 To.java 中 peekedToken 可能为 null 的问题

在并发场景下，window.peekAndConsume() 可能返回 null，
导致后续调用 peekedToken.finishConsume() 时抛出 NPE。

添加 null 检查以防止此类错误。

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* chore: 移除 AiStart.java 中未使用的 Logger

清理之前添加诊断代码时引入的 Logger 导入和声明。

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: CodeCasterX

framework/fel/java/fel-core/src/main/java/modelengine/fel/core/util/Tip.java

@@ -118,6 +118,9 @@ public Tip addAll(Map<String, Content> args) {
      * @return 表示当前的 {@link Tip}。
      */
     public Tip merge(Tip other) {
+        if (other == null) {
+            return this;
+        }
         return this.addAll(other.values);
     }
 

framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/activities/AiStart.java

@@ -592,10 +592,7 @@ public final AiState<Tip, D, Tip, RF, F> runnableParallel(Pattern<O, Tip>... pat
                     .orElseGet(() -> new AiParallel<>(this.start.parallel(), mineFlow).fork(branchProcessor));
         }
 
-        AiState<Tip, D, Tip, RF, F> state = aiFork.join(Tip::new, (acc, data) -> {
-            acc.merge(data);
-            return acc;
-        });
+        AiState<Tip, D, Tip, RF, F> state = aiFork.join(Tip::new, (acc, data) -> acc.merge(data));
         ((Processor<?, ?>) state.publisher()).displayAs("runnableParallel");
         return state;
     }

framework/fel/java/fel-flow/src/test/java/modelengine/fel/engine/operators/PatternTest.java

@@ -42,6 +42,7 @@
 import modelengine.fitframework.util.StringUtils;
 
 import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.RepeatedTest;
 import org.junit.jupiter.api.Test;
 
 import java.util.Collection;
@@ -97,6 +98,23 @@ void shouldOkWhenAiFlowWithExampleSelector() {
         assertThat(converse.offer("1+2").await().text()).isEqualTo("2+2=4\n2+3=5\n1+2=");
     }
 
+    @RepeatedTest(1000)
+    @DisplayName("测试 RunnableParallel 并发稳定性")
+    void shouldStableWhenRunnableParallelUnderConcurrency() {
+        Example[] examples = {new DefaultExample("2+2", "4"), new DefaultExample("2+3", "5")};
+        Conversation<String, Prompt> converse = AiFlows.<String>create()
+                .runnableParallel(question(),
+                        fewShot(ExampleSelector.builder()
+                                .template("{{q}}={{a}}", "q", "a")
+                                .delimiter("\n")
+                                .example(examples)
+                                .build()))
+                .prompt(Prompts.human("{{examples}}\n{{question}}="))
+                .close()
+                .converse();
+        assertThat(converse.offer("1+2").await().text()).isEqualTo("2+2=4\n2+3=5\n1+2=");
+    }
+
     @Test
     @DisplayName("测试 Retriever")
     void shouldOkWhenAiFlowWithRetriever() {

framework/waterflow/java/waterflow-core/src/main/java/modelengine/fit/waterflow/domain/states/Fork.java

@@ -93,7 +93,13 @@ public synchronized R process(FlowContext<O> input) {
                         acc = Tuple.from((R) "", 0);
                     }
                 }
-                acc = Tuple.from(processor.process(acc.first(), input.getData()), acc.second() + 1);
+
+                O inputData = input.getData();
+                if (inputData == null) {
+                    return null;
+                }
+                R processedResult = processor.process(acc.first(), inputData);
+                acc = Tuple.from(processedResult, acc.second() + 1);
                 accs.put(key, acc);
 
                 if (acc.second() == forkNumber.get()) {

framework/waterflow/java/waterflow-core/src/main/java/modelengine/fit/waterflow/domain/stream/nodes/To.java

@@ -937,7 +937,9 @@ public <T1, R1> List<FlowContext<R1>> process(To<T1, R1> to, List<FlowContext<T1
                             nextSession.getWindow().complete();
                         }
                     } else {
-                        peekedToken.finishConsume();
+                        if (peekedToken != null) {
+                            peekedToken.finishConsume();
+                        }
                         if (window.isDone()) {
                             window.tryFinish();
                         }