mirror of https://gitee.com/dromara/liteFlow
test(agent): v2 迁移全量回归 + vendor R4 核对 + in-flight interrupt 集成测试
Task 9（agentscope v2 RC3 迁移最后一任务）：
A. vendor Spec R4 核对：4 vendor 模块 -am compile 全 SUCCESS。
逐 vendor 核对 resolve() builder 方法名——现状已全部符合 R4
（OpenAI=generateOptions+stream；Anthropic/DashScope=defaultOptions+stream；
Gemini=defaultOptions+streamEnabled），未改 ModelSpec 抽象、未改 OpenAISpec。
B. 补 Task 7.1 推迟的 in-flight interrupt 集成测试（race-free，非降级）：
- BlockingChunksModel：Flux.create 两段阻塞 mock，emit chunk1 后阻塞在 gate。
- InFlightInterruptTest：异步 call → await chunk1 → interrupt(cid,akey)（flag=true）
→ releaseSecondChunk → chunk2 进 concatMap 的 checkInterrupted() → InterruptedException
→ ReActAgent.handleInterrupt 返回恢复 Msg。断言返回文本 == RC3 固定恢复消息。
- 时序无并发窗口（flag 写入严格早于 chunk2 到达检查点），5/5 GREEN，~150ms 稳定。
C. 全量回归：
- C.1 5 react-agent 模块 clean package -DskipTests → 9 模块 BUILD SUCCESS。
- C.2 testcase-el-react-agent v2 全测试（15 类显式列表）→ 49 tests PASS。
- C.3 compile-17+ profile react-agent-core + testcase test-compile → 12 模块 SUCCESS。
未改 skipTests（CLI -Dmaven.test.skip=false），未改生产代码，未 push。
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
everywhere.z
parent
bccf1426e8
commit
290ab03471
|
|
@ -0,0 +1,114 @@
|
|||
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.GenerateOptions;
|
||||
import io.agentscope.core.model.ToolSchema;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.FluxSink;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
/**
|
||||
* 可手动控制「分两段 emit」的阻塞型 {@link io.agentscope.core.model.Model}，专供
|
||||
* {@code InFlightInterruptTest}（Task 9）在<b>无网络、无真实 LLM</b> 的前提下端到端验证
|
||||
* 「in-flight {@code call()} 被外部 {@code interrupt(cid, akey)} 打断后以恢复 Msg 结束」。
|
||||
*
|
||||
* <p><b>设计依据（findings R-enh + RC3 源码通读）：</b>RC3 的 {@code ReActAgent} 在 reasoning
|
||||
* 阶段对模型流做 {@code concatMap(chunk -> checkInterrupted().thenReturn(chunk))}
|
||||
* （{@code ReActAgent.java} 第 2049 行）——<b>每处理完一个 chunk，下一个 chunk 进入 concatMap 时
|
||||
* 都会调 {@code checkInterrupted()} 读 session-scoped {@code InterruptControl} 的 flag</b>。
|
||||
* 故只要在 chunk1 被消费后、chunk2 到达 concatMap <b>之前</b>把 flag 置 true，chunk2 触发的
|
||||
* {@code checkInterrupted()} 就会抛 {@code InterruptedException}，经
|
||||
* {@code AgentBase.createErrorHandler} → {@code ReActAgent.handleInterrupt} 转成恢复 Msg
|
||||
* （"I noticed that you have interrupted me. What can I do for you?"）。
|
||||
*
|
||||
* <p><b>时序（race-free）：</b>
|
||||
* <ol>
|
||||
* <li>{@link #stream} emit chunk1（含一段文本），随后阻塞在 {@link #gate}（CountDownLatch，
|
||||
* 初始 closed）——chunk2 暂不发。</li>
|
||||
* <li>emit chunk1 后立即 {@code chunk1Emitted.countDown()}，让测试线程确认 chunk1 已离开模型。</li>
|
||||
* <li>测试线程等待 {@link #chunk1Emitted}，调 {@code component.interrupt(cid, akey)}（flag=true），
|
||||
* 再 {@code gate.countDown()} 放行 chunk2。</li>
|
||||
* <li>{@link #stream} 发 chunk2，进入 concatMap 的 {@code checkInterrupted()}——flag 已 true →
|
||||
* {@code InterruptedException} → 恢复 Msg。</li>
|
||||
* </ol>
|
||||
*
|
||||
* <p>整个序列对 reactor 时序<b>不敏感</b>：中断 flag 的写入严格早于 chunk2 到达
|
||||
* {@code checkInterrupted()}（两者都在测试线程内顺序执行，无并发窗口），故测试不 flaky。
|
||||
*
|
||||
* <p>无状态字段以外，{@link #gate}/{@link #chunk1Emitted} 是<b>每实例</b>的一次性夹具，
|
||||
* 一个实例只服务于一次 {@code call()}，不复用。线程安全（{@link FluxSink} 与 latch 均线程安全）。
|
||||
*/
|
||||
final class BlockingChunksModel implements io.agentscope.core.model.Model {
|
||||
|
||||
/** 恢复 Msg 的固定文本（RC3 {@code ReActAgent.handleInterrupt} 硬编码）。 */
|
||||
static final String RECOVERY_TEXT =
|
||||
"I noticed that you have interrupted me. What can I do for you?";
|
||||
|
||||
private final String modelName;
|
||||
private final CountDownLatch gate = new CountDownLatch(1);
|
||||
private final CountDownLatch chunk1Emitted = new CountDownLatch(1);
|
||||
private final AtomicReference<Throwable> streamError = new AtomicReference<>();
|
||||
|
||||
BlockingChunksModel(String modelName) {
|
||||
this.modelName = modelName;
|
||||
}
|
||||
|
||||
/** 放行 chunk2 的发出（测试线程在 interrupt 之后调用）。 */
|
||||
void releaseSecondChunk() {
|
||||
gate.countDown();
|
||||
}
|
||||
|
||||
/** 等待 chunk1 已被 emit（测试线程据此确认可以安全触发 interrupt）。 */
|
||||
boolean awaitFirstChunkEmitted(long timeoutMs) throws InterruptedException {
|
||||
return chunk1Emitted.await(timeoutMs, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
|
||||
/** 模型流内部若意外异常，记下供测试诊断。 */
|
||||
Throwable streamError() {
|
||||
return streamError.get();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Flux<ChatResponse> stream(
|
||||
List<Msg> messages, List<ToolSchema> tools, GenerateOptions options) {
|
||||
ChatResponse chunk1 = ChatResponse.builder()
|
||||
.id("blk-1-" + System.nanoTime())
|
||||
.content(List.<ContentBlock>of(TextBlock.builder().text("thinking...").build()))
|
||||
.build();
|
||||
ChatResponse chunk2 = ChatResponse.builder()
|
||||
.id("blk-2-" + System.nanoTime())
|
||||
.content(List.<ContentBlock>of(TextBlock.builder().text("final answer").build()))
|
||||
.finishReason("stop")
|
||||
.build();
|
||||
return Flux.<ChatResponse>create(sink -> {
|
||||
// chunk1 立即发——进入 concatMap 被 checkInterrupted() 检查（此时 flag=false，通过）。
|
||||
sink.next(chunk1);
|
||||
chunk1Emitted.countDown();
|
||||
// 阻塞等测试线程放行；放行前 chunk2 永不到达 concatMap，故 checkInterrupted 不会被
|
||||
// 第二次读到——flag 的写入严格早于 chunk2，无 race。
|
||||
try {
|
||||
gate.await();
|
||||
} catch (InterruptedException ie) {
|
||||
streamError.set(ie);
|
||||
sink.error(ie);
|
||||
return;
|
||||
}
|
||||
// 放行后发 chunk2——此刻 flag 已被 interrupt 置 true，concatMap 的 checkInterrupted 抛
|
||||
// InterruptedException → handleInterrupt → 恢复 Msg。
|
||||
sink.next(chunk2);
|
||||
sink.complete();
|
||||
}, FluxSink.OverflowStrategy.BUFFER);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getModelName() {
|
||||
return modelName;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,205 @@
|
|||
package com.yomahub.liteflow.test.agent.v2;
|
||||
|
||||
import com.yomahub.liteflow.agent.component.ReActAgentComponent;
|
||||
import com.yomahub.liteflow.agent.component.ReactAgentFactory;
|
||||
import com.yomahub.liteflow.agent.model.ModelSpec;
|
||||
import com.yomahub.liteflow.property.agent.AgentConfig;
|
||||
import io.agentscope.core.ReActAgent;
|
||||
import io.agentscope.core.agent.RuntimeContext;
|
||||
import io.agentscope.core.message.Msg;
|
||||
import io.agentscope.core.message.UserMessage;
|
||||
import io.agentscope.core.model.Model;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.TimeoutException;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
* Task 9 补遗：Task 7.1 推迟的<b>集成级</b> in-flight interrupt 测试。
|
||||
*
|
||||
* <p>用 {@link BlockingChunksModel}（race-free 两段阻塞模型，见其 javadoc 的 RC3 concatMap
|
||||
* 中断检查点分析）启动一个 agent {@code call()}，在 chunk1 被消费后、chunk2 到达前从外部调
|
||||
* {@link ReActAgentComponent#interrupt(String, String)}，断言该 call <b>在合理超时内完成</b>，
|
||||
* 且返回的 Msg 是 RC3 {@code ReActAgent.handleInterrupt} 产出的恢复消息
|
||||
* （{@link BlockingChunksModel#RECOVERY_TEXT}）——即"in-flight call 被中断后以恢复 Msg 结束"。
|
||||
*
|
||||
* <p><b>不依赖 Spring 上下文</b>：直接经 {@link ReactAgentFactory#getOrCreate} 取 agent 单例、
|
||||
* 手工 {@code agent.call(...)} 并阻塞，跳过 {@code process()} 的 slot/chain 装配。仅验证
|
||||
* agent + interrupt 的 reactor 链路，与 {@code InterruptTest}（单元级"不抛 + 参数透传"）互补。
|
||||
*
|
||||
* <p><b>时序（race-free，见 {@link BlockingChunksModel}）：</b>interrupt 的 flag 写入严格早于
|
||||
* chunk2 到达 {@code checkInterrupted()}——两者都在测试线程内顺序执行，无并发窗口，故不 flaky。
|
||||
*
|
||||
* <p><b>降级（brief Step 2 允许）：</b>若因 reactor 调度在某些 CI 环境下 chunk2 的 concatMap
|
||||
* 检查时序异常，{@link #AWAIT_TIMEOUT_MS} 超时后测试失败而非 hang；测试不依赖 wall-clock 精度。
|
||||
*/
|
||||
class InFlightInterruptTest {
|
||||
|
||||
/** call().get() 的硬超时（秒）；远大于正常完成所需，仅作防 hang 兜底。 */
|
||||
private static final long CALL_TIMEOUT_SECONDS = 15;
|
||||
/** 等模型 emit chunk1 的超时（毫秒）。 */
|
||||
private static final long CHUNK1_AWAIT_MS = 3000;
|
||||
|
||||
/** 最小可中断组件：buildModel 返回 BlockingChunksModel（每实例一次性夹具）。 */
|
||||
static class InterruptableCmp extends ReActAgentComponent {
|
||||
final BlockingChunksModel model;
|
||||
|
||||
InterruptableCmp(BlockingChunksModel model) {
|
||||
this.model = model;
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("rawtypes")
|
||||
protected ModelSpec model() {
|
||||
return new ModelSpec() {
|
||||
@Override
|
||||
public Model resolve(AgentConfig c) {
|
||||
return model;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Model buildModel() {
|
||||
return model;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected String systemPrompt() {
|
||||
return "test";
|
||||
}
|
||||
|
||||
@Override
|
||||
protected String userPrompt() {
|
||||
return "hello";
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean enableShellTool() {
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean enableWorkspaceFileTools() {
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean enableReActLogging() {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private ExecutorService executor;
|
||||
|
||||
@BeforeEach
|
||||
void reset() {
|
||||
ReactAgentFactory.resetForTesting();
|
||||
executor = Executors.newSingleThreadExecutor(r -> {
|
||||
Thread t = new Thread(r, "inflight-interrupt-call");
|
||||
t.setDaemon(true);
|
||||
return t;
|
||||
});
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
void cleanup() {
|
||||
if (executor != null) {
|
||||
executor.shutdownNow();
|
||||
}
|
||||
ReactAgentFactory.resetForTesting();
|
||||
}
|
||||
|
||||
/**
|
||||
* 主断言：in-flight {@code call()} 被外部 {@code interrupt(cid, akey)} 打断后，在合理超时内
|
||||
* 以 RC3 恢复 Msg（{@link BlockingChunksModel#RECOVERY_TEXT}）结束——而非 hang、抛异常、
|
||||
* 或返回 chunk2 的 "final answer"。
|
||||
*/
|
||||
@Test
|
||||
void inFlightCall_interruptedReturnsRecoveryMessage() throws Exception {
|
||||
BlockingChunksModel model = new BlockingChunksModel("blocking");
|
||||
AgentConfig cfg = HarnessFixture.minimalConfig();
|
||||
InterruptableCmp cmp = new InterruptableCmp(model);
|
||||
// 构建 agent 单例（按 cmp.getClass() 缓存）。
|
||||
ReActAgent agent = ReactAgentFactory.getOrCreate(cmp, cfg);
|
||||
assertNotNull(agent, "agent must be built & cached");
|
||||
|
||||
String cid = "conv-inflight";
|
||||
String akey = "agent-key-inflight";
|
||||
RuntimeContext rc = RuntimeContext.builder().userId(cid).sessionId(akey).build();
|
||||
|
||||
// 异步启动 call()：进入 reasoning → 模型 emit chunk1 → concatMap checkInterrupted(通过)
|
||||
// → 模型阻塞在 gate 等 chunk2 放行。
|
||||
AtomicReference<Msg> result = new AtomicReference<>();
|
||||
AtomicReference<Throwable> error = new AtomicReference<>();
|
||||
Future<?> callFuture = executor.submit(() -> {
|
||||
try {
|
||||
Msg reply = agent.call(List.of(new UserMessage("hello")), rc).block();
|
||||
result.set(reply);
|
||||
} catch (Throwable t) {
|
||||
error.set(t);
|
||||
}
|
||||
});
|
||||
|
||||
// 等模型 emit chunk1（确认 call 真正进入模型流、chunk1 已被 concatMap 消费）。
|
||||
boolean chunk1Sent = model.awaitFirstChunkEmitted(CHUNK1_AWAIT_MS);
|
||||
assertTrue(chunk1Sent,
|
||||
"model must emit chunk1 within " + CHUNK1_AWAIT_MS + "ms; stream error="
|
||||
+ model.streamError());
|
||||
|
||||
// 关键：在放行 chunk2 <b>之前</b>触发中断（flag=true）。这一步在测试线程内、chunk2 到达
|
||||
// concatMap 之前完成——race-free。
|
||||
cmp.interrupt(cid, akey);
|
||||
|
||||
// 放行 chunk2：模型 emit chunk2 → concatMap checkInterrupted → flag=true →
|
||||
// InterruptedException → handleInterrupt → 恢复 Msg。
|
||||
model.releaseSecondChunk();
|
||||
|
||||
// 等待 call 完成（硬超时兜底防 hang）。
|
||||
try {
|
||||
callFuture.get(CALL_TIMEOUT_SECONDS, TimeUnit.SECONDS);
|
||||
} catch (TimeoutException te) {
|
||||
failWithDiagnostics("call did not complete within " + CALL_TIMEOUT_SECONDS + "s", model, error);
|
||||
return;
|
||||
}
|
||||
|
||||
// 断言：call 以恢复 Msg 结束（不抛、不返回 chunk2 的 final answer）。
|
||||
assertNullError(error, "call should not throw on user interrupt; it should return recovery Msg");
|
||||
Msg reply = result.get();
|
||||
assertNotNull(reply, "call must return a (recovery) Msg after interrupt, not null");
|
||||
assertNotNull(reply.getTextContent(), "recovery Msg must carry text content");
|
||||
assertEquals(BlockingChunksModel.RECOVERY_TEXT, reply.getTextContent(),
|
||||
"call result must be RC3 handleInterrupt recovery message, not chunk2's final answer");
|
||||
|
||||
// 附加断言：未返回 chunk2 的 final answer（双保险，防误判）。
|
||||
assertFalse(reply.getTextContent().contains("final answer"),
|
||||
"interrupted call must not have returned chunk2's final answer text");
|
||||
}
|
||||
|
||||
private static void assertNullError(AtomicReference<Throwable> error, String msg) {
|
||||
Throwable t = error.get();
|
||||
if (t != null) {
|
||||
throw new AssertionError(msg + " — got: "
|
||||
+ t.getClass().getSimpleName() + ": " + t.getMessage(), t);
|
||||
}
|
||||
}
|
||||
|
||||
private static void failWithDiagnostics(String msg, BlockingChunksModel model,
|
||||
AtomicReference<Throwable> error) {
|
||||
throw new AssertionError(msg
|
||||
+ " | streamError=" + model.streamError()
|
||||
+ " | callError=" + (error.get() == null ? "<none>" : error.get()));
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue