中断控制流消息误入steer队列问题修改

[openjiuwen-sync-bot]

Paired: GitHub #104 ↔ GitCode !233

中断控制流消息误入 Steering 队列修复

一、问题描述

在 DeepAgent -> ReActAgent 的嵌套调用场景中，当底层 ReActAgent 因工具调用需要用户补充输入而产生中断时，结构化中断信息会被额外转换成 steering 文本，并在下一轮 LLM 请求中作为用户消息注入模型上下文。

develop 分支中的实际脏消息形态为：

role=user
content=[STEERING] {payload=InteractionOutput{id='ask-user-call', value=..., metadata={}}, type=__interaction__}

正常情况下，InteractionOutput 应作为结构化交互状态向上传递，供客户端展示以及后续恢复使用，不应被转换成用户 steering 指令。

实际现象是：

正常的结构化中断输出
        +
额外的 [STEERING] {payload=InteractionOutput{...}, type=__interaction__} 消息

这条额外消息没有业务含义，可能污染模型上下文、干扰模型判断，并增加不必要的 token 消耗。

二、正确语义

当前 TASK_INTERACTION 中可能包含两类内容：

内容	方向与语义	正确处理
显式文本 interaction	上层向运行中 Agent 追加 steering 指令	写入 steering 队列
结构化 interrupt	底层 Agent 向上层上报中断控制信息	作为结构化中断输出，不进入 steering 队列

develop 当前存在两种结构化中断表示：

type == "__interaction__"

或者：

result_type == "interrupt"

两类数据虽然都经过 handleTaskInteraction()，但不能执行相同的文本转换和入队逻辑。

三、问题产生链路

内部流式中断的完整路径如下：

ReActAgent 检测到工具调用中断
        ↓
创建结构化 InteractionOutput
        ↓
生成 OutputSchema
type = "__interaction__"
payload = InteractionOutput(...)
        ↓
CoreTaskLoopEventExecutor.toProcessingChunk()
        ↓
包装为 JsonDataFrame
data = {type=__interaction__, payload=InteractionOutput(...)}
        ↓
生成 TASK_INTERACTION 事件
        ↓
TaskLoopEventHandler.handleTaskInteraction()
        ↓
frameText(JsonDataFrame)
        ↓
String.valueOf(jsonDataFrame.data())
        ↓
得到 {payload=InteractionOutput{...}, type=__interaction__}
        ↓
interactionQueues.pushSteer(message)
        ↓
ReActAgent.drainSteering()
        ↓
包装为 UserMessage("[STEERING] ...")
        ↓
进入下一轮 LLM messages

除内部流式 __interaction__ 外，ReActAgent 的最终中断结果还会通过以下结构生成 TASK_INTERACTION：

JsonDataFrame.data = {
    result_type: "interrupt",
    ...
}

该结构同样会被 frameText() 转成字符串并进入 steering 队列，因此需要同时覆盖两种中断标识。

四、根因分析

根因位于 TaskLoopEventHandler.handleTaskInteraction()：原实现默认把 TASK_INTERACTION 的第一个 frame 全部当作可注入模型的 steering 文本处理。

原逻辑为：

if (event instanceof TaskInteractionEvent interactionEvent
        && !interactionEvent.getInteraction().isEmpty()) {
    message = frameText(interactionEvent.getInteraction().get(0));
}
if (!message.isBlank() && interactionQueues != null) {
    interactionQueues.pushSteer(message);
}

frameText() 对 JsonDataFrame 的处理是：

if (frame instanceof DataFrame.JsonDataFrame jsonDataFrame
        && jsonDataFrame.data() != null) {
    return String.valueOf(jsonDataFrame.data());
}

该逻辑只判断 frame 是否可以转换为非空文本，没有判断它属于显式 steering 还是结构化 interrupt。因此，带有 type=__interaction__ 或 result_type=interrupt 的控制流数据也会被字符串化并入队。

develop 中的 InteractionOutput 已实现 toString()，所以脏消息内容比早期版本更易读，但这并没有改变问题本质：内部中断状态仍被伪装成用户指令注入模型。

五、修复原则

本次修复遵循以下原则：

1. 结构化控制信息不进入 steering 队列

   type=__interaction__ 或 result_type=interrupt 的 frame 不执行文本转换和 steering 入队。

2. 正常显式文本 interaction 保持原有行为

   DeepAgent.steer() 发送的 TextDataFrame 仍按原逻辑写入 steering 队列。

3. 在 interaction 业务边界进行过滤

   判断放在 handleTaskInteraction() 中，不修改通用 frameText()。frameText() 还被 follow-up 和 completion 路径使用，在其中加入中断语义会影响无关逻辑。

4. 依据协议字段判断，不依赖具体 payload 类型

   使用 type 和 result_type 识别结构化中断，避免只对 InteractionOutput 生效而遗漏其他中断载荷形式。

5. 保持修改范围最小

   不修改事件结构、中断状态结构、恢复协议、steering 队列和通用 frame 文本转换规则。

六、具体代码修改

在 TaskLoopEventHandler.handleTaskInteraction() 中，先判断第一个 frame 是否为结构化中断，再决定是否执行 frameText()：

if (event instanceof TaskInteractionEvent interactionEvent
        && !interactionEvent.getInteraction().isEmpty()) {
    DataFrame frame = interactionEvent.getInteraction().get(0);
    if (!isStructuredInterrupt(frame)) {
        message = frameText(frame);
    }
}

新增结构化中断识别函数：

private static boolean isStructuredInterrupt(DataFrame frame) {
    if (!(frame instanceof DataFrame.JsonDataFrame jsonDataFrame)
            || jsonDataFrame.data() == null) {
        return false;
    }
    Map<String, Object> data = jsonDataFrame.data();
    return Constant.INTERACTION.equals(String.valueOf(data.get("type")))
            || "interrupt".equals(String.valueOf(data.get("result_type")));
}

处理结果如下：

输入	是否转换为 steering 文本	是否进入 steering 队列
type=__interaction__	否	否
result_type=interrupt	否	否
普通 TextDataFrame	是	是
其他普通 JsonDataFrame	按原逻辑	按原逻辑

七、为什么不采用其他处理方式

1. 不修改 InteractionOutput.toString()

develop 中 InteractionOutput 已经实现了可读的 toString()。修改其输出形式只能改变脏消息的显示内容，不能阻止控制信息进入 steering 队列。

2. 不删除 handleTaskInteraction() 的全部 steering 逻辑

当前 DeepAgent.steer() 在 task-loop 运行期间会发送包含 TextDataFrame 的 TaskInteractionEvent。完全禁止 TASK_INTERACTION 入 steering 会破坏正常的显式追加指令场景。

3. 不修改通用 frameText()

frameText() 同时用于：

handleTaskInteraction()
handleFollowUp()
extractCompletionResult()

将 interrupt 判断加入通用转换函数会把 task interaction 的控制语义扩散到 follow-up 和 completion，扩大修改影响范围。

4. 不仅判断 payload 是否为 InteractionOutput

中断既可能以 type=__interaction__ 形式出现，也可能以 result_type=interrupt 的最终结果形式出现。依据协议字段判断可以覆盖两条实际链路。

八、测试覆盖

在 TaskLoopEventHandlerTest 中新增两个回归测试：

1. structuredInteractionShouldNotEnterSteering

构造：

type = __interaction__
payload = InteractionOutput(...)

验证：

• handler 返回的 msg 为空；
• steering 队列为空；
• 结构化 payload 不再被转换成 steering 文本。

2. interruptResultShouldNotEnterSteering

构造：

result_type = interrupt
message = approval required

验证：

• 即使中断结果中存在 message 字段，也不会被当作 steering；
• steering 队列为空。

原有 handleTaskInteraction() 测试继续验证：

• 普通 TextDataFrame("change plan") 正常进入 steering 队列；
• 修复没有破坏显式 steering 行为。

执行命令：

mvn test \
  -Dtest=TaskLoopEventHandlerTest,TaskLoopEventExecutorPythonParityTest,TaskLoopPackageTest

验证结果：

Tests run: 15, Failures: 0, Errors: 0, Skipped: 0
BUILD SUCCESS

测试完全使用 agent-core-java 内部对象，不依赖 yoskills、jiuwen-test、Redis、MCP 或其他外部服务。

九、影响范围与边界

本次修改涉及两个文件：

src/main/java/com/openjiuwen/harness/task_loop/TaskLoopEventHandler.java
src/test/java/com/openjiuwen/harness/task_loop/TaskLoopEventHandlerTest.java

修改边界如下：

• 不改变 TASK_INTERACTION 事件结构；
• 不改变 InteractionOutput 数据结构及序列化行为；
• 不改变 interrupt state 的产生、输出和恢复逻辑；
• 不改变普通文本 interaction 的 steering 行为；
• 不改变 follow-up 和 completion 的文本转换逻辑；
• 不改变 ReActAgent 的正常工具结果回灌逻辑；
• 不涉及 agent-runtime-java；
• 不增加第三方依赖；
• 不涉及对外接口变更。

修复后，结构化 interrupt 仍按原链路向上传递和恢复，但不会再生成多余的 [STEERING] 用户消息。

What type of PR is this?

/kind bug

Self-checklist:（请自检，在[ ]内打上x，我们将检视你的完成情况，否则会导致pr无法合入）

• • 设计：PR对应的方案是否已经经过Maintainer评审，方案检视意见是否均已答复并完成方案修改
• • 测试：PR中的代码是否已有UT/ST测试用例进行充分的覆盖，新增测试用例是否随本PR一并上库或已经上库
• • 验证：PR描述信息中是否已包含对该PR对应的Feature、Refactor、Bugfix的预期目标达成情况的详细验证结果描述
• • 接口：是否涉及对外接口变更，相应变更已得到接口评审组织的通过，API对应的注释信息已经刷新正确
• • 文档：是否涉及官网文档修改，如果涉及请及时提交资料到Doc仓

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.

---

guoyangsen seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please add the email address used for this commit to your account.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[openjiuwen-collaboration-bot]

head_sha: 40c7e8b7da1ff770c113c688ce3194cfd7f5ff3b

TASK	STATUS	DETAILS
CodeCheck	✅SUCCESS	Click here
AntiPoison	✅SUCCESS	Click here
Software Composition Analysis	✅SUCCESS	Click here
Maven Build	❌FAILED	See CHECK tab

[openjiuwen-collaboration-bot]

head_sha: a54ca08f38b666483311f804ec9f7581cfb5ebf5

TASK	STATUS	DETAILS
CodeCheck	✅SUCCESS	Click here
AntiPoison	✅SUCCESS	Click here
Software Composition Analysis	✅SUCCESS	Click here
Maven Build	✅SUCCESS	See CHECK tab