harness - 子代理流式输出适配

当主代理通过 task / multitask 委派子任务时，子代理内部产生的推理事件（思考、工具调用、结果）默认不会出现在父流中。本文从原理到实践，完整介绍如何让子代理的实时输出透传给调用方。

1、为什么子代理事件需要特殊处理

父代理的流来自 engine.prompt(prompt).session(session).stream()，对外是一条 Flux<AgentEvent>（HarnessEngine 本身没有独立的 stream() 入口）。当主代理调用 task 或 multitask 工具时，TaskTalent 会为每个子任务构建独立的 ReActAgent 实例与 InMemoryAgentSession，并开启各自的子流。

需要说明的是：子代理自身始终是以 .stream().blockLast() 执行完毕的（TaskTalent 的两条分支都是如此），区别只在于子事件是否被包装转发给父流的 sink。所以"子流默认是封闭的"指的不是子代理没开流，而是没有把事件透传出去。若要让子代理的事件出现在父流中，需要满足两个条件：

1. 父请求处于流式模式：即父级 ReActTrace.getOptions().getStreamSink() 不为 null——只有在父请求调用了 .stream()、由 ReActRequest 注入 FluxSink 时才成立。父请求使用 .call() 时没有可供外部消费的 sink，子代理事件不会传播（父级只拿到 task / multitask 的工具结果）。
2. 消费者能识别 TaskWrapEvent：子代理事件被包装成 TaskWrapEvent 注入父流，消费者必须识别并解包这一包装层，才能还原出真实的子代理事件。

另外，父 sink 被取消时，子流会通过 takeUntil(r -> sink.isCancelled()) 主动停止，且每次转发前都会再判一次 sink.isCancelled()，避免取消后继续输出。

2、原理：TaskWrapEvent 与 FluxSink

流式模式下，子代理每产生一个 AgentEvent，TaskTalent 都会将它包裹在 TaskWrapEvent 中，通过父流的 FluxSink 注入主流：

主代理流（Flux<AgentEvent>）
  ├── RunStartEvent
  ├── ReasonDeltaEvent        ← 主代理思考
  ├── ToolCallStartEvent      → 触发 task / multitask 工具
  │
  │      TaskTalent 内部，子代理流运行：
  │
  │      ├── TaskWrapEvent(ReasonDeltaEvent)    ← 子代理思考
  │      ├── TaskWrapEvent(ToolCallStartEvent)  ← 子代理工具开始
  │      ├── TaskWrapEvent(ToolCallEndEvent)    ← 子代理工具结束
  │      └── TaskWrapEvent(RunEndEvent)         ← 子代理推理完成
  │
  ├── ToolCallEndEvent        ← task/multitask 工具返回汇总
  └── RunEndEvent             ← 主代理最终结果

TaskWrapEvent 携带的信息：

并发安全： multitask 会并行运行多个子代理，它们共享同一个 FluxSink。TaskTalent 通过 synchronized(sink) 保证多线程并发写入时的串行化：

private void emitTaskChunk(FluxSink<AgentEvent> sink, TaskWrapEvent wrapChunk) {
    synchronized (sink) {
        if (!sink.isCancelled()) {
            sink.next(wrapChunk);
        }
    }
}

3、如何在消费端适配

消费 .stream() 时，流中会混入 TaskWrapEvent。适配的核心逻辑是：遇到 TaskWrapEvent 就解包，取出真实事件后与主代理事件按统一方式处理，同时保存子任务元信息（agentName、taskId 等）用于来源标识。

3.1 基本适配模式

import org.noear.solon.ai.agent.AgentEvent;
import org.noear.solon.ai.agent.react.RunEndEvent;
import org.noear.solon.ai.agent.react.task.ReasonDeltaEvent;
import org.noear.solon.ai.agent.react.task.ToolCallEndEvent;
import org.noear.solon.ai.agent.react.task.ToolCallStartEvent;
import org.noear.solon.ai.chat.event.ChatEvent;
import org.noear.solon.ai.chat.event.ChatEventType;
import org.noear.solon.ai.harness.agent.TaskWrapEvent;

engine.prompt(prompt)
    .session(session)
    .stream()
    .flatMap(outerEvent -> {
        // ① 来源元信息（仅子代理事件才有）
        String taskAgentName = null;
        String taskId = null;
        String taskDescription = null;
        int taskIndex = 0;

        // ② 解包 TaskWrapEvent
        AgentEvent event = outerEvent;
        if (outerEvent instanceof TaskWrapEvent) {
            TaskWrapEvent wrap = (TaskWrapEvent) outerEvent;
            event = wrap.getRealEvent();

            // 只透传感兴趣的事件类型，其余忽略
            if (!(event instanceof ReasonDeltaEvent)
                    && !(event instanceof ToolCallStartEvent)
                    && !(event instanceof ToolCallEndEvent)
                    && !(event instanceof RunEndEvent)) {
                return Flux.empty();
            }

            taskAgentName   = wrap.getTaskAgentName();
            taskId          = wrap.getTaskId();
            taskDescription = wrap.getTaskDescription();
            taskIndex       = wrap.getTaskIndex();

            // 子代理流结束：可在此做一次"子任务完成"通知
            if (event instanceof RunEndEvent) {
                RunEndEvent runEnd = (RunEndEvent) event;
                notifyTaskDone(taskAgentName, taskId, taskDescription, runEnd.isAbnormal());
                return Flux.empty();
            }
        }

        // ③ 统一处理主代理 + 子代理的真实事件
        MyChunk chunk = null;

        if (event instanceof ReasonDeltaEvent) {
            ReasonDeltaEvent delta = (ReasonDeltaEvent) event;
            ChatEvent chatEvent = delta.getChatEvent();

            // 正文只取底层 TEXT_DELTA；思考只取底层 THINKING_DELTA；边界与媒体不混入
            if (chatEvent != null && delta.hasText()) {
                if (chatEvent.is(ChatEventType.THINKING_DELTA)) {
                    chunk = MyChunk.ofThinking(delta.getText());
                } else if (chatEvent.is(ChatEventType.TEXT_DELTA)) {
                    chunk = MyChunk.ofText(delta.getText());
                }
            }
        } else if (event instanceof ToolCallStartEvent) {
            ToolCallStartEvent e = (ToolCallStartEvent) event;
            chunk = MyChunk.ofToolStart(e.getToolName(), e.getArgs());
        } else if (event instanceof ToolCallEndEvent) {
            ToolCallEndEvent e = (ToolCallEndEvent) event;
            chunk = MyChunk.ofToolEnd(e.getToolName(), e.getText());
        } else if (event instanceof RunEndEvent) {
            chunk = MyChunk.ofDone(((RunEndEvent) event).getText());
        }

        if (chunk == null) return Flux.empty();

        // ④ 附加来源标记，供下游区分主代理/子代理
        if (taskAgentName != null) {
            chunk.setAgentName(taskAgentName);
            chunk.setTaskId(taskId);
            chunk.setTaskIndex(taskIndex);
            chunk.setTaskDescription(taskDescription);
        }

        return Flux.just(chunk);
    })
    .subscribe(chunk -> sseEmitter.send(chunk));

几个关键点：

• task / multitask 工具本身的 ToolCallStartEvent / ToolCallEndEvent 来自主代理，不会被包在 TaskWrapEvent 里——这两个工具事件通常应在 UI 上过滤掉（它们是调度指令，不是用户可见的操作）。
• 子代理的 RunEndEvent 包在 TaskWrapEvent 里，可借此在该子任务完成时立刻通知前端（不必等待整条主流结束），在并行任务场景下尤其有用。
• agentName（子代理名，如 explore）加上 taskId 是区分多个并行子任务输出来源的关键字段，前端可据此将不同子任务的 chunk 分组展示；agentName 可能重复，不能单独作为标识。
• 正文/思考必须按底层 ChatEventType 过滤：仅用 ReasonDeltaEvent.isThinking() 只能区分思考分组，无法排除 THINKING_START / THINKING_END 边界。

3.2 同步模式下的行为差异

若父请求不是流式模式（直接 .call()），父链路没有 FluxSink，TaskTalent 就退回到"不包装、不转发"的分支：

// 同步模式：子代理流仍被完整消费，但事件不外发，只取终态文本
RunEndEvent agentEnd = (RunEndEvent) agent.prompt(prompt)
        .session(session)
        .stream()
        .blockLast();

String result = agentEnd.getText();

同步模式下，子代理完整执行后只把最终文本回传给主代理，不产生任何包装的中间事件。主代理的调用方只能看到最终汇总结果，看不到子代理的工具调用过程。

选择依据： - 需要实时展示子代理操作过程 → 必须使用父级 .stream() 流式模式 - 只关心最终结果、不关心过程 → .call() 同步模式即可，更简单

4、参考示例：WebStreamBuilder

SolonCode 中的 WebStreamBuilder 是一个生产级的完整适配实现，将 AgentEvent（含 TaskWrapEvent）映射为 WebChunk 序列，通过 SSE 推送给 Web 前端。其对子代理流的处理逻辑与上文描述的模式一致，并在此基础上增加了：

• IM 通道同步转发（微信、飞书等）
• HITL（人机交互循环）挂起卡片推送
• 工具事件的来源前缀规则（agentName/toolName）
• 子代理 RunEndEvent 时发出"任务完成"信号让前端立即结算 task-group

可直接参考其 buildStreamFlux 方法中对 TaskWrapEvent 的解包与分发逻辑作为完整示例。