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>

Author: CodeCasterX

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

@@ -118,31 +118,11 @@ public Tip addAll(Map<String, Content> args) {
      * @return 表示当前的 {@link Tip}。
      */
     public Tip merge(Tip other) {
-        // === DIAGNOSTIC #5: Tip.merge() 开始 ===
-        System.err.println(String.format(
-            "[DIAG-Tip.merge-START] thread=%s, this=%s, other=%s, other_is_null=%b",
-            Thread.currentThread().getName(), this, other, (other == null)
-        ));
-
-        // 防御性处理：在并发场景下，Fork.join() 可能传入 null
-        // 参考：https://github.com/ModelEngine-Group/fit-framework/issues/247
+        // Issue #247: 防御性处理，在并发场景下 Fork.join() 可能传入 null
         if (other == null) {
-            System.err.println(String.format(
-                "[DIAG-Tip.merge-NULL] thread=%s, other is null, returning this=%s",
-                Thread.currentThread().getName(), this
-            ));
             return this;
         }
-
-        Tip result = this.addAll(other.values);
-
-        // === DIAGNOSTIC #6: Tip.merge() 结束 ===
-        System.err.println(String.format(
-            "[DIAG-Tip.merge-END] thread=%s, result=%s",
-            Thread.currentThread().getName(), result
-        ));
-
-        return result;
+        return this.addAll(other.values);
     }
 
     /**

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

@@ -594,23 +594,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) -> {
-            // === DIAGNOSTIC #3: AiStart reducer 调用 merge 之前 ===
-            System.err.println(String.format(
-                "[DIAG-AiStart:605-BEFORE] thread=%s, acc=%s, data=%s, data_is_null=%b",
-                Thread.currentThread().getName(), acc, data, (data == null)
-            ));
-
-            Tip mergeResult = acc.merge(data);  // Tip.merge() 内部会处理 null
-
-            // === DIAGNOSTIC #4: AiStart reducer 调用 merge 之后 ===
-            System.err.println(String.format(
-                "[DIAG-AiStart:605-AFTER] thread=%s, mergeResult=%s",
-                Thread.currentThread().getName(), mergeResult
-            ));
-
-            return mergeResult;
-        });
+        AiState<Tip, D, Tip, RF, F> state = aiFork.join(Tip::new, (acc, data) -> acc.merge(data));
         ((Processor<?, ?>) state.publisher()).displayAs("runnableParallel");
         return state;
     }

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

@@ -12,6 +12,7 @@
 import modelengine.fit.waterflow.domain.stream.reactive.Processor;
 import modelengine.fit.waterflow.domain.stream.reactive.Publisher;
 import modelengine.fit.waterflow.domain.utils.Tuple;
+import modelengine.fitframework.log.Logger;
 import modelengine.fitframework.util.ObjectUtils;
 
 import java.util.ArrayList;
@@ -33,6 +34,8 @@
  * @since 1.0
  */
 public class Fork<O, D, I, F extends Flow<D>> extends Activity<D, F> {
+    private static final Logger LOG = Logger.get(Fork.class);
+
     private final State<I, D, I, F> node;
 
     private final List<State<O, D, ?, F>> forks = new ArrayList<>();
@@ -94,21 +97,25 @@ public synchronized R process(FlowContext<O> input) {
                     }
                 }
 
-                // === DIAGNOSTIC #1: Fork.join wrapper 调用 processor 之前 ===
+                // Issue #247: 智能处理并发场景下的 null 数据
+                // 在某些竞态条件下，FlowContext.data 可能为 null
                 O inputData = input.getData();
-                System.err.println(String.format(
-                    "[DIAG-Fork:96-BEFORE] key=%s, thread=%s, branchCount=%d/%d, acc.first=%s, input.getData=%s, input.getData_is_null=%b",
-                    key, Thread.currentThread().getName(), acc.second(), forkNumber.get(),
-                    acc.first(), inputData, (inputData == null)
-                ));
+                if (inputData == null) {
+                    LOG.warn("[Fork.join] Received null FlowContext.data. "
+                            + "key={}, session={}, thread={}, branch={}/{}, acc={}",
+                            key, input.getSession().getId(), Thread.currentThread().getName(),
+                            acc.second() + 1, forkNumber.get(), acc.first());
 
-                R processedResult = processor.process(acc.first(), inputData);
+                    // 跳过此分支，不更新累加器
+                    // 如果是最后一个分支，返回已有数据（避免整个流程失败）
+                    if (acc.second() + 1 == forkNumber.get()) {
+                        accs.remove(key);
+                        return acc.first();
+                    }
+                    return null;
+                }
 
-                // === DIAGNOSTIC #2: Fork.join wrapper 调用 processor 之后 ===
-                System.err.println(String.format(
-                    "[DIAG-Fork:96-AFTER] key=%s, thread=%s, processedResult=%s, processedResult_is_null=%b",
-                    key, Thread.currentThread().getName(), processedResult, (processedResult == null)
-                ));
+                R processedResult = processor.process(acc.first(), inputData);
 
                 acc = Tuple.from(processedResult, acc.second() + 1);
                 accs.put(key, acc);