From 55adb1ccc4a2f7cc190baf43b180329b643345d0 Mon Sep 17 00:00:00 2001 From: kevinyang03 <152797417+kevinyang03@users.noreply.github.com> Date: Tue, 25 Aug 2026 00:14:47 +0800 Subject: [PATCH 1/3] fix(tool): honor execution config retries for tool exceptions --- .../core/model/ExecutionConfig.java | 7 + .../io/agentscope/core/tool/AgentTool.java | 23 + .../core/tool/ReflectiveFunctionTool.java | 9 + .../io/agentscope/core/tool/ToolExecutor.java | 117 +++- .../core/tool/ToolMethodInvoker.java | 55 +- .../io/agentscope/core/tool/mcp/McpTool.java | 52 +- .../agentscope/core/util/ExceptionUtils.java | 28 + .../core/tool/ToolExecutorTest.java | 586 ++++++++++++++++++ 8 files changed, 826 insertions(+), 51 deletions(-) diff --git a/agentscope-core/src/main/java/io/agentscope/core/model/ExecutionConfig.java b/agentscope-core/src/main/java/io/agentscope/core/model/ExecutionConfig.java index 1b269cc6ad..cabc31691f 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/model/ExecutionConfig.java +++ b/agentscope-core/src/main/java/io/agentscope/core/model/ExecutionConfig.java @@ -159,6 +159,13 @@ private static boolean isRetryableError(Throwable error) { *
  • Timeout: 5 minutes *
  • Max attempts: 1 (no retry) * + * + *

    Tool retry semantics: when {@code maxAttempts} is greater than 1, failures that + * surface as reactive error signals — such as exceptions thrown by the tool or its transport, + * and timeouts — are retried as decided by {@link #getRetryOn()}. Failures that a tool + * reports as a completed {@code ToolResultBlock} error result (for example MCP protocol-level + * {@code isError=true} responses or deliberately returned business errors) are never retried, + * since replaying a completed call may be unsafe for non-idempotent operations. */ public static final ExecutionConfig TOOL_DEFAULTS = builder().timeout(Duration.ofMinutes(5)).maxAttempts(1).build(); diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/AgentTool.java b/agentscope-core/src/main/java/io/agentscope/core/tool/AgentTool.java index ba2d9014e5..512ce32bf3 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/tool/AgentTool.java +++ b/agentscope-core/src/main/java/io/agentscope/core/tool/AgentTool.java @@ -127,4 +127,27 @@ default boolean isReadOnly() { * @return Mono containing ToolResultBlock */ Mono callAsync(ToolCallParam param); + + /** + * Executes the tool through the framework's execution infrastructure. + * + *

    This method is the channel used by {@link ToolExecutor} when running a tool call through + * the infrastructure layers (scheduling, timeout, retry, graceful-shutdown guard). Unlike + * {@link #callAsync(ToolCallParam)}, which converts failures into {@link ToolResultBlock} + * error results, this channel may surface failures as reactive error signals so that the + * infrastructure can decide whether to retry them; once retries are exhausted, the executor + * converts the final error into a {@link ToolResultBlock} error result. + * + *

    The default implementation delegates to {@link #callAsync(ToolCallParam)}, so existing + * implementations remain compatible without any changes. Built-in tools whose failures should + * be retryable (for example annotation-based tools and MCP tools) override this method to keep + * exceptions as error signals. Tools that deliberately report failures as + * {@link ToolResultBlock} error results keep them as completed results and are never retried. + * + * @param param The tool call parameters + * @return Mono containing ToolResultBlock, or signalling an error on failure + */ + default Mono callAsyncForExecution(ToolCallParam param) { + return callAsync(param); + } } diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/ReflectiveFunctionTool.java b/agentscope-core/src/main/java/io/agentscope/core/tool/ReflectiveFunctionTool.java index c92ef950df..a8f034586b 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/tool/ReflectiveFunctionTool.java +++ b/agentscope-core/src/main/java/io/agentscope/core/tool/ReflectiveFunctionTool.java @@ -165,6 +165,15 @@ public Mono callAsync(ToolCallParam param) { return methodInvoker.invokeAsync(toolObject, method, param, customConverter); } + @Override + public Mono callAsyncForExecution(ToolCallParam param) { + if (isExternalTool()) { + return Mono.just( + ToolResultBlock.suspended(param.getToolUseBlock(), new ToolSuspendException())); + } + return methodInvoker.invokeRawAsync(toolObject, method, param, customConverter); + } + Method getMethod() { return method; } diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/ToolExecutor.java b/agentscope-core/src/main/java/io/agentscope/core/tool/ToolExecutor.java index 7d9c13a681..72831d863a 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/tool/ToolExecutor.java +++ b/agentscope-core/src/main/java/io/agentscope/core/tool/ToolExecutor.java @@ -28,6 +28,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeoutException; import java.util.function.BiConsumer; import java.util.function.Predicate; import org.slf4j.Logger; @@ -159,13 +160,47 @@ private void invokeChunkCallback( // ==================== Single Tool Execution ==================== /** - * Execute a single tool call with full infrastructure support. + * Execute a single tool call without the infrastructure layers (scheduling, timeout, retry, + * shutdown guard). + * + *

    This is the compatibility entry point used by {@link Toolkit#callTool(ToolCallParam)}: it + * keeps the {@link AgentTool#callAsync(ToolCallParam)} contract that failures surface as + * {@link ToolResultBlock} error results instead of reactive error signals. * * @param param Tool call parameters * @return Mono containing execution result */ Mono execute(ToolCallParam param) { - return TracerRegistry.get().callTool(this.toolkit, param, () -> executeCore(param)); + return executeWithTracing(param, false) + .onErrorResume(e -> Mono.just(buildToolErrorResult(e))); + } + + /** + * Execute a single tool call through the raw execution channel, keeping failures as reactive + * error signals for the infrastructure layers (scheduling, timeout, retry, shutdown guard) to + * act on. Used only by {@link #executeWithInfrastructure}. + * + * @param param Tool call parameters + * @return Mono containing execution result, or signalling an error on failure + */ + private Mono executeRaw(ToolCallParam param) { + return executeWithTracing(param, true); + } + + /** + * Run the traced core execution for the given tool call. + * + * @param param Tool call parameters + * @param useExecutionPath whether to use the raw execution channel + * ({@link AgentTool#callAsyncForExecution(ToolCallParam)}) that keeps failures as error + * signals, or the compatibility channel ({@link AgentTool#callAsync(ToolCallParam)}) that + * converts failures into error results + * @return Mono containing execution result + */ + private Mono executeWithTracing( + ToolCallParam param, boolean useExecutionPath) { + return TracerRegistry.get() + .callTool(this.toolkit, param, () -> executeCore(param, useExecutionPath)); } /** @@ -181,7 +216,7 @@ Mono execute(ToolCallParam param) { *

  • Actual tool invocation
    • * */ - private Mono executeCore(ToolCallParam param) { + private Mono executeCore(ToolCallParam param, boolean useExecutionPath) { ToolUseBlock toolCall = param.getToolUseBlock(); AgentTool tool = toolRegistry.getTool(toolCall.getName()); @@ -263,8 +298,34 @@ private Mono executeCore(ToolCallParam param) { .emitter(toolEmitter) .build(); - return tool.callAsync(executionParam) - .onErrorResume( + // Invoke the tool through the selected channel. + // + // Compatibility channel (useExecutionPath=false): failures are converted into + // ToolResultBlock error results right here, preserving the AgentTool contract for callers + // that execute tools directly without infrastructure. + // + // Execution channel (useExecutionPath=true): failures stay as reactive error signals and + // flow through the timeout/retry/shutdown layers applied by executeWithInfrastructure, + // which converts them into ToolResultBlock error results only after retries are + // exhausted. Mono.defer re-invokes callAsyncForExecution on every subscription so each + // retry attempt rebuilds and re-runs the tool call instead of resubscribing to the same + // pre-assembled publisher. + Mono invocation; + if (useExecutionPath) { + invocation = + Mono.defer(() -> tool.callAsyncForExecution(executionParam)) + // Unwrap reflection/future wrappers so the retry predicate sees the + // original exception (IOException, transport errors, ...) instead of + // InvocationTargetException or ExecutionException. + .onErrorMap(ExceptionUtils::unwrapExecutionWrapper); + } else { + invocation = tool.callAsync(executionParam); + } + + // ToolSuspendException is business-level suspension, not a failure: both channels convert + // it to a suspended result without retrying. + Mono chain = + invocation.onErrorResume( ToolSuspendException.class, e -> { // Convert ToolSuspendException to suspended result @@ -273,21 +334,23 @@ private Mono executeCore(ToolCallParam param) { toolCall.getName(), e.getReason() != null ? e.getReason() : "no reason"); return Mono.just(ToolResultBlock.suspended(toolCall, e)); - }) - .onErrorResume( - e -> { - String errorMsg = - e.getMessage() != null - ? e.getMessage() - : e.getClass().getSimpleName(); - return Mono.just( - ToolResultBlock.error("Tool execution failed: " + errorMsg)); - }) - .switchIfEmpty( - Mono.just( - ToolResultBlock.error( - "Tool execution failed: Tool completed without returning a" - + " result"))); + }); + if (!useExecutionPath) { + chain = chain.onErrorResume(e -> Mono.just(buildToolErrorResult(e))); + } + return chain.switchIfEmpty( + Mono.just( + ToolResultBlock.error( + "Tool execution failed: Tool completed without returning a" + + " result"))); + } + + /** + * Build the standard {@link ToolResultBlock} error result for a failed tool call. + */ + private ToolResultBlock buildToolErrorResult(Throwable e) { + String errorMsg = e.getMessage() != null ? e.getMessage() : e.getClass().getSimpleName(); + return ToolResultBlock.error("Tool execution failed: " + errorMsg); } // ==================== Batch Tool Execution ==================== @@ -384,8 +447,10 @@ private Mono executeWithInfrastructure( .runtimeContext(agentRuntimeContext) .build(); - // Get core execution - Mono execution = execute(param); + // Use the raw execution channel so tool failures remain reactive error signals while + // the timeout/retry/shutdown layers below run; this final onErrorResume converts the + // error that remains after retries are exhausted into a ToolResultBlock error result. + Mono execution = executeRaw(param); // Apply infrastructure layers execution = applyScheduling(execution); @@ -424,9 +489,11 @@ private Mono applyTimeout( Duration timeout = config.getTimeout(); logger.debug("Applied timeout: {} for tool: {}", timeout, toolCall.getName()); + // TimeoutException (rather than a bare RuntimeException) so that + // ExecutionConfig.RETRYABLE_ERRORS recognizes tool timeouts as retryable. return execution.timeout( timeout, - Mono.error(new RuntimeException("Tool execution timeout after " + timeout))); + Mono.error(new TimeoutException("Tool execution timeout after " + timeout))); } private Mono applyRetry( @@ -450,6 +517,10 @@ private Mono applyRetry( .maxBackoff(maxBackoff) .jitter(0.5) .filter(retryOn) + // On exhaustion, propagate the last failure itself instead of Reactor's + // RetryExhaustedException wrapper so the error result surfaces the real + // failure message (e.g. the timeout or the tool exception). + .onRetryExhaustedThrow((spec, signal) -> signal.failure()) .doBeforeRetry( signal -> logger.warn( diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/ToolMethodInvoker.java b/agentscope-core/src/main/java/io/agentscope/core/tool/ToolMethodInvoker.java index 73bb16c3fe..fd78b9bfb3 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/tool/ToolMethodInvoker.java +++ b/agentscope-core/src/main/java/io/agentscope/core/tool/ToolMethodInvoker.java @@ -46,6 +46,10 @@ class ToolMethodInvoker { /** * Invoke tool method asynchronously with custom converter support. * + *

    Failures are converted into {@link ToolResultBlock} error results, so this entry point + * never signals an error to its caller. Direct callers of {@code AgentTool.callAsync} rely on + * this contract. + * * @param toolObject the object containing the method * @param method the method to invoke * @param param the tool call parameters containing input, toolUseBlock, agent, and context @@ -57,6 +61,31 @@ Mono invokeAsync( Method method, ToolCallParam param, ToolResultConverter customConverter) { + return invokeRawAsync(toolObject, method, param, customConverter) + .onErrorResume(this::handleError); + } + + /** + * Invoke tool method asynchronously while keeping failures as reactive error signals. + * + *

    This is the raw channel used by {@link ToolExecutor}'s execution infrastructure: errors + * stay as error signals so the timeout, retry and graceful-shutdown layers can act on them, + * and the executor converts them into {@link ToolResultBlock} error results only after the + * retry decision has been made. Parameter injection, argument conversion, reflection + * invocation, {@link CompletableFuture}/{@link Mono} adaptation and result conversion behave + * identically to {@link #invokeAsync}; only the error-to-result conversion is omitted. + * + * @param toolObject the object containing the method + * @param method the method to invoke + * @param param the tool call parameters containing input, toolUseBlock, agent, and context + * @param customConverter custom converter for this invocation (null to use default) + * @return Mono containing ToolResultBlock, or signalling an error on failure + */ + Mono invokeRawAsync( + Object toolObject, + Method method, + ToolCallParam param, + ToolResultConverter customConverter) { // Use custom converter if provided, otherwise use default final ToolResultConverter converter = customConverter != null ? customConverter : defaultConverter; @@ -87,9 +116,8 @@ Mono invokeAsync( .map( r -> converter.convert( - r, extractGenericType(method))) - .onErrorResume(this::handleError)) - .onErrorResume(this::handleError); + r, + extractGenericType(method)))); } else if (returnType == Mono.class) { // Async method returning Mono: invoke and flatMap @@ -105,22 +133,19 @@ r, extractGenericType(method))) }) .flatMap( mono -> - mono.map(r -> converter.convert(r, extractGenericType(method))) - .onErrorResume(this::handleError)) - .onErrorResume(this::handleError); + mono.map( + r -> converter.convert(r, extractGenericType(method)))); } else { // Sync method: wrap in Mono.fromCallable return Mono.fromCallable( - () -> { - method.setAccessible(true); - Object[] args = - convertParameters( - method, input, agent, runtimeContext, emitter); - Object result = method.invoke(toolObject, args); - return converter.convert(result, method.getGenericReturnType()); - }) - .onErrorResume(this::handleError); + () -> { + method.setAccessible(true); + Object[] args = + convertParameters(method, input, agent, runtimeContext, emitter); + Object result = method.invoke(toolObject, args); + return converter.convert(result, method.getGenericReturnType()); + }); } } diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpTool.java b/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpTool.java index b7b3a9ec49..4cb75d341f 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpTool.java +++ b/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpTool.java @@ -178,19 +178,7 @@ public Mono checkPermissions( */ @Override public Mono callAsync(ToolCallParam param) { - logger.debug("Calling MCP tool '{}' with input: {}", getName(), param.getInput()); - - // Merge preset arguments with input arguments - Map mergedArgs = mergeArguments(param.getInput()); - - // Extract MCP meta from ContextStore by McpMeta type namespace - Map metaMap = extractMcpMeta(param); - - return clientWrapper - .callTool(getName(), mergedArgs, metaMap) - .map(McpContentConverter::convertCallToolResult) - .doOnSuccess( - result -> logger.debug("MCP tool '{}' completed successfully", getName())) + return callAsyncForExecution(param) .onErrorResume( e -> { logger.error( @@ -203,6 +191,44 @@ public Mono callAsync(ToolCallParam param) { }); } + /** + * Executes this MCP tool through the framework's execution infrastructure. + * + *

    Transport-level failures (connection loss, network interruption, timeout, client + * exceptions) stay as reactive error signals so {@link io.agentscope.core.tool.ToolExecutor} + * can apply the configured retry policy. Protocol-level business errors (the MCP server + * completing the call with {@code isError=true}) remain {@link ToolResultBlock} error results + * and are never retried, since replaying a completed non-idempotent call is unsafe. + */ + @Override + public Mono callAsyncForExecution(ToolCallParam param) { + // Keep the historical contract that invoking with a null param fails fast. + Objects.requireNonNull(param, "param must not be null"); + // Mono.defer re-creates the whole attempt (argument merging, meta extraction and the + // remote call) for every subscription, so each retry runs a fresh attempt and synchronous + // failures during preparation also surface as reactive errors. + return Mono.defer( + () -> { + logger.debug( + "Calling MCP tool '{}' with input: {}", getName(), param.getInput()); + + // Merge preset arguments with input arguments + Map mergedArgs = mergeArguments(param.getInput()); + + // Extract MCP meta from ContextStore by McpMeta type namespace + Map metaMap = extractMcpMeta(param); + + return clientWrapper + .callTool(getName(), mergedArgs, metaMap) + .map(McpContentConverter::convertCallToolResult) + .doOnSuccess( + result -> + logger.debug( + "MCP tool '{}' completed successfully", + getName())); + }); + } + /** * Gets the name of the MCP client that provides this tool. * diff --git a/agentscope-core/src/main/java/io/agentscope/core/util/ExceptionUtils.java b/agentscope-core/src/main/java/io/agentscope/core/util/ExceptionUtils.java index 681793e6a0..321e05b7dc 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/util/ExceptionUtils.java +++ b/agentscope-core/src/main/java/io/agentscope/core/util/ExceptionUtils.java @@ -15,7 +15,10 @@ */ package io.agentscope.core.util; +import java.lang.reflect.InvocationTargetException; import java.util.IdentityHashMap; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutionException; /** * Utility methods for exception handling. @@ -79,4 +82,29 @@ public static boolean containsInterruptedException(Throwable error) { } return false; } + + /** + * Unwraps reflective and future execution wrappers to expose the original failure. + * + *

    {@code method.invoke} wraps business exceptions in {@link InvocationTargetException}, + * and failed {@link java.util.concurrent.CompletableFuture}s surface as {@link + * ExecutionException} or {@link CompletionException}. This method walks those wrappers (whose + * cause chains cannot cycle) so retry predicates such as {@code + * ExecutionConfig.RETRYABLE_ERRORS} can inspect the original {@code IOException}, {@code + * HttpTransportException} or domain exception. + * + * @param error the throwable to unwrap (may be {@code null}) + * @return the innermost non-wrapper throwable, or {@code null} if the input is {@code null} + */ + public static Throwable unwrapExecutionWrapper(Throwable error) { + Throwable current = error; + while (current != null + && (current instanceof InvocationTargetException + || current instanceof ExecutionException + || current instanceof CompletionException) + && current.getCause() != null) { + current = current.getCause(); + } + return current; + } } diff --git a/agentscope-core/src/test/java/io/agentscope/core/tool/ToolExecutorTest.java b/agentscope-core/src/test/java/io/agentscope/core/tool/ToolExecutorTest.java index 55ae0cb786..bf37e13db3 100644 --- a/agentscope-core/src/test/java/io/agentscope/core/tool/ToolExecutorTest.java +++ b/agentscope-core/src/test/java/io/agentscope/core/tool/ToolExecutorTest.java @@ -18,19 +18,32 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import io.agentscope.core.message.ContentBlock; import io.agentscope.core.message.TextBlock; import io.agentscope.core.message.ToolResultBlock; +import io.agentscope.core.message.ToolResultState; import io.agentscope.core.message.ToolUseBlock; +import io.agentscope.core.model.ExecutionConfig; import io.agentscope.core.model.ToolSchema; +import io.agentscope.core.tool.mcp.McpClientWrapper; +import io.agentscope.core.tool.mcp.McpTool; import io.agentscope.core.tool.test.SampleTools; import io.agentscope.core.tool.test.ToolTestUtils; import io.agentscope.core.util.JsonUtils; +import io.modelcontextprotocol.spec.McpSchema; +import java.io.IOException; import java.time.Duration; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; import java.util.stream.Collectors; @@ -669,6 +682,547 @@ public Mono callAsync(ToolCallParam param) { } } + // ==================== Retry Behavior Tests ==================== + + @Test + @DisplayName("Should retry annotation sync tools that throw exceptions") + void shouldRetryAnnotationSyncToolExceptions() { + FlakyTools flaky = new FlakyTools(); + toolkit.registerTool(flaky); + + List responses = + toolkit.callTools( + List.of(toolCall("call-flaky-sync", "flaky_sync")), + retryConfig(2), + null, + null) + .block(TIMEOUT); + + assertEquals(1, responses.size()); + assertEquals("\"recovered\"", extractFirstText(responses.get(0))); + assertEquals(2, flaky.calls.get()); + } + + @Test + @DisplayName("Should retry annotation Mono tools that fail with exceptions") + void shouldRetryAnnotationMonoToolExceptions() { + FlakyTools flaky = new FlakyTools(); + toolkit.registerTool(flaky); + + List responses = + toolkit.callTools( + List.of(toolCall("call-flaky-mono", "flaky_mono")), + retryConfig(2), + null, + null) + .block(TIMEOUT); + + assertEquals(1, responses.size()); + assertEquals("\"recovered\"", extractFirstText(responses.get(0))); + assertEquals(2, flaky.calls.get()); + } + + @Test + @DisplayName("Should retry annotation CompletableFuture tools that fail with exceptions") + void shouldRetryAnnotationFutureToolExceptions() { + FlakyTools flaky = new FlakyTools(); + toolkit.registerTool(flaky); + + List responses = + toolkit.callTools( + List.of(toolCall("call-flaky-future", "flaky_future")), + retryConfig(2), + null, + null) + .block(TIMEOUT); + + assertEquals(1, responses.size()); + assertEquals("\"recovered\"", extractFirstText(responses.get(0))); + assertEquals(2, flaky.calls.get()); + } + + @Test + @DisplayName("Should retry MCP tool calls on transport failures") + void shouldRetryMcpToolTransportFailures() { + McpClientWrapper wrapper = mock(McpClientWrapper.class); + when(wrapper.getName()).thenReturn("test-client"); + McpTool mcpTool = new McpTool("flaky_mcp", "Flaky MCP tool", emptySchema(), wrapper); + when(wrapper.callTool(eq("flaky_mcp"), any(), any())) + .thenReturn(Mono.error(new IOException("Network down"))) + .thenReturn( + Mono.just( + new McpSchema.CallToolResult( + List.of(new McpSchema.TextContent("mcp recovered")), + false))); + toolkit.registerTool(mcpTool); + + List responses = + toolkit.callTools( + List.of(toolCall("call-flaky-mcp", "flaky_mcp")), + retryConfig(2), + null, + null) + .block(TIMEOUT); + + assertEquals(1, responses.size()); + assertTrue(extractFirstText(responses.get(0)).contains("mcp recovered")); + verify(wrapper, times(2)).callTool(eq("flaky_mcp"), any(), any()); + } + + @Test + @DisplayName("Should not retry MCP protocol-level business errors") + void shouldNotRetryMcpProtocolErrors() { + McpClientWrapper wrapper = mock(McpClientWrapper.class); + when(wrapper.getName()).thenReturn("test-client"); + McpTool mcpTool = new McpTool("mcp_error", "Failing MCP tool", emptySchema(), wrapper); + when(wrapper.callTool(eq("mcp_error"), any(), any())) + .thenReturn( + Mono.just( + new McpSchema.CallToolResult( + List.of(new McpSchema.TextContent("boom")), true))); + toolkit.registerTool(mcpTool); + + List responses = + toolkit.callTools( + List.of(toolCall("call-mcp-error", "mcp_error")), + retryConfig(3), + null, + null) + .block(TIMEOUT); + + assertEquals(1, responses.size()); + assertEquals(ToolResultState.ERROR, responses.get(0).getState()); + verify(wrapper, times(1)).callTool(eq("mcp_error"), any(), any()); + } + + @Test + @DisplayName("Should retry custom AgentTool failures that surface as errors") + void shouldRetryCustomAgentToolErrors() { + AtomicInteger calls = new AtomicInteger(0); + toolkit.registerTool( + new AgentTool() { + @Override + public String getName() { + return "custom_flaky"; + } + + @Override + public String getDescription() { + return "Custom tool that fails once"; + } + + @Override + public Map getParameters() { + return emptySchema(); + } + + @Override + public Mono callAsync(ToolCallParam param) { + int attempt = calls.incrementAndGet(); + if (attempt == 1) { + return Mono.error(new IOException("Transient custom failure")); + } + return Mono.just(ToolResultBlock.text("custom recovered")); + } + }); + + List responses = + toolkit.callTools( + List.of(toolCall("call-custom-flaky", "custom_flaky")), + retryConfig(2), + null, + null) + .block(TIMEOUT); + + assertEquals(1, responses.size()); + assertEquals("custom recovered", extractFirstText(responses.get(0))); + assertEquals(2, calls.get()); + } + + @Test + @DisplayName("Should not retry tool-returned error results") + void shouldNotRetrySemanticErrorResults() { + AtomicInteger calls = new AtomicInteger(0); + toolkit.registerTool( + new AgentTool() { + @Override + public String getName() { + return "semantic_error"; + } + + @Override + public String getDescription() { + return "Tool that reports a business error result"; + } + + @Override + public Map getParameters() { + return emptySchema(); + } + + @Override + public Mono callAsync(ToolCallParam param) { + calls.incrementAndGet(); + return Mono.just(ToolResultBlock.error("Business failure")); + } + }); + + List responses = + toolkit.callTools( + List.of(toolCall("call-semantic-error", "semantic_error")), + retryConfig(3), + null, + null) + .block(TIMEOUT); + + assertEquals(1, responses.size()); + assertEquals(1, calls.get()); + assertTrue(extractFirstText(responses.get(0)).contains("Business failure")); + } + + @Test + @DisplayName("Should respect the retryOn predicate") + void shouldRespectRetryOnPredicate() { + AtomicInteger calls = new AtomicInteger(0); + toolkit.registerTool( + new AgentTool() { + @Override + public String getName() { + return "never_retry"; + } + + @Override + public String getDescription() { + return "Tool whose failures are filtered out by retryOn"; + } + + @Override + public Map getParameters() { + return emptySchema(); + } + + @Override + public Mono callAsync(ToolCallParam param) { + calls.incrementAndGet(); + return Mono.error(new IOException("Always failing")); + } + }); + + ExecutionConfig config = + ExecutionConfig.builder() + .maxAttempts(3) + .initialBackoff(Duration.ofMillis(1)) + .maxBackoff(Duration.ofMillis(10)) + .retryOn(error -> false) + .build(); + + List responses = + toolkit.callTools( + List.of(toolCall("call-never-retry", "never_retry")), + config, + null, + null) + .block(TIMEOUT); + + assertEquals(1, responses.size()); + assertEquals(1, calls.get()); + assertEquals(ToolResultState.ERROR, responses.get(0).getState()); + } + + @Test + @DisplayName("Should exhaust retries and return an error result with id and name") + void shouldExhaustRetriesWithErrorResult() { + AtomicInteger calls = new AtomicInteger(0); + toolkit.registerTool( + new AgentTool() { + @Override + public String getName() { + return "always_fail"; + } + + @Override + public String getDescription() { + return "Tool that always fails"; + } + + @Override + public Map getParameters() { + return emptySchema(); + } + + @Override + public Mono callAsync(ToolCallParam param) { + calls.incrementAndGet(); + return Mono.error(new IOException("Always failing")); + } + }); + + List responses = + toolkit.callTools( + List.of(toolCall("call-always-fail", "always_fail")), + retryConfig(3), + null, + null) + .block(TIMEOUT); + + assertEquals(1, responses.size()); + assertEquals(3, calls.get()); + ToolResultBlock result = responses.get(0); + assertEquals(ToolResultState.ERROR, result.getState()); + assertEquals("call-always-fail", result.getId()); + assertEquals("always_fail", result.getName()); + assertTrue(extractFirstText(result).contains("Always failing")); + } + + @Test + @DisplayName("Should retry tool execution timeouts when retryOn matches") + void shouldRetryTimeouts() { + AtomicInteger calls = new AtomicInteger(0); + toolkit.registerTool( + new AgentTool() { + @Override + public String getName() { + return "never_tool"; + } + + @Override + public String getDescription() { + return "Tool that never completes"; + } + + @Override + public Map getParameters() { + return emptySchema(); + } + + @Override + public Mono callAsync(ToolCallParam param) { + calls.incrementAndGet(); + return Mono.never(); + } + }); + + ExecutionConfig config = + ExecutionConfig.builder() + .timeout(Duration.ofMillis(100)) + .maxAttempts(2) + .initialBackoff(Duration.ofMillis(1)) + .maxBackoff(Duration.ofMillis(10)) + .retryOn(ExecutionConfig.RETRYABLE_ERRORS) + .build(); + + List responses = + toolkit.callTools(List.of(toolCall("call-never", "never_tool")), config, null, null) + .block(TIMEOUT); + + assertEquals(1, responses.size()); + assertEquals(2, calls.get()); + assertTrue(extractFirstText(responses.get(0)).contains("timeout")); + } + + @Test + @DisplayName("Should isolate retry exhaustion in parallel batches") + void shouldIsolateRetryExhaustionInParallelBatches() { + AtomicInteger calls = new AtomicInteger(0); + toolkit.registerTool( + new AgentTool() { + @Override + public String getName() { + return "parallel_fail"; + } + + @Override + public String getDescription() { + return "Tool that always fails in parallel batches"; + } + + @Override + public Map getParameters() { + return emptySchema(); + } + + @Override + public Mono callAsync(ToolCallParam param) { + calls.incrementAndGet(); + return Mono.error(new IOException("Always failing")); + } + }); + + Map addInput = Map.of("a", 1, "b", 2); + List responses = + toolkit.callTools( + List.of( + toolCall("call-parallel-fail", "parallel_fail"), + toolCall("call-parallel-add", "add", addInput)), + retryConfig(2), + null, + null) + .block(TIMEOUT); + + assertEquals(2, responses.size()); + Map byId = + responses.stream() + .collect(Collectors.toMap(ToolResultBlock::getId, Function.identity())); + assertEquals(2, calls.get()); + assertEquals(ToolResultState.ERROR, byId.get("call-parallel-fail").getState()); + assertEquals("3", extractFirstText(byId.get("call-parallel-add"))); + } + + @Test + @DisplayName("Should not retry external tool suspensions") + void shouldNotRetryExternalToolSuspensions() { + toolkit.registerSchema( + ToolSchema.builder() + .name("external_retry") + .description("External tool with retry config") + .parameters( + Map.of( + "type", + "object", + "properties", + Map.of("endpoint", Map.of("type", "string")))) + .build()); + + List responses = + toolkit.callTools( + List.of( + toolCall( + "call-external-retry", + "external_retry", + Map.of("endpoint", "/users"))), + retryConfig(3), + null, + null) + .block(TIMEOUT); + + assertEquals(1, responses.size()); + assertTrue(responses.get(0).isSuspended()); + } + + @Test + @DisplayName("Should not retry annotation tools that suspend via ToolSuspendException") + void shouldNotRetryToolSuspensionExceptions() { + AtomicInteger calls = new AtomicInteger(0); + toolkit.registerTool( + new Object() { + @Tool(name = "suspend_tool", description = "Tool that suspends") + public String suspend() { + calls.incrementAndGet(); + throw new ToolSuspendException("awaiting external execution"); + } + }); + + List responses = + toolkit.callTools( + List.of(toolCall("call-suspend", "suspend_tool")), + retryConfig(3), + null, + null) + .block(TIMEOUT); + + assertEquals(1, responses.size()); + assertEquals(1, calls.get()); + assertTrue(responses.get(0).isSuspended()); + } + + @Test + @DisplayName("Should keep the callAsync error-result contract on direct calls") + void shouldKeepCallAsyncContractOnDirectCalls() { + // Custom tool failing with an error signal: direct calls still receive an error result + toolkit.registerTool( + new AgentTool() { + @Override + public String getName() { + return "direct_error"; + } + + @Override + public String getDescription() { + return "Tool that fails with an error signal"; + } + + @Override + public Map getParameters() { + return emptySchema(); + } + + @Override + public Mono callAsync(ToolCallParam param) { + return Mono.error(new IllegalStateException("Direct failure")); + } + }); + ToolResultBlock result = + toolkit.callTool( + ToolCallParam.builder() + .toolUseBlock(toolCall("call-direct", "direct_error")) + .build()) + .block(TIMEOUT); + assertEquals("Error: Tool execution failed: Direct failure", extractFirstText(result)); + + // Annotation tools keep converting exceptions to error results on the direct path + ToolResultBlock annotationResult = + toolkit.callTool( + ToolCallParam.builder() + .toolUseBlock( + toolCall( + "call-direct-annotation", + "error_tool", + Map.of("message", "direct"))) + .input(Map.of("message", "direct")) + .build()) + .block(TIMEOUT); + assertEquals( + "Error: Tool execution failed: Tool error: direct", + extractFirstText(annotationResult)); + + // MCP tools keep their own error formatting on the direct path + McpClientWrapper wrapper = mock(McpClientWrapper.class); + when(wrapper.getName()).thenReturn("test-client"); + McpTool mcpTool = new McpTool("direct_mcp", "Direct MCP tool", emptySchema(), wrapper); + when(wrapper.callTool(eq("direct_mcp"), any(), any())) + .thenReturn(Mono.error(new RuntimeException("Network error"))); + toolkit.registerTool(mcpTool); + ToolResultBlock mcpResult = + toolkit.callTool( + ToolCallParam.builder() + .toolUseBlock(toolCall("call-direct-mcp", "direct_mcp")) + .build()) + .block(TIMEOUT); + String mcpText = extractFirstText(mcpResult); + assertTrue(mcpText.contains("MCP tool error")); + assertTrue(mcpText.contains("Network error")); + } + + // ==================== Test Helpers ==================== + + private ToolUseBlock toolCall(String id, String name) { + return toolCall(id, name, Map.of()); + } + + private ToolUseBlock toolCall(String id, String name, Map input) { + return ToolUseBlock.builder() + .id(id) + .name(name) + .input(input) + .content(JsonUtils.getJsonCodec().toJson(input)) + .build(); + } + + private ExecutionConfig retryConfig(int maxAttempts) { + return ExecutionConfig.builder() + .maxAttempts(maxAttempts) + .initialBackoff(Duration.ofMillis(1)) + .maxBackoff(Duration.ofMillis(10)) + .build(); + } + + private Map emptySchema() { + Map schema = new HashMap<>(); + schema.put("type", "object"); + schema.put("properties", new HashMap<>()); + return schema; + } + private String extractFirstText(ToolResultBlock response) { assertTrue( ToolTestUtils.isValidToolResultBlock(response), @@ -677,4 +1231,36 @@ private String extractFirstText(ToolResultBlock response) { if (outputs.isEmpty()) return ""; return ((TextBlock) outputs.get(0)).getText(); } + + /** Annotation-based tools that fail on the first call and recover afterwards. */ + public static class FlakyTools { + + final AtomicInteger calls = new AtomicInteger(0); + + @Tool(name = "flaky_sync", description = "Throws IOException on the first call") + public String flakySync() throws IOException { + if (calls.incrementAndGet() == 1) { + throw new IOException("Transient sync failure"); + } + return "recovered"; + } + + @Tool(name = "flaky_mono", description = "Fails with IOException on the first call") + public Mono flakyMono() { + int attempt = calls.incrementAndGet(); + if (attempt == 1) { + return Mono.error(new IOException("Transient mono failure")); + } + return Mono.just("recovered"); + } + + @Tool(name = "flaky_future", description = "Fails with IOException on the first call") + public CompletableFuture flakyFuture() { + int attempt = calls.incrementAndGet(); + if (attempt == 1) { + return CompletableFuture.failedFuture(new IOException("Transient future failure")); + } + return CompletableFuture.completedFuture("recovered"); + } + } } From 557a4fa9733d70392164a8ed8faea96689fe58f2 Mon Sep 17 00:00:00 2001 From: kevinyang03 <152797417+kevinyang03@users.noreply.github.com> Date: Tue, 25 Aug 2026 00:47:32 +0800 Subject: [PATCH 2/3] style(tool): fix inspections and trim comments on tool retry changes - Remove redundant null check in ExceptionUtils.unwrapExecutionWrapper - Fix inaccessible ToolExecutor javadoc link in McpTool - Trim verbose comments to match repository style --- .../core/model/ExecutionConfig.java | 8 ++- .../io/agentscope/core/tool/AgentTool.java | 18 +++---- .../io/agentscope/core/tool/ToolExecutor.java | 50 ++++++------------- .../core/tool/ToolMethodInvoker.java | 13 ++--- .../io/agentscope/core/tool/mcp/McpTool.java | 13 ++--- .../agentscope/core/util/ExceptionUtils.java | 24 ++++----- 6 files changed, 42 insertions(+), 84 deletions(-) diff --git a/agentscope-core/src/main/java/io/agentscope/core/model/ExecutionConfig.java b/agentscope-core/src/main/java/io/agentscope/core/model/ExecutionConfig.java index cabc31691f..4f5794b298 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/model/ExecutionConfig.java +++ b/agentscope-core/src/main/java/io/agentscope/core/model/ExecutionConfig.java @@ -161,11 +161,9 @@ private static boolean isRetryableError(Throwable error) { * * *

    Tool retry semantics: when {@code maxAttempts} is greater than 1, failures that - * surface as reactive error signals — such as exceptions thrown by the tool or its transport, - * and timeouts — are retried as decided by {@link #getRetryOn()}. Failures that a tool - * reports as a completed {@code ToolResultBlock} error result (for example MCP protocol-level - * {@code isError=true} responses or deliberately returned business errors) are never retried, - * since replaying a completed call may be unsafe for non-idempotent operations. + * surface as reactive error signals (tool or transport exceptions, timeouts) are retried as + * decided by {@link #getRetryOn()}. Failures reported as completed {@code ToolResultBlock} + * error results (e.g. MCP {@code isError=true}) are never retried. */ public static final ExecutionConfig TOOL_DEFAULTS = builder().timeout(Duration.ofMinutes(5)).maxAttempts(1).build(); diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/AgentTool.java b/agentscope-core/src/main/java/io/agentscope/core/tool/AgentTool.java index 512ce32bf3..59c6b3c9ad 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/tool/AgentTool.java +++ b/agentscope-core/src/main/java/io/agentscope/core/tool/AgentTool.java @@ -131,18 +131,12 @@ default boolean isReadOnly() { /** * Executes the tool through the framework's execution infrastructure. * - *

    This method is the channel used by {@link ToolExecutor} when running a tool call through - * the infrastructure layers (scheduling, timeout, retry, graceful-shutdown guard). Unlike - * {@link #callAsync(ToolCallParam)}, which converts failures into {@link ToolResultBlock} - * error results, this channel may surface failures as reactive error signals so that the - * infrastructure can decide whether to retry them; once retries are exhausted, the executor - * converts the final error into a {@link ToolResultBlock} error result. - * - *

    The default implementation delegates to {@link #callAsync(ToolCallParam)}, so existing - * implementations remain compatible without any changes. Built-in tools whose failures should - * be retryable (for example annotation-based tools and MCP tools) override this method to keep - * exceptions as error signals. Tools that deliberately report failures as - * {@link ToolResultBlock} error results keep them as completed results and are never retried. + *

    Used by {@link ToolExecutor} when running a tool call through the scheduling, timeout, + * retry and shutdown layers. Unlike {@link #callAsync(ToolCallParam)}, failures stay as + * reactive error signals so they can be retried; the executor converts the final error into + * an error result once retries are exhausted. The default implementation delegates to + * {@link #callAsync(ToolCallParam)}; built-in tools override it to keep exceptions as error + * signals. * * @param param The tool call parameters * @return Mono containing ToolResultBlock, or signalling an error on failure diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/ToolExecutor.java b/agentscope-core/src/main/java/io/agentscope/core/tool/ToolExecutor.java index 72831d863a..a24aea4e7c 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/tool/ToolExecutor.java +++ b/agentscope-core/src/main/java/io/agentscope/core/tool/ToolExecutor.java @@ -160,12 +160,10 @@ private void invokeChunkCallback( // ==================== Single Tool Execution ==================== /** - * Execute a single tool call without the infrastructure layers (scheduling, timeout, retry, - * shutdown guard). + * Execute a single tool call without the infrastructure layers. * - *

    This is the compatibility entry point used by {@link Toolkit#callTool(ToolCallParam)}: it - * keeps the {@link AgentTool#callAsync(ToolCallParam)} contract that failures surface as - * {@link ToolResultBlock} error results instead of reactive error signals. + *

    Compatibility entry point used by {@link Toolkit#callTool(ToolCallParam)}: failures + * surface as {@link ToolResultBlock} error results, not reactive error signals. * * @param param Tool call parameters * @return Mono containing execution result @@ -177,8 +175,7 @@ Mono execute(ToolCallParam param) { /** * Execute a single tool call through the raw execution channel, keeping failures as reactive - * error signals for the infrastructure layers (scheduling, timeout, retry, shutdown guard) to - * act on. Used only by {@link #executeWithInfrastructure}. + * error signals. Used only by {@link #executeWithInfrastructure}. * * @param param Tool call parameters * @return Mono containing execution result, or signalling an error on failure @@ -192,9 +189,8 @@ private Mono executeRaw(ToolCallParam param) { * * @param param Tool call parameters * @param useExecutionPath whether to use the raw execution channel - * ({@link AgentTool#callAsyncForExecution(ToolCallParam)}) that keeps failures as error - * signals, or the compatibility channel ({@link AgentTool#callAsync(ToolCallParam)}) that - * converts failures into error results + * ({@link AgentTool#callAsyncForExecution(ToolCallParam)}) instead of the compatibility + * channel ({@link AgentTool#callAsync(ToolCallParam)}) * @return Mono containing execution result */ private Mono executeWithTracing( @@ -298,32 +294,20 @@ private Mono executeCore(ToolCallParam param, boolean useExecut .emitter(toolEmitter) .build(); - // Invoke the tool through the selected channel. - // - // Compatibility channel (useExecutionPath=false): failures are converted into - // ToolResultBlock error results right here, preserving the AgentTool contract for callers - // that execute tools directly without infrastructure. - // - // Execution channel (useExecutionPath=true): failures stay as reactive error signals and - // flow through the timeout/retry/shutdown layers applied by executeWithInfrastructure, - // which converts them into ToolResultBlock error results only after retries are - // exhausted. Mono.defer re-invokes callAsyncForExecution on every subscription so each - // retry attempt rebuilds and re-runs the tool call instead of resubscribing to the same - // pre-assembled publisher. + // Compatibility channel converts failures to error results here; the execution channel + // keeps them as error signals for the timeout/retry layers. Mono.defer re-invokes the + // tool per subscription so each retry runs a fresh attempt. Mono invocation; if (useExecutionPath) { invocation = Mono.defer(() -> tool.callAsyncForExecution(executionParam)) - // Unwrap reflection/future wrappers so the retry predicate sees the - // original exception (IOException, transport errors, ...) instead of - // InvocationTargetException or ExecutionException. + // Unwrap reflection/future wrappers for the retry predicate .onErrorMap(ExceptionUtils::unwrapExecutionWrapper); } else { invocation = tool.callAsync(executionParam); } - // ToolSuspendException is business-level suspension, not a failure: both channels convert - // it to a suspended result without retrying. + // Suspension is not a failure: convert to a suspended result Mono chain = invocation.onErrorResume( ToolSuspendException.class, @@ -447,9 +431,8 @@ private Mono executeWithInfrastructure( .runtimeContext(agentRuntimeContext) .build(); - // Use the raw execution channel so tool failures remain reactive error signals while - // the timeout/retry/shutdown layers below run; this final onErrorResume converts the - // error that remains after retries are exhausted into a ToolResultBlock error result. + // Keep failures as error signals for the layers below; convert what remains after + // retries are exhausted into an error result. Mono execution = executeRaw(param); // Apply infrastructure layers @@ -489,8 +472,7 @@ private Mono applyTimeout( Duration timeout = config.getTimeout(); logger.debug("Applied timeout: {} for tool: {}", timeout, toolCall.getName()); - // TimeoutException (rather than a bare RuntimeException) so that - // ExecutionConfig.RETRYABLE_ERRORS recognizes tool timeouts as retryable. + // TimeoutException so ExecutionConfig.RETRYABLE_ERRORS recognizes timeouts as retryable return execution.timeout( timeout, Mono.error(new TimeoutException("Tool execution timeout after " + timeout))); @@ -517,9 +499,7 @@ private Mono applyRetry( .maxBackoff(maxBackoff) .jitter(0.5) .filter(retryOn) - // On exhaustion, propagate the last failure itself instead of Reactor's - // RetryExhaustedException wrapper so the error result surfaces the real - // failure message (e.g. the timeout or the tool exception). + // Propagate the last failure instead of RetryExhaustedException .onRetryExhaustedThrow((spec, signal) -> signal.failure()) .doBeforeRetry( signal -> diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/ToolMethodInvoker.java b/agentscope-core/src/main/java/io/agentscope/core/tool/ToolMethodInvoker.java index fd78b9bfb3..f2fef4381d 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/tool/ToolMethodInvoker.java +++ b/agentscope-core/src/main/java/io/agentscope/core/tool/ToolMethodInvoker.java @@ -46,9 +46,7 @@ class ToolMethodInvoker { /** * Invoke tool method asynchronously with custom converter support. * - *

    Failures are converted into {@link ToolResultBlock} error results, so this entry point - * never signals an error to its caller. Direct callers of {@code AgentTool.callAsync} rely on - * this contract. + *

    Failures are converted into {@link ToolResultBlock} error results. * * @param toolObject the object containing the method * @param method the method to invoke @@ -68,12 +66,9 @@ Mono invokeAsync( /** * Invoke tool method asynchronously while keeping failures as reactive error signals. * - *

    This is the raw channel used by {@link ToolExecutor}'s execution infrastructure: errors - * stay as error signals so the timeout, retry and graceful-shutdown layers can act on them, - * and the executor converts them into {@link ToolResultBlock} error results only after the - * retry decision has been made. Parameter injection, argument conversion, reflection - * invocation, {@link CompletableFuture}/{@link Mono} adaptation and result conversion behave - * identically to {@link #invokeAsync}; only the error-to-result conversion is omitted. + *

    Raw channel used by {@link ToolExecutor}'s execution infrastructure: only the + * error-to-result conversion of {@link #invokeAsync} is omitted, so the timeout and retry + * layers can act on failures. * * @param toolObject the object containing the method * @param method the method to invoke diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpTool.java b/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpTool.java index 4cb75d341f..ca34948910 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpTool.java +++ b/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpTool.java @@ -194,19 +194,14 @@ public Mono callAsync(ToolCallParam param) { /** * Executes this MCP tool through the framework's execution infrastructure. * - *

    Transport-level failures (connection loss, network interruption, timeout, client - * exceptions) stay as reactive error signals so {@link io.agentscope.core.tool.ToolExecutor} - * can apply the configured retry policy. Protocol-level business errors (the MCP server - * completing the call with {@code isError=true}) remain {@link ToolResultBlock} error results - * and are never retried, since replaying a completed non-idempotent call is unsafe. + *

    Transport-level failures stay as reactive error signals so the executor can apply its + * retry policy. Protocol-level business errors ({@code isError=true}) remain completed error + * results and are never retried. */ @Override public Mono callAsyncForExecution(ToolCallParam param) { - // Keep the historical contract that invoking with a null param fails fast. Objects.requireNonNull(param, "param must not be null"); - // Mono.defer re-creates the whole attempt (argument merging, meta extraction and the - // remote call) for every subscription, so each retry runs a fresh attempt and synchronous - // failures during preparation also surface as reactive errors. + // Mono.defer re-runs the whole attempt per subscription so each retry is a fresh attempt return Mono.defer( () -> { logger.debug( diff --git a/agentscope-core/src/main/java/io/agentscope/core/util/ExceptionUtils.java b/agentscope-core/src/main/java/io/agentscope/core/util/ExceptionUtils.java index 321e05b7dc..8a263a0412 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/util/ExceptionUtils.java +++ b/agentscope-core/src/main/java/io/agentscope/core/util/ExceptionUtils.java @@ -84,26 +84,22 @@ public static boolean containsInterruptedException(Throwable error) { } /** - * Unwraps reflective and future execution wrappers to expose the original failure. - * - *

    {@code method.invoke} wraps business exceptions in {@link InvocationTargetException}, - * and failed {@link java.util.concurrent.CompletableFuture}s surface as {@link - * ExecutionException} or {@link CompletionException}. This method walks those wrappers (whose - * cause chains cannot cycle) so retry predicates such as {@code - * ExecutionConfig.RETRYABLE_ERRORS} can inspect the original {@code IOException}, {@code - * HttpTransportException} or domain exception. + * Unwraps {@link InvocationTargetException}, {@link ExecutionException} and + * {@link CompletionException} wrappers to expose the original failure. * * @param error the throwable to unwrap (may be {@code null}) * @return the innermost non-wrapper throwable, or {@code null} if the input is {@code null} */ public static Throwable unwrapExecutionWrapper(Throwable error) { Throwable current = error; - while (current != null - && (current instanceof InvocationTargetException - || current instanceof ExecutionException - || current instanceof CompletionException) - && current.getCause() != null) { - current = current.getCause(); + while (current instanceof InvocationTargetException + || current instanceof ExecutionException + || current instanceof CompletionException) { + Throwable cause = current.getCause(); + if (cause == null) { + break; + } + current = cause; } return current; } From fbad5bd652439ca918b4dac472998c47951c0fd7 Mon Sep 17 00:00:00 2001 From: kevinyang03 <152797417+kevinyang03@users.noreply.github.com> Date: Tue, 25 Aug 2026 01:31:33 +0800 Subject: [PATCH 3/3] fix(tool): detect wrapped ToolSuspendException before retrying - Search the full cause chain (identity-guarded) for ToolSuspendException so suspensions hidden behind arbitrary wrappers are never retried - Drop the redundant Mono.defer in McpTool.callAsyncForExecution and the unreachable external-tool branch in ReflectiveFunctionTool - Document the execution-channel contract on AgentTool and clarify TOOL_DEFAULTS retry semantics when no retry predicate is configured - Add regression tests for Mono/Future/wrapped suspensions and for RETRYABLE_ERRORS not retrying deterministic failures --- .../core/model/ExecutionConfig.java | 7 +- .../io/agentscope/core/tool/AgentTool.java | 16 +- .../core/tool/ReflectiveFunctionTool.java | 4 - .../io/agentscope/core/tool/ToolExecutor.java | 23 +-- .../io/agentscope/core/tool/mcp/McpTool.java | 31 ++-- .../agentscope/core/util/ExceptionUtils.java | 22 +++ .../core/tool/ToolExecutorTest.java | 151 ++++++++++++++++++ 7 files changed, 213 insertions(+), 41 deletions(-) diff --git a/agentscope-core/src/main/java/io/agentscope/core/model/ExecutionConfig.java b/agentscope-core/src/main/java/io/agentscope/core/model/ExecutionConfig.java index 4f5794b298..aa1f357400 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/model/ExecutionConfig.java +++ b/agentscope-core/src/main/java/io/agentscope/core/model/ExecutionConfig.java @@ -160,10 +160,9 @@ private static boolean isRetryableError(Throwable error) { *

  • Max attempts: 1 (no retry) * * - *

    Tool retry semantics: when {@code maxAttempts} is greater than 1, failures that - * surface as reactive error signals (tool or transport exceptions, timeouts) are retried as - * decided by {@link #getRetryOn()}. Failures reported as completed {@code ToolResultBlock} - * error results (e.g. MCP {@code isError=true}) are never retried. + *

    When {@code maxAttempts} is greater than 1, failures that surface as reactive error + * signals are retried (every failure when no retry predicate is configured); failures + * reported as completed {@code ToolResultBlock} error results are never retried. */ public static final ExecutionConfig TOOL_DEFAULTS = builder().timeout(Duration.ofMinutes(5)).maxAttempts(1).build(); diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/AgentTool.java b/agentscope-core/src/main/java/io/agentscope/core/tool/AgentTool.java index 59c6b3c9ad..74b48b6d5b 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/tool/AgentTool.java +++ b/agentscope-core/src/main/java/io/agentscope/core/tool/AgentTool.java @@ -131,12 +131,16 @@ default boolean isReadOnly() { /** * Executes the tool through the framework's execution infrastructure. * - *

    Used by {@link ToolExecutor} when running a tool call through the scheduling, timeout, - * retry and shutdown layers. Unlike {@link #callAsync(ToolCallParam)}, failures stay as - * reactive error signals so they can be retried; the executor converts the final error into - * an error result once retries are exhausted. The default implementation delegates to - * {@link #callAsync(ToolCallParam)}; built-in tools override it to keep exceptions as error - * signals. + *

    Called by {@link ToolExecutor} when running a tool call through the scheduling, timeout, + * retry and shutdown layers; application code should keep using + * {@link #callAsync(ToolCallParam)}. Unlike {@link #callAsync(ToolCallParam)}, failures stay + * as reactive error signals so they can be retried; the executor converts the final error + * into an error result once retries are exhausted. + * + *

    Each subscription must be a fresh attempt, business failures must stay completed + * {@link ToolResultBlock} error results, and {@link ToolSuspendException} must propagate as + * an error signal. The default implementation delegates to {@link #callAsync(ToolCallParam)}; + * built-in tools override it to keep exceptions as error signals. * * @param param The tool call parameters * @return Mono containing ToolResultBlock, or signalling an error on failure diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/ReflectiveFunctionTool.java b/agentscope-core/src/main/java/io/agentscope/core/tool/ReflectiveFunctionTool.java index a8f034586b..0962445f3b 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/tool/ReflectiveFunctionTool.java +++ b/agentscope-core/src/main/java/io/agentscope/core/tool/ReflectiveFunctionTool.java @@ -167,10 +167,6 @@ public Mono callAsync(ToolCallParam param) { @Override public Mono callAsyncForExecution(ToolCallParam param) { - if (isExternalTool()) { - return Mono.just( - ToolResultBlock.suspended(param.getToolUseBlock(), new ToolSuspendException())); - } return methodInvoker.invokeRawAsync(toolObject, method, param, customConverter); } diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/ToolExecutor.java b/agentscope-core/src/main/java/io/agentscope/core/tool/ToolExecutor.java index a24aea4e7c..5b87c90dbd 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/tool/ToolExecutor.java +++ b/agentscope-core/src/main/java/io/agentscope/core/tool/ToolExecutor.java @@ -191,7 +191,7 @@ private Mono executeRaw(ToolCallParam param) { * @param useExecutionPath whether to use the raw execution channel * ({@link AgentTool#callAsyncForExecution(ToolCallParam)}) instead of the compatibility * channel ({@link AgentTool#callAsync(ToolCallParam)}) - * @return Mono containing execution result + * @return Mono containing execution result, or signalling an error on failure */ private Mono executeWithTracing( ToolCallParam param, boolean useExecutionPath) { @@ -307,17 +307,22 @@ private Mono executeCore(ToolCallParam param, boolean useExecut invocation = tool.callAsync(executionParam); } - // Suspension is not a failure: convert to a suspended result Mono chain = invocation.onErrorResume( - ToolSuspendException.class, e -> { - // Convert ToolSuspendException to suspended result - logger.debug( - "Tool '{}' suspended: {}", - toolCall.getName(), - e.getReason() != null ? e.getReason() : "no reason"); - return Mono.just(ToolResultBlock.suspended(toolCall, e)); + // Convert any wrapped ToolSuspendException to a suspended result + ToolSuspendException suspended = + ExceptionUtils.findToolSuspendException(e); + if (suspended != null) { + logger.debug( + "Tool '{}' suspended: {}", + toolCall.getName(), + suspended.getReason() != null + ? suspended.getReason() + : "no reason"); + return Mono.just(ToolResultBlock.suspended(toolCall, suspended)); + } + return Mono.error(e); }); if (!useExecutionPath) { chain = chain.onErrorResume(e -> Mono.just(buildToolErrorResult(e))); diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpTool.java b/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpTool.java index ca34948910..dcb6e19110 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpTool.java +++ b/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpTool.java @@ -197,31 +197,26 @@ public Mono callAsync(ToolCallParam param) { *

    Transport-level failures stay as reactive error signals so the executor can apply its * retry policy. Protocol-level business errors ({@code isError=true}) remain completed error * results and are never retried. + * + * @param param The tool call parameters containing toolUseBlock, input, and agent + * @return a Mono that emits the tool result, or signals an error on failure */ @Override public Mono callAsyncForExecution(ToolCallParam param) { Objects.requireNonNull(param, "param must not be null"); - // Mono.defer re-runs the whole attempt per subscription so each retry is a fresh attempt - return Mono.defer( - () -> { - logger.debug( - "Calling MCP tool '{}' with input: {}", getName(), param.getInput()); + logger.debug("Calling MCP tool '{}' with input: {}", getName(), param.getInput()); - // Merge preset arguments with input arguments - Map mergedArgs = mergeArguments(param.getInput()); + // Merge preset arguments with input arguments + Map mergedArgs = mergeArguments(param.getInput()); - // Extract MCP meta from ContextStore by McpMeta type namespace - Map metaMap = extractMcpMeta(param); + // Extract MCP meta from ContextStore by McpMeta type namespace + Map metaMap = extractMcpMeta(param); - return clientWrapper - .callTool(getName(), mergedArgs, metaMap) - .map(McpContentConverter::convertCallToolResult) - .doOnSuccess( - result -> - logger.debug( - "MCP tool '{}' completed successfully", - getName())); - }); + return clientWrapper + .callTool(getName(), mergedArgs, metaMap) + .map(McpContentConverter::convertCallToolResult) + .doOnSuccess( + result -> logger.debug("MCP tool '{}' completed successfully", getName())); } /** diff --git a/agentscope-core/src/main/java/io/agentscope/core/util/ExceptionUtils.java b/agentscope-core/src/main/java/io/agentscope/core/util/ExceptionUtils.java index 8a263a0412..272f995f9a 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/util/ExceptionUtils.java +++ b/agentscope-core/src/main/java/io/agentscope/core/util/ExceptionUtils.java @@ -15,6 +15,7 @@ */ package io.agentscope.core.util; +import io.agentscope.core.tool.ToolSuspendException; import java.lang.reflect.InvocationTargetException; import java.util.IdentityHashMap; import java.util.concurrent.CompletionException; @@ -103,4 +104,25 @@ public static Throwable unwrapExecutionWrapper(Throwable error) { } return current; } + + /** + * Finds a {@link ToolSuspendException} anywhere in the given throwable's cause chain. + * + *

    The cause chain is walked with an identity set guard so circular causes cannot cause an + * infinite loop. + * + * @param error the throwable to inspect (may be {@code null}) + * @return the first {@link ToolSuspendException} found, or {@code null} if none is present + */ + public static ToolSuspendException findToolSuspendException(Throwable error) { + IdentityHashMap visited = new IdentityHashMap<>(); + Throwable current = error; + while (current != null && visited.put(current, Boolean.TRUE) == null) { + if (current instanceof ToolSuspendException suspendException) { + return suspendException; + } + current = current.getCause(); + } + return null; + } } diff --git a/agentscope-core/src/test/java/io/agentscope/core/tool/ToolExecutorTest.java b/agentscope-core/src/test/java/io/agentscope/core/tool/ToolExecutorTest.java index bf37e13db3..da8c048885 100644 --- a/agentscope-core/src/test/java/io/agentscope/core/tool/ToolExecutorTest.java +++ b/agentscope-core/src/test/java/io/agentscope/core/tool/ToolExecutorTest.java @@ -1125,6 +1125,157 @@ public String suspend() { assertTrue(responses.get(0).isSuspended()); } + @Test + @DisplayName("Should not retry Mono tools that suspend via ToolSuspendException") + void shouldNotRetryMonoToolSuspension() { + AtomicInteger calls = new AtomicInteger(0); + toolkit.registerTool( + new Object() { + @Tool(name = "suspend_mono", description = "Mono tool that suspends") + public Mono suspend() { + calls.incrementAndGet(); + return Mono.error(new ToolSuspendException("awaiting external execution")); + } + }); + + List responses = + toolkit.callTools( + List.of(toolCall("call-suspend-mono", "suspend_mono")), + retryConfig(3), + null, + null) + .block(TIMEOUT); + + assertEquals(1, responses.size()); + assertEquals(1, calls.get()); + assertTrue(responses.get(0).isSuspended()); + } + + @Test + @DisplayName("Should not retry CompletableFuture tools that suspend via ToolSuspendException") + void shouldNotRetryFutureToolSuspension() { + AtomicInteger calls = new AtomicInteger(0); + toolkit.registerTool( + new Object() { + @Tool(name = "suspend_future", description = "Future tool that suspends") + public CompletableFuture suspend() { + calls.incrementAndGet(); + return CompletableFuture.failedFuture( + new ToolSuspendException("awaiting external execution")); + } + }); + + List responses = + toolkit.callTools( + List.of(toolCall("call-suspend-future", "suspend_future")), + retryConfig(3), + null, + null) + .block(TIMEOUT); + + assertEquals(1, responses.size()); + assertEquals(1, calls.get()); + assertTrue(responses.get(0).isSuspended()); + } + + @Test + @DisplayName("Should not retry ToolSuspendException wrapped in a non-standard exception") + void shouldNotRetryWrappedToolSuspension() { + AtomicInteger calls = new AtomicInteger(0); + toolkit.registerTool( + new AgentTool() { + @Override + public String getName() { + return "wrapped_suspend"; + } + + @Override + public String getDescription() { + return "Tool that suspends behind a custom wrapper"; + } + + @Override + public Map getParameters() { + return emptySchema(); + } + + @Override + public Mono callAsync(ToolCallParam param) { + return Mono.just(ToolResultBlock.error("unused")); + } + + @Override + public Mono callAsyncForExecution(ToolCallParam param) { + calls.incrementAndGet(); + return Mono.error( + new IllegalStateException( + "wrapped", + new ToolSuspendException("awaiting external execution"))); + } + }); + + List responses = + toolkit.callTools( + List.of(toolCall("call-wrapped-suspend", "wrapped_suspend")), + retryConfig(3), + null, + null) + .block(TIMEOUT); + + assertEquals(1, responses.size()); + assertEquals(1, calls.get()); + assertTrue(responses.get(0).isSuspended()); + } + + @Test + @DisplayName("Should not retry deterministic failures with the retryable-errors predicate") + void shouldNotRetryDeterministicFailureWithRetryableErrors() { + AtomicInteger calls = new AtomicInteger(0); + toolkit.registerTool( + new AgentTool() { + @Override + public String getName() { + return "misconfigured_tool"; + } + + @Override + public String getDescription() { + return "Tool that fails deterministically"; + } + + @Override + public Map getParameters() { + return emptySchema(); + } + + @Override + public Mono callAsync(ToolCallParam param) { + calls.incrementAndGet(); + return Mono.error(new IllegalStateException("Not initialized")); + } + }); + + ExecutionConfig config = + ExecutionConfig.builder() + .maxAttempts(3) + .initialBackoff(Duration.ofMillis(1)) + .maxBackoff(Duration.ofMillis(10)) + .retryOn(ExecutionConfig.RETRYABLE_ERRORS) + .build(); + + List responses = + toolkit.callTools( + List.of(toolCall("call-misconfigured", "misconfigured_tool")), + config, + null, + null) + .block(TIMEOUT); + + assertEquals(1, responses.size()); + assertEquals(1, calls.get()); + assertEquals(ToolResultState.ERROR, responses.get(0).getState()); + } + @Test @DisplayName("Should keep the callAsync error-result contract on direct calls") void shouldKeepCallAsyncContractOnDirectCalls() {