package com.easyagents.agent.runtime.agentscope;

import com.easyagents.agent.runtime.*;
import com.easyagents.agent.runtime.event.*;
import com.easyagents.agent.runtime.event.interceptor.AutoContextInterceptor;
import com.easyagents.agent.runtime.event.interceptor.ToolHitlInterceptor;
import com.easyagents.agent.runtime.event.observer.AgentRuntimeErrorObserver;
import com.easyagents.agent.runtime.event.observer.ReasoningLifecycleObserver;
import com.easyagents.agent.runtime.event.observer.SkillExecutionObserver;
import com.easyagents.agent.runtime.event.observer.ToolExecutionObserver;
import com.easyagents.agent.runtime.hitl.AgentResumeToken;
import com.easyagents.agent.runtime.knowledge.AgentKnowledgeDocument;
import com.easyagents.agent.runtime.knowledge.AgentKnowledgeRegistration;
import com.easyagents.agent.runtime.knowledge.AgentKnowledgeRetrievalResult;
import com.easyagents.agent.runtime.knowledge.AgentKnowledgeSpec;
import com.easyagents.agent.runtime.memory.AgentMemoryCompressionParameter;
import com.easyagents.agent.runtime.memory.AgentMemoryPolicy;
import com.easyagents.agent.runtime.memory.AgentMemorySnapshot;
import com.easyagents.agent.runtime.message.AgentMessage;
import com.easyagents.agent.runtime.message.AgentMessageRole;
import com.easyagents.agent.runtime.message.AgentMediaBlock;
import com.easyagents.agent.runtime.model.AgentModelSpec;
import com.easyagents.agent.runtime.persistence.session.memory.InMemoryAgentSessionStore;
import com.easyagents.agent.runtime.skill.AgentSkillBoxSpec;
import com.easyagents.agent.runtime.skill.AgentSkillRuntimeContext;
import com.easyagents.agent.runtime.skill.AgentSkillSpec;
import com.easyagents.agent.runtime.tool.AgentToolCategory;
import com.easyagents.agent.runtime.tool.AgentToolResult;
import com.easyagents.agent.runtime.tool.AgentToolSpec;
import com.easyagents.agent.runtime.tool.operate.AgentOperateToolAdapter;
import com.easyagents.agent.runtime.tool.operate.AgentOperateToolSpec;
import com.easyagents.agent.runtime.tool.operate.AgentOperateToolType;
import io.agentscope.core.ReActAgent;
import io.agentscope.core.hook.*;
import io.agentscope.core.memory.autocontext.AutoContextHook;
import io.agentscope.core.memory.autocontext.AutoContextMemory;
import io.agentscope.core.message.*;
import io.agentscope.core.model.ChatResponse;
import io.agentscope.core.model.GenerateOptions;
import io.agentscope.core.model.Model;
import io.agentscope.core.model.ToolSchema;
import io.agentscope.core.skill.SkillBox;
import io.agentscope.core.tool.AgentTool;
import io.agentscope.core.tool.Toolkit;
import io.agentscope.core.tool.mcp.McpClientWrapper;
import io.modelcontextprotocol.spec.McpSchema;
import org.junit.Assert;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;

import java.lang.reflect.Field;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BooleanSupplier;

/**
 * 测试有状态 AgentScope 运行时。
 */
public class AgentScopeStatefulRuntimeTest {

    @Test(expected = AgentRuntimeException.class)
    public void shouldRejectStreamBeforeInit() {
        AgentScopeReActRuntime runtime = fakeRuntime();

        runtime.stream(AgentMessage.text(AgentMessageRole.USER, "hello"));
    }

    @Test(expected = AgentRuntimeException.class)
    public void shouldRejectEmptyStreamMessage() {
        AgentScopeReActRuntime runtime = fakeRuntime();
        AgentMessage message = new AgentMessage();
        runtime.init(initRequest());

        runtime.stream(message);
    }

    @Test(expected = AgentRuntimeException.class)
    public void shouldRejectImageWhenModelCapabilityIsDisabled() {
        AgentScopeReActRuntime runtime = fakeRuntime();
        AgentInitRequest request = initRequest();
        runtime.init(request);
        AgentMessage message = new AgentMessage();
        message.setRole(AgentMessageRole.USER);
        AgentMediaBlock image = new AgentMediaBlock("image");
        image.setReference("draft:upload-1");
        message.setContentBlocks(List.of(image));

        runtime.stream(message);
    }

    @Test(expected = AgentRuntimeException.class)
    public void shouldRejectDuplicateInit() {
        AgentScopeReActRuntime runtime = fakeRuntime();
        AgentInitRequest request = initRequest();

        runtime.init(request);
        runtime.init(request);
    }

    @Test(expected = AgentRuntimeException.class)
    public void shouldRequireSessionStore() {
        AgentInitRequest request = initRequest();
        request.setSessionStore(null);

        fakeRuntime().init(request);
    }

    @Test
    public void shouldRegisterUnifiedRuntimeHookForStatefulAutoContextRuntime() {
        AgentInitRequest request = initRequest();
        request.getAgentDefinition().setMemoryPolicy(AgentMemoryPolicy.autoContext());
        AgentScopeReActRuntime runtime = fakeRuntime();

        runtime.init(request);

        Assert.assertTrue(runtime.getAgent().getHooks().stream().anyMatch(this::isRuntimeHook));
        Assert.assertFalse(runtime.getAgent().getHooks().stream().anyMatch(AutoContextHook.class::isInstance));
    }

    @Test
    public void shouldKeepSystemPromptUnchangedWithoutAsyncTool() {
        AgentScopeReActRuntime runtime = fakeRuntime();

        runtime.init(initRequest());

        Assert.assertEquals("system", runtime.getAgent().getSysPrompt());
    }

    @Test
    public void shouldAppendAsyncToolProtocolPromptWhenAsyncToolsExist() {
        AgentInitRequest request = initRequest();
        AgentToolSpec submit = new AgentToolSpec();
        submit.setName("demo_submit");
        submit.setDescription("submit demo");
        submit.getMetadata().put("asyncTool", true);
        submit.getMetadata().put("asyncToolPhase", "submit");
        request.getAgentDefinition().setToolSpecs(List.of(submit));
        request.setToolInvokers(Map.of("demo_submit", (arguments, context) -> AgentToolResult.success("submitted")));
        AgentScopeReActRuntime runtime = fakeRuntime();

        runtime.init(request);

        Assert.assertTrue(runtime.getAgent().getSysPrompt().startsWith("system"));
        Assert.assertTrue(runtime.getAgent().getSysPrompt().contains("Async tool protocol:"));
        Assert.assertTrue(runtime.getAgent().getSysPrompt().contains("These are internal execution phases."));
        Assert.assertTrue(runtime.getAgent().getSysPrompt().contains("call its submit sub-tool first with the user-provided arguments by default"));
        Assert.assertTrue(runtime.getAgent().getSysPrompt().contains("immediately call observe"));
    }

    @Test
    public void shouldAppendKnowledgeToolProtocolPromptWhenKnowledgeToolsExist() {
        AgentInitRequest request = initRequest();
        AgentKnowledgeSpec knowledgeSpec = new AgentKnowledgeSpec();
        knowledgeSpec.setKnowledgeId("knowledge-faq");
        knowledgeSpec.setRuntimeName("faq");
        request.getAgentDefinition().setKnowledgeSpecs(List.of(knowledgeSpec));
        request.setKnowledgeRegistrations(List.of(new AgentKnowledgeRegistration(
                knowledgeSpec, retrievalRequest -> AgentKnowledgeRetrievalResult.of(List.of()))));
        AgentScopeReActRuntime runtime = fakeRuntime();

        runtime.init(request);

        Assert.assertTrue(runtime.getAgent().getSysPrompt()
                .startsWith("system\n\nKnowledge tool protocol:"));
        Assert.assertTrue(runtime.getAgent().getSysPrompt()
                .contains("call the most relevant knowledge tool first"));
        Assert.assertTrue(runtime.getAgent().getSysPrompt()
                .contains("reformulate the query and retry once"));
    }

    @Test
    public void shouldEmitSideEventWithRuntimeIdentityFromBridge() throws Exception {
        AgentRuntimeExecutionContext context = new AgentRuntimeExecutionContext();
        AgentDefinition definition = new AgentDefinition();
        definition.setAgentId("agent-bridge");
        context.setAgentDefinition(definition);
        context.setRequestId("request-bridge");
        context.setTraceId("trace-bridge");
        context.setSessionId("session-bridge");
        context.setMetadata(Map.of("tenantId", "tenant-1"));
        Sinks.Many<AgentRuntimeEvent> sink = Sinks.many().replay().all();
        AgentRuntimeEventBridge bridge = AgentRuntimeEventBridge.fixed(context, sink);

        bridge.emit(AgentRuntimeEvent.of(AgentRuntimeEventType.KNOWLEDGE_RETRIEVAL));
        AgentRuntimeEvent event = sink.asFlux().take(1).collectList().toFuture().get(3, TimeUnit.SECONDS).get(0);

        Assert.assertEquals("trace-bridge", event.getTraceId());
        Assert.assertEquals("session-bridge", event.getSessionId());
        Assert.assertEquals("agent-bridge", event.getAgentId());
        Assert.assertEquals("request-bridge", event.getMetadata().get("requestId"));
        Assert.assertEquals("tenant-1", event.getMetadata().get("tenantId"));
    }

    @Test
    public void shouldKeepAutoContextInterceptorInsideUnifiedRuntimeHook() throws Exception {
        AgentInitRequest request = initRequest();
        request.getAgentDefinition().setMemoryPolicy(AgentMemoryPolicy.autoContext());
        AgentScopeReActRuntime runtime = fakeRuntime();

        runtime.init(request);
        AgentScopeRuntimeHook runtimeHook = runtime.getAgent().getHooks().stream()
                .filter(AgentScopeRuntimeHook.class::isInstance)
                .map(AgentScopeRuntimeHook.class::cast)
                .findFirst()
                .orElseThrow();
        Field managerField = AgentScopeRuntimeHook.class.getDeclaredField("observationManager");
        managerField.setAccessible(true);
        AgentRuntimeObservationManager manager = (AgentRuntimeObservationManager) managerField.get(runtimeHook);
        Field interceptorsField = AgentRuntimeObservationManager.class.getDeclaredField("interceptors");
        interceptorsField.setAccessible(true);
        @SuppressWarnings("unchecked")
        List<AgentRuntimeInterceptor> interceptors = (List<AgentRuntimeInterceptor>) interceptorsField.get(manager);

        Assert.assertTrue(interceptors.stream().anyMatch(AutoContextInterceptor.class::isInstance));
    }

    @Test
    public void shouldKeepToolHitlInterceptorInsideUnifiedRuntimeHook() throws Exception {
        AgentInitRequest request = initRequest();
        AgentToolSpec toolSpec = new AgentToolSpec();
        toolSpec.setName("search");
        toolSpec.setDescription("search");
        toolSpec.setApprovalRequired(true);
        request.getAgentDefinition().setToolSpecs(List.of(toolSpec));
        request.setToolInvokers(Map.of("search", (arguments, context) -> AgentToolResult.success("tool result")));
        AgentScopeReActRuntime runtime = runtimeWithModel(List.of(ChatResponse.builder()
                .id("tool-call-message")
                .content(List.of(ToolUseBlock.builder()
                        .id("call-search")
                        .name("search")
                        .input(Map.of("q", "easyflow"))
                        .build()))
                .finishReason("tool_calls")
                .build()));

        runtime.init(request);
        AgentScopeRuntimeHook runtimeHook = runtime.getAgent().getHooks().stream()
                .filter(AgentScopeRuntimeHook.class::isInstance)
                .map(AgentScopeRuntimeHook.class::cast)
                .findFirst()
                .orElseThrow();
        Field managerField = AgentScopeRuntimeHook.class.getDeclaredField("observationManager");
        managerField.setAccessible(true);
        AgentRuntimeObservationManager manager = (AgentRuntimeObservationManager) managerField.get(runtimeHook);
        Field interceptorsField = AgentRuntimeObservationManager.class.getDeclaredField("interceptors");
        interceptorsField.setAccessible(true);
        @SuppressWarnings("unchecked")
        List<AgentRuntimeInterceptor> interceptors = (List<AgentRuntimeInterceptor>) interceptorsField.get(manager);

        Assert.assertTrue(interceptors.stream().anyMatch(ToolHitlInterceptor.class::isInstance));
    }

    /**
     * 验证调用方提供的历史快照会在 Agent 首次构建时进入模型记忆。
     */
    @Test
    public void shouldAttachInitialConversationHistoryToAgentMemory() {
        AgentInitRequest request = initRequest();
        AgentMemorySnapshot snapshot = new AgentMemorySnapshot();
        snapshot.addMessage(AgentMessage.text(AgentMessageRole.USER, "previous user question"));
        snapshot.addMessage(AgentMessage.text(AgentMessageRole.ASSISTANT, "previous assistant answer"));
        request.setMemorySnapshot(snapshot);
        AgentScopeReActRuntime runtime = fakeRuntime();

        runtime.init(request);

        List<Msg> messages = runtime.getAgent().getMemory().getMessages();
        Assert.assertEquals(2, messages.size());
        Assert.assertEquals("previous user question", messages.get(0).getTextContent());
        Assert.assertEquals("previous assistant answer", messages.get(1).getTextContent());
    }

    @Test
    public void shouldRegisterOperateToolsIntoToolkit() {
        AgentInitRequest request = initRequest();
        request.getAgentDefinition().setOperateToolSpecs(List.of(
                operateToolSpec(AgentOperateToolType.READ_FILE),
                operateToolSpec(AgentOperateToolType.WRITE_FILE),
                operateToolSpec(AgentOperateToolType.PATCH),
                operateToolSpec(AgentOperateToolType.SHELL)));
        AgentScopeReActRuntime runtime = fakeRuntime();

        runtime.init(request);

        Toolkit toolkit = runtime.getAgent().getToolkit();
        Assert.assertNotNull(toolkit.getTool(AgentOperateToolAdapter.VIEW_TEXT_FILE_TOOL));
        Assert.assertNotNull(toolkit.getTool(AgentOperateToolAdapter.LIST_DIRECTORY_TOOL));
        Assert.assertNotNull(toolkit.getTool(AgentOperateToolAdapter.WRITE_TEXT_FILE_TOOL));
        Assert.assertNotNull(toolkit.getTool(AgentOperateToolAdapter.INSERT_TEXT_FILE_TOOL));
        Assert.assertNotNull(toolkit.getTool(AgentOperateToolAdapter.APPLY_PATCH_TOOL));
        Assert.assertNotNull(toolkit.getTool(AgentOperateToolAdapter.EXECUTE_SHELL_COMMAND_TOOL));
    }

    @Test
    public void shouldSuspendShellOperateToolWithToolHitlInterceptor() {
        AgentInitRequest request = initRequest();
        AgentOperateToolSpec shell = operateToolSpec(AgentOperateToolType.SHELL);
        request.getAgentDefinition().setOperateToolSpecs(List.of(shell));
        AgentScopeReActRuntime runtime = runtimeWithModel(List.of(ChatResponse.builder()
                .id("shell-call-message")
                .content(List.of(ToolUseBlock.builder()
                        .id("call-shell")
                        .name(AgentOperateToolAdapter.EXECUTE_SHELL_COMMAND_TOOL)
                        .input(Map.of("command", "pwd"))
                        .build()))
                .finishReason("tool_calls")
                .build()));

        runtime.init(request);
        List<AgentRuntimeEvent> events = runtime.stream(AgentMessage.text(AgentMessageRole.USER, "run shell"))
                .collectList()
                .block(Duration.ofSeconds(5));

        Assert.assertNotNull(events);
        Assert.assertTrue(events.stream().anyMatch(event -> event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED));
        Assert.assertTrue(events.stream().anyMatch(event -> event.getEventType() == AgentRuntimeEventType.SUSPENDED));
    }

    @Test
    public void shouldBypassRemoveApprovalWhenShellApprovalIsDisabled() {
        AgentInitRequest request = initRequest();
        AgentOperateToolSpec shell = operateToolSpec(AgentOperateToolType.SHELL);
        shell.setApprovalRequired(false);
        request.getAgentDefinition().setOperateToolSpecs(List.of(shell));
        AgentScopeReActRuntime runtime = runtimeWithModel(List.of(ChatResponse.builder()
                .id("remove-message")
                .content(List.of(ToolUseBlock.builder()
                        .id("call-remove")
                        .name(AgentOperateToolAdapter.EXECUTE_SHELL_COMMAND_TOOL)
                        .input(Map.of("command", "'rm' removable.txt"))
                        .build()))
                .finishReason("tool_calls")
                .build()));

        runtime.init(request);
        List<AgentRuntimeEvent> events = runtime.stream(AgentMessage.text(AgentMessageRole.USER, "remove file"))
                .collectList()
                .block(Duration.ofSeconds(5));

        Assert.assertNotNull(events);
        Assert.assertFalse(events.stream()
                .anyMatch(event -> event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED));
        Assert.assertFalse(events.stream()
                .anyMatch(event -> event.getEventType() == AgentRuntimeEventType.SUSPENDED));
        Assert.assertTrue(events.stream()
                .anyMatch(event -> event.getEventType() == AgentRuntimeEventType.TOOL_RESULT));
    }

    @Test(expected = AgentRuntimeException.class)
    public void shouldRejectOperateToolNameConflictWithBusinessTool() {
        AgentInitRequest request = initRequest();
        AgentToolSpec toolSpec = new AgentToolSpec();
        toolSpec.setName(AgentOperateToolAdapter.EXECUTE_SHELL_COMMAND_TOOL);
        toolSpec.setDescription("conflict");
        request.getAgentDefinition().setToolSpecs(List.of(toolSpec));
        request.setToolInvokers(Map.of(AgentOperateToolAdapter.EXECUTE_SHELL_COMMAND_TOOL,
                (arguments, context) -> AgentToolResult.success("ok")));
        request.getAgentDefinition().setOperateToolSpecs(List.of(operateToolSpec(AgentOperateToolType.SHELL)));

        fakeRuntime().init(request);
    }

    @Test
    public void shouldEnablePendingToolRecoveryForRejectedHitlContinuation() {
        AgentScopeReActRuntime runtime = fakeRuntime();

        runtime.init(initRequest());

        Assert.assertTrue(runtime.getAgent().getHooks().stream()
                .anyMatch(PendingToolRecoveryHook.class::isInstance));
    }

    @Test
    public void shouldCloseMcpClientsWhenRuntimeCloses() throws Exception {
        AgentScopeReActRuntime runtime = fakeRuntime();
        FakeMcpClientWrapper client = new FakeMcpClientWrapper("mcp-close");
        Field mcpClientsField = AgentScopeReActRuntime.class.getDeclaredField("mcpClients");
        mcpClientsField.setAccessible(true);
        @SuppressWarnings("unchecked")
        List<McpClientWrapper> clients = (List<McpClientWrapper>) mcpClientsField.get(runtime);
        clients.add(client);

        runtime.close();

        Assert.assertTrue(client.closed.get());
        Assert.assertTrue(clients.isEmpty());
    }

    @Test
    public void shouldCloseMcpClientsEvenWhenRuntimeIsStreaming() throws Exception {
        AgentScopeReActRuntime runtime = fakeRuntime();
        FakeMcpClientWrapper client = new FakeMcpClientWrapper("mcp-close-streaming");
        Field mcpClientsField = AgentScopeReActRuntime.class.getDeclaredField("mcpClients");
        mcpClientsField.setAccessible(true);
        @SuppressWarnings("unchecked")
        List<McpClientWrapper> clients = (List<McpClientWrapper>) mcpClientsField.get(runtime);
        clients.add(client);
        Field runningField = AgentScopeReActRuntime.class.getDeclaredField("running");
        runningField.setAccessible(true);
        AtomicBoolean running = (AtomicBoolean) runningField.get(runtime);
        running.set(true);

        runtime.close();

        Assert.assertTrue(client.closed.get());
        Assert.assertTrue(clients.isEmpty());
    }

    @Test
    public void shouldEmitAutoContextCompressionEventsFromInterceptor() throws Exception {
        AgentInitRequest request = initRequest();
        AgentMemoryCompressionParameter parameter = new AgentMemoryCompressionParameter();
        parameter.setMsgThreshold(3);
        parameter.setLastKeep(1);
        parameter.setLargePayloadThreshold(20);
        AgentMemoryPolicy policy = AgentMemoryPolicy.autoContext();
        policy.setCompressionParameter(parameter);
        policy.setMaxAttachedMessageCount(10);
        request.getAgentDefinition().setMemoryPolicy(policy);
        AgentScopeMemoryBuildResult memoryResult = new AgentScopeMemoryAdapter()
                .createMemoryResult(compressionSnapshot(), policy, new FakeAgentScopeModelFactory()
                        .create(request.getAgentDefinition().getModelSpec(), request.getAgentDefinition().getGenerationOptions()));
        ReActAgent agent = ReActAgent.builder()
                .name("agent-name")
                .sysPrompt("system")
                .model(new FakeAgentScopeModelFactory()
                        .create(request.getAgentDefinition().getModelSpec(), request.getAgentDefinition().getGenerationOptions()))
                .memory(memoryResult.getMemory())
                .toolkit(new Toolkit())
                .build();
        Sinks.Many<AgentRuntimeEvent> sink = Sinks.many().replay().all();
        AutoContextInterceptor interceptor = new AutoContextInterceptor(
                AgentRuntimeEventBridge.fixed(executionContext(), sink), memoryResult.getAutoContextConfig());
        CompletableFuture<List<AgentRuntimeEvent>> eventsFuture = sink.asFlux().take(2).collectList().toFuture();
        PreReasoningEvent event = new PreReasoningEvent(agent, "fake-model", null,
                List.of(Msg.builder()
                        .role(MsgRole.SYSTEM)
                        .name("system")
                        .content(TextBlock.builder().text("system").build())
                        .build()));

        interceptor.intercept(event).block();
        List<AgentRuntimeEvent> events = eventsFuture.get(3, TimeUnit.SECONDS);

        Assert.assertEquals(AgentRuntimeEventType.MEMORY_COMPRESSION_STARTED, events.get(0).getEventType());
        Assert.assertEquals(AgentRuntimeEventType.MEMORY_COMPRESSION_COMPLETED, events.get(1).getEventType());
        Assert.assertEquals("memory-compression", events.get(0).getPayload().get("statusKey"));
        Assert.assertEquals(Boolean.TRUE, events.get(1).getPayload().get("compressed"));
        Assert.assertTrue(((Number) events.get(1).getPayload().get("eventCount")).intValue() > 0);
        @SuppressWarnings("unchecked")
        List<Map<String, Object>> compressionEvents = (List<Map<String, Object>>) events.get(1).getPayload().get("events");
        Assert.assertFalse(compressionEvents.isEmpty());
        Assert.assertTrue(event.getInputMessages().get(0).getTextContent().contains("CONTEXT_OFFLOAD"));
        Assert.assertTrue(((AutoContextMemory) agent.getMemory()).getMessages().stream()
                .anyMatch(message -> message.getTextContent() != null
                        && message.getTextContent().contains("CONTEXT_OFFLOAD")));
    }

    @Test
    public void shouldNotEmitAutoContextCompressionEventsWhenThresholdIsNotReached() {
        AgentInitRequest request = initRequest();
        AgentMemoryCompressionParameter parameter = new AgentMemoryCompressionParameter();
        parameter.setMsgThreshold(10);
        AgentMemoryPolicy policy = AgentMemoryPolicy.autoContext();
        policy.setCompressionParameter(parameter);
        request.getAgentDefinition().setMemoryPolicy(policy);
        AgentScopeMemoryBuildResult memoryResult = new AgentScopeMemoryAdapter()
                .createMemoryResult(null, policy, new FakeAgentScopeModelFactory()
                        .create(request.getAgentDefinition().getModelSpec(), request.getAgentDefinition().getGenerationOptions()));
        memoryResult.getMemory().addMessage(Msg.builder().role(MsgRole.USER).textContent("hello").build());
        ReActAgent agent = ReActAgent.builder()
                .name("agent-name")
                .sysPrompt("system")
                .model(new FakeAgentScopeModelFactory()
                        .create(request.getAgentDefinition().getModelSpec(), request.getAgentDefinition().getGenerationOptions()))
                .memory(memoryResult.getMemory())
                .toolkit(new Toolkit())
                .build();
        Sinks.Many<AgentRuntimeEvent> sink = Sinks.many().replay().all();
        AutoContextInterceptor interceptor = new AutoContextInterceptor(
                AgentRuntimeEventBridge.fixed(executionContext(), sink), memoryResult.getAutoContextConfig());
        PreReasoningEvent event = new PreReasoningEvent(agent, "fake-model", null,
                List.of(Msg.builder().role(MsgRole.SYSTEM).textContent("system").build()));

        interceptor.intercept(event).block();

        Assert.assertTrue(sink.asFlux().take(Duration.ofMillis(100)).collectList().block().isEmpty());
        Assert.assertTrue(event.getInputMessages().get(0).getTextContent().contains("CONTEXT_OFFLOAD"));
    }

    @Test
    public void shouldEmitNoCompressionCompletedEventWhenCompressionIsSkipped() throws Exception {
        AgentInitRequest request = initRequest();
        AgentMemoryCompressionParameter parameter = new AgentMemoryCompressionParameter();
        parameter.setMsgThreshold(1);
        parameter.setLastKeep(1);
        parameter.setMinCompressionTokenThreshold(300);
        AgentMemoryPolicy policy = AgentMemoryPolicy.autoContext();
        policy.setCompressionParameter(parameter);
        request.getAgentDefinition().setMemoryPolicy(policy);
        AgentScopeMemoryBuildResult memoryResult = new AgentScopeMemoryAdapter()
                .createMemoryResult(null, policy, new FakeAgentScopeModelFactory()
                        .create(request.getAgentDefinition().getModelSpec(), request.getAgentDefinition().getGenerationOptions()));
        memoryResult.getMemory().addMessage(Msg.builder().role(MsgRole.USER).textContent("hi").build());
        ReActAgent agent = ReActAgent.builder()
                .name("agent-name")
                .sysPrompt("system")
                .model(new FakeAgentScopeModelFactory()
                        .create(request.getAgentDefinition().getModelSpec(), request.getAgentDefinition().getGenerationOptions()))
                .memory(memoryResult.getMemory())
                .toolkit(new Toolkit())
                .build();
        Sinks.Many<AgentRuntimeEvent> sink = Sinks.many().replay().all();
        AutoContextInterceptor interceptor = new AutoContextInterceptor(
                AgentRuntimeEventBridge.fixed(executionContext(), sink), memoryResult.getAutoContextConfig());
        CompletableFuture<List<AgentRuntimeEvent>> eventsFuture = sink.asFlux().take(2).collectList().toFuture();
        PreReasoningEvent event = new PreReasoningEvent(agent, "fake-model", null,
                List.of(Msg.builder().role(MsgRole.SYSTEM).textContent("system").build()));

        interceptor.intercept(event).block();
        List<AgentRuntimeEvent> events = eventsFuture.get(3, TimeUnit.SECONDS);

        Assert.assertEquals(AgentRuntimeEventType.MEMORY_COMPRESSION_STARTED, events.get(0).getEventType());
        Assert.assertEquals(AgentRuntimeEventType.MEMORY_COMPRESSION_COMPLETED, events.get(1).getEventType());
        Assert.assertEquals(Boolean.FALSE, events.get(1).getPayload().get("compressed"));
        Assert.assertEquals("无需压缩上下文", events.get(1).getPayload().get("label"));
        Assert.assertEquals(0, ((Number) events.get(1).getPayload().get("eventCount")).intValue());
        Assert.assertTrue(event.getInputMessages().get(0).getTextContent().contains("CONTEXT_OFFLOAD"));
    }

    @Test
    public void shouldRegisterContextReloadToolFromAutoContextInterceptor() {
        AgentMemoryPolicy policy = AgentMemoryPolicy.autoContext();
        AgentScopeMemoryBuildResult memoryResult = new AgentScopeMemoryAdapter()
                .createMemoryResult(null, policy, new FakeAgentScopeModelFactory().create(new AgentModelSpec(), null));
        ReActAgent agent = ReActAgent.builder()
                .name("agent-name")
                .model(new FakeAgentScopeModelFactory().create(new AgentModelSpec(), null))
                .memory(memoryResult.getMemory())
                .toolkit(new Toolkit())
                .build();
        Sinks.Many<AgentRuntimeEvent> sink = Sinks.many().replay().all();
        AutoContextInterceptor interceptor = new AutoContextInterceptor(
                AgentRuntimeEventBridge.fixed(executionContext(), sink), memoryResult.getAutoContextConfig());

        interceptor.intercept(new PreCallEvent(agent, List.of())).block();

        Assert.assertTrue(interceptor.isRegistered());
        Assert.assertNotNull(agent.getToolkit().getTool("context_reload"));
    }

    @Test
    public void shouldEmitToolCallAndResultFromToolExecutionObserver() throws Exception {
        AgentRuntimeExecutionContext context = executionContext();
        Sinks.Many<AgentRuntimeEvent> sink = Sinks.many().replay().all();
        AgentRuntimeEventBridge bridge = AgentRuntimeEventBridge.fixed(context, sink);
        AgentToolSpec toolSpec = new AgentToolSpec();
        toolSpec.setName("search");
        toolSpec.getMetadata().put("toolDisplayName", "Search Tool");
        toolSpec.getMetadata().put("rawMcpToolName", "search");
        ToolExecutionObserver observer = new ToolExecutionObserver(bridge, null, List.of(toolSpec));
        ReActAgent agent = initializedAgent();
        Toolkit toolkit = agent.getToolkit();
        ToolUseBlock toolUse = ToolUseBlock.builder()
                .id("call-1")
                .name("search")
                .input(Map.of("q", "sentinel-secret-input"))
                .metadata(Map.of("authorization", "sentinel-secret-metadata"))
                .build();
        ToolResultBlock toolResult = ToolResultBlock.of("call-1", "search",
                TextBlock.builder().text("sentinel-secret-result").build(),
                Map.of("success", true, "token", "sentinel-secret-result-metadata"));

        observer.observe(new PreActingEvent(agent, toolkit, toolUse)).block();
        observer.observe(new PostActingEvent(agent, toolkit, toolUse, toolResult)).block();
        List<AgentRuntimeEvent> events = sink.asFlux().take(2).collectList().toFuture().get(3, TimeUnit.SECONDS);

        Assert.assertEquals(AgentRuntimeEventType.TOOL_CALL, events.get(0).getEventType());
        Assert.assertEquals("RUNNING", events.get(0).getPayload().get("status"));
        Assert.assertEquals("Search Tool", events.get(0).getPayload().get("toolDisplayName"));
        Assert.assertFalse(events.get(0).getPayload().containsKey("input"));
        Assert.assertFalse(events.get(0).getPayload().containsKey("content"));
        Assert.assertFalse(events.get(0).getMetadata().toString().contains("sentinel-secret"));
        Assert.assertEquals(AgentRuntimeEventType.TOOL_RESULT, events.get(1).getEventType());
        Assert.assertEquals("SUCCESS", events.get(1).getPayload().get("status"));
        Assert.assertEquals("Search Tool", events.get(1).getPayload().get("toolDisplayName"));
        Assert.assertFalse(events.get(1).getPayload().containsKey("text"));
        Assert.assertFalse(events.get(1).getMetadata().toString().contains("sentinel-secret"));
        Assert.assertFalse(events.toString().contains("sentinel-secret"));
    }

    /**
     * 验证知识库工具复用统一工具生命周期，并携带可供调用方投影的知识库分类。
     *
     * @throws Exception 等待旁路事件失败时抛出
     */
    @Test
    public void shouldEmitKnowledgeToolLifecycleFromToolExecutionObserver() throws Exception {
        AgentRuntimeExecutionContext context = executionContext();
        Sinks.Many<AgentRuntimeEvent> sink = Sinks.many().replay().all();
        AgentRuntimeEventBridge bridge = AgentRuntimeEventBridge.fixed(context, sink);
        AgentToolSpec toolSpec = new AgentToolSpec();
        toolSpec.setName("retrieve_knowledge_homeinn_faq");
        toolSpec.setCategory(AgentToolCategory.KNOWLEDGE);
        toolSpec.getMetadata().put("toolDisplayName", "如家 FAQ");
        toolSpec.getMetadata().put("knowledgeId", "knowledge-1");
        toolSpec.getMetadata().put("knowledgeName", "如家 FAQ");
        toolSpec.getMetadata().put("knowledgeRuntimeName", "homeinn_faq");
        ToolExecutionObserver observer = new ToolExecutionObserver(bridge, null, List.of(toolSpec));
        ReActAgent agent = initializedAgent();
        Toolkit toolkit = agent.getToolkit();
        ToolUseBlock toolUse = ToolUseBlock.builder()
                .id("knowledge-call-1")
                .name("retrieve_knowledge_homeinn_faq")
                .input(Map.of("query", "几点退房"))
                .build();
        ToolResultBlock toolResult = ToolResultBlock.of(
                "knowledge-call-1",
                "retrieve_knowledge_homeinn_faq",
                TextBlock.builder().text("中午十二点退房").build(),
                Map.of("success", true));

        observer.observe(new PreActingEvent(agent, toolkit, toolUse)).block();
        observer.observe(new PostActingEvent(agent, toolkit, toolUse, toolResult)).block();
        List<AgentRuntimeEvent> events = sink.asFlux().take(2).collectList().toFuture()
                .get(3, TimeUnit.SECONDS);

        Assert.assertEquals(AgentRuntimeEventType.TOOL_CALL, events.get(0).getEventType());
        Assert.assertEquals("KNOWLEDGE", events.get(0).getPayload().get("toolCategory"));
        Assert.assertEquals("knowledge-1", events.get(0).getPayload().get("knowledgeId"));
        Assert.assertEquals("RUNNING", events.get(0).getPayload().get("status"));
        Assert.assertEquals(AgentRuntimeEventType.TOOL_RESULT, events.get(1).getEventType());
        Assert.assertEquals("KNOWLEDGE", events.get(1).getPayload().get("toolCategory"));
        Assert.assertEquals("SUCCESS", events.get(1).getPayload().get("status"));
    }

    /**
     * 验证 AgentScope 的文本错误结果会投影为失败工具状态。
     *
     * @throws Exception 等待旁路事件失败时抛出
     */
    @Test
    public void shouldMarkAgentScopeKnowledgeToolErrorAsFailed() throws Exception {
        AgentRuntimeExecutionContext context = executionContext();
        Sinks.Many<AgentRuntimeEvent> sink = Sinks.many().replay().all();
        AgentRuntimeEventBridge bridge = AgentRuntimeEventBridge.fixed(context, sink);
        AgentToolSpec toolSpec = new AgentToolSpec();
        toolSpec.setName("retrieve_knowledge_homeinn_faq");
        toolSpec.setCategory(AgentToolCategory.KNOWLEDGE);
        ToolExecutionObserver observer = new ToolExecutionObserver(bridge, null, List.of(toolSpec));
        ReActAgent agent = initializedAgent();
        ToolUseBlock toolUse = ToolUseBlock.builder()
                .id("knowledge-call-error")
                .name("retrieve_knowledge_homeinn_faq")
                .input(Map.of("query", "异常查询"))
                .build();
        ToolResultBlock toolResult = ToolResultBlock.error("retriever unavailable")
                .withIdAndName("knowledge-call-error", "retrieve_knowledge_homeinn_faq");

        observer.observe(new PostActingEvent(agent, agent.getToolkit(), toolUse, toolResult)).block();
        AgentRuntimeEvent event = sink.asFlux().next().toFuture().get(3, TimeUnit.SECONDS);

        Assert.assertEquals(AgentRuntimeEventType.TOOL_RESULT, event.getEventType());
        Assert.assertEquals("FAILED", event.getPayload().get("status"));
        Assert.assertEquals(Boolean.FALSE, event.getPayload().get("success"));
    }

    @Test
    public void shouldEmitSkillLifecycleEventsFromSkillExecutionObserver() throws Exception {
        AgentRuntimeExecutionContext context = executionContext();
        Sinks.Many<AgentRuntimeEvent> sink = Sinks.many().replay().all();
        AgentRuntimeEventBridge bridge = AgentRuntimeEventBridge.fixed(context, sink);
        AgentSkillRuntimeContext skillContext = AgentSkillRuntimeContext.from(skillBoxSpec());
        Toolkit toolkit = new Toolkit();
        SkillBox skillBox = skillBox(toolkit);
        SkillExecutionObserver observer = new SkillExecutionObserver(bridge, skillContext, skillBox);
        ReActAgent agent = initializedAgent();
        toolkit.updateToolGroups(List.of("skill-1_skill_tools"), true);
        ToolUseBlock loadSkill = ToolUseBlock.builder()
                .id("skill-call-1")
                .name(AgentSkillRuntimeContext.LOAD_SKILL_TOOL_NAME)
                .input(Map.of("skillId", "skill-1", "path", "SKILL.md"))
                .build();
        ToolResultBlock loadResult = ToolResultBlock.of("skill-call-1",
                AgentSkillRuntimeContext.LOAD_SKILL_TOOL_NAME,
                TextBlock.builder().text("Successfully loaded skill: skill-1").build(), Map.of("success", true));
        ToolUseBlock skillToolUse = ToolUseBlock.builder()
                .id("skill-step-1")
                .name("search")
                .input(Map.of("q", "easyflow"))
                .build();
        ToolResultBlock skillToolResult = ToolResultBlock.of("skill-step-1", "search",
                TextBlock.builder().text("done").build(), Map.of("success", true));

        observer.observe(new PreActingEvent(agent, toolkit, loadSkill)).block();
        observer.observe(new PostActingEvent(agent, toolkit, loadSkill, loadResult)).block();
        observer.observe(new PreActingEvent(agent, toolkit, skillToolUse)).block();
        observer.observe(new PostActingEvent(agent, toolkit, skillToolUse, skillToolResult)).block();
        List<AgentRuntimeEvent> events = sink.asFlux().take(4).collectList().toFuture().get(3, TimeUnit.SECONDS);

        Assert.assertEquals(AgentRuntimeEventType.SKILL_CALL, events.get(0).getEventType());
        Assert.assertEquals(AgentRuntimeEventType.SKILL_RESULT, events.get(1).getEventType());
        Assert.assertTrue(skillContext.isSkillActive("skill-1"));
        Assert.assertEquals(AgentRuntimeEventType.SKILL_STEP, events.get(2).getEventType());
        Assert.assertEquals("TOOL_CALL", events.get(2).getPayload().get("stepType"));
        Assert.assertEquals(AgentRuntimeEventType.SKILL_STEP, events.get(3).getEventType());
        Assert.assertEquals("TOOL_RESULT", events.get(3).getPayload().get("stepType"));
        Assert.assertEquals("skill-1", events.get(3).getPayload().get("skillId"));
    }

    @Test
    public void shouldUseAgentScopeSkillStateWhenSkillLoadReturnsErrorText() throws Exception {
        AgentRuntimeExecutionContext context = executionContext();
        Sinks.Many<AgentRuntimeEvent> sink = Sinks.many().replay().all();
        AgentRuntimeEventBridge bridge = AgentRuntimeEventBridge.fixed(context, sink);
        AgentSkillRuntimeContext skillContext = AgentSkillRuntimeContext.from(skillBoxSpec());
        Toolkit toolkit = new Toolkit();
        SkillBox skillBox = skillBox(toolkit);
        SkillExecutionObserver observer = new SkillExecutionObserver(bridge, skillContext, skillBox);
        ReActAgent agent = initializedAgent();
        ToolUseBlock loadSkill = ToolUseBlock.builder()
                .id("skill-call-error")
                .name(AgentSkillRuntimeContext.LOAD_SKILL_TOOL_NAME)
                .input(Map.of("skillId", "skill-1", "path", "missing.md"))
                .build();

        skillBox.getSkill("skill-1").getResource("missing.md");
        toolkit.updateToolGroups(List.of("skill-1_skill_tools"), true);
        ToolResultBlock loadResult = ToolResultBlock.of("skill-call-error",
                AgentSkillRuntimeContext.LOAD_SKILL_TOOL_NAME,
                TextBlock.builder().text("Error: Resource not found").build(), Map.of("success", true));
        observer.observe(new PreActingEvent(agent, toolkit, loadSkill)).block();
        observer.observe(new PostActingEvent(agent, toolkit, loadSkill, loadResult)).block();
        List<AgentRuntimeEvent> events = sink.asFlux().take(2).collectList().toFuture().get(3, TimeUnit.SECONDS);

        Assert.assertEquals(AgentRuntimeEventType.SKILL_RESULT, events.get(1).getEventType());
        Assert.assertEquals(Boolean.TRUE, events.get(1).getPayload().get("active"));
        Assert.assertTrue(skillContext.isSkillActive("skill-1"));
    }

    @Test
    public void shouldEmitSkillStepFromRestoredToolkitActiveGroup() throws Exception {
        AgentRuntimeExecutionContext context = executionContext();
        Sinks.Many<AgentRuntimeEvent> sink = Sinks.many().replay().all();
        AgentRuntimeEventBridge bridge = AgentRuntimeEventBridge.fixed(context, sink);
        AgentSkillRuntimeContext skillContext = AgentSkillRuntimeContext.from(skillBoxSpec());
        Toolkit toolkit = new Toolkit();
        skillBox(toolkit);
        toolkit.updateToolGroups(List.of("skill-1_skill_tools"), true);
        SkillExecutionObserver observer = new SkillExecutionObserver(bridge, skillContext, null);
        ReActAgent agent = initializedAgent();
        ToolUseBlock skillToolUse = ToolUseBlock.builder()
                .id("skill-step-restored")
                .name("search")
                .input(Map.of("q", "easyflow"))
                .build();
        ToolResultBlock skillToolResult = ToolResultBlock.of("skill-step-restored", "search",
                TextBlock.builder().text("done").build(), Map.of("success", true));

        observer.observe(new PreActingEvent(agent, toolkit, skillToolUse)).block();
        observer.observe(new PostActingEvent(agent, toolkit, skillToolUse, skillToolResult)).block();
        List<AgentRuntimeEvent> events = sink.asFlux().take(2).collectList().toFuture().get(3, TimeUnit.SECONDS);

        Assert.assertEquals(AgentRuntimeEventType.SKILL_STEP, events.get(0).getEventType());
        Assert.assertEquals("TOOL_CALL", events.get(0).getPayload().get("stepType"));
        Assert.assertEquals(AgentRuntimeEventType.SKILL_STEP, events.get(1).getEventType());
        Assert.assertEquals("TOOL_RESULT", events.get(1).getPayload().get("stepType"));
        Assert.assertTrue(skillContext.isSkillActive("skill-1"));
    }

    @Test
    public void shouldEmitReasoningLifecycleEvents() throws Exception {
        AgentRuntimeExecutionContext context = executionContext();
        Sinks.Many<AgentRuntimeEvent> sink = Sinks.many().replay().all();
        AgentRuntimeEventBridge bridge = AgentRuntimeEventBridge.fixed(context, sink);
        ReasoningLifecycleObserver observer = new ReasoningLifecycleObserver(bridge);
        ReActAgent agent = initializedAgent();
        Msg input = Msg.builder().role(MsgRole.USER).textContent("hello").build();
        Msg reasoning = Msg.builder().role(MsgRole.ASSISTANT).textContent("thinking done").build();

        observer.observe(new PreReasoningEvent(agent, "fake-model", null, List.of(input))).block();
        observer.observe(new PostReasoningEvent(agent, "fake-model", null, reasoning)).block();
        List<AgentRuntimeEvent> events = sink.asFlux().take(2).collectList().toFuture().get(3, TimeUnit.SECONDS);

        Assert.assertEquals(AgentRuntimeEventType.REASONING_STARTED, events.get(0).getEventType());
        Assert.assertEquals("PRE_REASONING", events.get(0).getPayload().get("phase"));
        Assert.assertEquals(AgentRuntimeEventType.REASONING_COMPLETED, events.get(1).getEventType());
        Assert.assertEquals("POST_REASONING", events.get(1).getPayload().get("phase"));
        Assert.assertEquals("thinking done", events.get(1).getPayload().get("text"));
    }

    @Test
    public void shouldEmitFailedFromRuntimeErrorObserver() throws Exception {
        AgentRuntimeExecutionContext context = executionContext();
        Sinks.Many<AgentRuntimeEvent> sink = Sinks.many().replay().all();
        AgentRuntimeEventBridge bridge = AgentRuntimeEventBridge.fixed(context, sink);
        AgentRuntimeErrorObserver observer = new AgentRuntimeErrorObserver(bridge);
        ReActAgent agent = initializedAgent();

        observer.observe(new ErrorEvent(agent, new IllegalStateException("boom"))).block();
        AgentRuntimeEvent event = sink.asFlux().take(1).collectList().toFuture().get(3, TimeUnit.SECONDS).get(0);

        Assert.assertEquals(AgentRuntimeEventType.FAILED, event.getEventType());
        Assert.assertEquals("HOOK", event.getPayload().get("source"));
        Assert.assertEquals("ERROR", event.getPayload().get("phase"));
        Assert.assertEquals(IllegalStateException.class.getName(), event.getPayload().get("errorType"));
        Assert.assertEquals("boom", event.getPayload().get("message"));
    }

    @Test
    public void shouldStreamStatefulStartedMessageAndCompletedEvents() {
        InMemoryAgentSessionStore sessionStore = new InMemoryAgentSessionStore();
        AgentInitRequest request = initRequest();
        request.setSessionStore(sessionStore);
        AgentScopeReActRuntime runtime = fakeRuntime();
        runtime.init(request);

        List<AgentRuntimeEvent> events = runtime.stream(AgentMessage.text(AgentMessageRole.USER, "hello"))
                .collectList()
                .block();

        Assert.assertFalse(events.isEmpty());
        Assert.assertEquals(AgentRuntimeEventType.STARTED, events.get(0).getEventType());
        Assert.assertTrue(events.stream().anyMatch(event -> event.getEventType() == AgentRuntimeEventType.MESSAGE_DELTA));
        String streamedText = events.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.MESSAGE_DELTA)
                .map(event -> String.valueOf(event.getPayload().getOrDefault("text", "")))
                .reduce("", String::concat);
        Assert.assertEquals("fake answer", streamedText);
        AgentRuntimeEvent completed = events.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.COMPLETED)
                .findFirst()
                .orElseThrow();
        Assert.assertEquals("fake answer", completed.getPayload().get("text"));
        Assert.assertEquals("trace-1", completed.getTraceId());
        Assert.assertEquals("session-1", completed.getSessionId());
        Assert.assertTrue(sessionStore.exists("session-1"));
    }

    /**
     * 验证增量模式会保留 URL 中连续出现的相同字符。
     */
    @Test
    public void shouldPreserveRepeatedIdenticalTextDeltas() {
        String expectedUrl = "http://127.0.0.1:39000/easyflow/file.docx";
        AgentScopeReActRuntime runtime = runtimeWithStreamingModel(List.of(
                ChatResponse.builder()
                        .id("url-response")
                        .content(List.of(TextBlock.builder().text("http://127.0.0.1:39").build()))
                        .build(),
                ChatResponse.builder()
                        .id("url-response")
                        .content(List.of(TextBlock.builder().text("0").build()))
                        .build(),
                ChatResponse.builder()
                        .id("url-response")
                        .content(List.of(TextBlock.builder().text("0").build()))
                        .build(),
                ChatResponse.builder()
                        .id("url-response")
                        .content(List.of(TextBlock.builder().text("0/easyflow/file.docx").build()))
                        .finishReason("stop")
                        .build()));
        runtime.init(initRequest());

        List<AgentRuntimeEvent> events = runtime.stream(
                        AgentMessage.text(AgentMessageRole.USER, "create file"))
                .collectList()
                .block();
        String streamedText = events.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.MESSAGE_DELTA)
                .map(event -> String.valueOf(event.getPayload().getOrDefault("text", "")))
                .reduce("", String::concat);

        Assert.assertEquals(expectedUrl, streamedText);
        AgentRuntimeEvent completed = events.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.COMPLETED)
                .findFirst()
                .orElseThrow();
        Assert.assertEquals(expectedUrl, completed.getPayload().get("text"));
    }

    @Test
    public void shouldAllowNextStreamAfterPreviousStreamCompleted() {
        AgentScopeReActRuntime runtime = fakeRuntime();
        runtime.init(initRequest());

        runtime.stream(AgentMessage.text(AgentMessageRole.USER, "first")).collectList().block();
        List<AgentRuntimeEvent> events = runtime.stream(AgentMessage.text(AgentMessageRole.USER, "second"))
                .collectList()
                .block();

        Assert.assertTrue(events.stream().anyMatch(event -> event.getEventType() == AgentRuntimeEventType.COMPLETED));
    }

    @Test
    public void shouldRejectConcurrentStatefulStream() {
        AgentScopeReActRuntime runtime = runtimeWithModel(List.of(
                ChatResponse.builder()
                        .id("slow-response")
                        .content(List.of(TextBlock.builder().text("still running").build()))
                        .finishReason("stop")
                        .build()), Duration.ofSeconds(1));
        runtime.init(initRequest());

        reactor.core.Disposable disposable = runtime.stream(AgentMessage.text(AgentMessageRole.USER, "first"))
                .subscribe();
        try {
            runtime.stream(AgentMessage.text(AgentMessageRole.USER, "second")).collectList().block();
            Assert.fail("Concurrent stream should be rejected.");
        } catch (AgentRuntimeException expected) {
            Assert.assertTrue(expected.getMessage().contains("already streaming"));
        } finally {
            disposable.dispose();
        }
    }

    @Test
    public void shouldPersistPartialAssistantMessageWhenStreamIsCancelled() throws Exception {
        InMemoryAgentSessionStore sessionStore = new InMemoryAgentSessionStore();
        AgentInitRequest request = initRequest();
        request.setSessionStore(sessionStore);
        AgentScopeReActRuntime runtime = runtimeWithModel(List.of(
                ChatResponse.builder()
                        .id("partial-message")
                        .content(List.of(TextBlock.builder().text("partial answer").build()))
                        .finishReason("stop")
                        .build()),
                Duration.ofMillis(500));
        runtime.init(request);

        CompletableFuture<AgentRuntimeEvent> firstDelta = new CompletableFuture<>();
        reactor.core.Disposable disposable = runtime.stream(AgentMessage.text(AgentMessageRole.USER, "first"))
                .subscribe(event -> {
                    if (event.getEventType() == AgentRuntimeEventType.MESSAGE_DELTA) {
                        firstDelta.complete(event);
                    }
                }, firstDelta::completeExceptionally);
        AgentRuntimeEvent delta = firstDelta.get(3, TimeUnit.SECONDS);
        Assert.assertEquals("partial answer", delta.getPayload().get("text"));

        disposable.dispose();
        awaitCondition(() -> sessionStore.exists("session-1"));
        AgentScopeReActRuntime restoredRuntime = fakeRuntime();
        restoredRuntime.init(request);

        Assert.assertTrue(restoredRuntime.getAgent().getMemory().getMessages().stream()
                .anyMatch(message -> message.getRole() == MsgRole.ASSISTANT
                        && "partial answer".equals(message.getTextContent())));
    }

    @Test
    public void shouldPersistUserMessageWhenModelFails() {
        InMemoryAgentSessionStore sessionStore = new InMemoryAgentSessionStore();
        AgentInitRequest request = initRequest();
        request.setSessionStore(sessionStore);
        AgentScopeReActRuntime runtime = runtimeWithError(new IllegalStateException("model unavailable"));
        runtime.init(request);

        List<AgentRuntimeEvent> events = runtime.stream(
                        AgentMessage.text(AgentMessageRole.USER, "remember this request"))
                .collectList()
                .block();

        Assert.assertTrue(events.stream().anyMatch(event -> event.getEventType() == AgentRuntimeEventType.FAILED));
        Assert.assertTrue(sessionStore.exists("session-1"));
        AgentScopeReActRuntime restoredRuntime = fakeRuntime();
        restoredRuntime.init(request);
        Assert.assertTrue(restoredRuntime.getAgent().getMemory().getMessages().stream()
                .anyMatch(message -> message.getRole() == MsgRole.USER
                        && "remember this request".equals(message.getTextContent())));
    }

    @Test
    public void shouldPersistUserMessageWhenReasoningOnlyStreamIsCancelled() throws Exception {
        InMemoryAgentSessionStore sessionStore = new InMemoryAgentSessionStore();
        AgentInitRequest request = initRequest();
        request.setSessionStore(sessionStore);
        AgentScopeReActRuntime runtime = runtimeWithModel(List.of(
                ChatResponse.builder()
                        .id("reasoning-only")
                        .content(List.of(ThinkingBlock.builder().thinking("still thinking").build()))
                        .build()), Duration.ofSeconds(5));
        runtime.init(request);

        CompletableFuture<AgentRuntimeEvent> firstReasoning = new CompletableFuture<>();
        reactor.core.Disposable disposable = runtime.stream(
                        AgentMessage.text(AgentMessageRole.USER, "remember reasoning request"))
                .subscribe(event -> {
                    if (event.getEventType() == AgentRuntimeEventType.REASONING_DELTA) {
                        firstReasoning.complete(event);
                    }
                }, firstReasoning::completeExceptionally);
        firstReasoning.get(3, TimeUnit.SECONDS);

        disposable.dispose();
        awaitCondition(() -> sessionStore.exists("session-1"));
        AgentScopeReActRuntime restoredRuntime = fakeRuntime();
        restoredRuntime.init(request);
        Assert.assertTrue(restoredRuntime.getAgent().getMemory().getMessages().stream()
                .anyMatch(message -> message.getRole() == MsgRole.USER
                        && "remember reasoning request".equals(message.getTextContent())));
    }

    @Test
    public void shouldNotDuplicateNormalToolEventsFromMainStream() {
        AgentInitRequest request = initRequest();
        AgentToolSpec toolSpec = new AgentToolSpec();
        toolSpec.setName("search");
        toolSpec.setDescription("search");
        request.getAgentDefinition().setToolSpecs(List.of(toolSpec));
        AtomicBoolean invoked = new AtomicBoolean(false);
        request.setToolInvokers(Map.of("search", (arguments, context) -> {
            invoked.set(true);
            return AgentToolResult.success("tool result");
        }));
        AgentScopeReActRuntime runtime = runtimeWithModel(List.of(
                ChatResponse.builder()
                        .id("tool-call-message")
                        .content(List.of(ToolUseBlock.builder()
                                .id("call-search")
                                .name("search")
                                .input(Map.of("q", "easyflow"))
                                .build()))
                        .finishReason("tool_calls")
                        .build(),
                ChatResponse.builder()
                        .id("final-message")
                        .content(List.of(TextBlock.builder().text("done").build()))
                        .finishReason("stop")
                        .build()));
        runtime.init(request);

        List<AgentRuntimeEvent> events = runtime.stream(AgentMessage.text(AgentMessageRole.USER, "use tool"))
                .collectList()
                .block();

        Assert.assertTrue(invoked.get());
        Assert.assertEquals(1, events.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.TOOL_CALL)
                .count());
        Assert.assertEquals(1, events.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.TOOL_RESULT)
                .count());
    }

    @Test
    public void shouldSuspendBeforeApprovedToolExecutionFromToolHitlInterceptor() {
        AgentInitRequest request = initRequest();
        AgentToolSpec toolSpec = new AgentToolSpec();
        toolSpec.setName("search");
        toolSpec.setDescription("search");
        toolSpec.setApprovalRequired(true);
        toolSpec.getApprovalRequest().setApprovalPrompt("Approve search?");
        toolSpec.getMetadata().put("toolDisplayName", "MCP Search - search");
        toolSpec.getMetadata().put("rawMcpToolName", "search");
        request.getAgentDefinition().setToolSpecs(List.of(toolSpec));
        AtomicBoolean invoked = new AtomicBoolean(false);
        request.setToolInvokers(Map.of("search", (arguments, context) -> {
            invoked.set(true);
            return AgentToolResult.success("tool result");
        }));
        AgentScopeReActRuntime runtime = runtimeWithModel(List.of(ChatResponse.builder()
                .id("tool-call-message")
                .content(List.of(ToolUseBlock.builder()
                        .id("call-search")
                        .name("search")
                        .input(Map.of("q", "easyflow"))
                        .build()))
                .finishReason("tool_calls")
                .build()));
        runtime.init(request);

        List<AgentRuntimeEvent> events = runtime.stream(AgentMessage.text(AgentMessageRole.USER, "use tool"))
                .collectList()
                .block();

        Assert.assertFalse(invoked.get());
        Assert.assertTrue(events.stream()
                .anyMatch(event -> event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED));
        AgentRuntimeEvent approval = events.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED)
                .findFirst()
                .orElseThrow();
        Assert.assertEquals("MCP Search - search", approval.getPayload().get("toolDisplayName"));
        Assert.assertEquals("search", approval.getPayload().get("rawMcpToolName"));
        AgentRuntimeEvent suspended = events.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.SUSPENDED)
                .findFirst()
                .orElseThrow();
        Assert.assertEquals("TOOL_APPROVAL_REQUIRED", suspended.getPayload().get("reason"));
        Assert.assertFalse(((List<?>) suspended.getPayload().get("pendingApprovals")).isEmpty());
        Assert.assertFalse(events.stream().anyMatch(event -> event.getEventType() == AgentRuntimeEventType.COMPLETED));
        Assert.assertFalse(events.stream().anyMatch(event -> event.getEventType() == AgentRuntimeEventType.TOOL_RESULT));
        Assert.assertEquals(1, events.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.SUSPENDED)
                .count());
    }

    @Test(expected = AgentRuntimeException.class)
    public void shouldRejectResumeBeforeInit() {
        AgentResumeRequest request = new AgentResumeRequest();
        AgentResumeToken token = new AgentResumeToken();
        token.setValue("resume-token");
        request.setResumeToken(token);
        request.setApproved(true);

        fakeRuntime().resume(request).collectList().block();
    }

    @Test(expected = AgentRuntimeException.class)
    public void shouldRejectResumeWithInvalidToken() {
        AgentScopeReActRuntime runtime = fakeRuntime();
        runtime.init(initRequest());
        AgentResumeRequest request = new AgentResumeRequest();
        AgentResumeToken token = new AgentResumeToken();
        token.setValue("missing-token");
        request.setResumeToken(token);
        request.setApproved(true);

        runtime.resume(request).collectList().block();
    }

    @Test
    public void shouldResumeApprovedToolExecutionFromSuspendedRuntime() {
        InMemoryAgentSessionStore sessionStore = new InMemoryAgentSessionStore();
        AgentInitRequest request = initRequest();
        request.setSessionStore(sessionStore);
        AgentToolSpec toolSpec = new AgentToolSpec();
        toolSpec.setName("search");
        toolSpec.setDescription("search");
        toolSpec.setApprovalRequired(true);
        request.getAgentDefinition().setToolSpecs(List.of(toolSpec));
        AtomicBoolean invoked = new AtomicBoolean(false);
        request.setToolInvokers(Map.of("search", (arguments, context) -> {
            invoked.set(true);
            return AgentToolResult.success("tool result");
        }));
        AgentScopeReActRuntime runtime = runtimeWithModel(List.of(
                ChatResponse.builder()
                        .id("tool-call-message")
                        .content(List.of(ToolUseBlock.builder()
                                .id("call-search")
                                .name("search")
                                .input(Map.of("q", "easyflow"))
                                .build()))
                        .finishReason("tool_calls")
                        .build(),
                ChatResponse.builder()
                        .id("final-message")
                        .content(List.of(TextBlock.builder().text("done").build()))
                        .finishReason("stop")
                        .build()));
        runtime.init(request);
        List<AgentRuntimeEvent> suspendedEvents = runtime.stream(AgentMessage.text(AgentMessageRole.USER, "use tool"))
                .collectList()
                .block();
        AgentRuntimeEvent approval = suspendedEvents.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED)
                .findFirst()
                .orElseThrow();
        AgentResumeRequest resumeRequest = new AgentResumeRequest();
        AgentResumeToken token = new AgentResumeToken();
        token.setValue(String.valueOf(approval.getPayload().get("resumeToken")));
        resumeRequest.setResumeToken(token);
        resumeRequest.setApproved(true);

        List<AgentRuntimeEvent> resumeEvents = runtime.resume(resumeRequest)
                .collectList()
                .block();

        Assert.assertTrue(invoked.get());
        Assert.assertEquals(AgentRuntimeEventType.STARTED, resumeEvents.get(0).getEventType());
        Assert.assertEquals(Boolean.TRUE, resumeEvents.get(0).getPayload().get("resume"));
        Assert.assertTrue(resumeEvents.stream().anyMatch(event -> event.getEventType() == AgentRuntimeEventType.TOOL_RESULT));
        Assert.assertTrue(resumeEvents.stream().anyMatch(event -> event.getEventType() == AgentRuntimeEventType.MESSAGE_DELTA));
        Assert.assertTrue(resumeEvents.stream().anyMatch(event -> event.getEventType() == AgentRuntimeEventType.COMPLETED));
        Assert.assertFalse(resumeEvents.stream().anyMatch(event -> event.getEventType() == AgentRuntimeEventType.SUSPENDED));
        Assert.assertTrue(sessionStore.exists("session-1"));
    }

    /**
     * 验证同一 Turn 内同一 MCP 的后续工具复用一次批准，新 Turn 会重新请求批准。
     */
    @Test
    public void shouldReuseMcpApprovalWithinTurnAndResetForNextTurn() {
        AgentInitRequest request = initRequest();
        AgentToolSpec resolveSpec = approvalRequiredMcpTool(
                "mcp_101_resolve_library_id", "101");
        AgentToolSpec querySpec = approvalRequiredMcpTool(
                "mcp_101_query_docs", "101");
        request.getAgentDefinition().setToolSpecs(List.of(resolveSpec, querySpec));
        AtomicInteger invocationCount = new AtomicInteger();
        request.setToolInvokers(Map.of(
                resolveSpec.getName(), (arguments, context) -> {
                    invocationCount.incrementAndGet();
                    return AgentToolResult.success("library-id");
                },
                querySpec.getName(), (arguments, context) -> {
                    invocationCount.incrementAndGet();
                    return AgentToolResult.success("docs");
                }));
        AgentScopeReActRuntime runtime = runtimeWithSequentialModel(List.of(
                toolResponse("resolve-call", "call-resolve", resolveSpec.getName()),
                toolResponse("query-call", "call-query", querySpec.getName()),
                ChatResponse.builder()
                        .id("final-message")
                        .content(List.of(TextBlock.builder().text("done").build()))
                        .finishReason("stop")
                        .build(),
                toolResponse("next-turn-call", "call-next", resolveSpec.getName())));
        runtime.init(request);

        List<AgentRuntimeEvent> initialEvents = runtime.stream(
                        AgentMessage.text(AgentMessageRole.USER, "介绍 AG-UI"))
                .collectList()
                .block();
        AgentRuntimeEvent approval = initialEvents.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED)
                .findFirst()
                .orElseThrow();

        List<AgentRuntimeEvent> resumeEvents = runtime.resume(resumeFromApproval(approval, true))
                .collectList()
                .block();

        Assert.assertEquals(2, invocationCount.get());
        Assert.assertFalse(resumeEvents.stream()
                .anyMatch(event -> event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED));
        Assert.assertTrue(resumeEvents.stream()
                .anyMatch(event -> event.getEventType() == AgentRuntimeEventType.COMPLETED));

        List<AgentRuntimeEvent> nextTurnEvents = runtime.stream(
                        AgentMessage.text(AgentMessageRole.USER, "再查一次"))
                .collectList()
                .block();

        Assert.assertEquals(1, nextTurnEvents.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED)
                .count());
        Assert.assertTrue(nextTurnEvents.stream()
                .anyMatch(event -> event.getEventType() == AgentRuntimeEventType.SUSPENDED));
    }

    /**
     * 验证同一推理消息中同一 MCP 的多个工具只生成一个审批请求。
     */
    @Test
    public void shouldRequestOneApprovalForParallelToolsFromSameMcp() {
        AgentInitRequest request = initRequest();
        AgentToolSpec resolveSpec = approvalRequiredMcpTool(
                "mcp_101_resolve_library_id", "101");
        AgentToolSpec querySpec = approvalRequiredMcpTool(
                "mcp_101_query_docs", "101");
        request.getAgentDefinition().setToolSpecs(List.of(resolveSpec, querySpec));
        AtomicInteger invocationCount = new AtomicInteger();
        request.setToolInvokers(Map.of(
                resolveSpec.getName(), (arguments, context) -> {
                    invocationCount.incrementAndGet();
                    return AgentToolResult.success("library-id");
                },
                querySpec.getName(), (arguments, context) -> {
                    invocationCount.incrementAndGet();
                    return AgentToolResult.success("docs");
                }));
        AgentScopeReActRuntime runtime = runtimeWithSequentialModel(List.of(
                ChatResponse.builder()
                        .id("parallel-mcp-tools")
                        .content(List.of(
                                ToolUseBlock.builder()
                                        .id("call-resolve")
                                        .name(resolveSpec.getName())
                                        .input(Map.of())
                                        .build(),
                                ToolUseBlock.builder()
                                        .id("call-query")
                                        .name(querySpec.getName())
                                        .input(Map.of())
                                        .build()))
                        .finishReason("tool_calls")
                        .build(),
                ChatResponse.builder()
                        .id("parallel-final")
                        .content(List.of(TextBlock.builder().text("done").build()))
                        .finishReason("stop")
                        .build()));
        runtime.init(request);

        List<AgentRuntimeEvent> initialEvents = runtime.stream(
                        AgentMessage.text(AgentMessageRole.USER, "并行查询"))
                .collectList()
                .block();
        List<AgentRuntimeEvent> approvals = initialEvents.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED)
                .toList();

        Assert.assertEquals(1, approvals.size());
        List<AgentRuntimeEvent> resumeEvents = runtime.resume(
                        resumeFromApproval(approvals.get(0), true))
                .collectList()
                .block();

        Assert.assertEquals(2, invocationCount.get());
        Assert.assertTrue(resumeEvents.stream()
                .anyMatch(event -> event.getEventType() == AgentRuntimeEventType.COMPLETED));
    }

    /**
     * 验证模型返回的 ToolUse 元数据不能覆盖 ToolSpec 中受信任的 MCP 审批作用域。
     */
    @Test
    public void shouldIgnoreForgedMcpScopeFromToolUseMetadata() {
        AgentInitRequest request = initRequest();
        AgentToolSpec resolveSpec = approvalRequiredMcpTool(
                "mcp_101_resolve_library_id", "101");
        AgentToolSpec querySpec = approvalRequiredMcpTool(
                "mcp_101_query_docs", "101");
        request.getAgentDefinition().setToolSpecs(List.of(resolveSpec, querySpec));
        request.setToolInvokers(Map.of(
                resolveSpec.getName(), (arguments, context) -> AgentToolResult.success("library-id"),
                querySpec.getName(), (arguments, context) -> AgentToolResult.success("docs")));
        AgentScopeReActRuntime runtime = runtimeWithSequentialModel(List.of(
                ChatResponse.builder()
                        .id("forged-scope-call")
                        .content(List.of(ToolUseBlock.builder()
                                .id("call-resolve")
                                .name(resolveSpec.getName())
                                .input(Map.of())
                                .metadata(Map.of("toolType", "MCP", "mcpId", "forged"))
                                .build()))
                        .finishReason("tool_calls")
                        .build(),
                toolResponse("query-call", "call-query", querySpec.getName()),
                ChatResponse.builder()
                        .id("final-message")
                        .content(List.of(TextBlock.builder().text("done").build()))
                        .finishReason("stop")
                        .build()));
        runtime.init(request);

        List<AgentRuntimeEvent> initialEvents = runtime.stream(
                        AgentMessage.text(AgentMessageRole.USER, "介绍 AG-UI"))
                .collectList()
                .block();
        AgentRuntimeEvent approval = initialEvents.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED)
                .findFirst()
                .orElseThrow();
        @SuppressWarnings("unchecked")
        Map<String, Object> approvalMetadata =
                (Map<String, Object>) approval.getPayload().get("approvalMetadata");

        Assert.assertEquals("101", approvalMetadata.get("mcpId"));
        List<AgentRuntimeEvent> resumeEvents = runtime.resume(resumeFromApproval(approval, true))
                .collectList()
                .block();

        Assert.assertFalse(resumeEvents.stream()
                .anyMatch(event -> event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED));
        Assert.assertTrue(resumeEvents.stream()
                .anyMatch(event -> event.getEventType() == AgentRuntimeEventType.COMPLETED));
    }

    /**
     * 验证同一轮推理包含多个审批工具时，全部批准前不会执行任何工具。
     */
    @Test
    public void shouldWaitForAllToolApprovalsBeforeExecutingBatch() {
        AgentInitRequest request = initRequest();
        AgentToolSpec searchSpec = new AgentToolSpec();
        searchSpec.setName("search");
        searchSpec.setDescription("search");
        searchSpec.setApprovalRequired(true);
        AgentToolSpec auditSpec = new AgentToolSpec();
        auditSpec.setName("audit");
        auditSpec.setDescription("audit");
        auditSpec.setApprovalRequired(true);
        request.getAgentDefinition().setToolSpecs(List.of(searchSpec, auditSpec));
        AtomicInteger invocationCount = new AtomicInteger();
        request.setToolInvokers(Map.of(
                "search", (arguments, context) -> {
                    invocationCount.incrementAndGet();
                    return AgentToolResult.success("search result");
                },
                "audit", (arguments, context) -> {
                    invocationCount.incrementAndGet();
                    return AgentToolResult.success("audit result");
                }));
        AgentScopeReActRuntime runtime = runtimeWithModel(List.of(
                ChatResponse.builder()
                        .id("tool-call-message")
                        .content(List.of(
                                ToolUseBlock.builder()
                                        .id("call-search")
                                        .name("search")
                                        .input(Map.of("q", "easyflow"))
                                        .build(),
                                ToolUseBlock.builder()
                                        .id("call-audit")
                                        .name("audit")
                                        .input(Map.of("scope", "current"))
                                        .build()))
                        .finishReason("tool_calls")
                        .build(),
                ChatResponse.builder()
                        .id("final-message")
                        .content(List.of(TextBlock.builder().text("done").build()))
                        .finishReason("stop")
                        .build()));
        runtime.init(request);

        List<AgentRuntimeEvent> initialEvents = runtime.stream(
                        AgentMessage.text(AgentMessageRole.USER, "use tools"))
                .collectList()
                .block();
        List<AgentRuntimeEvent> approvals = initialEvents.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED)
                .toList();

        Assert.assertEquals(2, approvals.size());
        Assert.assertEquals(1, initialEvents.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.SUSPENDED)
                .count());
        List<AgentRuntimeEvent> firstResumeEvents = runtime.resume(resumeFromApproval(approvals.get(0), true))
                .collectList()
                .block();

        Assert.assertEquals(0, invocationCount.get());
        AgentRuntimeEvent waitingEvent = firstResumeEvents.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.SUSPENDED)
                .findFirst()
                .orElseThrow();
        Assert.assertEquals(1, ((List<?>) waitingEvent.getPayload().get("pendingApprovals")).size());
        Assert.assertFalse(firstResumeEvents.stream()
                .anyMatch(event -> event.getEventType() == AgentRuntimeEventType.TOOL_CALL));

        List<AgentRuntimeEvent> secondResumeEvents = runtime.resume(resumeFromApproval(approvals.get(1), true))
                .collectList()
                .block();

        Assert.assertEquals(2, invocationCount.get());
        Assert.assertEquals(2, secondResumeEvents.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.TOOL_RESULT)
                .count());
        Assert.assertTrue(secondResumeEvents.stream()
                .anyMatch(event -> event.getEventType() == AgentRuntimeEventType.COMPLETED));
    }

    /**
     * 验证不同 toolCallId 即使名称和入参相同也会分别审批和执行。
     */
    @Test
    public void shouldPreserveDistinctApprovedToolCallsWithIdenticalInput() {
        AgentInitRequest request = initRequest();
        AgentToolSpec toolSpec = new AgentToolSpec();
        toolSpec.setName("search");
        toolSpec.setDescription("search");
        toolSpec.setApprovalRequired(true);
        request.getAgentDefinition().setToolSpecs(List.of(toolSpec));
        AtomicInteger invocationCount = new AtomicInteger();
        request.setToolInvokers(Map.of("search", (arguments, context) -> {
            invocationCount.incrementAndGet();
            return AgentToolResult.success("tool result");
        }));
        Map<String, Object> input = Map.of("q", "easyflow");
        AgentScopeReActRuntime runtime = runtimeWithModel(List.of(
                ChatResponse.builder()
                        .id("tool-call-message")
                        .content(List.of(
                                ToolUseBlock.builder()
                                        .id("call-search")
                                        .name("search")
                                        .input(input)
                                        .build(),
                                ToolUseBlock.builder()
                                        .id("call-search-duplicate")
                                        .name("search")
                                        .input(input)
                                        .build()))
                        .finishReason("tool_calls")
                        .build(),
                ChatResponse.builder()
                        .id("final-message")
                        .content(List.of(TextBlock.builder().text("done").build()))
                        .finishReason("stop")
                        .build()));
        runtime.init(request);

        List<AgentRuntimeEvent> suspendedEvents = runtime.stream(
                        AgentMessage.text(AgentMessageRole.USER, "use tool"))
                .collectList()
                .block();
        List<AgentRuntimeEvent> approvals = suspendedEvents.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED)
                .toList();

        Assert.assertEquals(2, approvals.size());
        List<AgentRuntimeEvent> firstResumeEvents = runtime.resume(resumeFromApproval(approvals.get(0), true))
                .collectList()
                .block();

        Assert.assertEquals(0, invocationCount.get());
        Assert.assertTrue(firstResumeEvents.stream()
                .anyMatch(event -> event.getEventType() == AgentRuntimeEventType.SUSPENDED));
        List<AgentRuntimeEvent> secondResumeEvents = runtime.resume(resumeFromApproval(approvals.get(1), true))
                .collectList()
                .block();

        Assert.assertEquals(2, invocationCount.get());
        Assert.assertEquals(2, secondResumeEvents.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.TOOL_RESULT)
                .count());
        Assert.assertTrue(secondResumeEvents.stream()
                .anyMatch(event -> event.getEventType() == AgentRuntimeEventType.COMPLETED));
    }

    /**
     * 验证相同 toolCallId 的同轮重放只审批并执行一次。
     */
    @Test
    public void shouldDeduplicateRepeatedToolCallIdWithinReasoning() {
        AgentInitRequest request = initRequest();
        AgentToolSpec toolSpec = new AgentToolSpec();
        toolSpec.setName("search");
        toolSpec.setDescription("search");
        toolSpec.setApprovalRequired(true);
        request.getAgentDefinition().setToolSpecs(List.of(toolSpec));
        AtomicInteger invocationCount = new AtomicInteger();
        request.setToolInvokers(Map.of("search", (arguments, context) -> {
            invocationCount.incrementAndGet();
            return AgentToolResult.success("tool result");
        }));
        Map<String, Object> input = Map.of("q", "easyflow");
        AgentScopeReActRuntime runtime = runtimeWithModel(List.of(
                ChatResponse.builder()
                        .id("tool-call-message")
                        .content(List.of(
                                ToolUseBlock.builder()
                                        .id("call-search")
                                        .name("search")
                                        .input(input)
                                        .build(),
                                ToolUseBlock.builder()
                                        .id("call-search")
                                        .name("search")
                                        .input(input)
                                        .build()))
                        .finishReason("tool_calls")
                        .build(),
                ChatResponse.builder()
                        .id("final-message")
                        .content(List.of(TextBlock.builder().text("done").build()))
                        .finishReason("stop")
                        .build()));
        runtime.init(request);

        List<AgentRuntimeEvent> suspendedEvents = runtime.stream(
                        AgentMessage.text(AgentMessageRole.USER, "use tool"))
                .collectList()
                .block();
        List<AgentRuntimeEvent> approvals = suspendedEvents.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED)
                .toList();

        Assert.assertEquals(1, approvals.size());
        List<AgentRuntimeEvent> resumeEvents = runtime.resume(resumeFromApproval(approvals.get(0), true))
                .collectList()
                .block();

        Assert.assertEquals(1, invocationCount.get());
        Assert.assertEquals(1, resumeEvents.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.TOOL_RESULT)
                .count());
        Assert.assertTrue(resumeEvents.stream()
                .anyMatch(event -> event.getEventType() == AgentRuntimeEventType.COMPLETED));
    }

    /**
     * 验证模型未提供 toolCallId 时审批链路仍会获得完整的调用身份。
     */
    @Test
    public void shouldProvideApprovalIdentityWhenModelOmitsToolCallId() {
        AgentInitRequest request = initRequest();
        AgentToolSpec toolSpec = new AgentToolSpec();
        toolSpec.setName("search");
        toolSpec.setDescription("search");
        toolSpec.setApprovalRequired(true);
        request.getAgentDefinition().setToolSpecs(List.of(toolSpec));
        request.setToolInvokers(Map.of("search", (arguments, context) ->
                AgentToolResult.success("tool result")));
        AgentScopeReActRuntime runtime = runtimeWithModel(List.of(
                ChatResponse.builder()
                        .id("tool-call-message")
                        .content(List.of(ToolUseBlock.builder()
                                .name("search")
                                .input(Map.of("q", "easyflow"))
                                .build()))
                        .finishReason("tool_calls")
                        .build(),
                ChatResponse.builder()
                        .id("final-message")
                        .content(List.of(TextBlock.builder().text("done").build()))
                        .finishReason("stop")
                        .build()));
        runtime.init(request);

        List<AgentRuntimeEvent> events = runtime.stream(
                        AgentMessage.text(AgentMessageRole.USER, "use tool"))
                .collectList()
                .block();
        AgentRuntimeEvent approval = events.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED)
                .findFirst()
                .orElseThrow();

        String toolCallId = String.valueOf(approval.getPayload().get("toolCallId"));
        Assert.assertFalse(toolCallId.isBlank());
        @SuppressWarnings("unchecked")
        Map<String, Object> metadata =
                (Map<String, Object>) approval.getPayload().get("approvalMetadata");
        String approvalBatchId = String.valueOf(metadata.get("approvalBatchId"));
        Assert.assertFalse(approvalBatchId.isBlank());
    }

    @Test
    public void shouldCancelRejectedToolResumeWithoutExecutingTool() {
        InMemoryAgentSessionStore sessionStore = new InMemoryAgentSessionStore();
        AgentInitRequest request = initRequest();
        request.setSessionStore(sessionStore);
        AgentToolSpec toolSpec = new AgentToolSpec();
        toolSpec.setName("search");
        toolSpec.setDescription("search");
        toolSpec.setApprovalRequired(true);
        request.getAgentDefinition().setToolSpecs(List.of(toolSpec));
        AtomicBoolean invoked = new AtomicBoolean(false);
        request.setToolInvokers(Map.of("search", (arguments, context) -> {
            invoked.set(true);
            return AgentToolResult.success("tool result");
        }));
        AgentScopeReActRuntime runtime = runtimeWithModel(List.of(ChatResponse.builder()
                .id("tool-call-message")
                .content(List.of(ToolUseBlock.builder()
                        .id("call-search")
                        .name("search")
                        .input(Map.of("q", "easyflow"))
                        .build()))
                .finishReason("tool_calls")
                .build()));
        runtime.init(request);
        List<AgentRuntimeEvent> suspendedEvents = runtime.stream(AgentMessage.text(AgentMessageRole.USER, "use tool"))
                .collectList()
                .block();
        AgentRuntimeEvent approval = suspendedEvents.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED)
                .findFirst()
                .orElseThrow();
        AgentResumeRequest resumeRequest = new AgentResumeRequest();
        AgentResumeToken token = new AgentResumeToken();
        token.setValue(String.valueOf(approval.getPayload().get("resumeToken")));
        resumeRequest.setResumeToken(token);
        resumeRequest.setApproved(false);
        resumeRequest.setRejectReason("not allowed");

        List<AgentRuntimeEvent> resumeEvents = runtime.resume(resumeRequest)
                .collectList()
                .block();

        Assert.assertFalse(invoked.get());
        AgentRuntimeEvent cancelled = resumeEvents.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.CANCELLED)
                .findFirst()
                .orElseThrow();
        Assert.assertEquals("not allowed", cancelled.getPayload().get("reason"));
        Assert.assertEquals(Boolean.TRUE, cancelled.getPayload().get("resume"));
        Assert.assertTrue(sessionStore.exists("session-1"));
    }

    @Test
    public void shouldRejectStreamWhileResumeIsRunning() {
        AgentInitRequest request = initRequest();
        AgentToolSpec toolSpec = new AgentToolSpec();
        toolSpec.setName("search");
        toolSpec.setDescription("search");
        toolSpec.setApprovalRequired(true);
        request.getAgentDefinition().setToolSpecs(List.of(toolSpec));
        CompletableFuture<Void> toolStarted = new CompletableFuture<>();
        request.setToolInvokers(Map.of("search", (arguments, context) -> {
            toolStarted.complete(null);
            try {
                Thread.sleep(1_000L);
            } catch (InterruptedException interruptedException) {
                Thread.currentThread().interrupt();
            }
            return AgentToolResult.success("tool result");
        }));
        AgentScopeReActRuntime runtime = runtimeWithModel(List.of(
                ChatResponse.builder()
                        .id("tool-call-message")
                        .content(List.of(ToolUseBlock.builder()
                                .id("call-search")
                                .name("search")
                                .input(Map.of("q", "easyflow"))
                                .build()))
                        .finishReason("tool_calls")
                        .build(),
                ChatResponse.builder()
                        .id("final-message")
                        .content(List.of(TextBlock.builder().text("done").build()))
                        .finishReason("stop")
                        .build()));
        runtime.init(request);
        List<AgentRuntimeEvent> suspendedEvents = runtime.stream(AgentMessage.text(AgentMessageRole.USER, "use tool"))
                .collectList()
                .block();
        AgentRuntimeEvent approval = suspendedEvents.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED)
                .findFirst()
                .orElseThrow();
        AgentResumeRequest resumeRequest = new AgentResumeRequest();
        AgentResumeToken token = new AgentResumeToken();
        token.setValue(String.valueOf(approval.getPayload().get("resumeToken")));
        resumeRequest.setResumeToken(token);
        resumeRequest.setApproved(true);

        reactor.core.Disposable disposable = runtime.resume(resumeRequest).subscribe();
        try {
            toolStarted.get(3, TimeUnit.SECONDS);
            runtime.stream(AgentMessage.text(AgentMessageRole.USER, "second")).collectList().block();
            Assert.fail("Concurrent stream should be rejected.");
        } catch (AgentRuntimeException expected) {
            Assert.assertTrue(expected.getMessage().contains("already streaming"));
        } catch (Exception error) {
            throw new AssertionError(error);
        } finally {
            disposable.dispose();
        }
    }

    @Test
    public void shouldAttachKnowledgeReferencesToCompletedMessage() {
        AgentInitRequest request = initRequest();
        AgentKnowledgeSpec knowledgeSpec = new AgentKnowledgeSpec();
        knowledgeSpec.setKnowledgeId("knowledge-1");
        knowledgeSpec.setRuntimeName("faq");
        knowledgeSpec.setName("知识库");
        request.getAgentDefinition().setKnowledgeSpecs(List.of(knowledgeSpec));
        AtomicBoolean knowledgeInvoked = new AtomicBoolean(false);
        request.setKnowledgeRegistrations(List.of(new AgentKnowledgeRegistration(knowledgeSpec, retrievalRequest -> {
            knowledgeInvoked.set(true);
            AgentKnowledgeDocument document = new AgentKnowledgeDocument();
            document.setDocumentId("doc-1");
            document.setDocumentName("说明文档");
            document.setChunkId("chunk-1");
            document.setContent("fake answer");
            document.setScore(0.9D);
            return AgentKnowledgeRetrievalResult.of(List.of(document));
        })));
        AgentScopeReActRuntime runtime = runtimeWithModel(List.of(
                ChatResponse.builder()
                        .id("knowledge-tool-call")
                        .content(List.of(ToolUseBlock.builder()
                                .id("call-search")
                                .name("retrieve_knowledge_faq")
                                .input(Map.of("query", "fake answer"))
                                .content("{\"query\":\"fake answer\"}")
                                .build()))
                        .finishReason("tool_calls")
                        .build(),
                ChatResponse.builder()
                        .id("knowledge-final")
                        .content(List.of(TextBlock.builder().text("fake answer").build()))
                        .finishReason("stop")
                        .build()));
        runtime.init(request);
        Assert.assertNotNull(runtime.getAgent().getToolkit().getTool("retrieve_knowledge_faq"));
        Assert.assertTrue(runtime.getAgent().getToolkit().getToolSchemas().stream()
                .anyMatch(schema -> "retrieve_knowledge_faq".equals(schema.getName())));

        List<AgentRuntimeEvent> events = runtime.stream(AgentMessage.text(AgentMessageRole.USER, "query knowledge"))
                .collectList()
                .block();

        AgentRuntimeEvent completed = events.stream()
                .filter(event -> event.getEventType() == AgentRuntimeEventType.COMPLETED)
                .findFirst()
                .orElseThrow();
        Assert.assertNotNull(completed.getMessage());
        Assert.assertTrue("knowledgeInvoked=" + knowledgeInvoked.get() + ", events="
                        + events.stream().map(AgentRuntimeEvent::getEventType).toList(),
                events.stream().anyMatch(event ->
                        event.getEventType() == AgentRuntimeEventType.KNOWLEDGE_RETRIEVAL));
        Assert.assertTrue(events.stream().anyMatch(event ->
                event.getEventType() == AgentRuntimeEventType.TOOL_CALL
                        && "KNOWLEDGE".equals(event.getPayload().get("toolCategory"))));
        Assert.assertTrue(events.stream().anyMatch(event ->
                event.getEventType() == AgentRuntimeEventType.TOOL_RESULT
                        && "KNOWLEDGE".equals(event.getPayload().get("toolCategory"))));
        Assert.assertEquals("chunk-1", completed.getMessage().getKnowledgeReferences().get(0).getChunkId());
    }

    /**
     * 验证知识库生成工具与普通工具同名时拒绝初始化，避免 Toolkit 静默覆盖。
     */
    @Test(expected = AgentRuntimeException.class)
    public void shouldRejectKnowledgeToolNameConflictWithRegularTool() {
        AgentInitRequest request = initRequest();
        AgentKnowledgeSpec knowledgeSpec = new AgentKnowledgeSpec();
        knowledgeSpec.setKnowledgeId("knowledge-1");
        knowledgeSpec.setRuntimeName("faq");
        knowledgeSpec.setName("知识库");
        request.getAgentDefinition().setKnowledgeSpecs(List.of(knowledgeSpec));
        request.setKnowledgeRegistrations(List.of(new AgentKnowledgeRegistration(
                knowledgeSpec,
                retrievalRequest -> AgentKnowledgeRetrievalResult.of(List.of()))));
        AgentToolSpec regularTool = new AgentToolSpec();
        regularTool.setName("retrieve_knowledge_faq");
        regularTool.setDescription("conflicting tool");
        request.getAgentDefinition().setToolSpecs(List.of(regularTool));
        request.setToolInvokers(Map.of(
                regularTool.getName(),
                (arguments, context) -> AgentToolResult.success("done")));

        fakeRuntime().init(request);
    }

    private AgentScopeReActRuntime fakeRuntime() {
        return new AgentScopeReActRuntime(new FakeAgentScopeModelFactory(), new AgentScopeToolAdapter(),
                new AgentScopeKnowledgeAdapter(), new AgentScopeMemoryAdapter(), new AgentScopeSkillAdapter(),
                new AgentScopeMessageAdapter());
    }

    private AgentScopeReActRuntime runtimeWithModel(List<ChatResponse> responses) {
        return runtimeWithModel(responses, Duration.ZERO);
    }

    private AgentScopeReActRuntime runtimeWithModel(List<ChatResponse> responses, Duration completionDelay) {
        AgentScopeModelFactory modelFactory = new AgentScopeModelFactory() {
            @Override
            public Model create(AgentModelSpec modelSpec,
                                com.easyagents.agent.runtime.model.AgentGenerationOptions generationOptions) {
                return new ScriptedModel(modelSpec == null ? "fake-model" : modelSpec.getModelName(), responses, completionDelay);
            }
        };
        return new AgentScopeReActRuntime(modelFactory, new AgentScopeToolAdapter(),
                new AgentScopeKnowledgeAdapter(), new AgentScopeMemoryAdapter(), new AgentScopeSkillAdapter(),
                new AgentScopeMessageAdapter());
    }

    /**
     * 创建模型调用直接失败的运行时。
     *
     * @param error 模型异常
     * @return 测试运行时
     */
    private AgentScopeReActRuntime runtimeWithError(Throwable error) {
        AgentScopeModelFactory modelFactory = new AgentScopeModelFactory() {
            @Override
            public Model create(AgentModelSpec modelSpec,
                                com.easyagents.agent.runtime.model.AgentGenerationOptions generationOptions) {
                return new Model() {
                    @Override
                    public Flux<ChatResponse> stream(List<Msg> messages,
                                                     List<ToolSchema> toolSchemas,
                                                     GenerateOptions options) {
                        return Flux.error(error);
                    }

                    @Override
                    public String getModelName() {
                        return modelSpec == null ? "fake-model" : modelSpec.getModelName();
                    }
                };
            }
        };
        return new AgentScopeReActRuntime(modelFactory, new AgentScopeToolAdapter(),
                new AgentScopeKnowledgeAdapter(), new AgentScopeMemoryAdapter(), new AgentScopeSkillAdapter(),
                new AgentScopeMessageAdapter());
    }

    /**
     * 创建每次模型调用仅返回下一条预设响应的运行时。
     *
     * @param responses 按模型调用顺序排列的响应
     * @return 测试运行时
     */
    private AgentScopeReActRuntime runtimeWithSequentialModel(List<ChatResponse> responses) {
        AgentScopeModelFactory modelFactory = new AgentScopeModelFactory() {
            @Override
            public Model create(AgentModelSpec modelSpec,
                                com.easyagents.agent.runtime.model.AgentGenerationOptions generationOptions) {
                return new SequentialScriptedModel(
                        modelSpec == null ? "fake-model" : modelSpec.getModelName(),
                        responses);
            }
        };
        return new AgentScopeReActRuntime(modelFactory, new AgentScopeToolAdapter(),
                new AgentScopeKnowledgeAdapter(), new AgentScopeMemoryAdapter(), new AgentScopeSkillAdapter(),
                new AgentScopeMessageAdapter());
    }

    /**
     * 创建单次模型调用返回多个增量响应的运行时。
     *
     * @param responses 同一次模型调用中的响应增量
     * @return 测试运行时
     */
    private AgentScopeReActRuntime runtimeWithStreamingModel(List<ChatResponse> responses) {
        AgentScopeModelFactory modelFactory = new AgentScopeModelFactory() {
            @Override
            public Model create(AgentModelSpec modelSpec,
                                com.easyagents.agent.runtime.model.AgentGenerationOptions generationOptions) {
                return new StreamingScriptedModel(
                        modelSpec == null ? "fake-model" : modelSpec.getModelName(),
                        responses);
            }
        };
        return new AgentScopeReActRuntime(modelFactory, new AgentScopeToolAdapter(),
                new AgentScopeKnowledgeAdapter(), new AgentScopeMemoryAdapter(), new AgentScopeSkillAdapter(),
                new AgentScopeMessageAdapter());
    }

    /**
     * 根据审批事件创建恢复请求。
     *
     * @param approvalEvent 工具审批事件
     * @param approved 是否批准
     * @return 恢复请求
     */
    private AgentResumeRequest resumeFromApproval(AgentRuntimeEvent approvalEvent, boolean approved) {
        AgentResumeRequest request = new AgentResumeRequest();
        AgentResumeToken token = new AgentResumeToken();
        token.setValue(String.valueOf(approvalEvent.getPayload().get("resumeToken")));
        request.setResumeToken(token);
        request.setApproved(approved);
        return request;
    }

    /**
     * 创建需要批准且归属于指定 MCP 的工具定义。
     *
     * @param toolName 工具名称
     * @param mcpId MCP 标识
     * @return MCP 工具定义
     */
    private AgentToolSpec approvalRequiredMcpTool(String toolName, String mcpId) {
        AgentToolSpec spec = new AgentToolSpec();
        spec.setName(toolName);
        spec.setDescription(toolName);
        spec.setApprovalRequired(true);
        spec.getMetadata().put("toolType", "MCP");
        spec.getMetadata().put("mcpId", mcpId);
        spec.getMetadata().put("mcpTitle", "Context7");
        return spec;
    }

    /**
     * 创建包含一次工具调用的模型响应。
     *
     * @param messageId 响应消息标识
     * @param toolCallId 工具调用标识
     * @param toolName 工具名称
     * @return 模型响应
     */
    private ChatResponse toolResponse(String messageId, String toolCallId, String toolName) {
        return ChatResponse.builder()
                .id(messageId)
                .content(List.of(ToolUseBlock.builder()
                        .id(toolCallId)
                        .name(toolName)
                        .input(Map.of())
                        .build()))
                .finishReason("tool_calls")
                .build();
    }

    private static class ScriptedModel implements Model {

        private final String modelName;
        private final List<ChatResponse> responses;
        private final Duration completionDelay;

        private ScriptedModel(String modelName, List<ChatResponse> responses, Duration completionDelay) {
            this.modelName = modelName;
            this.responses = responses;
            this.completionDelay = completionDelay == null ? Duration.ZERO : completionDelay;
        }

        @Override
        public Flux<ChatResponse> stream(List<Msg> messages, List<ToolSchema> toolSchemas, GenerateOptions options) {
            boolean hasToolResult = messages.stream()
                    .flatMap(message -> message.getContentBlocks(ToolResultBlock.class).stream())
                    .anyMatch(result -> "call-search".equals(result.getId()));
            List<ChatResponse> selectedResponses = hasToolResult && responses.size() > 1
                    ? responses.subList(1, responses.size())
                    : responses.subList(0, 1);
            Flux<ChatResponse> responseFlux = Flux.fromIterable(selectedResponses);
            if (completionDelay.isZero() || completionDelay.isNegative()) {
                return responseFlux;
            }
            return responseFlux.concatWith(Flux.never()).timeout(completionDelay, Flux.empty());
        }

        @Override
        public String getModelName() {
            return modelName;
        }
    }

    /**
     * 每次调用按顺序返回一条响应的测试模型。
     */
    private static class SequentialScriptedModel implements Model {

        private final AtomicInteger invocationIndex = new AtomicInteger();
        private final String modelName;
        private final List<ChatResponse> responses;

        /**
         * 创建顺序响应模型。
         *
         * @param modelName 模型名称
         * @param responses 按调用顺序排列的响应
         */
        private SequentialScriptedModel(String modelName, List<ChatResponse> responses) {
            this.modelName = modelName;
            this.responses = responses;
        }

        /**
         * 返回当前模型调用对应的单条响应。
         *
         * @param messages 输入消息
         * @param toolSchemas 工具定义
         * @param options 生成配置
         * @return 单条响应流
         */
        @Override
        public Flux<ChatResponse> stream(List<Msg> messages,
                                         List<ToolSchema> toolSchemas,
                                         GenerateOptions options) {
            int index = invocationIndex.getAndIncrement();
            if (index >= responses.size()) {
                return Flux.error(new IllegalStateException("No scripted response for invocation " + index));
            }
            return Flux.just(responses.get(index));
        }

        /**
         * 返回模型名称。
         *
         * @return 模型名称
         */
        @Override
        public String getModelName() {
            return modelName;
        }
    }

    /**
     * 单次调用按顺序返回全部响应增量的测试模型。
     */
    private static class StreamingScriptedModel implements Model {

        private final String modelName;
        private final List<ChatResponse> responses;

        /**
         * 创建流式测试模型。
         *
         * @param modelName 模型名称
         * @param responses 响应增量
         */
        private StreamingScriptedModel(String modelName, List<ChatResponse> responses) {
            this.modelName = modelName;
            this.responses = responses;
        }

        /**
         * 返回预设的响应增量。
         *
         * @param messages 输入消息
         * @param toolSchemas 工具定义
         * @param options 生成配置
         * @return 响应流
         */
        @Override
        public Flux<ChatResponse> stream(List<Msg> messages,
                                         List<ToolSchema> toolSchemas,
                                         GenerateOptions options) {
            return Flux.fromIterable(responses);
        }

        /**
         * 获取模型名称。
         *
         * @return 模型名称
         */
        @Override
        public String getModelName() {
            return modelName;
        }
    }

    private AgentInitRequest initRequest() {
        AgentModelSpec modelSpec = new AgentModelSpec();
        modelSpec.setModelName("fake-model");
        AgentDefinition definition = new AgentDefinition();
        definition.setAgentId("agent-1");
        definition.setAgentName("agent-name");
        definition.setSystemPrompt("system");
        definition.setModelSpec(modelSpec);
        definition.setMemoryPolicy(AgentMemoryPolicy.inMemory());

        AgentInitRequest request = new AgentInitRequest();
        request.setSessionId("session-1");
        request.setSessionStore(new InMemoryAgentSessionStore());
        request.setAgentDefinition(definition);
        AgentRuntimeContext runtimeContext = new AgentRuntimeContext();
        runtimeContext.setTraceId("trace-1");
        request.setRuntimeContext(runtimeContext);
        return request;
    }

    private AgentMemorySnapshot compressionSnapshot() {
        AgentMemorySnapshot snapshot = new AgentMemorySnapshot();
        snapshot.addMessage(AgentMessage.text(AgentMessageRole.USER,
                "large previous message " + "y".repeat(80)));
        snapshot.addMessage(AgentMessage.text(AgentMessageRole.ASSISTANT,
                "old assistant answer"));
        snapshot.addMessage(AgentMessage.text(AgentMessageRole.USER, "latest user"));
        return snapshot;
    }

    private AgentSkillBoxSpec skillBoxSpec() {
        AgentSkillSpec skill = new AgentSkillSpec();
        skill.setSkillId("skill-1");
        skill.setName("Search Skill");
        skill.setDescription("Search related information with the search tool.");
        skill.setSkillContent("Use search.");
        AgentSkillBoxSpec spec = new AgentSkillBoxSpec();
        spec.setSkillBoxId("box-1");
        spec.setSkills(List.of(skill));
        spec.setToolBindings(Map.of("skill-1", List.of("search")));
        return spec;
    }

    private AgentOperateToolSpec operateToolSpec(AgentOperateToolType type) {
        AgentOperateToolSpec spec = new AgentOperateToolSpec();
        spec.setType(type);
        spec.setBaseDir(System.getProperty("java.io.tmpdir"));
        return spec;
    }

    private SkillBox skillBox(Toolkit toolkit) {
        return new AgentScopeSkillAdapter().createSkillBox(skillBoxSpec(), toolkit,
                Map.of("skill-1", List.of(new NoopAgentTool("search"))));
    }

    private record NoopAgentTool(String name) implements AgentTool {

        @Override
        public String getName() {
            return name;
        }

        @Override
        public String getDescription() {
            return name;
        }

        @Override
        public Map<String, Object> getParameters() {
            return Map.of();
        }

        @Override
        public reactor.core.publisher.Mono<ToolResultBlock> callAsync(io.agentscope.core.tool.ToolCallParam param) {
            return reactor.core.publisher.Mono.just(ToolResultBlock.text("ok"));
        }
    }

    private static class FakeMcpClientWrapper extends McpClientWrapper {

        private final AtomicBoolean closed = new AtomicBoolean(false);

        private FakeMcpClientWrapper(String name) {
            super(name);
        }

        @Override
        public Mono<Void> initialize() {
            initialized = true;
            return Mono.empty();
        }

        @Override
        public Mono<List<McpSchema.Tool>> listTools() {
            return Mono.just(List.of());
        }

        @Override
        public Mono<McpSchema.CallToolResult> callTool(String toolName, Map<String, Object> arguments) {
            return Mono.just(new McpSchema.CallToolResult(List.of(), false));
        }

        @Override
        public void close() {
            closed.set(true);
            initialized = false;
        }
    }

    private AgentRuntimeExecutionContext executionContext() {
        AgentRuntimeExecutionContext context = new AgentRuntimeExecutionContext();
        AgentDefinition definition = new AgentDefinition();
        AgentModelSpec modelSpec = new AgentModelSpec();
        modelSpec.setModelName("fake-model");
        definition.setAgentId("agent-1");
        definition.setAgentName("agent-name");
        definition.setSystemPrompt("system");
        definition.setModelSpec(modelSpec);
        definition.setMemoryPolicy(AgentMemoryPolicy.inMemory());
        context.setAgentDefinition(definition);
        context.setTraceId("trace-1");
        context.setSessionId("session-1");
        context.setRequestId("request-1");
        return context;
    }

    private ReActAgent initializedAgent() {
        AgentScopeReActRuntime runtime = fakeRuntime();
        runtime.init(initRequest());
        return runtime.getAgent();
    }

    private boolean isRuntimeHook(Hook hook) {
        return hook instanceof AgentScopeRuntimeHook;
    }

    private void awaitCondition(BooleanSupplier condition) throws Exception {
        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(3);
        while (System.nanoTime() < deadline) {
            if (condition.getAsBoolean()) {
                return;
            }
            Thread.sleep(20L);
        }
        Assert.fail("Condition was not met in time.");
    }
}
