From 199d8b8fc0b63da0d1d6907162bc413877bee15a Mon Sep 17 00:00:00 2001 From: "everywhere.z" Date: Sat, 20 Jun 2026 11:48:34 +0800 Subject: [PATCH] fix(react-agent): migrate SkillTrackingMiddleware to reactor Context (cross-thread) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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、 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 --- .../agent/component/ReActAgentComponent.java | 37 +++--- .../middleware/SkillTrackingMiddleware.java | 108 ++++++++++++++++-- .../agent/v2/SkillTrackingMiddlewareTest.java | 53 +++++++++ .../test/agent/v2/StreamingReplyModel.java | 11 +- 4 files changed, 185 insertions(+), 24 deletions(-) diff --git a/liteflow-react-agent/liteflow-react-agent-core/src/main/java/com/yomahub/liteflow/agent/component/ReActAgentComponent.java b/liteflow-react-agent/liteflow-react-agent-core/src/main/java/com/yomahub/liteflow/agent/component/ReActAgentComponent.java index 81f65682a..ef59c31b9 100644 --- a/liteflow-react-agent/liteflow-react-agent-core/src/main/java/com/yomahub/liteflow/agent/component/ReActAgentComponent.java +++ b/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 { *
  • {@code finally} 中摘除 ctx，避免跨 invocation 悬挂引用。
    • * * - *

    ChatUsage 线程安全（findings R-stream）：真实 vendor 模型下， - * {@code ChatUsageMiddleware.onModelCall} 在 reactor 调度线程（boundedElastic）上 - * 执行，不在 {@code bind()} 的 HTTP 线程上。故 {@code bind()} 返回的累加器 - * 同时经 {@link ChatUsageMiddleware#bindToContext} 注入 reactor Context，使 middleware - * 能在调度线程上经 {@code deferContextual} 读到。两条路径都注入（非流式也走同一 - * Context 传播机制，行为一致）。 + *

    ChatUsage / SkillTracking 线程安全（findings R-stream）：真实 vendor 模型下， + * {@code ChatUsageMiddleware.onModelCall} / {@code SkillTrackingMiddleware.onActing} 在 + * reactor 调度线程（boundedElastic）上执行，不在 {@code bind()} 的 HTTP 线程上。 + * 故 {@code bindAndReturn()} 返回的累加器/已用技能集合同时经 + * {@link ChatUsageMiddleware#bindToContext} / {@link SkillTrackingMiddleware#bindToContext} + * 注入 reactor Context，使 middleware 能在调度线程上经 {@code deferContextual} 读到。 + * 两条路径都注入（非流式也走同一 Context 传播机制，行为一致）。 * *

    签名保持 {@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 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 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 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 { diff --git a/liteflow-react-agent/liteflow-react-agent-core/src/main/java/com/yomahub/liteflow/agent/middleware/SkillTrackingMiddleware.java b/liteflow-react-agent/liteflow-react-agent-core/src/main/java/com/yomahub/liteflow/agent/middleware/SkillTrackingMiddleware.java index 7df262354..3f526554c 100644 --- a/liteflow-react-agent/liteflow-react-agent-core/src/main/java/com/yomahub/liteflow/agent/middleware/SkillTrackingMiddleware.java +++ b/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; *

    （对比 1.0：旧的 {@code SkillTrackingHook} 监听 {@code PostActingEvent}，取 * {@code toolUse.getName()/getInput()}，逻辑等价——只是 v2 把这层从 hook 挪到了 middleware。） * - *

    per-invocation 绑定

    + *

    per-invocation 绑定 —— 双源（reactor Context 优先，ThreadLocal 回退）

    * {@code ReActAgent} 单例、跨 {@code process()} 复用；本 middleware 同样单例。故已用技能集 * 用 {@link ThreadLocal} 持有，{@link #bind()} 在 {@code process()} 入口 push、 * {@link #unbind()} 在出口 pop。{@link #usedSkills()} 读当前线程绑定的集合（未 bind 返回空）。 * + *

    线程模型（findings R-stream）：真实 vendor 模型（{@code OpenAIChatModel.stream} + * 实测第 180 行 {@code .subscribeOn(Schedulers.boundedElastic())}）下，middleware 的 + * {@code onActing} 在 reactor 调度线程（boundedElastic）上执行（在 {@code reasoningStream} + * 的 {@code flatMap} 内），不在 {@code process()} 调用 {@code bind()} 的 HTTP + * 线程上。故单纯的 ThreadLocal 在真实模型下 {@code BOUND.get()} 返回 null，技能记录被丢弃， + * {@code usedSkills()} 恒空。与 {@code ChatUsageMiddleware} 同构的修复方案： + *

      + *
    • {@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}）；
      • + *
    • {@code onActing} 用 {@code Flux.deferContextual} 先读 reactor Context 里的集合； + * 没有（例如纯单元测试无 Context）才回退 ThreadLocal。
      • + *
    + * reactor Context 在 reactor 链上向上游传播、不受 {@code subscribeOn} 线程切换影响。 + * + *

    集合仍是 {@code synchronized} 的共享对象：写端（boundedElastic 线程的 add）与读端 + * （HTTP 线程的 {@link #usedSkills()}）互斥、可见。ThreadLocal / Context 仅提供 + * per-invocation 隔离，不保证单线程访问。 + * *

    skillId → name 映射

    * 构造期从注入的 {@link AgentSkillRepository}（若有）遍历一次，建 {@code skillId → name} 表。 * 用户在 {@code cmp.skills()} 写的是裸 name（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 skillIdToName; @@ -77,9 +106,25 @@ public class SkillTrackingMiddleware implements MiddlewareBase { /** * 在当前线程绑定一个新的空"已用技能"集合。{@code process()} 入口调用。 + * + *

    返回类型保持 {@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 bindAndReturn() { + Set 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} 读到。 + * + *

    用法（{@code process()} 内）： + * 

    {@code
+     * Set used = SkillTrackingMiddleware.bindAndReturn();
+     * Msg reply = SkillTrackingMiddleware.bindToContext(
+     *         agent.call(msgs, rc), used).block();
+     * }
    + * + * @param publisher 要附加 Context 的 reactor 源（call 返回的 Mono / streamEvents 返回的 Flux） + * @param used 本次 invocation 的已用技能集合（{@link #bindAndReturn()} 返回值） + * @param Mono/Flux 元素类型 + * @return 带 {@link #SKILL_CONTEXT_KEY} 注入的同一源（contextWrite 返回新实例） + */ + public static reactor.core.publisher.Mono bindToContext( + reactor.core.publisher.Mono publisher, Set used) { + return used == null ? publisher : publisher.contextWrite(c -> c.put(SKILL_CONTEXT_KEY, used)); + } + + /** {@link #bindToContext(reactor.core.publisher.Mono, Set)} 的 Flux 重载。 */ + public static Flux bindToContext(Flux publisher, Set used) { + return used == null ? publisher : publisher.contextWrite(c -> c.put(SKILL_CONTEXT_KEY, used)); + } + + /** 构造一个 {@code synchronized} 包装的 {@link LinkedHashSet}（保证写/读跨线程互斥、可见）。 */ + private static Set newSet() { + return Collections.synchronizedSet(new LinkedHashSet<>()); + } + /** * 返回当前线程已用技能名列表（保持插入顺序）。未 bind 时返回空列表。 * @@ -112,13 +188,31 @@ public class SkillTrackingMiddleware implements MiddlewareBase { RuntimeContext ctx, ActingInput input, Function> next) { - Set used = BOUND.get(); - if (used != null && input != null && input.toolCalls() != null) { - for (ToolUseBlock t : input.toolCalls()) { - recordIfSkillLoad(t, used); + // 先在 caller 线程取 ThreadLocal 集合（回退路径：纯单元测试、或非 process() 触发的 acting）。 + Set threadLocalSet = BOUND.get(); + // 用 deferContextual 先读 reactor Context 里的集合（真实模型下 middleware 在 + // boundedElastic 调度线程执行，ThreadLocal 读不到——findings R-stream）；没有则回退 + // ThreadLocal。无 toolCalls 或无集合时透传。 + return Flux.deferContextual(cv -> { + Set 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 resolveUsedSet(ContextView cv, Set threadLocalSet) { + Object fromCtx = cv == null ? null : cv.getOrDefault(SKILL_CONTEXT_KEY, null); + if (fromCtx instanceof Set) { + @SuppressWarnings("unchecked") + Set fromCtxSet = (Set) fromCtx; + return fromCtxSet; } - return next.apply(input); + return threadLocalSet; } private void recordIfSkillLoad(ToolUseBlock toolUse, Set used) { diff --git a/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/SkillTrackingMiddlewareTest.java b/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/SkillTrackingMiddlewareTest.java index d42fdfd48..6967ed188 100644 --- a/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/SkillTrackingMiddlewareTest.java +++ b/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 集合已被摘除）"); } + /** + * 跨线程集成断言（findings R-stream）：真实 vendor 模型下 + * {@code SkillTrackingMiddleware.onActing} 在 reactor 调度线程（boundedElastic）上执行 + * （{@code reasoningStream} 的 {@code flatMap} 内构造 middleware 链），不在 + * {@code process()} 调 {@code bind()} 的 HTTP 线程上——单纯的 ThreadLocal 在调度线程读 + * 不到（{@code BOUND.get()} 返回 null），记录被跳过，{@code usedSkills()} 恒空。 + * + *

    本用例复现该场景并验证 reactor Context 方案生效：把 {@code middleware.onActing(...)} 的 + * 调用 + 订阅整体放进 {@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()} 里做的事）。 + * + *

    订阅完成后，同一 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> noopNext = input -> Flux.empty(); + + Set 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 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 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 并同步消费完整个流（触发记录副作用）。 */ diff --git a/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/StreamingReplyModel.java b/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/StreamingReplyModel.java index 2e6d08dee..71e4a247c 100644 --- a/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/StreamingReplyModel.java +++ b/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 调度线程）。 * + *

    跨线程（findings R-stream）：{@code stream(...)} 返回的 Flux 末尾加 + * {@code .subscribeOn(Schedulers.boundedElastic())}——镜像真实 vendor 模型 + * （{@code OpenAIChatModel.stream} 第 180 行）的线程行为，使 middleware + * （{@code ChatUsageMiddleware}/{@code SkillTrackingMiddleware}）在 boundedElastic + * 调度线程上执行，不在 {@code process()} 调 {@code bind()} 的 HTTP 线程上。 + * 这样测试就能真正复现并守住 ThreadLocal 跨线程失效的场景（否则用 {@code Flux.just} + * 无 subscribeOn 的 mock 会掩盖该 bug）。 + * *

    无状态、线程安全；不依赖任何凭据。 */ 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