diff --git a/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/BlockingChunksModel.java b/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/BlockingChunksModel.java new file mode 100644 index 000000000..73726bdcd --- /dev/null +++ b/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/BlockingChunksModel.java @@ -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）在无网络、无真实 LLM 的前提下端到端验证 + * 「in-flight {@code call()} 被外部 {@code interrupt(cid, akey)} 打断后以恢复 Msg 结束」。 + * + *

设计依据（findings R-enh + RC3 源码通读）：RC3 的 {@code ReActAgent} 在 reasoning + * 阶段对模型流做 {@code concatMap(chunk -> checkInterrupted().thenReturn(chunk))} + * （{@code ReActAgent.java} 第 2049 行）——每处理完一个 chunk，下一个 chunk 进入 concatMap 时 + * 都会调 {@code checkInterrupted()} 读 session-scoped {@code InterruptControl} 的 flag。 + * 故只要在 chunk1 被消费后、chunk2 到达 concatMap 之前把 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?"）。 + * + *

时序（race-free）： + *

    + *
  1. {@link #stream} emit chunk1（含一段文本），随后阻塞在 {@link #gate}（CountDownLatch， + * 初始 closed）——chunk2 暂不发。
    2. + *
  2. emit chunk1 后立即 {@code chunk1Emitted.countDown()}，让测试线程确认 chunk1 已离开模型。
    3. + *
  3. 测试线程等待 {@link #chunk1Emitted}，调 {@code component.interrupt(cid, akey)}（flag=true）， + * 再 {@code gate.countDown()} 放行 chunk2。
    4. + *
  4. {@link #stream} 发 chunk2，进入 concatMap 的 {@code checkInterrupted()}——flag 已 true → + * {@code InterruptedException} → 恢复 Msg。
    5. + *
+ * + *

整个序列对 reactor 时序不敏感：中断 flag 的写入严格早于 chunk2 到达 + * {@code checkInterrupted()}（两者都在测试线程内顺序执行，无并发窗口），故测试不 flaky。 + * + *

无状态字段以外，{@link #gate}/{@link #chunk1Emitted} 是每实例的一次性夹具， + * 一个实例只服务于一次 {@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 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 stream( + List messages, List tools, GenerateOptions options) { + ChatResponse chunk1 = ChatResponse.builder() + .id("blk-1-" + System.nanoTime()) + .content(List.of(TextBlock.builder().text("thinking...").build())) + .build(); + ChatResponse chunk2 = ChatResponse.builder() + .id("blk-2-" + System.nanoTime()) + .content(List.of(TextBlock.builder().text("final answer").build())) + .finishReason("stop") + .build(); + return Flux.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; + } +} diff --git a/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/InFlightInterruptTest.java b/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/InFlightInterruptTest.java new file mode 100644 index 000000000..f44983eee --- /dev/null +++ b/liteflow-testcase-el/liteflow-testcase-el-react-agent/src/test/java/com/yomahub/liteflow/test/agent/v2/InFlightInterruptTest.java @@ -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 推迟的集成级 in-flight interrupt 测试。 + * + *

用 {@link BlockingChunksModel}（race-free 两段阻塞模型，见其 javadoc 的 RC3 concatMap + * 中断检查点分析）启动一个 agent {@code call()}，在 chunk1 被消费后、chunk2 到达前从外部调 + * {@link ReActAgentComponent#interrupt(String, String)}，断言该 call 在合理超时内完成， + * 且返回的 Msg 是 RC3 {@code ReActAgent.handleInterrupt} 产出的恢复消息 + * （{@link BlockingChunksModel#RECOVERY_TEXT}）——即"in-flight call 被中断后以恢复 Msg 结束"。 + * + *

不依赖 Spring 上下文：直接经 {@link ReactAgentFactory#getOrCreate} 取 agent 单例、 + * 手工 {@code agent.call(...)} 并阻塞，跳过 {@code process()} 的 slot/chain 装配。仅验证 + * agent + interrupt 的 reactor 链路，与 {@code InterruptTest}（单元级"不抛 + 参数透传"）互补。 + * + *

时序（race-free，见 {@link BlockingChunksModel}）：interrupt 的 flag 写入严格早于 + * chunk2 到达 {@code checkInterrupted()}——两者都在测试线程内顺序执行，无并发窗口，故不 flaky。 + * + *

降级（brief Step 2 允许）：若因 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 result = new AtomicReference<>(); + AtomicReference 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 之前触发中断（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 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 error) { + throw new AssertionError(msg + + " | streamError=" + model.streamError() + + " | callError=" + (error.get() == null ? "" : error.get())); + } +}