fix(react-agent): migrate SkillTrackingMiddleware to reactor Context (cross-thread)

Task 6.1 遗留疑虑：SkillTrackingMiddleware 与 ChatUsage 同构的 ThreadLocal 跨线程
bug。真实 vendor 模型下 onActing 在 boundedElastic 调度线程执行（reasoningStream
的 flatMap 内构造 middleware 链），BOUND.get() 读不到 HTTP 线程写的 ThreadLocal →
usedSkills() 恒空。原 SkillTrackingMiddlewareTest 用进程内 mock（无 subscribeOn）
掩盖此 bug。

修复（镜像 ChatUsage 的 reactor Context 双源方案）：
- SkillTrackingMiddleware：新增 SKILL_CONTEXT_KEY 常量、bindAndReturn():Set<String>、
  bindToContext(Mono/Flux,Set)；onActing 改用 Flux.deferContextual 先读 Context
  再回退 ThreadLocal；保留 void bind() 维持二进制兼容。
- ReActAgentComponent.process()：流式 + 非流式两路径把 skill tracking 集合经
  bindToContext 注入 reactor Context（与 ChatUsage contextWrite 并列）。
- StreamingReplyModel.stream(...) 末尾加 .subscribeOn(boundedElastic)，镜像真实
  vendor 模型线程行为，消除"mock 掩盖 bug"。
- SkillTrackingMiddlewareTest 新增跨线程集成断言
  recordsSkillsAcrossSchedulerThreadJumpViaReactorContext（经回归窗口验证为有效断言）。

Tests run: 13, Failures: 0 (StreamingBridge/SkillTracking/ChatUsage/ProcessIntegration)。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

liteflow-react-agent/liteflow-react-agent-core/src/main/java/com/yomahub/liteflow/agent/component/ReActAgentComponent.java

@@ -23,6 +23,7 @@ import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 /**
  * 封装 agentscope agent 的 LiteFlow 抽象组件。
@@ -280,12 +281,13 @@ public abstract class ReActAgentComponent extends NodeComponent {
      *   <li>{@code finally} 中摘除 ctx，避免跨 invocation 悬挂引用。</li>
      * </ol>
      *
-     * <p><b>ChatUsage 线程安全（findings R-stream）：</b>真实 vendor 模型下，
-     * {@code ChatUsageMiddleware.onModelCall} 在 reactor 调度线程（boundedElastic）上
-     * 执行，<b>不在</b> {@code bind()} 的 HTTP 线程上。故 {@code bind()} 返回的累加器
-     * 同时经 {@link ChatUsageMiddleware#bindToContext} 注入 reactor Context，使 middleware
-     * 能在调度线程上经 {@code deferContextual} 读到。两条路径都注入（非流式也走同一
-     * Context 传播机制，行为一致）。
+     * <p><b>ChatUsage / SkillTracking 线程安全（findings R-stream）：</b>真实 vendor 模型下，
+     * {@code ChatUsageMiddleware.onModelCall} / {@code SkillTrackingMiddleware.onActing} 在
+     * reactor 调度线程（boundedElastic）上执行，<b>不在</b> {@code bind()} 的 HTTP 线程上。
+     * 故 {@code bindAndReturn()} 返回的累加器/已用技能集合同时经
+     * {@link ChatUsageMiddleware#bindToContext} / {@link SkillTrackingMiddleware#bindToContext}
+     * 注入 reactor Context，使 middleware 能在调度线程上经 {@code deferContextual} 读到。
+     * 两条路径都注入（非流式也走同一 Context 传播机制，行为一致）。
      *
      * <p>签名保持 {@code final}，与 1.0 一致。
      */
@@ -305,25 +307,28 @@ public abstract class ReActAgentComponent extends NodeComponent {
         slot.setAttachment(ctxKey(), ctx);
         // per-invocation 绑定 ChatUsage / Skill 累加器——agent 是单例、跨 process() 复用，
         // 不能跨 invocation 累加。在入口 bind、出口 finally unbind（见 findings R5）。
-        // bind() 返回的 ChatUsage 累加器还要注入 reactor Context，使流式/非流式路径下
+        // bind() 返回的累加器/集合还要注入 reactor Context，使流式/非流式路径下
         // middleware 在调度线程上都能读到（findings R-stream）。
         ChatUsageMiddleware.Accumulator usageAcc = ChatUsageMiddleware.bindAndReturn();
-        SkillTrackingMiddleware.bind();
+        Set<String> usedSkills = SkillTrackingMiddleware.bindAndReturn();
         try {
             Msg reply;
             if (FlowEventPublisher.hasListener(slot)) {
                 // 流式：streamEvents → AgentEventBridge → FlowEvent 发布；末尾 AGENT_RESULT 的 Msg 交回。
                 Msg userMsg = new UserMessage(userPrompt());
-                reply = ChatUsageMiddleware.bindToContext(
-                        AgentEventBridge.streamAndPublish(
-                                agent, userMsg, rc, slot,
-                                slot.getChainId(), getNodeId(), slot.getRequestId(), cid),
-                        usageAcc).block();
+                reactor.core.publisher.Mono<Msg> streamed = AgentEventBridge.streamAndPublish(
+                        agent, userMsg, rc, slot,
+                        slot.getChainId(), getNodeId(), slot.getRequestId(), cid);
+                streamed = ChatUsageMiddleware.bindToContext(streamed, usageAcc);
+                streamed = SkillTrackingMiddleware.bindToContext(streamed, usedSkills);
+                reply = streamed.block();
             } else {
                 // 非流式：call().block()。
-                reply = ChatUsageMiddleware.bindToContext(
-                        agent.call(List.of(new UserMessage(userPrompt())), rc),
-                        usageAcc).block();
+                reactor.core.publisher.Mono<Msg> called =
+                        agent.call(List.of(new UserMessage(userPrompt())), rc);
+                called = ChatUsageMiddleware.bindToContext(called, usageAcc);
+                called = SkillTrackingMiddleware.bindToContext(called, usedSkills);
+                reply = called.block();
             }
             handleReply(reply);
         } finally {

liteflow-react-agent/liteflow-react-agent-core/src/main/java/com/yomahub/liteflow/agent/middleware/SkillTrackingMiddleware.java

@@ -9,6 +9,7 @@ import io.agentscope.core.middleware.MiddlewareBase;
 import io.agentscope.core.skill.AgentSkill;
 import io.agentscope.core.skill.repository.AgentSkillRepository;
 import reactor.core.publisher.Flux;
+import reactor.util.context.ContextView;
 
 import java.util.Collections;
 import java.util.LinkedHashMap;
@@ -36,11 +37,30 @@ import java.util.function.Function;
  * <p>（对比 1.0：旧的 {@code SkillTrackingHook} 监听 {@code PostActingEvent}，取
  * {@code toolUse.getName()/getInput()}，逻辑等价——只是 v2 把这层从 hook 挪到了 middleware。）
  *
- * <h2>per-invocation 绑定</h2>
+ * <h2>per-invocation 绑定 —— 双源（reactor Context 优先，ThreadLocal 回退）</h2>
  * {@code ReActAgent} 单例、跨 {@code process()} 复用；本 middleware 同样单例。故已用技能集
  * 用 {@link ThreadLocal} 持有，{@link #bind()} 在 {@code process()} 入口 push、
  * {@link #unbind()} 在出口 pop。{@link #usedSkills()} 读当前线程绑定的集合（未 bind 返回空）。
  *
+ * <p><b>线程模型（findings R-stream）：</b>真实 vendor 模型（{@code OpenAIChatModel.stream}
+ * 实测第 180 行 {@code .subscribeOn(Schedulers.boundedElastic())}）下，middleware 的
+ * {@code onActing} 在 reactor 调度线程（boundedElastic）上执行（在 {@code reasoningStream}
+ * 的 {@code flatMap} 内），<b>不在</b> {@code process()} 调用 {@code bind()} 的 HTTP
+ * 线程上。故单纯的 ThreadLocal 在真实模型下 {@code BOUND.get()} 返回 null，技能记录被丢弃，
+ * {@code usedSkills()} 恒空。与 {@code ChatUsageMiddleware} 同构的修复方案：
+ * <ul>
+ *   <li>{@code process()} 在订阅前用 {@code contextWrite} 把已用技能集合塞进 reactor
+ *       {@link reactor.util.context.Context}（{@link #bindToContext(reactor.core.publisher.Mono, Set)}
+ *       / {@link #bindToContext(Flux, Set)} / {@link #SKILL_CONTEXT_KEY}）；</li>
+ *   <li>{@code onActing} 用 {@code Flux.deferContextual} 先读 reactor Context 里的集合；
+ *       没有（例如纯单元测试无 Context）才回退 ThreadLocal。</li>
+ * </ul>
+ * reactor Context 在 reactor 链上向上游传播、不受 {@code subscribeOn} 线程切换影响。
+ *
+ * <p>集合仍是 {@code synchronized} 的共享对象：写端（boundedElastic 线程的 add）与读端
+ * （HTTP 线程的 {@link #usedSkills()}）互斥、可见。ThreadLocal / Context 仅提供
+ * per-invocation 隔离，<b>不</b>保证单线程访问。
+ *
  * <h2>skillId → name 映射</h2>
  * 构造期从注入的 {@link AgentSkillRepository}（若有）遍历一次，建 {@code skillId → name} 表。
  * 用户在 {@code cmp.skills()} 写的是<b>裸 name</b>（findings R-skill (c)），故 {@code usedSkills()}
@@ -65,6 +85,15 @@ public class SkillTrackingMiddleware implements MiddlewareBase {
     public static final String LOAD_SKILL_TOOL_NAME = "load_skill_through_path";
     private static final String SKILL_ID_INPUT_KEY = "skillId";
 
+    /**
+     * reactor {@link reactor.util.context.Context} 上携带 per-invocation 已用技能集合的 key。
+     * {@code process()} 在 {@code call()}/{@code streamEvents()} 返回的 Mono/Flux 上
+     * {@code contextWrite(c -> c.put(SKILL_CONTEXT_KEY, set))} 注入；middleware 在
+     * {@link #onActing} 经 {@code deferContextual} 读取。
+     */
+    public static final String SKILL_CONTEXT_KEY =
+            "io.agentscope.liteflow.SkillTrackingMiddleware.usedSkills";
+
     /** skillId → 裸 name 映射（不可变；构造期一次性建立）。 */
     private final Map<String, String> skillIdToName;
 
@@ -77,9 +106,25 @@ public class SkillTrackingMiddleware implements MiddlewareBase {
 
     /**
      * 在当前线程绑定一个新的空"已用技能"集合。{@code process()} 入口调用。
+     *
+     * <p>返回类型保持 {@code void}（既有契约，二进制兼容）。需拿到集合引用（用于 reactor
+     * Context 注入）的调用方改用 {@link #bindAndReturn()}。
      */
     public static void bind() {
-        BOUND.set(Collections.synchronizedSet(new LinkedHashSet<>()));
+        BOUND.set(newSet());
+    }
+
+    /**
+     * 与 {@link #bind()} 相同，但返回新建的已用技能集合——供 {@code process()} 再通过
+     * {@link #bindToContext(reactor.core.publisher.Mono, Set)} 注入到 reactor Context，
+     * 使 middleware 在调度线程上能读到。
+     *
+     * @return 本次 invocation 新建的集合（{@code synchronized} 包装的 {@link LinkedHashSet}）
+     */
+    public static Set<String> bindAndReturn() {
+        Set<String> set = newSet();
+        BOUND.set(set);
+        return set;
     }
 
     /**
@@ -89,6 +134,37 @@ public class SkillTrackingMiddleware implements MiddlewareBase {
         BOUND.remove();
     }
 
+    /**
+     * 把已用技能集合注入 reactor {@link reactor.util.context.Context}，使下游任意调度线程上的
+     * middleware {@link #onActing} 都能经 {@code deferContextual} 读到。
+     *
+     * <p>用法（{@code process()} 内）：
+     * <pre>{@code
+     * Set<String> used = SkillTrackingMiddleware.bindAndReturn();
+     * Msg reply = SkillTrackingMiddleware.bindToContext(
+     *         agent.call(msgs, rc), used).block();
+     * }</pre>
+     *
+     * @param publisher 要附加 Context 的 reactor 源（call 返回的 Mono / streamEvents 返回的 Flux）
+     * @param used      本次 invocation 的已用技能集合（{@link #bindAndReturn()} 返回值）
+     * @param <T>       Mono/Flux 元素类型
+     * @return 带 {@link #SKILL_CONTEXT_KEY} 注入的同一源（contextWrite 返回新实例）
+     */
+    public static <T> reactor.core.publisher.Mono<T> bindToContext(
+            reactor.core.publisher.Mono<T> publisher, Set<String> used) {
+        return used == null ? publisher : publisher.contextWrite(c -> c.put(SKILL_CONTEXT_KEY, used));
+    }
+
+    /** {@link #bindToContext(reactor.core.publisher.Mono, Set)} 的 Flux 重载。 */
+    public static <T> Flux<T> bindToContext(Flux<T> publisher, Set<String> used) {
+        return used == null ? publisher : publisher.contextWrite(c -> c.put(SKILL_CONTEXT_KEY, used));
+    }
+
+    /** 构造一个 {@code synchronized} 包装的 {@link LinkedHashSet}（保证写/读跨线程互斥、可见）。 */
+    private static Set<String> newSet() {
+        return Collections.synchronizedSet(new LinkedHashSet<>());
+    }
+
     /**
      * 返回当前线程已用技能名列表（保持插入顺序）。未 bind 时返回空列表。
      *
@@ -112,13 +188,31 @@ public class SkillTrackingMiddleware implements MiddlewareBase {
             RuntimeContext ctx,
             ActingInput input,
             Function<ActingInput, Flux<AgentEvent>> next) {
-        Set<String> used = BOUND.get();
-        if (used != null && input != null && input.toolCalls() != null) {
-            for (ToolUseBlock t : input.toolCalls()) {
-                recordIfSkillLoad(t, used);
+        // 先在 caller 线程取 ThreadLocal 集合（回退路径：纯单元测试、或非 process() 触发的 acting）。
+        Set<String> threadLocalSet = BOUND.get();
+        // 用 deferContextual 先读 reactor Context 里的集合（真实模型下 middleware 在
+        // boundedElastic 调度线程执行，ThreadLocal 读不到——findings R-stream）；没有则回退
+        // ThreadLocal。无 toolCalls 或无集合时透传。
+        return Flux.deferContextual(cv -> {
+            Set<String> used = resolveUsedSet(cv, threadLocalSet);
+            if (used != null && input != null && input.toolCalls() != null) {
+                for (ToolUseBlock t : input.toolCalls()) {
+                    recordIfSkillLoad(t, used);
+                }
             }
+            return next.apply(input);
+        });
+    }
+
+    /** reactor Context 里的集合优先；没有则回退 ThreadLocal（兼容无 Context 的调用）。 */
+    private static Set<String> resolveUsedSet(ContextView cv, Set<String> threadLocalSet) {
+        Object fromCtx = cv == null ? null : cv.getOrDefault(SKILL_CONTEXT_KEY, null);
+        if (fromCtx instanceof Set) {
+            @SuppressWarnings("unchecked")
+            Set<String> fromCtxSet = (Set<String>) fromCtx;
+            return fromCtxSet;
         }
-        return next.apply(input);
+        return threadLocalSet;
     }
 
     private void recordIfSkillLoad(ToolUseBlock toolUse, Set<String> used) {

liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/SkillTrackingMiddlewareTest.java

@@ -10,10 +10,12 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
 
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.function.Function;
 
 import static org.mockito.Mockito.mock;
@@ -188,6 +190,57 @@ class SkillTrackingMiddlewareTest {
                 "unbind 后 usedSkills() 应为空（per-invocation 集合已被摘除）");
     }
 
+    /**
+     * <b>跨线程集成断言（findings R-stream）：</b>真实 vendor 模型下
+     * {@code SkillTrackingMiddleware.onActing} 在 reactor 调度线程（boundedElastic）上执行
+     * （{@code reasoningStream} 的 {@code flatMap} 内构造 middleware 链），<b>不在</b>
+     * {@code process()} 调 {@code bind()} 的 HTTP 线程上——单纯的 ThreadLocal 在调度线程读
+     * 不到（{@code BOUND.get()} 返回 null），记录被跳过，{@code usedSkills()} 恒空。
+     *
+     * <p>本用例复现该场景并验证 reactor Context 方案生效：把 {@code middleware.onActing(...)} 的
+     * <b>调用 + 订阅</b>整体放进 {@code Flux.defer(...).subscribeOn(boundedElastic)}——这样
+     * {@code onActing} 的方法体（含 {@code BOUND.get()} 的捕获与 {@code deferContextual} 的
+     * 求值）都在 boundedElastic 调度线程上执行，与 HTTP 线程（测试线程、bind 的线程）不同。
+     * 外层再 {@code contextWrite(c -> c.put(SKILL_CONTEXT_KEY, used))}（即
+     * {@link SkillTrackingMiddleware#bindToContext(Flux, Set)} 在 {@code process()} 里做的事）。
+     *
+     * <p>订阅完成后，同一 used 集合（也由 ThreadLocal 持有，供 {@code usedSkills()} 读）应非空。
+     * 该用例在 ThreadLocal-only 实现下会失败（{@code BOUND.get()} 在 boundedElastic 线程返回
+     * null → 记录被跳过 → usedSkills() 空）——守住这个回归窗口。
+     */
+    @Test
+    void recordsSkillsAcrossSchedulerThreadJumpViaReactorContext() {
+        AgentSkillRepository repo = mockRepo(List.of(stubSkill("research", "src")));
+        middleware = new SkillTrackingMiddleware(repo);
+
+        // next 透传，空 Flux 即可（onActing 记录发生在订阅链构造阶段，不依赖 next 的发射）。
+        Function<ActingInput, Flux<AgentEvent>> noopNext = input -> Flux.empty();
+
+        Set<String> used = SkillTrackingMiddleware.bindAndReturn();
+        try {
+            ToolUseBlock loadSkill = new ToolUseBlock(
+                    "cross-thread-1",
+                    SkillTrackingMiddleware.LOAD_SKILL_TOOL_NAME,
+                    Map.of("skillId", "research_src"));
+            ActingInput input = new ActingInput(List.of(loadSkill));
+
+            // 把 onActing 的调用 + 订阅整体放到 boundedElastic 调度线程，复现真实模型
+            // （middleware 链在 boundedElastic 上构造）。外层 contextWrite 注入 used 集合。
+            Flux<AgentEvent> pipeline = Flux.defer(() -> middleware.onActing(null, null, input, noopNext))
+                    .subscribeOn(Schedulers.boundedElastic())
+                    .contextWrite(c -> c.put(SkillTrackingMiddleware.SKILL_CONTEXT_KEY, used));
+            pipeline.collectList().block();
+
+            List<String> recorded = SkillTrackingMiddleware.usedSkills();
+            Assertions.assertEquals(List.of("research"), recorded,
+                    "跨调度线程（boundedElastic）下 reactor Context 必须把 used 集合传到 "
+                            + "onActing，使 load_skill_through_path 被正确记录；ThreadLocal-only "
+                            + "实现在此场景下 usedSkills() 会恒空（bug 回归窗口）");
+        } finally {
+            SkillTrackingMiddleware.unbind();
+        }
+    }
+
     /* ----- helpers ----- */
 
     /** 调 middleware.onActing 并同步消费完整个流（触发记录副作用）。 */

liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/StreamingReplyModel.java

@@ -8,6 +8,7 @@ import io.agentscope.core.model.ChatUsage;
 import io.agentscope.core.model.GenerateOptions;
 import io.agentscope.core.model.ToolSchema;
 import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
 
 import java.util.List;
 
@@ -29,6 +30,14 @@ import java.util.List;
  * 末尾 {@code agent.result}(last=true)，且流式路径下 {@code ctx.getChatUsage()} 应为
  * 100/40（验证 reactor Context 把累加器传到 middleware 调度线程）。
  *
+ * <p><b>跨线程（findings R-stream）：</b>{@code stream(...)} 返回的 Flux 末尾加
+ * {@code .subscribeOn(Schedulers.boundedElastic())}——镜像真实 vendor 模型
+ * （{@code OpenAIChatModel.stream} 第 180 行）的线程行为，使 middleware
+ * （{@code ChatUsageMiddleware}/{@code SkillTrackingMiddleware}）在 boundedElastic
+ * 调度线程上执行，<b>不在</b> {@code process()} 调 {@code bind()} 的 HTTP 线程上。
+ * 这样测试就能真正复现并守住 ThreadLocal 跨线程失效的场景（否则用 {@code Flux.just}
+ * 无 subscribeOn 的 mock 会掩盖该 bug）。
+ *
  * <p>无状态、线程安全；不依赖任何凭据。
  */
 final class StreamingReplyModel implements io.agentscope.core.model.Model {
@@ -63,7 +72,7 @@ final class StreamingReplyModel implements io.agentscope.core.model.Model {
                         .build())
                 .finishReason("stop")
                 .build();
-        return Flux.just(chunk1, chunk2);
+        return Flux.just(chunk1, chunk2).subscribeOn(Schedulers.boundedElastic());
     }
 
     @Override