From 4ef57103feeb906ff90ba3fb24508cca5fe53176 Mon Sep 17 00:00:00 2001 From: Buktal Date: Mon, 27 Jul 2026 17:23:09 +0800 Subject: [PATCH] fix(core): reconcile dangling tool_use on interrupt before persisting AgentState When an interrupt is signalled between reasoning (which writes the assistant tool_use) and acting (which would produce the tool result), handleInterrupt appended its recovery message without reconciling the pending tool call. The persisted AgentState ended up with a ToolUseBlock that had no matching ToolResultBlock, immediately followed by the recovery message, so the next resumed run inherited an inconsistent context and providers (e.g. MiniMax-M3) returned an empty response. maybePatchPendingToolCalls already existed to synthesize error results, but getPendingToolUseIds only inspects the last assistant message, so once the recovery message became the last assistant message the patch no longer detected the pending call. Fix: in handleInterrupt, synthesize error ToolResultBlocks for pending tool calls before appending the recovery message, reusing the existing getPendingToolUseIds/findLastAssistantMsg/buildErrorToolResult helpers. Unconditional (not gated by enablePendingToolRecovery, which defaults to false) because the dangling tool_use is produced by the framework's own interrupt path and otherwise always corrupts the next resumed run. Closes #2409 --- .../java/io/agentscope/core/ReActAgent.java | 48 ++++ .../ReActAgentInterruptToolCallTest.java | 216 ++++++++++++++++++ 2 files changed, 264 insertions(+) create mode 100644 agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentInterruptToolCallTest.java 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 003bbfeb15..2c04ab53ab 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java +++ b/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java @@ -1749,6 +1749,50 @@ private void maybePatchPendingToolCalls(List msgs) { } } + /** + * Synthesize error {@link ToolResultBlock}s for pending tool calls in the last assistant + * message that have no matching results. + * + *

Called from {@link ReActAgent#handleInterrupt} before the recovery message is + * appended, so that an interrupt signalled between reasoning (which writes the assistant + * {@code tool_use}) and acting (which would produce the tool results) does not persist a + * {@link ToolUseBlock} with no matching {@link ToolResultBlock}. Without this, the next + * resumed run inherits an inconsistent context and providers may return an empty response. + * + *

Must run before the recovery message is added, otherwise that recovery + * message becomes the last assistant message and {@link #getPendingToolUseIds()} no longer + * detects the pending calls. + */ + private void synthesizeErrorResultsForPendingToolCalls() { + Set pendingIds = getPendingToolUseIds(); + if (pendingIds.isEmpty()) { + return; + } + Msg lastAssistant = findLastAssistantMsg(); + if (lastAssistant == null) { + return; + } + List pendingToolCalls = + lastAssistant.getContentBlocks(ToolUseBlock.class).stream() + .filter(toolUse -> pendingIds.contains(toolUse.getId())) + .toList(); + for (ToolUseBlock toolCall : pendingToolCalls) { + ToolResultBlock errorResult = + buildErrorToolResult( + toolCall.getId(), + "Tool execution was interrupted before it could run. Tool: " + + toolCall.getName()); + Msg toolResultMsg = + ToolResultMessageBuilder.buildToolResultMsg( + errorResult, toolCall, getName()); + state.contextMutable().add(toolResultMsg); + log.info( + "Synthesized interrupted-result for pending tool call: {} ({})", + toolCall.getName(), + toolCall.getId()); + } + } + private void publishEvent(AgentEvent event) { FluxSink sink = eventSink; if (sink != null) { @@ -3562,6 +3606,10 @@ protected Mono handleInterrupt(InterruptContext context, Msg... originalArg shutdownManager.saveOnInterruptObserved(requestId); return Mono.error(new AgentShuttingDownException()); } + // Reconcile any pending tool calls before appending the recovery message. + // The tool_use written during reasoning would otherwise be persisted with no + // matching tool result, leaving the AgentState inconsistent for the next run. + scope.synthesizeErrorResultsForPendingToolCalls(); String recoveryText = "I noticed that you have interrupted me. What can I do for you?"; Msg recoveryMsg = diff --git a/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentInterruptToolCallTest.java b/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentInterruptToolCallTest.java new file mode 100644 index 0000000000..95c3dfe41c --- /dev/null +++ b/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentInterruptToolCallTest.java @@ -0,0 +1,216 @@ +/* + * 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.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.agentscope.core.ReActAgent; +import io.agentscope.core.event.AgentEvent; +import io.agentscope.core.event.ModelCallEndEvent; +import io.agentscope.core.message.ContentBlock; +import io.agentscope.core.message.GenerateReason; +import io.agentscope.core.message.Msg; +import io.agentscope.core.message.MsgRole; +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.middleware.MiddlewareBase; +import io.agentscope.core.middleware.ModelCallInput; +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.tool.Toolkit; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; +import java.util.function.Supplier; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; + +/** + * Tests for {@link ReActAgent#handleInterrupt} reconciling a dangling {@link ToolUseBlock} when an + * interrupt is signalled between reasoning (tool_use emitted) and acting (tool executed). See + * issue #2409. + */ +class ReActAgentInterruptToolCallTest { + + private static final class ScriptedModel extends ChatModelBase { + private final List>> scripts; + private final AtomicInteger idx = new AtomicInteger(); + + ScriptedModel(List>> scripts) { + this.scripts = scripts; + } + + @Override + public String getModelName() { + return "scripted"; + } + + @Override + protected Flux doStream( + List messages, List tools, GenerateOptions options) { + int i = idx.getAndIncrement(); + if (i >= scripts.size()) { + return Flux.just(textResponse("")); + } + return scripts.get(i).get(); + } + } + + private static ChatResponse textResponse(String text) { + return ChatResponse.builder() + .content(List.of(TextBlock.builder().text(text).build())) + .build(); + } + + private static ChatResponse toolUseResponse(String toolId, String toolName) { + Map input = new HashMap<>(); + input.put("query", "x"); + return ChatResponse.builder() + .content( + List.of( + ToolUseBlock.builder() + .id(toolId) + .name(toolName) + .input(input) + .build())) + .build(); + } + + /** + * Interrupts the agent right after the first model call completes, simulating a post-generation + * resource check (e.g. quota/budget exhaustion): the model has emitted a {@link ToolUseBlock}, + * but the agent is interrupted before acting executes it. + */ + private static final class InterruptAfterModelCallMiddleware implements MiddlewareBase { + private final AtomicBoolean fired = new AtomicBoolean(); + + @Override + public Flux onModelCall( + Agent agent, + RuntimeContext ctx, + ModelCallInput input, + Function> next) { + return next.apply(input) + .concatMap( + event -> { + if (event instanceof ModelCallEndEvent + && fired.compareAndSet(false, true)) { + if (agent instanceof ReActAgent r) { + r.interrupt(ctx); + } else { + agent.interrupt(); + } + } + return Flux.just(event); + }); + } + } + + private static ReActAgent buildAgent(ChatModelBase model, MiddlewareBase... middlewares) { + return ReActAgent.builder() + .name("asst") + .model(model) + .toolkit(new Toolkit()) + .middlewares(List.of(middlewares)) + .build(); + } + + private static Msg userMsg(String text) { + return Msg.builder() + .name("user") + .role(MsgRole.USER) + .content(TextBlock.builder().text(text).build()) + .build(); + } + + @Test + void interruptAfterToolUseSynthesizesErrorResultAndKeepsContextConsistent() { + ChatModelBase model = + new ScriptedModel( + List.of( + () -> Flux.just(toolUseResponse("tc1", "search")), + () -> Flux.just(textResponse("resumed")))); + ReActAgent agent = buildAgent(model, new InterruptAfterModelCallMiddleware()); + + Msg result = agent.call(List.of(userMsg("do something"))).block(); + assertNotNull(result); + assertEquals(GenerateReason.INTERRUPTED, result.getGenerateReason()); + + List context = agent.getAgentState().getContext(); + List toolUses = + context.stream() + .flatMap(m -> m.getContentBlocks(ToolUseBlock.class).stream()) + .toList(); + List toolResults = + context.stream() + .flatMap(m -> m.getContentBlocks(ToolResultBlock.class).stream()) + .toList(); + + assertEquals(1, toolUses.size(), "one tool_use should have been emitted"); + assertEquals( + 1, + toolResults.size(), + "an error result must be synthesized for the pending tool_use"); + assertEquals( + toolUses.get(0).getId(), + toolResults.get(0).getId(), + "synthesized result must match the pending tool_use id"); + assertEquals( + ToolResultState.ERROR, + toolResults.get(0).getState(), + "synthesized result must be ERROR"); + + Msg lastAssistant = null; + for (int i = context.size() - 1; i >= 0; i--) { + if (context.get(i).getRole() == MsgRole.ASSISTANT) { + lastAssistant = context.get(i); + break; + } + } + assertNotNull(lastAssistant, "a recovery assistant message must exist"); + assertTrue( + lastAssistant.getContentBlocks(ToolUseBlock.class).isEmpty(), + "the last assistant message must be the recovery message, not the tool_use"); + assertEquals(GenerateReason.INTERRUPTED, lastAssistant.getGenerateReason()); + } + + @Test + void resumedRunAfterInterruptRepliesNormally() { + ChatModelBase model = + new ScriptedModel( + List.of( + () -> Flux.just(toolUseResponse("tc1", "search")), + () -> Flux.just(textResponse("resumed-ok")))); + ReActAgent agent = buildAgent(model, new InterruptAfterModelCallMiddleware()); + + Msg first = agent.call(List.of(userMsg("do something"))).block(); + assertNotNull(first); + assertEquals(GenerateReason.INTERRUPTED, first.getGenerateReason()); + + Msg second = agent.call(List.of(userMsg("continue"))).block(); + assertNotNull(second); + assertEquals("resumed-ok", second.getTextContent()); + } +}