feat(agui): configure agent interruption on disconnect (#2719)

Closes #2715
---------
Co-authored-by: jujn <2087687391@qq.com>

Author: hexly666

agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/processor/AguiRequestProcessor.java

@@ -15,6 +15,7 @@
  */
 package io.agentscope.core.agui.processor;
 
+import io.agentscope.core.ReActAgent;
 import io.agentscope.core.agent.Agent;
 import io.agentscope.core.agent.RuntimeContext;
 import io.agentscope.core.agui.adapter.AguiAdapterConfig;
@@ -85,7 +86,29 @@ private AguiRequestProcessor(Builder builder) {
      * @param agent The resolved agent instance
      * @param events The event stream
      */
-    public record ProcessResult(Agent agent, Flux<AguiEvent> events) {}
+    public record ProcessResult(Agent agent, Flux<AguiEvent> events) {
+
+        /**
+         * Interrupt this request's active session.
+         *
+         * <p>AG-UI uses {@code threadId} as the session id. For a multi-session
+         * {@link ReActAgent}, preserve the caller's user id and target that session instead of
+         * invoking the deprecated no-argument interrupt method, which always targets the default
+         * session.
+         *
+         * @param threadId The AG-UI thread id for this request
+         * @param runtimeContext The caller-provided runtime context, may be null
+         */
+        public void interrupt(String threadId, RuntimeContext runtimeContext) {
+            if (agent instanceof ReActAgent reActAgent) {
+                RuntimeContext interruptContext =
+                        RuntimeContext.builder(runtimeContext).sessionId(threadId).build();
+                reActAgent.interrupt(interruptContext);
+            } else {
+                agent.interrupt();
+            }
+        }
+    }
 
     /**
      * Process an AG-UI request and return the result containing agent and event stream.

agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/common/AguiProperties.java

@@ -40,6 +40,7 @@
  *     enable-reasoning: false
  *     emit-token-usage: false
  *     emit-run-finished-after-error: false
+ *     interrupt-on-disconnect: true
  * </pre>
  */
 @ConfigurationProperties(prefix = "agentscope.agui")
@@ -125,6 +126,9 @@ public class AguiProperties {
      */
     private long sseTimeout = 600000L;
 
+    /** Whether to interrupt the agent when the client disconnects. */
+    private boolean interruptOnDisconnect = true;
+
     public String getPathPrefix() {
         return pathPrefix;
     }
@@ -260,4 +264,12 @@ public long getSseTimeout() {
     public void setSseTimeout(long sseTimeout) {
         this.sseTimeout = sseTimeout;
     }
+
+    public boolean isInterruptOnDisconnect() {
+        return interruptOnDisconnect;
+    }
+
+    public void setInterruptOnDisconnect(boolean interruptOnDisconnect) {
+        this.interruptOnDisconnect = interruptOnDisconnect;
+    }
 }

agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/mvc/AgentscopeAguiMvcAutoConfiguration.java

@@ -105,6 +105,7 @@ public AguiMvcController aguiMvcController(
                 .serverSideMemory(props.isServerSideMemory())
                 .agentIdHeader(props.getAgentIdHeader())
                 .sseTimeout(props.getSseTimeout())
+                .interruptOnDisconnect(props.isInterruptOnDisconnect())
                 .runtimeContextResolver(runtimeContextResolverProvider.getIfAvailable())
                 .adapterFactory(adapterFactoryProvider.getIfAvailable())
                 .config(config)

agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/mvc/AguiMvcController.java

@@ -76,6 +76,7 @@ public class AguiMvcController {
     private final AguiEventEncoder encoder;
     private final String agentIdHeader;
     private final long sseTimeout;
+    private final boolean interruptOnDisconnect;
     private final ExecutorService executorService;
     private final AguiRuntimeContextResolver runtimeContextResolver;
 
@@ -98,6 +99,7 @@ private AguiMvcController(Builder builder) {
         this.agentIdHeader =
                 builder.agentIdHeader != null ? builder.agentIdHeader : DEFAULT_AGENT_ID_HEADER;
         this.sseTimeout = builder.sseTimeout > 0 ? builder.sseTimeout : 600000L;
+        this.interruptOnDisconnect = builder.interruptOnDisconnect;
         this.executorService = Executors.newCachedThreadPool();
         this.runtimeContextResolver = builder.runtimeContextResolver;
     }
@@ -170,34 +172,47 @@ private SseEmitter handleInternal(
                     Disposable subscription = null;
                     try {
                         // Process request - returns both agent and event stream
+                        RuntimeContext runtimeContext =
+                                resolveRuntimeContext(input, headerAgentId, pathAgentId, request);
                         AguiRequestProcessor.ProcessResult result =
                                 processor.process(
-                                        input,
-                                        headerAgentId,
-                                        pathAgentId,
-                                        resolveRuntimeContext(
-                                                input, headerAgentId, pathAgentId, request));
+                                        input, headerAgentId, pathAgentId, runtimeContext);
 
                         // Set up callbacks for client disconnect handling
                         // using the same agent instance from the result
                         emitter.onCompletion(
                                 () -> logger.debug("SSE connection completed for run {}", runId));
                         emitter.onTimeout(
                                 () -> {
-                                    logger.info(
-                                            "SSE connection timed out for run {}, interrupting"
-                                                    + " agent",
-                                            runId);
-                                    result.agent().interrupt();
+                                    if (interruptOnDisconnect) {
+                                        logger.info(
+                                                "SSE connection timed out for run {}, interrupting"
+                                                        + " agent",
+                                                runId);
+                                        result.interrupt(threadId, runtimeContext);
+                                    } else {
+                                        logger.info(
+                                                "SSE connection timed out for run {}, agent"
+                                                        + " continues running",
+                                                runId);
+                                    }
                                 });
                         emitter.onError(
                                 (ex) -> {
-                                    logger.info(
-                                            "SSE connection error for run {}: {}, interrupting"
-                                                    + " agent",
-                                            runId,
-                                            ex.getMessage());
-                                    result.agent().interrupt();
+                                    if (interruptOnDisconnect) {
+                                        logger.info(
+                                                "SSE connection error for run {}: {}, interrupting"
+                                                        + " agent",
+                                                runId,
+                                                ex.getMessage());
+                                        result.interrupt(threadId, runtimeContext);
+                                    } else {
+                                        logger.info(
+                                                "SSE connection error for run {}: {}, agent"
+                                                        + " continues running",
+                                                runId,
+                                                ex.getMessage());
+                                    }
                                 });
 
                         // Subscribe to event stream from the same result
@@ -342,6 +357,7 @@ public static class Builder {
         private boolean serverSideMemory = false;
         private String agentIdHeader;
         private long sseTimeout = 600000L;
+        private boolean interruptOnDisconnect = true;
         private AguiRuntimeContextResolver runtimeContextResolver;
         private AguiAgentAdapterFactory adapterFactory;
 
@@ -411,6 +427,17 @@ public Builder sseTimeout(long sseTimeout) {
             return this;
         }
 
+        /**
+         * Set whether to interrupt the agent when the client disconnects.
+         *
+         * @param interruptOnDisconnect whether to interrupt the agent
+         * @return This builder
+         */
+        public Builder interruptOnDisconnect(boolean interruptOnDisconnect) {
+            this.interruptOnDisconnect = interruptOnDisconnect;
+            return this;
+        }
+
         /**
          * Set the runtime context resolver.
          *

agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/webflux/AgentscopeAguiWebFluxAutoConfiguration.java

@@ -107,6 +107,7 @@ public AguiWebFluxHandler aguiWebFluxHandler(
                 .sessionManager(sessionManager)
                 .serverSideMemory(props.isServerSideMemory())
                 .agentIdHeader(props.getAgentIdHeader())
+                .interruptOnDisconnect(props.isInterruptOnDisconnect())
                 .runtimeContextResolver(runtimeContextResolverProvider.getIfAvailable())
                 .adapterFactory(adapterFactoryProvider.getIfAvailable())
                 .config(config)

agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/webflux/AguiWebFluxHandler.java

@@ -80,6 +80,7 @@ public class AguiWebFluxHandler {
     private final AguiRequestProcessor processor;
     private final AguiEventEncoder encoder;
     private final String agentIdHeader;
+    private final boolean interruptOnDisconnect;
     private final AguiRuntimeContextResolver runtimeContextResolver;
 
     private AguiWebFluxHandler(Builder builder) {
@@ -100,6 +101,7 @@ private AguiWebFluxHandler(Builder builder) {
         this.encoder = new AguiEventEncoder();
         this.agentIdHeader =
                 builder.agentIdHeader != null ? builder.agentIdHeader : DEFAULT_AGENT_ID_HEADER;
+        this.interruptOnDisconnect = builder.interruptOnDisconnect;
         this.runtimeContextResolver = builder.runtimeContextResolver;
     }
 
@@ -144,31 +146,39 @@ private Mono<ServerResponse> processInput(
         try {
             // Get header agent ID
             String headerAgentId = request.headers().firstHeader(agentIdHeader);
+            RuntimeContext runtimeContext =
+                    resolveRuntimeContext(input, headerAgentId, pathAgentId, request);
 
             // Process request - returns both agent and event stream
             AguiRequestProcessor.ProcessResult result =
-                    processor.process(
-                            input,
-                            headerAgentId,
-                            pathAgentId,
-                            resolveRuntimeContext(input, headerAgentId, pathAgentId, request));
+                    processor.process(input, headerAgentId, pathAgentId, runtimeContext);
 
             // Create SSE stream using ServerSentEvent for proper streaming behavior
+            Flux<AguiEvent> events =
+                    interruptOnDisconnect
+                            ? result.events()
+                            : result.events().publish().autoConnect(1);
             Flux<ServerSentEvent<String>> sseStream =
-                    result.events()
-                            .map(
+                    events.map(
                                     event ->
                                             ServerSentEvent.<String>builder()
                                                     .data(encoder.encodeToJson(event).trim())
                                                     .build())
-                            // When client closes connection (cancels stream), interrupt the agent
+                            // When the client closes the connection, optionally interrupt the agent
                             .doOnCancel(
                                     () -> {
-                                        logger.info(
-                                                "SSE stream cancelled for run {}, interrupting"
-                                                        + " agent",
-                                                runId);
-                                        result.agent().interrupt();
+                                        if (interruptOnDisconnect) {
+                                            logger.info(
+                                                    "SSE stream cancelled for run {}, interrupting"
+                                                            + " agent",
+                                                    runId);
+                                            result.interrupt(threadId, runtimeContext);
+                                        } else {
+                                            logger.info(
+                                                    "SSE stream cancelled for run {}, agent"
+                                                            + " continues running",
+                                                    runId);
+                                        }
                                     });
 
             return ServerResponse.ok()
@@ -284,6 +294,7 @@ public static class Builder {
         private AguiAdapterConfig config;
         private boolean serverSideMemory = false;
         private String agentIdHeader;
+        private boolean interruptOnDisconnect = true;
         private AguiRuntimeContextResolver runtimeContextResolver;
         private AguiAgentAdapterFactory adapterFactory;
 
@@ -342,6 +353,17 @@ public Builder agentIdHeader(String agentIdHeader) {
             return this;
         }
 
+        /**
+         * Set whether to interrupt the agent when the client disconnects.
+         *
+         * @param interruptOnDisconnect whether to interrupt the agent
+         * @return This builder
+         */
+        public Builder interruptOnDisconnect(boolean interruptOnDisconnect) {
+            this.interruptOnDisconnect = interruptOnDisconnect;
+            return this;
+        }
+
         /**
          * Set the runtime context resolver.
          *

agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/test/java/io/agentscope/spring/boot/agui/common/AguiAdapterConfigAutoConfigurationTest.java

@@ -85,9 +85,34 @@ void testDefaultAdapterConfig() {
                     assertFalse(config.isBaseEventPropertiesEnricherEnabled());
                     assertTrue(config.getEventConverters().isEmpty());
                     assertTrue(config.getEventEnrichers().isEmpty());
+                    assertTrue(interruptOnDisconnect(ctx.getBean(AguiMvcController.class)));
                 });
     }
 
+    @Test
+    @DisplayName("Should bind interrupt-on-disconnect property for MVC")
+    void testMvcInterruptOnDisconnectProperty() {
+        mvcContextRunner
+                .withPropertyValues("agentscope.agui.interrupt-on-disconnect=false")
+                .run(
+                        ctx ->
+                                assertFalse(
+                                        interruptOnDisconnect(
+                                                ctx.getBean(AguiMvcController.class))));
+    }
+
+    @Test
+    @DisplayName("Should bind interrupt-on-disconnect property for WebFlux")
+    void testWebFluxInterruptOnDisconnectProperty() {
+        webFluxContextRunner
+                .withPropertyValues("agentscope.agui.interrupt-on-disconnect=false")
+                .run(
+                        ctx ->
+                                assertFalse(
+                                        interruptOnDisconnect(
+                                                ctx.getBean(AguiWebFluxHandler.class))));
+    }
+
     @Test
     @DisplayName("Should bind emit-token-usage property for MVC")
     void testMvcEmitTokenUsageProperty() {
@@ -376,6 +401,10 @@ private static AguiRuntimeContextResolver webFluxRuntimeContextResolver(
                 ReflectionTestUtils.getField(handler, "runtimeContextResolver");
     }
 
+    private static boolean interruptOnDisconnect(Object handler) {
+        return (boolean) ReflectionTestUtils.getField(handler, "interruptOnDisconnect");
+    }
+
     private static AguiAgentAdapterFactory mvcAdapterFactory(AguiMvcController controller) {
         Object processor = ReflectionTestUtils.getField(controller, "processor");
         return (AguiAgentAdapterFactory) ReflectionTestUtils.getField(processor, "adapterFactory");

docs/v2/en/integration/protocol/agui.md

@@ -220,8 +220,14 @@ agentscope:
     enable-reasoning: false
     emit-run-finished-after-error: false
     server-side-memory: false
+    interrupt-on-disconnect: true
 ```
 
+`interrupt-on-disconnect` controls whether an Agent run is interrupted when the MVC/WebFlux SSE
+connection is closed, times out, or fails while sending an event. It defaults to `true` for
+backward compatibility. Set it to `false` to let the Agent continue running after the client
+disconnects; events produced while the connection is closed are not replayed by the starter.
+
 You can extend the default chain with beans:
 
 - `AgentEventConverter`: custom event semantic mapping.

docs/v2/zh/integration/protocol/agui.md

@@ -220,8 +220,13 @@ agentscope:
     enable-reasoning: false
     emit-run-finished-after-error: false
     server-side-memory: false
+    interrupt-on-disconnect: true
 ```
 
+`interrupt-on-disconnect` 用于控制 MVC/WebFlux 的 SSE 连接关闭、超时或发送事件失败时是否中断
+Agent run。默认值为 `true`，用于保持现有行为兼容。设置为 `false` 后，客户端断开时 Agent
+会继续执行；连接关闭期间产生的事件不会由 starter 重放。
+
 可以通过 bean 扩展默认链路：
 
 - `AgentEventConverter`：注册自定义事件语义映射。