> modelCallCore = mci -> modelCallStream(ctx, mci, true);
+ return MiddlewareChain.build(middlewares, this, rc, MiddlewareBase::onModelCall, modelCallCore)
+ .apply(new ModelCallInput(msgs, tools, options, model))
+ .doOnNext(this::publishEvent);
+}
+```
+
+`MiddlewareChain.build(...).apply(input)` 是同步的 Java 函数链构造——它在此刻
+调用 `middleware.onModelCall(...)`。该调用发生在订阅 {@code reasoningStream} 的那条线程上。
+
+### (c) 谁在订阅 reasoningStream？—— `.subscribeOn(Schedulers.boundedElastic())`
+
+`modelCallStream` 内部 `mci.model().stream(...)`，对真实 vendor 模型（实测
+`OpenAIChatModel.stream`，第 180 行）结尾是 `.subscribeOn(Schedulers.boundedElastic())`。
+故：当 reasoning 循环订阅 `reasoningStream` 时，整个上游（包括 `onModelCall` 的
+链构造）在 `boundedElastic` 工作线程上被订阅——不在 `process()` 调用
+`bind()` 的 HTTP 线程上。
+
+**关键结论：真实 vendor 模型下，无论 `call()` 还是 `streamEvents()`，middleware
+`onModelCall`/`onActing` 都在 `boundedElastic` 调度线程上被调用，而非 caller 线程。**
+（Task 5.1 的 `ChatUsageMiddlewareTest` / `ProcessIntegrationTest` 之所以"无线程跳变"
+通过，是因为它们用 `CannedReplyModel`——一个 `Flux.just(resp)` 的 mock，无 `subscribeOn`，
+故 emit 留在 caller 线程上。真实 HTTP 模型不满足此前提。）
+
+### (d) 这对 ThreadLocal 累加器意味着什么
+
+`ChatUsageMiddleware.onModelCall` 在方法体（链构造）里读 `BOUND.get()`——这是
+在 boundedElastic 线程上读的，而 `bind()` 在 HTTP 线程上写——读不到，返回 null，
+usage 丢失。（`Accumulator` 的 `synchronized` 只保证写端/读端互斥，不解决"读的线程
+压根没有累加器"的问题。）
+
+`SkillTrackingMiddleware.onActing` 同构问题：`BOUND.get()` 在 boundedElastic 上返回 null，
+`usedSkills()` 恒为空（Task 6.1 仅修复 ChatUsage；SkillTracking 的修复推迟，影响只是
+`usedSkills()` 在真实模型下为空，无 token 计费正确性问题）。
+
+### (e) 修复方案（采用 b：reactor Context）
+
+Reactor `Context` 在 reactor 链上向上游传播（upstream propagation），不受
+`.subscribeOn`/`.publishOn` 线程切换影响——故只要 `process()` 在订阅前用
+`contextWrite` 把累加器塞进 Context，下游任意调度线程上的 middleware 都能读到。
+
+RC3 `buildAgentStream` 已经 `.contextWrite(c -> c.put(RUNTIME_CONTEXT_KEY, context))`
+和 `.contextWrite(c -> c.put(EVENT_SINK_KEY, sink))`——证明 Context 传播在 RC3 设计中
+是首选机制。Task 6.1 的 ChatUsage 累加器走同一路径：
+
+- `ChatUsageMiddleware` 的累加器改为"先看 reactor ContextView，没有再回退 ThreadLocal"
+ （保留 ThreadLocal 回退使 {@code ChatUsageMiddlewareTest} 这种无 Context 的纯单元测试
+ 继续可用——它直接 `bind()` 后调 `onModelCall`，无 reactor Context）。
+- `process()` 在 `call(...)` 和 `streamEvents(...)` 返回的 `Mono`/`Flux` 上都
+ `.contextWrite(c -> c.put(USAGE_ACC_KEY, acc))`，`acc` 是 bind 时创建的同一个实例，
+ 同时被 ThreadLocal 持有（供 HTTP 线程上的 `snapshot()` 读）。
+- 累加器仍是带 `synchronized` 的共享对象，HTTP 线程读 / boundedElastic 线程写互斥安全。
+
+### (f) 验证
+
+`StreamingBridgeTest` 用 mock 模型回放 `ModelCallEndEvent`（带 ChatUsage）+
+`AgentResultEvent`，断言 `streamEvents` 路径下 `ctx.getChatUsage()` 仍拿到累计值——
+即 reactor Context 方案在流式路径下生效。`ChatUsageMiddlewareTest`（ThreadLocal 回退路径）
+保持不变、继续 PASS。
+
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 9f22d95fe..81f65682a 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
@@ -1,10 +1,12 @@
package com.yomahub.liteflow.agent.component;
+import com.yomahub.liteflow.agent.event.AgentEventBridge;
import com.yomahub.liteflow.agent.exception.AgentConfigException;
import com.yomahub.liteflow.agent.middleware.ChatUsageMiddleware;
import com.yomahub.liteflow.agent.middleware.SkillTrackingMiddleware;
import com.yomahub.liteflow.agent.model.ModelSpec;
import com.yomahub.liteflow.core.NodeComponent;
+import com.yomahub.liteflow.flow.FlowEventPublisher;
import com.yomahub.liteflow.property.LiteflowConfigGetter;
import com.yomahub.liteflow.property.agent.AgentConfig;
import com.yomahub.liteflow.slot.Slot;
@@ -256,7 +258,8 @@ public abstract class ReActAgentComponent extends NodeComponent {
/* ===== 框架 final 执行体 ===== */
/**
- * 端到端非流式执行（RC3）。
+ * 端到端执行（RC3）：有 {@code FlowEvent} 监听者时走流式（{@code streamEvents} →
+ * {@link AgentEventBridge} → {@code FlowEvent} 发布），否则走非流式 {@code call()}。
*
* 流程：
*
@@ -267,14 +270,22 @@ public abstract class ReActAgentComponent extends NodeComponent {
* - 解析 {@code conversationId} / {@code agentKey}，写回 slot，并构造对应的
* {@link RuntimeContext}（{@code userId=conversationId, sessionId=agentKey}）。
* - 把 {@link ReActAgentContext}（含 runtimeContext）挂到 slot attachment 上，
- * 使 {@link #ctx()} 在 {@code agent.call(...)} 触发的工具回调内可用。
- * - 调 {@code agent.call(List.of(new UserMessage(userPrompt())), rc).block()}
- * 阻塞拿回复，交 {@link #handleReply(Msg)} 处理。
+ * 使 {@link #ctx()} 在 {@code agent.call(...)}/{@code streamEvents(...)} 触发的
+ * 工具回调内可用。
+ * - 分流：{@link FlowEventPublisher#hasListener} 为真 →
+ * {@link AgentEventBridge#streamAndPublish}（订阅 {@code streamEvents}，映射成
+ * {@code agent.reasoning/tool_result/summary/result} 等事件发布）；
+ * 否则 {@code agent.call(...).block()}（非流式）。
+ * - 两条路径的回复都交 {@link #handleReply(Msg)} 处理。
* - {@code finally} 中摘除 ctx，避免跨 invocation 悬挂引用。
*
*
- * 非流式：本方法始终用 {@code call(...)}。流式（{@code streamEvents}）桥接
- * 由 Task 6.1 单独实现，不在本方法范围内。
+ *
ChatUsage 线程安全（findings R-stream）：真实 vendor 模型下，
+ * {@code ChatUsageMiddleware.onModelCall} 在 reactor 调度线程（boundedElastic）上
+ * 执行，不在 {@code bind()} 的 HTTP 线程上。故 {@code bind()} 返回的累加器
+ * 同时经 {@link ChatUsageMiddleware#bindToContext} 注入 reactor Context，使 middleware
+ * 能在调度线程上经 {@code deferContextual} 读到。两条路径都注入（非流式也走同一
+ * Context 传播机制，行为一致）。
*
*
签名保持 {@code final}，与 1.0 一致。
*/
@@ -294,10 +305,26 @@ public abstract class ReActAgentComponent extends NodeComponent {
slot.setAttachment(ctxKey(), ctx);
// per-invocation 绑定 ChatUsage / Skill 累加器——agent 是单例、跨 process() 复用，
// 不能跨 invocation 累加。在入口 bind、出口 finally unbind（见 findings R5）。
- ChatUsageMiddleware.bind();
+ // bind() 返回的 ChatUsage 累加器还要注入 reactor Context，使流式/非流式路径下
+ // middleware 在调度线程上都能读到（findings R-stream）。
+ ChatUsageMiddleware.Accumulator usageAcc = ChatUsageMiddleware.bindAndReturn();
SkillTrackingMiddleware.bind();
try {
- Msg reply = agent.call(List.of(new UserMessage(userPrompt())), rc).block();
+ 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();
+ } else {
+ // 非流式：call().block()。
+ reply = ChatUsageMiddleware.bindToContext(
+ agent.call(List.of(new UserMessage(userPrompt())), rc),
+ usageAcc).block();
+ }
handleReply(reply);
} finally {
ChatUsageMiddleware.unbind();
diff --git a/liteflow-react-agent/liteflow-react-agent-core/src/main/java/com/yomahub/liteflow/agent/event/AgentEventBridge.java b/liteflow-react-agent/liteflow-react-agent-core/src/main/java/com/yomahub/liteflow/agent/event/AgentEventBridge.java
new file mode 100644
index 000000000..9f92fcf76
--- /dev/null
+++ b/liteflow-react-agent/liteflow-react-agent-core/src/main/java/com/yomahub/liteflow/agent/event/AgentEventBridge.java
@@ -0,0 +1,191 @@
+package com.yomahub.liteflow.agent.event;
+
+import com.yomahub.liteflow.flow.FlowEvent;
+import com.yomahub.liteflow.flow.FlowEventPublisher;
+import com.yomahub.liteflow.slot.Slot;
+import io.agentscope.core.ReActAgent;
+import io.agentscope.core.agent.RuntimeContext;
+import io.agentscope.core.event.AgentEventType;
+import io.agentscope.core.event.AgentResultEvent;
+import io.agentscope.core.event.HintBlockEvent;
+import io.agentscope.core.event.RequireExternalExecutionEvent;
+import io.agentscope.core.event.RequireUserConfirmEvent;
+import io.agentscope.core.event.TextBlockDeltaEvent;
+import io.agentscope.core.event.ToolResultDataDeltaEvent;
+import io.agentscope.core.event.ToolResultEndEvent;
+import io.agentscope.core.event.ToolResultTextDeltaEvent;
+import io.agentscope.core.message.Msg;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.List;
+
+/**
+ * 把 {@link ReActAgent#streamEvents} 的细粒度 {@link io.agentscope.core.event.AgentEvent}
+ * 流映射成 LiteFlow {@link FlowEvent}，经 {@link FlowEventPublisher} 发布给当次 {@link Slot}
+ * 上注册的监听者（{@code ExecuteOption.eventListener}）。
+ *
+ *
这是 Task 6.1 恢复的流式路径：{@code ReActAgentComponent.process()} 在 slot 有监听者时
+ * 调 {@link #streamAndPublish}，否则仍走非流式 {@code agent.call(...)}。
+ *
+ *
事件映射（保对外 type 字符串与 1.0 一致）
+ *
+ * - {@link AgentEventType#TEXT_BLOCK_DELTA}（reasoning 文本增量）→
+ * {@code agent.reasoning}（last=false）。
+ * - {@link AgentEventType#TOOL_RESULT_TEXT_DELTA} /
+ * {@link AgentEventType#TOOL_RESULT_DATA_DELTA} /
+ * {@link AgentEventType#TOOL_RESULT_END} →
+ * {@code agent.tool_result}（last=false；delta 透传文本，end 透传工具名）。
+ * - {@link AgentEventType#HINT_BLOCK}（RC3 的 summary/hint 信号）→
+ * {@code agent.summary}（last=false）。
+ * - {@link AgentEventType#AGENT_RESULT}（最终回复）→
+ * {@code agent.result}（last=true，text 取 {@link AgentResultEvent#getResult()} 的
+ * 文本；并把该 Msg 作为 {@link Mono} 的返回值交回 {@code process()}）。
+ * - {@link AgentEventType#REQUIRE_USER_CONFIRM} /
+ * {@link AgentEventType#REQUIRE_EXTERNAL_EXECUTION}（HITL 类，RC3 已存在）→
+ * {@code agent.hitl.confirm} / {@code agent.hitl.external_exec}（last=false，
+ * 可选透传，data 携带原始事件，便于业务侧自定义 HITL 处理）。
+ *
+ *
+ * 其余事件类型（{@code MODEL_CALL_*}、{@code TEXT_BLOCK_START/END}、
+ * {@code TOOL_CALL_*}、{@code THINKING_BLOCK_*} 等）不直接映射成 FlowEvent——它们是
+ * 中间态信号，对外暴露的"用户可观测事件"语义由 reasoning/tool_result/summary/result 承担。
+ *
+ *
线程安全
+ * {@code streamEvents} 的 emit 在 reactor 调度线程上（findings R-stream），但
+ * {@link FlowEventPublisher#publish} 是无状态静态方法（仅读 slot attachment 调监听者回调），
+ * 监听者实现（由调用方经 {@code ExecuteOption.eventListener} 提供）自行保证线程安全——
+ * 典型实现用 {@code CopyOnWriteArrayList} 收集事件。本桥不持有跨 invocation 状态。
+ */
+public final class AgentEventBridge {
+
+ /** HITL：要求用户确认（透传用，可选）。 */
+ public static final String FLOW_EVENT_TYPE_HITL_CONFIRM = "agent.hitl.confirm";
+ /** HITL：要求外部执行（透传用，可选）。 */
+ public static final String FLOW_EVENT_TYPE_HITL_EXTERNAL_EXEC = "agent.hitl.external_exec";
+
+ private AgentEventBridge() {
+ }
+
+ /**
+ * 流式执行 agent 并把 {@link io.agentscope.core.event.AgentEvent} 桥接成
+ * {@link FlowEvent} 发布；捕获 {@link AgentResultEvent} 的最终 Msg 作为返回值。
+ *
+ * @param agent 已构建好的 {@link ReActAgent} 单例
+ * @param userMsg 本次用户输入（{@code process()} 构造的 UserMessage）
+ * @param rc 本次调用的 {@link RuntimeContext}（{@code userId=conversationId,
+ * sessionId=agentKey}）
+ * @param slot 当次执行 slot（事件经 {@link FlowEventPublisher#publish} 发到其监听者）
+ * @param chainId chain id（填入 FlowEvent.chainId）
+ * @param nodeId 组件 nodeId（填入 FlowEvent.nodeId）
+ * @param requestId 请求 id（填入 FlowEvent.requestId，可空）
+ * @param conversationId 会话 id（填入 FlowEvent.conversationId）
+ * @return {@link Mono}，emit 流完成后携带 {@link AgentResultEvent} 的最终 Msg；
+ * 若流未发 {@code AGENT_RESULT}，则携带一个空文本 Msg 兜底
+ */
+ public static Mono streamAndPublish(
+ ReActAgent agent,
+ Msg userMsg,
+ RuntimeContext rc,
+ Slot slot,
+ String chainId,
+ String nodeId,
+ String requestId,
+ String conversationId) {
+ return agent.streamEvents(List.of(userMsg), rc)
+ .doOnNext(event -> publishMapped(slot, event, chainId, nodeId, requestId, conversationId))
+ .filter(event -> event.getType() == AgentEventType.AGENT_RESULT)
+ .next()
+ .map(event -> ((AgentResultEvent) event).getResult())
+ .switchIfEmpty(Mono.fromSupplier(() -> Msg.builder().textContent("").build()));
+ }
+
+ /** 把单个 {@link io.agentscope.core.event.AgentEvent} 映射并发布成 {@link FlowEvent}。 */
+ private static void publishMapped(
+ Slot slot,
+ io.agentscope.core.event.AgentEvent event,
+ String chainId,
+ String nodeId,
+ String requestId,
+ String conversationId) {
+ AgentEventType type = event.getType();
+ if (type == null) {
+ return;
+ }
+ switch (type) {
+ case TEXT_BLOCK_DELTA: {
+ String delta = event instanceof TextBlockDeltaEvent d ? d.getDelta() : null;
+ publish(slot, com.yomahub.liteflow.agent.component.ReActAgentComponent.FLOW_EVENT_TYPE_REASONING,
+ delta, false, null, chainId, nodeId, requestId, conversationId);
+ break;
+ }
+ case TOOL_RESULT_TEXT_DELTA: {
+ String delta = event instanceof ToolResultTextDeltaEvent d ? d.getDelta() : null;
+ publish(slot, com.yomahub.liteflow.agent.component.ReActAgentComponent.FLOW_EVENT_TYPE_TOOL_RESULT,
+ delta, false, null, chainId, nodeId, requestId, conversationId);
+ break;
+ }
+ case TOOL_RESULT_DATA_DELTA: {
+ Object data = event instanceof ToolResultDataDeltaEvent d ? d.getData() : null;
+ publish(slot, com.yomahub.liteflow.agent.component.ReActAgentComponent.FLOW_EVENT_TYPE_TOOL_RESULT,
+ null, false, data, chainId, nodeId, requestId, conversationId);
+ break;
+ }
+ case TOOL_RESULT_END: {
+ String toolName = event instanceof ToolResultEndEvent e ? e.getToolCallName() : null;
+ publish(slot, com.yomahub.liteflow.agent.component.ReActAgentComponent.FLOW_EVENT_TYPE_TOOL_RESULT,
+ toolName, false, null, chainId, nodeId, requestId, conversationId);
+ break;
+ }
+ case HINT_BLOCK: {
+ String hint = event instanceof HintBlockEvent h ? h.getHint() : null;
+ publish(slot, com.yomahub.liteflow.agent.component.ReActAgentComponent.FLOW_EVENT_TYPE_SUMMARY,
+ hint, false, null, chainId, nodeId, requestId, conversationId);
+ break;
+ }
+ case AGENT_RESULT: {
+ Msg result = event instanceof AgentResultEvent r ? r.getResult() : null;
+ String text = result == null ? null : result.getTextContent();
+ publish(slot, com.yomahub.liteflow.agent.component.ReActAgentComponent.FLOW_EVENT_TYPE_RESULT,
+ text, true, result, chainId, nodeId, requestId, conversationId);
+ break;
+ }
+ case REQUIRE_USER_CONFIRM: {
+ // RequireUserConfirmEvent 透传为 data（业务侧自定义 HITL 处理读 data 即可）。
+ if (event instanceof RequireUserConfirmEvent) {
+ publish(slot, FLOW_EVENT_TYPE_HITL_CONFIRM, null, false, event,
+ chainId, nodeId, requestId, conversationId);
+ }
+ break;
+ }
+ case REQUIRE_EXTERNAL_EXECUTION: {
+ // 仅在事件确实是 RequireExternalExecutionEvent 时透传（保持 data 类型一致）。
+ if (event instanceof RequireExternalExecutionEvent) {
+ publish(slot, FLOW_EVENT_TYPE_HITL_EXTERNAL_EXEC, null, false, event,
+ chainId, nodeId, requestId, conversationId);
+ }
+ break;
+ }
+ default:
+ // 其余事件类型（MODEL_CALL_*、TEXT_BLOCK_START/END、TOOL_CALL_*、
+ // THINKING_BLOCK_*、DATA_BLOCK_*、AGENT_START/END 等）不映射成 FlowEvent。
+ break;
+ }
+ }
+
+ private static void publish(
+ Slot slot, String type, String text, boolean last, Object data,
+ String chainId, String nodeId, String requestId, String conversationId) {
+ FlowEvent event = FlowEvent.builder()
+ .type(type)
+ .chainId(chainId)
+ .nodeId(nodeId)
+ .requestId(requestId)
+ .conversationId(conversationId)
+ .text(text)
+ .last(last)
+ .data(data)
+ .build();
+ FlowEventPublisher.publish(slot, event);
+ }
+}
diff --git a/liteflow-react-agent/liteflow-react-agent-core/src/main/java/com/yomahub/liteflow/agent/middleware/ChatUsageMiddleware.java b/liteflow-react-agent/liteflow-react-agent-core/src/main/java/com/yomahub/liteflow/agent/middleware/ChatUsageMiddleware.java
index f771c7417..b564bbbd2 100644
--- a/liteflow-react-agent/liteflow-react-agent-core/src/main/java/com/yomahub/liteflow/agent/middleware/ChatUsageMiddleware.java
+++ b/liteflow-react-agent/liteflow-react-agent-core/src/main/java/com/yomahub/liteflow/agent/middleware/ChatUsageMiddleware.java
@@ -8,6 +8,8 @@ import io.agentscope.core.middleware.MiddlewareBase;
import io.agentscope.core.middleware.ModelCallInput;
import io.agentscope.core.model.ChatUsage;
import reactor.core.publisher.Flux;
+import reactor.util.context.Context;
+import reactor.util.context.ContextView;
import java.util.function.Function;
@@ -21,17 +23,29 @@ import java.util.function.Function;
* {@code next.apply(input)} 返回的 {@code Flux} 里订阅
* {@link ModelCallEndEvent} 并把当步 usage 累加到一个 per-invocation 累加器。
*
- * per-invocation 绑定
+ * per-invocation 绑定 —— 双源（reactor Context 优先，ThreadLocal 回退）
* {@code ReActAgent} 被 {@code ReactAgentFactory} 按 cmp 子类缓存为单例、跨
* {@code process()} 复用，不能把累加器做成实例字段直接累加——否则会把上次
- * 调用的余量带入下一次。故累加器用 {@link ThreadLocal} 持有，{@link #bind()} 在
- * {@code process()} 入口调用（push 新累加器），{@link #unbind()} 在出口 {@code finally}
- * 调用（pop 并丢弃）。
+ * 调用的余量带入下一次。
*
- * 之所以用 ThreadLocal 而非 RuntimeContext：RC3-core 下 {@code process()} 用
- * {@code .block()} 同步执行整条 ReAct 循环，模型调用在同一线程上完成；middleware
- * 在该线程上被调用。子类若未来引入异步流式（Task 6.1），需相应把累加器改成随
- * reactor {@code Context} 传播——RC3-core 不在此范围内。
+ *
线程模型（findings R-stream）：真实 vendor 模型（{@code OpenAIChatModel.stream}
+ * 实测第 180 行 {@code .subscribeOn(Schedulers.boundedElastic())}）下，middleware 的
+ * {@code onModelCall} 链构造在 reactor 调度线程（boundedElastic）上执行，不在
+ * {@code process()} 调用 {@code bind()} 的 HTTP 线程上。故单纯的 ThreadLocal 在真实模型
+ * 下读不到累加器（usage 丢失）。修复方案：
+ *
+ * - {@code process()} 在订阅前用 {@code contextWrite} 把累加器塞进 reactor
+ * {@link Context}（{@link #bindToContext(Flux, Accumulator)} / {@link #USAGE_CONTEXT_KEY}）；
+ * - {@code onModelCall} 用 {@code Flux.deferContextual} 先读 reactor Context 里的累加器；
+ * 没有（例如纯单元测试无 Context）才回退 ThreadLocal。
+ *
+ * reactor Context 在 reactor 链上向上游传播、不受 {@code subscribeOn} 线程切换影响——
+ * 这是 RC3 内部 {@code buildAgentStream} 传 {@code EVENT_SINK_KEY}/{@code RUNTIME_CONTEXT_KEY}
+ * 的同一机制。
+ *
+ * 累加器仍是带 {@code synchronized} 的共享对象：写端（boundedElastic 线程的 add）与
+ * 读端（HTTP 线程的 {@link #snapshot()}，即 {@code ctx.getChatUsage()}）互斥、可见。
+ * ThreadLocal / Context 仅提供 per-invocation 隔离，不保证单线程访问。
*
*
读累计值
* {@link com.yomahub.liteflow.agent.component.ReActAgentContext#getChatUsage()} 通过
@@ -40,6 +54,15 @@ import java.util.function.Function;
*/
public class ChatUsageMiddleware implements MiddlewareBase {
+ /**
+ * reactor {@link Context} 上携带 per-invocation {@link Accumulator} 的 key。
+ * {@code process()} 在 {@code call()}/{@code streamEvents()} 返回的 Mono/Flux 上
+ * {@code contextWrite(c -> c.put(USAGE_CONTEXT_KEY, acc))} 注入；middleware 在
+ * {@link #onModelCall} 经 {@code deferContextual} 读取。
+ */
+ public static final String USAGE_CONTEXT_KEY =
+ "io.agentscope.liteflow.ChatUsageMiddleware.accumulator";
+
/** per-invocation 累加器栈：bind push、unbind pop（栈结构支持嵌套，虽当前 process() 不嵌套）。 */
private static final ThreadLocal BOUND = new ThreadLocal<>();
@@ -49,11 +72,27 @@ public class ChatUsageMiddleware implements MiddlewareBase {
/**
* 在当前线程绑定一个新的 per-invocation 累加器。必须在 {@code process()} 入口调用，
* 出口 {@link #unbind()} 清零。
+ *
+ * 返回类型保持 {@code void}（Task 5.1 既定契约，二进制兼容）。需拿到累加器引用
+ * （用于 reactor Context 注入）的调用方改用 {@link #bindAndReturn()}。
*/
public static void bind() {
BOUND.set(new Accumulator());
}
+ /**
+ * 与 {@link #bind()} 相同，但返回新建的累加器实例——供 {@code process()} 再通过
+ * {@link #bindToContext(reactor.core.publisher.Mono, Accumulator)} 注入到 reactor
+ * Context，使 middleware 在调度线程上能读到。
+ *
+ * @return 本次 invocation 新建的累加器
+ */
+ public static Accumulator bindAndReturn() {
+ Accumulator acc = new Accumulator();
+ BOUND.set(acc);
+ return acc;
+ }
+
/**
* 摘除当前线程绑定的累加器。必须在 {@code process()} 出口（{@code finally}）调用，
* 避免单例 middleware 跨 invocation 累加。
@@ -66,8 +105,8 @@ public class ChatUsageMiddleware implements MiddlewareBase {
* 返回当前线程累加器截至当前累计的 token 用量；未 bind 或未观察到任何 usage 时
* 返回 {@code null}。
*
- *
静态访问：累加器本身是 ThreadLocal，与具体 middleware 实例无关；故
- * {@link com.yomahub.liteflow.agent.component.ReActAgentContext#getChatUsage()}
+ *
静态访问：累加器本身是 ThreadLocal（+ reactor Context 双源），与具体 middleware
+ * 实例无关；故 {@link com.yomahub.liteflow.agent.component.ReActAgentContext#getChatUsage()}
* 可直接读，无需持有 middleware 引用。
*/
public static ChatUsage snapshot() {
@@ -75,40 +114,86 @@ public class ChatUsageMiddleware implements MiddlewareBase {
return acc == null ? null : acc.snapshot();
}
+ /**
+ * 把累加器注入 reactor {@link Context}，使下游任意调度线程上的 middleware
+ * {@link #onModelCall} 都能经 {@code deferContextual} 读到。
+ *
+ *
用法（{@code process()} 内）：
+ *
{@code
+ * Accumulator acc = ChatUsageMiddleware.bind();
+ * Msg reply = ChatUsageMiddleware.bindToContext(
+ * agent.call(msgs, rc), acc).block();
+ * }
+ *
+ * @param publisher 要附加 Context 的 reactor 源（call 返回的 Mono / streamEvents 返回的 Flux）
+ * @param acc 本次 invocation 的累加器（{@link #bind()} 返回值）
+ * @param Mono/Flux 元素类型
+ * @return 带 {@link #USAGE_CONTEXT_KEY} 注入的同一源（contextWrite 返回新实例）
+ */
+ public static reactor.core.publisher.Mono bindToContext(
+ reactor.core.publisher.Mono publisher, Accumulator acc) {
+ return acc == null ? publisher : publisher.contextWrite(c -> c.put(USAGE_CONTEXT_KEY, acc));
+ }
+
+ /**
+ * {@link #bindToContext(reactor.core.publisher.Mono, Accumulator)} 的 Flux 重载。
+ */
+ public static Flux bindToContext(Flux publisher, Accumulator acc) {
+ return acc == null ? publisher : publisher.contextWrite(c -> c.put(USAGE_CONTEXT_KEY, acc));
+ }
+
@Override
public Flux onModelCall(
Agent agent,
RuntimeContext ctx,
ModelCallInput input,
Function> next) {
- Flux downstream = next.apply(input);
- Accumulator acc = BOUND.get();
- if (acc == null) {
- // 未 bind（例如被独立使用、或 process() 外触发的模型调用）——不累加，透传。
- return downstream;
- }
- // 订阅下游流：每个 ModelCallEndEvent 累加到当前线程的累加器；不影响事件本身。
- return downstream.doOnNext(event -> {
- if (event instanceof ModelCallEndEvent end) {
- ChatUsage usage = end.getUsage();
- if (usage != null) {
- acc.add(usage);
- }
+ // 先在 caller 线程取 ThreadLocal 累加器（回退路径：纯单元测试、或非 process() 触发的模型调用）。
+ Accumulator threadLocalAcc = BOUND.get();
+ return Flux.deferContextual(cv -> {
+ Accumulator acc = resolveAccumulator(cv, threadLocalAcc);
+ if (acc == null) {
+ // 既无 reactor Context 累加器，也无 ThreadLocal——不累加，透传。
+ return next.apply(input);
}
+ // 订阅下游流：每个 ModelCallEndEvent 累加到本次 invocation 的累加器；不影响事件本身。
+ // 该 doOnNext 在模型流所在的 reactor 调度线程上执行（findings R-stream），
+ // 累加器的 synchronized 保证与 HTTP 线程的 snapshot 互斥、可见。
+ return next.apply(input).doOnNext(event -> {
+ if (event instanceof ModelCallEndEvent end) {
+ ChatUsage usage = end.getUsage();
+ if (usage != null) {
+ acc.add(usage);
+ }
+ }
+ });
});
}
+ /** reactor Context 里的累加器优先；没有则回退 ThreadLocal（兼容无 Context 的调用）。 */
+ private static Accumulator resolveAccumulator(ContextView cv, Accumulator threadLocalAcc) {
+ Object fromCtx = cv == null ? null : cv.getOrDefault(USAGE_CONTEXT_KEY, null);
+ if (fromCtx instanceof Accumulator) {
+ return (Accumulator) fromCtx;
+ }
+ return threadLocalAcc;
+ }
+
/* ----- 累加器（{@code add}/{@code snapshot} 特意 synchronized）-----
- * 旧注释写"线程不安全、仅由 ThreadLocal 保证单线程访问"具有误导性：实际上
* {@link #add} 在 {@code onModelCall} 的 {@code doOnNext} 回调里被调用，该回调
- * 运行在模型流所在的 reactor 调度线程上——它通常与 {@code process()} 调用
- * {@code bind()} 的 HTTP 线程不同（流可能被 publishOn 切换线程）。因此两个方法
- * 特意加 {@code synchronized}：写端（流线程的 add）与读端（HTTP 线程的 snapshot，
- * 即 {@code ctx.getChatUsage()}）之间保证可见性与互斥。ThreadLocal 仍提供
- * per-invocation 隔离（单例 agent 跨 invocation 不串），但不保证单线程访问。
+ * 运行在模型流所在的 reactor 调度线程上（boundedElastic），与 {@code process()}
+ * 调用 {@code bind()} 的 HTTP 线程不同（流可能被 subscribeOn/publishOn 切换线程）。
+ * 故两个方法特意加 {@code synchronized}：写端与读端（HTTP 线程的 snapshot，即
+ * {@code ctx.getChatUsage()}）之间保证可见性与互斥。ThreadLocal / reactor Context
+ * 提供 per-invocation 隔离（单例 agent 跨 invocation 不串），但不保证单线程访问。
*/
- private static final class Accumulator {
+ /**
+ * per-invocation token 用量累加器。{@link #bind()} 创建、可同时绑到 ThreadLocal 与
+ * reactor Context；{@code onModelCall} 跨任意调度线程累加，{@code snapshot()} 由 HTTP
+ * 线程读。公开为静态嵌套类以便 {@code process()} 持有其引用并 {@link #bindToContext}。
+ */
+ public static final class Accumulator {
private int inputTokens;
private int outputTokens;
private double time;
diff --git a/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/StreamingBridgeCmp.java b/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/StreamingBridgeCmp.java
new file mode 100644
index 000000000..30dcc22a8
--- /dev/null
+++ b/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/StreamingBridgeCmp.java
@@ -0,0 +1,89 @@
+package com.yomahub.liteflow.test.agent.v2;
+
+import com.yomahub.liteflow.agent.component.ReActAgentComponent;
+import com.yomahub.liteflow.agent.model.ModelSpec;
+import com.yomahub.liteflow.property.agent.AgentConfig;
+import io.agentscope.core.model.Model;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * {@code StreamingBridgeTest}（Task 6.1）用的 ReActAgentComponent 子类：
+ *
+ * - {@link #buildModel()} escape hatch 返回 {@link StreamingReplyModel}，绕开真实 LLM，
+ * 且模型会发出多个 chunk（→ 多条 reasoning 增量）+ 末尾 usage；
+ * - 关闭 shell / workspace 工具，最小化 toolkit；
+ * - 记录 userPrompt / handleReply 调用次数，供断言。
+ *
+ *
+ * 该组件配合 {@code ExecuteOption.eventListener(...)} 触发
+ * {@code ReActAgentComponent.process()} 的流式分流：有监听者时走
+ * {@code AgentEventBridge.streamAndPublish}（{@code streamEvents}）而非 {@code call()}。
+ */
+@Component("streamingBridgeAgent")
+public class StreamingBridgeCmp extends ReActAgentComponent {
+
+ public static final AtomicInteger USER_PROMPT_COUNT = new AtomicInteger();
+ public static final AtomicInteger HANDLE_REPLY_COUNT = new AtomicInteger();
+ public static volatile io.agentscope.core.model.ChatUsage LAST_CHAT_USAGE;
+
+ public static void reset() {
+ USER_PROMPT_COUNT.set(0);
+ HANDLE_REPLY_COUNT.set(0);
+ LAST_CHAT_USAGE = null;
+ }
+
+ /** 仅满足抽象方法签名；buildModel() 被覆写后这里不会被调用。 */
+ @Override
+ @SuppressWarnings("rawtypes")
+ protected ModelSpec model() {
+ return new ModelSpec() {
+ @Override
+ public Model resolve(AgentConfig c) {
+ return new StreamingReplyModel("streaming-mock");
+ }
+ };
+ }
+
+ @Override
+ protected Model buildModel() {
+ return new StreamingReplyModel("streaming-mock");
+ }
+
+ @Override
+ protected String systemPrompt() {
+ return "test system prompt for streaming bridge";
+ }
+
+ @Override
+ protected String userPrompt() {
+ USER_PROMPT_COUNT.incrementAndGet();
+ Object reqData = getSlot().getChainReqData(getSlot().getChainId());
+ return reqData == null ? "hi" : reqData.toString();
+ }
+
+ @Override
+ protected boolean enableShellTool() {
+ return false;
+ }
+
+ @Override
+ protected boolean enableWorkspaceFileTools() {
+ return false;
+ }
+
+ @Override
+ protected boolean enableReActLogging() {
+ return false;
+ }
+
+ @Override
+ protected void handleReply(io.agentscope.core.message.Msg reply) {
+ HANDLE_REPLY_COUNT.incrementAndGet();
+ // 在 handleReply 时快照流式累计的 ChatUsage（验证 reactor Context 把累加器传到
+ // middleware 调度线程后，HTTP 线程仍能读到正确的累计值）。
+ LAST_CHAT_USAGE = ctx().getChatUsage();
+ super.handleReply(reply);
+ }
+}
diff --git a/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/StreamingBridgeTest.java b/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/StreamingBridgeTest.java
new file mode 100644
index 000000000..2f0c7290a
--- /dev/null
+++ b/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/StreamingBridgeTest.java
@@ -0,0 +1,130 @@
+package com.yomahub.liteflow.test.agent.v2;
+
+import com.yomahub.liteflow.agent.component.ReActAgentComponent;
+import com.yomahub.liteflow.core.ExecuteOption;
+import com.yomahub.liteflow.flow.FlowEvent;
+import com.yomahub.liteflow.flow.LiteflowResponse;
+import com.yomahub.liteflow.test.agent.support.LiveTestSupport;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.test.context.TestPropertySource;
+
+import javax.annotation.Resource;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.stream.Collectors;
+
+/**
+ * Task 6.1 端到端流式集成测试：验证 {@code ReActAgentComponent.process()} 在
+ * {@code ExecuteOption.eventListener} 注册了监听者时走 {@code AgentEventBridge.streamAndPublish}
+ * （{@code streamEvents}）路径，把 {@code TextBlockDeltaEvent} → {@code agent.reasoning}、
+ * {@code AgentResultEvent} → {@code agent.result}(last=true) 桥接成 {@link FlowEvent} 推给监听者。
+ *
+ *
无真实 LLM：{@link StreamingBridgeCmp} 覆写 {@code buildModel()} 返回
+ * {@link StreamingReplyModel}（确定性回放：两个 TextBlock chunk + 末尾 ChatUsage），
+ * 整个测试不需要任何凭据。
+ *
+ *
断言：
+ *
+ * - 收到多条 {@code agent.reasoning} 增量，文本拼接还原为 "Hello " + "world"；
+ * - 末尾收到一条 {@code agent.result}(isLast=true)，nodeId 为组件 nodeId；
+ * - 流式路径下 {@code ctx.getChatUsage()}（在 handleReply 内快照）正确累加为
+ * 100 input / 40 output —— 验证 reactor Context 把累加器传到 middleware 调度线程
+ * 后，HTTP 线程仍能读到（R-stream 方案 b）；
+ * - {@code response.isSuccess()} 为 true。
+ *
+ */
+@TestPropertySource("classpath:/feature/streamingbridge/application.properties")
+@SpringBootTest(classes = StreamingBridgeTest.class)
+@EnableAutoConfiguration
+@ComponentScan("com.yomahub.liteflow.test.agent.v2")
+public class StreamingBridgeTest {
+
+ @Resource
+ private com.yomahub.liteflow.core.FlowExecutor flowExecutor;
+
+ @Resource
+ private com.yomahub.liteflow.property.LiteflowConfig liteflowConfig;
+
+ @BeforeEach
+ public void resetRuntime() {
+ LiveTestSupport.resetAgentSessionManager();
+ StreamingBridgeCmp.reset();
+ }
+
+ @Test
+ public void testStreamingPathPublishesReasoningAndResultEvents() {
+ List events = new CopyOnWriteArrayList<>();
+
+ LiteflowResponse response = flowExecutor.execute2Resp(
+ "streamingBridgeChain", "ping",
+ ExecuteOption.of().eventListener(events::add));
+
+ Assertions.assertTrue(response.isSuccess(),
+ "chain failed: " + (response.getCause() == null
+ ? ""
+ : toString(response.getCause())));
+
+ // 1) 收到多条 agent.reasoning 增量，文本拼接还原为 mock 模型的两段回复。
+ List reasoning = events.stream()
+ .filter(e -> ReActAgentComponent.FLOW_EVENT_TYPE_REASONING.equals(e.getType()))
+ .collect(Collectors.toList());
+ Assertions.assertFalse(reasoning.isEmpty(),
+ "stream listener should receive at least one agent.reasoning event");
+ String joined = reasoning.stream()
+ .map(FlowEvent::getText)
+ .reduce("", String::concat);
+ Assertions.assertEquals(StreamingReplyModel.FULL_REPLY, joined,
+ "reasoning deltas should reconstruct the model's streamed text");
+
+ // 所有 reasoning 事件的 nodeId 都应为该 agent 的 nodeId，且非 last。
+ for (FlowEvent e : reasoning) {
+ Assertions.assertEquals("streamingBridgeAgent", e.getNodeId(),
+ "reasoning event nodeId must be the agent's nodeId");
+ Assertions.assertFalse(e.isLast(),
+ "reasoning deltas must not be marked last");
+ }
+
+ // 2) 末尾收到一条 agent.result(last=true)。
+ List results = events.stream()
+ .filter(e -> ReActAgentComponent.FLOW_EVENT_TYPE_RESULT.equals(e.getType()))
+ .collect(Collectors.toList());
+ Assertions.assertEquals(1, results.size(),
+ "exactly one final agent.result event expected");
+ FlowEvent finalEvent = results.get(0);
+ Assertions.assertTrue(finalEvent.isLast(),
+ "final agent.result must be marked last=true");
+ Assertions.assertEquals("streamingBridgeAgent", finalEvent.getNodeId(),
+ "final result event nodeId must be the agent's nodeId");
+
+ // 3) handleReply 被调用一次（证明流式路径末尾 Msg 也走了 handleReply）。
+ Assertions.assertEquals(1, StreamingBridgeCmp.HANDLE_REPLY_COUNT.get(),
+ "handleReply must be called once with the final streamed Msg");
+
+ // 4) 流式路径下 ctx.getChatUsage() 正确累加（R-stream 方案 b 生效）。
+ io.agentscope.core.model.ChatUsage usage = StreamingBridgeCmp.LAST_CHAT_USAGE;
+ Assertions.assertNotNull(usage,
+ "streamed path must accumulate ChatUsage (reactor Context propagation)");
+ Assertions.assertEquals(StreamingReplyModel.USAGE_INPUT, usage.getInputTokens(),
+ "accumulated inputTokens must match the model's final chunk usage");
+ Assertions.assertEquals(StreamingReplyModel.USAGE_OUTPUT, usage.getOutputTokens(),
+ "accumulated outputTokens must match the model's final chunk usage");
+ }
+
+ private static String toString(Throwable t) {
+ StringBuilder sb = new StringBuilder();
+ Throwable cur = t;
+ while (cur != null) {
+ sb.append(cur.getClass().getSimpleName()).append(": ").append(cur.getMessage());
+ cur = cur.getCause();
+ if (cur != null) {
+ sb.append(" || caused by: ");
+ }
+ }
+ return sb.toString();
+ }
+}
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
new file mode 100644
index 000000000..2e6d08dee
--- /dev/null
+++ b/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/StreamingReplyModel.java
@@ -0,0 +1,73 @@
+package com.yomahub.liteflow.test.agent.v2;
+
+import io.agentscope.core.message.ContentBlock;
+import io.agentscope.core.message.Msg;
+import io.agentscope.core.message.TextBlock;
+import io.agentscope.core.model.ChatResponse;
+import io.agentscope.core.model.ChatUsage;
+import io.agentscope.core.model.GenerateOptions;
+import io.agentscope.core.model.ToolSchema;
+import reactor.core.publisher.Flux;
+
+import java.util.List;
+
+/**
+ * 确定性的「流式回放」{@link io.agentscope.core.model.Model} 实现，专供
+ * {@code AgentEventBridge}（Task 6.1）流式集成测试在无网络、无真实 LLM 的前提下端到端跑通。
+ *
+ * {@link #stream(List, List, GenerateOptions)} 发出 两个 {@link ChatResponse}：
+ *
+ * - 第一个：单 {@link TextBlock}（"Hello "），不带 usage；
+ * - 第二个：单 {@link TextBlock}（"world"）+ {@link ChatUsage}（inputTokens=100,
+ * outputTokens=40）+ {@code finishReason="stop"}。
+ *
+ *
+ * ReActAgent 把每个 chunk 的 {@link TextBlock} 转成 {@code TextBlockDeltaEvent}
+ * （→ {@code agent.reasoning}），并在模型调用结束时发 {@code ModelCallEndEvent}
+ * （携带最后 chunk 的 usage）+ 最终 {@code AgentResultEvent}（携带聚合 Msg）。
+ * 故订阅 {@code ExecuteOption.eventListener} 应能收到 2 条 reasoning 增量 + 1 条
+ * 末尾 {@code agent.result}(last=true)，且流式路径下 {@code ctx.getChatUsage()} 应为
+ * 100/40（验证 reactor Context 把累加器传到 middleware 调度线程）。
+ *
+ *
无状态、线程安全；不依赖任何凭据。
+ */
+final class StreamingReplyModel implements io.agentscope.core.model.Model {
+
+ static final String DELTA_1 = "Hello ";
+ static final String DELTA_2 = "world";
+ static final String FULL_REPLY = DELTA_1 + DELTA_2;
+ static final int USAGE_INPUT = 100;
+ static final int USAGE_OUTPUT = 40;
+
+ private final String modelName;
+
+ StreamingReplyModel(String modelName) {
+ this.modelName = modelName;
+ }
+
+ @Override
+ public Flux stream(
+ List messages, List tools, GenerateOptions options) {
+ ChatResponse chunk1 = ChatResponse.builder()
+ .id("stream-1-" + System.nanoTime())
+ .content(List.of(TextBlock.builder().text(DELTA_1).build()))
+ .finishReason("stop")
+ .build();
+ ChatResponse chunk2 = ChatResponse.builder()
+ .id("stream-2-" + System.nanoTime())
+ .content(List.of(TextBlock.builder().text(DELTA_2).build()))
+ .usage(ChatUsage.builder()
+ .inputTokens(USAGE_INPUT)
+ .outputTokens(USAGE_OUTPUT)
+ .time(0.5)
+ .build())
+ .finishReason("stop")
+ .build();
+ return Flux.just(chunk1, chunk2);
+ }
+
+ @Override
+ public String getModelName() {
+ return modelName;
+ }
+}
diff --git a/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/resources/feature/streamingbridge/application.properties b/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/resources/feature/streamingbridge/application.properties
new file mode 100644
index 000000000..f648ee771
--- /dev/null
+++ b/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/resources/feature/streamingbridge/application.properties
@@ -0,0 +1,11 @@
+liteflow.rule-source=feature/streamingbridge/flow.el.xml
+liteflow.print-banner=false
+
+# 最小 agent 配置：workspace 指向 tmp（StreamingReplyModel 不实际写盘）；memory=NONE；
+# shell=disabled；skills 关闭。本测试不接触真实 LLM。
+liteflow.agent.workspace.root=target/wk/v2_streaming_bridge_test
+liteflow.agent.workspace.auto-create=true
+liteflow.agent.shell.mode=disabled
+liteflow.agent.defaults.max-iterations=3
+liteflow.agent.logging.react-enabled=false
+liteflow.agent.skills.enabled=false
diff --git a/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/resources/feature/streamingbridge/flow.el.xml b/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/resources/feature/streamingbridge/flow.el.xml
new file mode 100644
index 000000000..13cc0aabf
--- /dev/null
+++ b/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/resources/feature/streamingbridge/flow.el.xml
@@ -0,0 +1,6 @@
+
+
+
+ THEN(streamingBridgeAgent);
+
+