diff --git a/agentscope-extensions/agentscope-extensions-autocontext-memory/src/main/java/io/agentscope/core/memory/autocontext/AutoContextHook.java b/agentscope-extensions/agentscope-extensions-autocontext-memory/src/main/java/io/agentscope/core/memory/autocontext/AutoContextHook.java index d948449881..50d39a166a 100644 --- a/agentscope-extensions/agentscope-extensions-autocontext-memory/src/main/java/io/agentscope/core/memory/autocontext/AutoContextHook.java +++ b/agentscope-extensions/agentscope-extensions-autocontext-memory/src/main/java/io/agentscope/core/memory/autocontext/AutoContextHook.java @@ -209,8 +209,14 @@ private Mono handlePreCall(PreCallEvent event) { *

This ensures compression happens at a deterministic point (before reasoning) * and the LLM receives the compressed context. * + *

If compression fails with a non-interruption exception (e.g. model timeout, rate limit), + * the error is logged at WARN level and the original event is returned unchanged, allowing + * the agent to continue with uncompressed messages (graceful degradation). If the exception + * wraps an {@link InterruptedException} anywhere in its cause chain, {@code Mono.error} is + * returned to propagate the interruption signal correctly through the reactive pipeline. + * * @param event the PreReasoningEvent - * @return Mono containing the potentially modified event + * @return Mono containing the potentially modified event, or {@code Mono.error} if interrupted */ private Mono handlePreReasoning(PreReasoningEvent event) { Agent agent = event.getAgent(); @@ -231,6 +237,17 @@ private Mono handlePreReasoning(PreReasoningEvent event) { event.setInputMessages(buildInputMessages(event, autoContextMemory)); return event; }) + .onErrorResume( + e -> { + if (isInterruptedException(e)) { + return Mono.error(e); + } + log.warn( + "Failed to compress context in handlePreReasoning, continuing" + + " with original messages", + e); + return Mono.just(event); + }) .subscribeOn(Schedulers.boundedElastic()); } @@ -284,4 +301,15 @@ private List buildInputMessages( newInputMessages.addAll(autoContextMemory.getMessages()); return newInputMessages; } + + private static boolean isInterruptedException(Throwable t) { + Throwable current = t; + while (current != null) { + if (current instanceof InterruptedException) { + return true; + } + current = current.getCause(); + } + return false; + } } diff --git a/agentscope-extensions/agentscope-extensions-autocontext-memory/src/test/java/io/agentscope/core/memory/autocontext/AutoContextHookTest.java b/agentscope-extensions/agentscope-extensions-autocontext-memory/src/test/java/io/agentscope/core/memory/autocontext/AutoContextHookTest.java index 1080a5b541..c1e5c63839 100644 --- a/agentscope-extensions/agentscope-extensions-autocontext-memory/src/test/java/io/agentscope/core/memory/autocontext/AutoContextHookTest.java +++ b/agentscope-extensions/agentscope-extensions-autocontext-memory/src/test/java/io/agentscope/core/memory/autocontext/AutoContextHookTest.java @@ -18,6 +18,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; @@ -629,6 +630,102 @@ void testPreReasoningEventPreservesSystemPromptOnNonBlockingScheduler() { assertTrue(updatedSystemPrompt.contains("context_reload")); } + @Test + @DisplayName("Should handle compression error gracefully and return original event") + void testPreReasoningEventHandlesCompressionError() { + FailingModel failingModel = new FailingModel(new RuntimeException("model timeout")); + AutoContextMemory failingMemory = + new AutoContextMemory(createCompressionConfig(), failingModel); + populateCompressibleConversation(failingMemory, 3); + failingMemory.addMessage( + Msg.builder() + .role(MsgRole.USER) + .name("user") + .content(TextBlock.builder().text("Final user message").build()) + .build()); + + ReActAgent agent = + ReActAgent.builder() + .name("TestAgent") + .model(mockModel) + .memory(failingMemory) + .toolkit(toolkit) + .build(); + + List inputMessages = new ArrayList<>(); + inputMessages.add( + Msg.builder() + .role(MsgRole.SYSTEM) + .name("system") + .content(TextBlock.builder().text("You are a helpful assistant.").build()) + .build()); + inputMessages.add( + Msg.builder() + .role(MsgRole.USER) + .name("user") + .content(TextBlock.builder().text("Hello").build()) + .build()); + + PreReasoningEvent event = new PreReasoningEvent(agent, "test-model", null, inputMessages); + + PreReasoningEvent result = hook.onEvent(event).block(); + + assertNotNull(result, "Should return event, not throw"); + assertEquals(event, result, "Should return original event unchanged on error"); + } + + @Test + @DisplayName("Should propagate wrapped InterruptedException via Mono.error") + void testPreReasoningEventPreservesWrappedInterruptedException() { + FailingModel failingModel = + new FailingModel( + new RuntimeException(new InterruptedException("thread interrupted"))); + AutoContextMemory failingMemory = + new AutoContextMemory(createCompressionConfig(), failingModel); + populateCompressibleConversation(failingMemory, 3); + failingMemory.addMessage( + Msg.builder() + .role(MsgRole.USER) + .name("user") + .content(TextBlock.builder().text("Final user message").build()) + .build()); + + ReActAgent agent = + ReActAgent.builder() + .name("TestAgent") + .model(mockModel) + .memory(failingMemory) + .toolkit(toolkit) + .build(); + + List inputMessages = new ArrayList<>(); + inputMessages.add( + Msg.builder() + .role(MsgRole.USER) + .name("user") + .content(TextBlock.builder().text("Hello").build()) + .build()); + + PreReasoningEvent event = new PreReasoningEvent(agent, "test-model", null, inputMessages); + + RuntimeException thrown = + assertThrows(RuntimeException.class, () -> hook.onEvent(event).block()); + assertTrue( + isInterruptedInCauseChain(thrown), + "Should propagate exception with InterruptedException in cause chain"); + } + + private static boolean isInterruptedInCauseChain(Throwable t) { + Throwable current = t; + while (current != null) { + if (current instanceof InterruptedException) { + return true; + } + current = current.getCause(); + } + return false; + } + private AutoContextConfig createCompressionConfig() { return AutoContextConfig.builder() .msgThreshold(5) @@ -720,4 +817,23 @@ void reset() { callCount = 0; } } + + private static class FailingModel implements Model { + private final RuntimeException exception; + + FailingModel(RuntimeException exception) { + this.exception = exception; + } + + @Override + public Flux stream( + List messages, List tools, GenerateOptions options) { + throw exception; + } + + @Override + public String getModelName() { + return "failing-model"; + } + } }