diff --git a/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java b/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java index 0bb6b49af..52c78707b 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java +++ b/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java @@ -710,6 +710,7 @@ public void interrupt(InterruptSource source) { * * @param ctx the runtime context identifying the session to interrupt */ + @Override public void interrupt(RuntimeContext ctx) { interrupt(ctx, null); } @@ -721,6 +722,7 @@ public void interrupt(RuntimeContext ctx) { * @param ctx the runtime context identifying the session to interrupt * @param msg optional user message to attach to the interrupt signal */ + @Override public void interrupt(RuntimeContext ctx, Msg msg) { String uid = ctx != null ? ctx.getUserId() : null; String sid = ctx != null ? ctx.getSessionId() : null; diff --git a/agentscope-core/src/main/java/io/agentscope/core/agent/Agent.java b/agentscope-core/src/main/java/io/agentscope/core/agent/Agent.java index 438d1a32f..886d850b2 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/agent/Agent.java +++ b/agentscope-core/src/main/java/io/agentscope/core/agent/Agent.java @@ -71,21 +71,61 @@ default String getDescription() { /** * Interrupt the current agent execution. - * This method sets an interrupt flag that will be checked by the agent at appropriate - * checkpoints during execution. The interruption is cooperative and may not take effect - * immediately. + * + * @deprecated since 2.0.0, use {@link #interrupt(RuntimeContext)} with explicit runtime + * context to ensure the interruption targets the correct session slot. */ + @Deprecated(since = "2.0.0", forRemoval = true) void interrupt(); /** * Interrupt the current agent execution with a user message. - * This method sets an interrupt flag and associates a user message with the interruption. - * The interruption is cooperative and may not take effect immediately. * - * @param msg User message associated with the interruption + * @deprecated since 2.0.0, use {@link #interrupt(RuntimeContext, Msg)} with explicit runtime + * context to ensure the interruption targets the correct session slot. */ + @Deprecated(since = "2.0.0", forRemoval = true) void interrupt(Msg msg); + /** + * Interrupt the agent execution for the session identified by the given {@link RuntimeContext}. + * This is the session-aware variant of {@link #interrupt()} that targets a specific + * {@code (userId, sessionId)} session instead of a hardcoded default session. + * + *

When called from a middleware or tool that holds both {@code Agent agent} and + * {@code RuntimeContext ctx}, this method ensures the interruption targets the current + * in-flight call's session rather than a different session. + * + *

Default implementation delegates to {@link #interrupt()} for backward compatibility. + * Subclasses that maintain per-session state (e.g., {@link io.agentscope.core.ReActAgent}) + * should override this to trigger the interrupt signal on the correct session slot. + * + * @param ctx the runtime context identifying the session to interrupt; if {@code null} or + * {@code ctx.getSessionId()} is {@code null}/{@code blank}, implementations may + * fall back to a default session + */ + default void interrupt(RuntimeContext ctx) { + interrupt(); + } + + /** + * Interrupt the agent execution for the session identified by the given {@link RuntimeContext} + * with an associated user message. + * + *

This is the session-aware variant of {@link #interrupt(Msg)} that targets a specific + * {@code (userId, sessionId)} session instead of a hardcoded default session. + * + *

Default implementation delegates to {@link #interrupt(Msg)} for backward compatibility. + * Subclasses that maintain per-session state (e.g., {@link io.agentscope.core.ReActAgent}) + * should override this to trigger the interrupt signal on the correct session slot. + * + * @param ctx the runtime context identifying the session to interrupt + * @param msg optional user message to associate with the interruption + */ + default void interrupt(RuntimeContext ctx, Msg msg) { + interrupt(msg); + } + /** * Returns the agent's runtime {@link io.agentscope.core.state.AgentState}, or {@code null} if * this agent type does not maintain one. diff --git a/agentscope-core/src/main/java/io/agentscope/core/agent/AgentBase.java b/agentscope-core/src/main/java/io/agentscope/core/agent/AgentBase.java index a6b0248b6..e5f006143 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/agent/AgentBase.java +++ b/agentscope-core/src/main/java/io/agentscope/core/agent/AgentBase.java @@ -440,28 +440,44 @@ public static void removeSystemHook(Hook hook) { systemHooks.remove(hook); } - /** @deprecated Subclasses should implement per-session interrupt via RuntimeContext. */ - @Deprecated + /** + * @deprecated since 2.0.0, use {@link #interrupt(RuntimeContext)} with explicit runtime context + * to ensure the interruption targets the correct session slot. + */ + @Deprecated(since = "2.0.0", forRemoval = true) @Override public void interrupt() {} - /** @deprecated Subclasses should implement per-session interrupt via RuntimeContext. */ - @Deprecated + /** + * @deprecated since 2.0.0, use {@link #interrupt(RuntimeContext, Msg)} with explicit runtime + * context to ensure the interruption targets the correct session slot. + */ + @Deprecated(since = "2.0.0", forRemoval = true) @Override public void interrupt(Msg msg) {} - /** @deprecated Subclasses should implement per-session interrupt via RuntimeContext. */ - @Deprecated + /** + * @deprecated since 2.0.0, use {@link #interrupt(RuntimeContext)} with explicit runtime + * context to ensure the interruption targets the correct session slot. + */ + @Deprecated(since = "2.0.0", forRemoval = true) public void interrupt(InterruptSource source) {} - /** @deprecated No longer needed; ReActAgent uses per-session InterruptControl. */ - @Deprecated + /** + * @deprecated since 2.0.0; no longer needed. ReActAgent uses per-session + * {@link io.agentscope.core.interruption.InterruptControl} on its per-call + * {@link io.agentscope.core.state.AgentState}. + */ + @Deprecated(since = "2.0.0", forRemoval = true) protected Mono checkInterruptedAsync() { return Mono.empty(); } - /** @deprecated No-op; per-session interrupt state is managed by AgentState.interruptControl(). */ - @Deprecated + /** + * @deprecated since 2.0.0; no-op. Per-session interrupt state is managed by + * {@link io.agentscope.core.interruption.InterruptControl}. + */ + @Deprecated(since = "2.0.0", forRemoval = true) protected void resetInterruptFlag() {} private InterruptContext createInterruptContext() { @@ -502,14 +518,21 @@ private Function> createErrorHandler(Msg... originalArgs) { }; } - /** @deprecated No-op stub. */ - @Deprecated + /** + * @deprecated since 2.0.0; returns a detached flag that is never read. Use + * {@link io.agentscope.core.interruption.InterruptControl#isInterrupted()} on the + * session's {@link io.agentscope.core.state.AgentState} instead. + */ + @Deprecated(since = "2.0.0", forRemoval = true) protected AtomicBoolean getInterruptFlag() { return new AtomicBoolean(false); } - /** @deprecated Returns USER. Per-session interrupt source is on AgentState.interruptControl(). */ - @Deprecated + /** + * @deprecated since 2.0.0; always returns {@link InterruptSource#USER}. Per-session + * interrupt source is on {@link io.agentscope.core.interruption.InterruptControl}. + */ + @Deprecated(since = "2.0.0", forRemoval = true) protected InterruptSource getInterruptSource() { return InterruptSource.USER; } diff --git a/agentscope-core/src/test/java/io/agentscope/core/agent/AgentDefaultInterruptTest.java b/agentscope-core/src/test/java/io/agentscope/core/agent/AgentDefaultInterruptTest.java new file mode 100644 index 000000000..1685e0e68 --- /dev/null +++ b/agentscope-core/src/test/java/io/agentscope/core/agent/AgentDefaultInterruptTest.java @@ -0,0 +1,159 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.agentscope.core.agent; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; + +import com.fasterxml.jackson.databind.JsonNode; +import io.agentscope.core.message.Msg; +import io.agentscope.core.message.MsgRole; +import io.agentscope.core.message.TextBlock; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * Backward-compatibility contract for the {@link Agent#interrupt(RuntimeContext)} and + * {@link Agent#interrupt(RuntimeContext, Msg)} default methods. + * + *

Agent implementations that predate the session-aware overloads must continue to work: when + * they do not override the new {@code interrupt(RuntimeContext, ...)} methods, calls MUST fall + * through to the legacy {@link Agent#interrupt()} / {@link Agent#interrupt(Msg)} methods + * unchanged. + */ +@DisplayName("Agent default interrupt(RuntimeContext) — legacy fallback") +class AgentDefaultInterruptTest { + + /** + * A minimal {@link Agent} implementation that only overrides the legacy interrupt methods. + * The two {@code interrupt(RuntimeContext, ...)} methods are inherited from the interface's + * default implementations and must delegate here. + */ + private static final class RecordingAgent implements Agent { + final AtomicInteger legacyNoArgCalls = new AtomicInteger(); + final AtomicInteger legacyMsgCalls = new AtomicInteger(); + final AtomicReference lastLegacyMsg = new AtomicReference<>(); + + @Override + public String getAgentId() { + return "recording"; + } + + @Override + public String getName() { + return "recording"; + } + + @Override + public void interrupt() { + legacyNoArgCalls.incrementAndGet(); + } + + @Override + public void interrupt(Msg msg) { + legacyMsgCalls.incrementAndGet(); + lastLegacyMsg.set(msg); + } + + @Override + public Mono call(List msgs) { + return Mono.empty(); + } + + @Override + public Mono call(List msgs, Class structuredModel) { + return Mono.empty(); + } + + @Override + public Mono call(List msgs, JsonNode schema) { + return Mono.empty(); + } + + @Override + public Flux stream(List msgs, StreamOptions options) { + return Flux.empty(); + } + + @Override + public Flux stream(List msgs, StreamOptions options, Class structuredModel) { + return Flux.empty(); + } + + @Override + public Flux stream(List msgs, StreamOptions options, JsonNode schema) { + return Flux.empty(); + } + + @Override + public Mono observe(Msg msg) { + return Mono.empty(); + } + + @Override + public Mono observe(List msgs) { + return Mono.empty(); + } + } + + @Test + @DisplayName("interrupt(RuntimeContext) falls back to legacy interrupt() by default") + void defaultInterruptWithContext_delegatesToLegacyNoArg() { + RecordingAgent agent = new RecordingAgent(); + RuntimeContext ctx = RuntimeContext.builder().userId("u1").sessionId("sessA").build(); + + agent.interrupt(ctx); + + assertEquals(1, agent.legacyNoArgCalls.get(), "default must delegate to interrupt()"); + assertEquals(0, agent.legacyMsgCalls.get(), "no-arg default must not touch interrupt(Msg)"); + } + + @Test + @DisplayName("interrupt(RuntimeContext, Msg) falls back to legacy interrupt(Msg) by default") + void defaultInterruptWithContextAndMsg_delegatesToLegacyWithMsg() { + RecordingAgent agent = new RecordingAgent(); + RuntimeContext ctx = RuntimeContext.builder().userId("u1").sessionId("sessA").build(); + Msg attached = + Msg.builder() + .name("user") + .role(MsgRole.USER) + .content(TextBlock.builder().text("please stop").build()) + .build(); + + agent.interrupt(ctx, attached); + + assertEquals(0, agent.legacyNoArgCalls.get(), "msg default must not touch interrupt()"); + assertEquals(1, agent.legacyMsgCalls.get(), "default must delegate to interrupt(Msg)"); + assertSame(attached, agent.lastLegacyMsg.get(), "attached message must be forwarded"); + } + + @Test + @DisplayName("null RuntimeContext still delegates to the legacy method") + void defaultInterrupt_toleratesNullContext() { + RecordingAgent agent = new RecordingAgent(); + + agent.interrupt((RuntimeContext) null); + agent.interrupt(null, null); + + assertEquals(1, agent.legacyNoArgCalls.get()); + assertEquals(1, agent.legacyMsgCalls.get()); + } +} diff --git a/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentInterruptRuntimeContextTest.java b/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentInterruptRuntimeContextTest.java new file mode 100644 index 000000000..e2c745312 --- /dev/null +++ b/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentInterruptRuntimeContextTest.java @@ -0,0 +1,156 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.agentscope.core.agent; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.agentscope.core.ReActAgent; +import io.agentscope.core.interruption.InterruptSource; +import io.agentscope.core.message.ContentBlock; +import io.agentscope.core.message.Msg; +import io.agentscope.core.message.MsgRole; +import io.agentscope.core.message.TextBlock; +import io.agentscope.core.model.ChatModelBase; +import io.agentscope.core.model.ChatResponse; +import io.agentscope.core.model.GenerateOptions; +import io.agentscope.core.model.ToolSchema; +import io.agentscope.core.state.InMemoryAgentStateStore; +import java.util.List; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; + +/** + * Semantic contract of the {@link ReActAgent} overrides of + * {@link io.agentscope.core.agent.Agent#interrupt(RuntimeContext)} and + * {@link io.agentscope.core.agent.Agent#interrupt(RuntimeContext, Msg)}. + * + *

The interrupt signal must land on the {@link io.agentscope.core.interruption.InterruptControl} + * belonging to the exact {@code (userId, sessionId)} slot carried by the {@link RuntimeContext}, + * and MUST NOT leak into other slots. When the context is {@code null} or its session id is + * blank, the agent falls back to the configured {@code defaultSessionId}. + */ +@DisplayName("ReActAgent.interrupt(RuntimeContext) — per-session slot targeting") +class ReActAgentInterruptRuntimeContextTest { + + private static final class NoopModel extends ChatModelBase { + @Override + public String getModelName() { + return "noop"; + } + + @Override + protected Flux doStream( + List messages, List tools, GenerateOptions options) { + return Flux.just( + ChatResponse.builder() + .content(List.of(TextBlock.builder().text("ok").build())) + .build()); + } + } + + private ReActAgent agent() { + return ReActAgent.builder() + .name("asst") + .sysPrompt("hi") + .model(new NoopModel()) + .stateStore(new InMemoryAgentStateStore()) + .build(); + } + + @Test + @DisplayName("interrupt(ctx) targets the (userId, sessionId) slot carried by the context") + void interruptWithContext_targetsMatchingSlot() { + ReActAgent agent = agent(); + RuntimeContext ctx = RuntimeContext.builder().userId("u1").sessionId("sessA").build(); + + agent.interrupt(ctx); + + assertTrue( + agent.getAgentState("u1", "sessA").interruptControl().isInterrupted(), + "matching slot must observe the interrupt signal"); + assertFalse( + agent.getAgentState("u1", "sessB").interruptControl().isInterrupted(), + "sibling slot must NOT observe the signal"); + assertFalse( + agent.getAgentState("u2", "sessA").interruptControl().isInterrupted(), + "sibling user must NOT observe the signal"); + } + + @Test + @DisplayName("interrupt(ctx, msg) attaches the user message on the correct slot") + void interruptWithContextAndMsg_attachesMessage() { + ReActAgent agent = agent(); + RuntimeContext ctx = RuntimeContext.builder().userId("u1").sessionId("sessA").build(); + Msg attached = + Msg.builder() + .name("user") + .role(MsgRole.USER) + .content(TextBlock.builder().text("please stop").build()) + .build(); + + agent.interrupt(ctx, attached); + + var control = agent.getAgentState("u1", "sessA").interruptControl(); + assertTrue(control.isInterrupted()); + assertSame(attached, control.getUserMessage(), "the attached msg must be recorded"); + assertSame( + InterruptSource.USER, + control.getSource(), + "session-aware interrupt is a USER-source signal"); + } + + @Test + @DisplayName("interrupt(null) falls back to the default session slot") + void interruptWithNullContext_fallsBackToDefaultSession() { + ReActAgent agent = + ReActAgent.builder() + .name("asst") + .sysPrompt("hi") + .model(new NoopModel()) + .stateStore(new InMemoryAgentStateStore()) + .defaultSessionId("fallback-slot") + .build(); + + agent.interrupt((RuntimeContext) null); + + assertTrue( + agent.getAgentState(null, "fallback-slot").interruptControl().isInterrupted(), + "null ctx must land on the configured defaultSessionId slot"); + } + + @Test + @DisplayName("interrupt(ctx) with blank sessionId falls back to the default session slot") + void interruptWithBlankSessionId_fallsBackToDefault() { + ReActAgent agent = + ReActAgent.builder() + .name("asst") + .sysPrompt("hi") + .model(new NoopModel()) + .stateStore(new InMemoryAgentStateStore()) + .defaultSessionId("fallback-slot") + .build(); + RuntimeContext ctx = RuntimeContext.builder().userId("u1").sessionId(" ").build(); + + agent.interrupt(ctx); + + assertTrue( + agent.getAgentState("u1", "fallback-slot").interruptControl().isInterrupted(), + "blank sessionId must fall back to the configured defaultSessionId slot"); + } +} diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java index d669184cb..9932f81bc 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java @@ -487,6 +487,16 @@ public void interrupt(Msg msg) { delegate.interrupt(msg); } + @Override + public void interrupt(RuntimeContext ctx) { + delegate.interrupt(ctx); + } + + @Override + public void interrupt(RuntimeContext ctx, Msg msg) { + delegate.interrupt(ctx, msg); + } + // ----------------------------------------------------------------- // Channel / Gateway // ----------------------------------------------------------------- diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/HarnessAgentTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/HarnessAgentTest.java index 8618448ce..14fc240bf 100644 --- a/agentscope-harness/src/test/java/io/agentscope/harness/agent/HarnessAgentTest.java +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/HarnessAgentTest.java @@ -1181,6 +1181,53 @@ void agentSpecLoader_markdownDeclaration_sharedMode_noPath_inlineBody() throws E "body should be inline agents body when no workspace.path"); } + @Test + void interruptWithContext_delegatesToInnerReActAgent() throws Exception { + Files.createDirectories(workspace); + HarnessAgent agent = + HarnessAgent.builder() + .name("t") + .model(stubModel("ok")) + .workspace(workspace) + .abstractFilesystem(new LocalFilesystem(workspace)) + .build(); + RuntimeContext ctx = RuntimeContext.builder().userId("u1").sessionId("sessA").build(); + + agent.interrupt(ctx); + + assertTrue( + agent.getDelegate().getAgentState("u1", "sessA").interruptControl().isInterrupted(), + "HarnessAgent.interrupt(ctx) must reach the inner ReActAgent's matching slot"); + assertFalse( + agent.getDelegate().getAgentState("u1", "sessB").interruptControl().isInterrupted(), + "unrelated slot must not observe the signal"); + } + + @Test + void interruptWithContextAndMsg_forwardsMessageThroughDelegate() throws Exception { + Files.createDirectories(workspace); + HarnessAgent agent = + HarnessAgent.builder() + .name("t") + .model(stubModel("ok")) + .workspace(workspace) + .abstractFilesystem(new LocalFilesystem(workspace)) + .build(); + RuntimeContext ctx = RuntimeContext.builder().userId("u1").sessionId("sessA").build(); + Msg attached = + Msg.builder() + .name("user") + .role(MsgRole.USER) + .content(TextBlock.builder().text("please stop").build()) + .build(); + + agent.interrupt(ctx, attached); + + var control = agent.getDelegate().getAgentState("u1", "sessA").interruptControl(); + assertTrue(control.isInterrupted()); + assertSame(attached, control.getUserMessage()); + } + private static Model stubModel(String assistantText) { Model model = mock(Model.class); when(model.getModelName()).thenReturn("stub-model");