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 c016735a78..794d6a0f02 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java +++ b/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java @@ -756,6 +756,12 @@ protected Object beforeAgentExecution(List msgs, RuntimeContext rc) { return scope; } + @Override + protected List prepareInputMessages(List msgs, Object callScope) { + CallExecution scope = (CallExecution) callScope; + return scope.selectReplayedMessages(msgs); + } + @Override protected Mono seedSystemMsg(Object callExectution) { RuntimeContext rc = @@ -1757,20 +1763,24 @@ private Mono doCallInner(List msgs) { Set pendingIds = getPendingToolUseIds(); - // No pending tools -> normal processing - if (pendingIds.isEmpty()) { - addToContext(msgs); - return coreAgent(); - } - // Permission HITL: if any pending tool is ASKING, the caller MUST supply // ConfirmResults (via Msg.METADATA_CONFIRM_RESULTS) before we can proceed. - List asking = askingToolCalls(); + List asking = pendingIds.isEmpty() ? List.of() : askingToolCalls(); if (!asking.isEmpty()) { validateAndAcceptConfirmResults(msgs, asking); return resumeAgent(); } + // Transcript-replaying clients may resend results already consumed by this session. + // Normalize against the call-scoped state before either recovery or normal input + // handling, including calls with no pending tools. Unknown IDs remain invalid. + msgs = removeConsumedToolResults(msgs, pendingIds); + + if (pendingIds.isEmpty()) { + addToContext(msgs); + return coreAgent(); + } + // Pending-tool-call recovery: auto-patch orphaned pending tool calls with synthetic // error results so the agent can continue instead of crashing. This must happen after // the permission HITL flow so ASKING tool calls are handled by confirmation first. @@ -2050,6 +2060,133 @@ private void applyToolUseBlockReplacements(Map replacement } } + private List selectReplayedMessages(List msgs) { + if (rc == null + || !Boolean.TRUE.equals(rc.get(RuntimeContext.REPLAYED_INPUT)) + || msgs == null + || msgs.isEmpty()) { + return msgs; + } + int lastAssistant = -1; + for (int i = msgs.size() - 1; i >= 0; i--) { + if (msgs.get(i).getRole() == MsgRole.ASSISTANT) { + lastAssistant = i; + break; + } + } + if (lastAssistant < 0) { + return msgs; + } + + Set pendingIds = getPendingToolUseIds(); + List selected = new ArrayList<>(); + // Clients can split one server tool round into several local assistant turns. Do + // not let those turns hide a result for a call the server is still waiting for. + for (int i = 0; i <= lastAssistant; i++) { + Msg msg = msgs.get(i); + List results = + msg.getContent().stream() + .filter( + block -> + block instanceof ToolResultBlock result + && pendingIds.contains(result.getId())) + .toList(); + if (!results.isEmpty()) { + selected.add(msg.withContent(results)); + } + } + selected.addAll(msgs.subList(lastAssistant + 1, msgs.size())); + // Preserve regenerate/continue behavior, but never replay the old user prompt in + // place of a pending tool result found in the middle of the transcript. + if (selected.isEmpty()) { + for (int i = lastAssistant - 1; i >= 0; i--) { + if (msgs.get(i).getRole() == MsgRole.USER) { + selected.add(msgs.get(i)); + break; + } + } + } + return selected; + } + + private List removeConsumedToolResults(List msgs, Set pendingIds) { + if (msgs == null + || msgs.stream().noneMatch(m -> m.hasContentBlocks(ToolResultBlock.class))) { + return msgs; + } + + Set consumedIds = + state.contextMutable().stream() + .flatMap(m -> m.getContentBlocks(ToolResultBlock.class).stream()) + .map(ToolResultBlock::getId) + .collect(Collectors.toSet()); + Set historicalToolIds = + state.contextMutable().stream() + .filter(m -> m.getRole() == MsgRole.ASSISTANT) + .flatMap(m -> m.getContentBlocks(ToolUseBlock.class).stream()) + .map(ToolUseBlock::getId) + .collect(Collectors.toSet()); + consumedIds.retainAll(historicalToolIds); + // A stateless caller can supply complete tool-call/result history in its input. + boolean initialTranscript = state.contextMutable().isEmpty(); + Set inputToolIds = new HashSet<>(); + Set providedIds = new HashSet<>(); + Set invalidIds = new HashSet<>(); + for (Msg msg : msgs) { + if (initialTranscript && msg.getRole() == MsgRole.ASSISTANT) { + msg.getContentBlocks(ToolUseBlock.class).stream() + .map(ToolUseBlock::getId) + .forEach(inputToolIds::add); + } + for (ToolResultBlock result : msg.getContentBlocks(ToolResultBlock.class)) { + String id = result.getId(); + if (!providedIds.add(id)) { + throw new IllegalStateException("Duplicate tool result ID: " + id); + } + if (!pendingIds.contains(id) + && !consumedIds.contains(id) + && !inputToolIds.contains(id)) { + invalidIds.add(id); + } + } + } + if (!invalidIds.isEmpty()) { + throw new IllegalStateException( + "Invalid tool result IDs: " + invalidIds + ". Expected: " + pendingIds); + } + + Set replayedIds = new HashSet<>(providedIds); + replayedIds.retainAll(consumedIds); + if (!replayedIds.isEmpty()) { + log.debug("Ignoring previously consumed tool result IDs: {}", replayedIds); + } + + List cleaned = new ArrayList<>(); + for (Msg msg : msgs) { + List content = + msg.getContent().stream() + .filter( + block -> + !(block instanceof ToolResultBlock result) + || !consumedIds.contains(result.getId())) + .toList(); + if (content.size() == msg.getContent().size()) { + cleaned.add(msg); + } else if (!content.isEmpty() + || (msg.getMetadata() != null + && msg.getMetadata().containsKey(Msg.METADATA_CONFIRM_RESULTS))) { + cleaned.add(msg.withContent(content)); + } + } + if (cleaned.isEmpty()) { + throw new IllegalStateException( + "Input contains only previously consumed tool results. Provide new" + + " messages or results for pending IDs: " + + pendingIds); + } + return cleaned; + } + private void maybePatchPendingToolCalls(List msgs, Set pendingIds) { if (pendingIds.isEmpty()) { return; 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 a6b0248b6d..9c84902bdf 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 @@ -307,18 +307,19 @@ private Mono runLifecycleBody( // shutdown interrupts / saves the exact (userId, sessionId) session rather than the agent's // no-arg "most-recently-active" accessors. GracefulShutdownManager.getInstance().bindRequestState(requestId, stateForCall(scope)); + List preparedMsgs = prepareInputMessages(msgs, scope); Mono body = TracerRegistry.get() .callAgent( this, - msgs, + preparedMsgs, () -> - notifyPreCall(msgs, scope) + notifyPreCall(preparedMsgs, scope) .flatMap(doCallFn) .flatMap(this::notifyPostCall) .onErrorResume( createErrorHandler( - msgs.toArray(new Msg[0])))); + preparedMsgs.toArray(new Msg[0])))); return scope == null ? body : body.contextWrite(c -> c.put(CALL_SCOPE_KEY, scope)); } @@ -336,6 +337,19 @@ protected Object callSerializationKey(RuntimeContext rc) { return null; } + /** + * Prepares caller input after session activation and before tracing and pre-call hooks. + * Subclasses can reconcile replayed input against the call-scoped state while holding the + * session serialization gate. The default preserves the original input. + * + * @param msgs caller input + * @param callScope the scope returned by {@code beforeAgentExecution} + * @return the messages to process + */ + protected List prepareInputMessages(List msgs, Object callScope) { + return msgs; + } + /** * Serializes {@code action} against other actions sharing {@code key}: this call waits for the * previously-enqueued call with the same key to terminate before running, then becomes the tail diff --git a/agentscope-core/src/main/java/io/agentscope/core/agent/RuntimeContext.java b/agentscope-core/src/main/java/io/agentscope/core/agent/RuntimeContext.java index 69c1a71f7d..6cfd0b9755 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/agent/RuntimeContext.java +++ b/agentscope-core/src/main/java/io/agentscope/core/agent/RuntimeContext.java @@ -32,6 +32,15 @@ */ public class RuntimeContext { + /** + * Boolean attribute indicating that input contains a replayed client transcript while the + * server owns conversation history. ReActAgent selects follow-up messages after the last + * client assistant turn, plus results for currently pending calls anywhere in the transcript. + * Selection happens after the authoritative session state is loaded. Omit this attribute for + * incremental input or stateless, client-owned history. + */ + public static final String REPLAYED_INPUT = "agentscope_replayed_input"; + private static final String TYPED_DEFAULT_KEY = ""; private final String sessionId; diff --git a/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentToolResultReplayTest.java b/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentToolResultReplayTest.java new file mode 100644 index 0000000000..2c0ad9866e --- /dev/null +++ b/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentToolResultReplayTest.java @@ -0,0 +1,486 @@ +/* + * 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.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.agentscope.core.ReActAgent; +import io.agentscope.core.event.AgentEvent; +import io.agentscope.core.event.ExternalExecutionResultEvent; +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.message.ToolCallState; +import io.agentscope.core.message.ToolResultBlock; +import io.agentscope.core.message.ToolResultState; +import io.agentscope.core.message.ToolUseBlock; +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.AgentState; +import io.agentscope.core.state.InMemoryAgentStateStore; +import io.agentscope.core.tool.Toolkit; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import reactor.core.publisher.Flux; + +class ReActAgentToolResultReplayTest { + private static final RuntimeContext CONTEXT = + RuntimeContext.builder().userId("alice").sessionId("thread").build(); + + private static final class CapturingModel extends ChatModelBase { + final List> inputs = new ArrayList<>(); + + @Override + public String getModelName() { + return "replay-test"; + } + + @Override + protected Flux doStream( + List messages, List tools, GenerateOptions options) { + inputs.add(List.copyOf(messages)); + return Flux.just( + ChatResponse.builder() + .content(List.of(TextBlock.builder().text("done").build())) + .build()); + } + } + + private static ReActAgent agent(CapturingModel model, boolean recovery) { + Toolkit toolkit = new Toolkit(); + toolkit.registerSchema( + ToolSchema.builder() + .name("external") + .description("Execute outside the runtime") + .parameters(Map.of("type", "object", "properties", Map.of())) + .build()); + return ReActAgent.builder() + .name("asst") + .model(model) + .toolkit(toolkit) + .enablePendingToolRecovery(recovery) + .build(); + } + + private static Msg calls(String... ids) { + return Msg.builder() + .role(MsgRole.ASSISTANT) + .content( + java.util.Arrays.stream(ids) + .map( + id -> + (ContentBlock) + ToolUseBlock.builder() + .id(id) + .name("external") + .input(Map.of()) + .build()) + .toList()) + .metadata(Map.of(Msg.METADATA_EXTERNAL_EXECUTION_REQUEST_REPLY_ID, "reply")) + .build(); + } + + private static Msg result(String id) { + return Msg.builder() + .role(MsgRole.TOOL) + .content( + ToolResultBlock.builder() + .id(id) + .name("external") + .output(TextBlock.builder().text("result-" + id).build()) + .state(ToolResultState.SUCCESS) + .build()) + .build(); + } + + private static Msg user(String text) { + return Msg.builder().role(MsgRole.USER).textContent(text).build(); + } + + private static void seed(ReActAgent agent, String... pending) { + List context = agent.getAgentState(CONTEXT).contextMutable(); + context.addAll(List.of(calls("old"), result("old"))); + if (pending.length > 0) { + context.add(calls(pending)); + } + } + + private static List results(List msgs) { + return msgs.stream() + .flatMap(m -> m.getContentBlocks(ToolResultBlock.class).stream()) + .toList(); + } + + @Test + void consumedAndCurrentResultsResumeWithoutDuplicatingHistoryOrEvents() { + CapturingModel model = new CapturingModel(); + try (ReActAgent agent = agent(model, false)) { + seed(agent, "pending"); + List events = + agent.streamEvents( + List.of(result("old"), result("pending"), user("continue")), + CONTEXT) + .collectList() + .block(); + assertEquals( + List.of("old", "pending"), + results(model.inputs.get(0)).stream().map(ToolResultBlock::getId).toList()); + ExternalExecutionResultEvent event = + events.stream() + .filter(ExternalExecutionResultEvent.class::isInstance) + .map(ExternalExecutionResultEvent.class::cast) + .findFirst() + .orElseThrow(); + assertEquals("reply", event.getReplyId()); + assertEquals( + List.of("pending"), + event.getToolResults().stream().map(ToolResultBlock::getId).toList()); + } + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void consumedResultAndFreshInstructionRespectRecoverySwitch(boolean recovery) { + CapturingModel model = new CapturingModel(); + try (ReActAgent agent = agent(model, recovery)) { + seed(agent, "pending"); + List input = List.of(result("old"), user("new instruction")); + if (recovery) { + assertEquals("done", agent.call(input, CONTEXT).block().getTextContent()); + List results = results(model.inputs.get(0)); + assertEquals( + List.of("old", "pending"), + results.stream().map(ToolResultBlock::getId).toList()); + assertEquals(ToolResultState.ERROR, results.get(1).getState()); + assertTrue( + model.inputs.get(0).stream() + .anyMatch(m -> "new instruction".equals(m.getTextContent()))); + } else { + IllegalStateException error = + assertThrows( + IllegalStateException.class, + () -> agent.call(input, CONTEXT).block()); + assertTrue(error.getMessage().contains("enablePendingToolRecovery")); + assertEquals(3, agent.getAgentState(CONTEXT).getContext().size()); + assertTrue(model.inputs.isEmpty()); + } + } + } + + @Test + void repeatedConsumedResultsDoNotPolluteContextWhenNoToolsArePending() { + CapturingModel model = new CapturingModel(); + try (ReActAgent agent = agent(model, true)) { + seed(agent); + for (int i = 0; i < 2; i++) { + agent.call(List.of(result("old"), user("turn-" + i)), CONTEXT).block(); + } + assertEquals(1, results(agent.getAgentState(CONTEXT).getContext()).size()); + assertEquals(2, model.inputs.size()); + } + } + + @Test + void cleaningMixedMessagePreservesRemainingContentAndIdentity() { + CapturingModel model = new CapturingModel(); + try (ReActAgent agent = agent(model, true)) { + seed(agent, "pending"); + Msg mixed = + Msg.builder() + .id("mixed") + .name("alice") + .role(MsgRole.TOOL) + .metadata(Map.of("trace", "value")) + .content( + List.of( + result("old") + .getFirstContentBlock(ToolResultBlock.class), + TextBlock.builder().text("new instruction").build())) + .build(); + agent.call(List.of(mixed), CONTEXT).block(); + Msg cleaned = + agent.getAgentState(CONTEXT).getContext().stream() + .filter(m -> "mixed".equals(m.getId())) + .findFirst() + .orElseThrow(); + assertEquals("new instruction", cleaned.getTextContent()); + assertEquals(mixed.getMetadata(), cleaned.getMetadata()); + assertEquals(mixed.getTimestamp(), cleaned.getTimestamp()); + assertEquals(mixed.getRole(), cleaned.getRole()); + assertEquals(mixed.getName(), cleaned.getName()); + assertFalse(cleaned.hasContentBlocks(ToolResultBlock.class)); + assertEquals(2, mixed.getContent().size()); + } + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void unknownIdsAreRejectedEvenWithoutPendingTools(boolean pending) { + CapturingModel model = new CapturingModel(); + try (ReActAgent agent = agent(model, true)) { + seed(agent, pending ? new String[] {"pending"} : new String[] {}); + List before = agent.getAgentState(CONTEXT).getContext(); + IllegalStateException error = + assertThrows( + IllegalStateException.class, + () -> + agent.call( + List.of(result("unknown"), user("continue")), + CONTEXT) + .block()); + assertTrue(error.getMessage().contains("Invalid tool result IDs")); + assertEquals(before, agent.getAgentState(CONTEXT).getContext()); + assertTrue(model.inputs.isEmpty()); + } + } + + @ParameterizedTest + @ValueSource(strings = {"old", "pending"}) + void duplicateIdsAreRejectedBeforeConsumedResultsAreRemoved(String id) { + CapturingModel model = new CapturingModel(); + try (ReActAgent agent = agent(model, true)) { + seed(agent, "pending"); + IllegalStateException error = + assertThrows( + IllegalStateException.class, + () -> + agent.call( + List.of( + result(id), + result(id), + user("continue")), + CONTEXT) + .block()); + assertTrue(error.getMessage().contains("Duplicate tool result ID")); + assertEquals(3, agent.getAgentState(CONTEXT).getContext().size()); + } + } + + @Test + void partialResultsAndTextStillFailAfterRemovingConsumedResults() { + CapturingModel model = new CapturingModel(); + try (ReActAgent agent = agent(model, true)) { + seed(agent, "one", "two"); + IllegalStateException error = + assertThrows( + IllegalStateException.class, + () -> + agent.call( + List.of( + result("old"), + result("one"), + user("continue")), + CONTEXT) + .block()); + assertTrue(error.getMessage().contains("partial tool results")); + assertEquals(3, agent.getAgentState(CONTEXT).getContext().size()); + } + } + + @Test + void consumedResultsAloneDoNotExecutePendingToolsOrCallModel() { + CapturingModel model = new CapturingModel(); + try (ReActAgent agent = agent(model, true)) { + seed(agent, "pending"); + assertThrows( + IllegalStateException.class, + () -> agent.call(List.of(result("old")), CONTEXT).block()); + assertEquals(3, agent.getAgentState(CONTEXT).getContext().size()); + assertTrue(model.inputs.isEmpty()); + } + } + + @Test + void statelessCompleteHistoryStillReachesModel() { + CapturingModel model = new CapturingModel(); + try (ReActAgent agent = agent(model, false)) { + List history = + List.of(user("first"), calls("client"), result("client"), user("next")); + agent.call(history, CONTEXT).block(); + assertTrue(model.inputs.get(0).containsAll(history)); + } + } + + @Test + void replaySelectionUsesReloadedStateAndDoesNotReplayRegeneratePrompt() { + CapturingModel model = new CapturingModel(); + InMemoryAgentStateStore store = new InMemoryAgentStateStore(); + try (ReActAgent agent = + ReActAgent.builder().name("asst").model(model).stateStore(store).build()) { + // The cached state is deliberately obsolete when the invocation starts. + agent.getAgentState(CONTEXT).contextMutable().add(calls("obsolete")); + AgentState authoritative = AgentState.builder().sessionId("thread").build(); + authoritative.contextMutable().add(calls("pending")); + store.save("alice", "thread", "agent_state", authoritative); + RuntimeContext replay = + RuntimeContext.builder(CONTEXT) + .put(RuntimeContext.REPLAYED_INPUT, true) + .build(); + List transcript = + List.of( + Msg.builder() + .role(MsgRole.SYSTEM) + .textContent("client system history") + .build(), + user("old prompt"), + calls("pending"), + result("pending"), + Msg.builder() + .role(MsgRole.ASSISTANT) + .textContent("client split turn") + .build()); + agent.call(transcript, replay).block(); + assertEquals( + List.of("pending"), + results(model.inputs.get(0)).stream().map(ToolResultBlock::getId).toList()); + assertFalse( + model.inputs.get(0).stream() + .anyMatch( + m -> + "old prompt".equals(m.getTextContent()) + || "client system history" + .equals(m.getTextContent()))); + } + } + + @Test + void replayDoesNotBypassAskingToolWithAResultAndFreshText() { + CapturingModel model = new CapturingModel(); + try (ReActAgent agent = agent(model, true)) { + seed(agent); + agent.getAgentState(CONTEXT) + .contextMutable() + .add( + Msg.builder() + .role(MsgRole.ASSISTANT) + .content( + ToolUseBlock.builder() + .id("approval") + .name("external") + .input(Map.of()) + .state(ToolCallState.ASKING) + .build()) + .build()); + RuntimeContext replay = + RuntimeContext.builder(CONTEXT) + .put(RuntimeContext.REPLAYED_INPUT, true) + .build(); + IllegalStateException error = + assertThrows( + IllegalStateException.class, + () -> + agent.call( + List.of( + calls("approval"), + result("approval"), + user("continue")), + replay) + .block()); + assertTrue(error.getMessage().contains("ASKING")); + assertEquals(1, results(agent.getAgentState(CONTEXT).getContext()).size()); + assertTrue(model.inputs.isEmpty()); + } + } + + @Test + void storedOrphanResultDoesNotMakeAnUnknownIdValid() { + CapturingModel model = new CapturingModel(); + try (ReActAgent agent = agent(model, true)) { + agent.getAgentState(CONTEXT).contextMutable().add(result("unknown")); + assertThrows( + IllegalStateException.class, + () -> + agent.call(List.of(result("unknown"), user("continue")), CONTEXT) + .block()); + assertTrue(model.inputs.isEmpty()); + } + } + + @Test + void existingSessionCannotValidateUnknownResultWithAnInputToolCall() { + CapturingModel model = new CapturingModel(); + try (ReActAgent agent = agent(model, true)) { + seed(agent); + assertThrows( + IllegalStateException.class, + () -> + agent.call( + List.of( + calls("unknown"), + result("unknown"), + user("continue")), + CONTEXT) + .block()); + assertEquals(2, agent.getAgentState(CONTEXT).getContext().size()); + } + } + + @Test + void initialTranscriptRequiresToolCallBeforeResult() { + CapturingModel model = new CapturingModel(); + try (ReActAgent agent = agent(model, false)) { + assertThrows( + IllegalStateException.class, + () -> + agent.call( + List.of( + result("unknown"), + calls("unknown"), + user("continue")), + CONTEXT) + .block()); + assertTrue(agent.getAgentState(CONTEXT).getContext().isEmpty()); + } + } + + @Test + void consumedResultsFromAnotherUserAreNotAccepted() { + CapturingModel model = new CapturingModel(); + try (ReActAgent agent = agent(model, true)) { + seed(agent); + RuntimeContext otherUser = + RuntimeContext.builder() + .userId("bob") + .sessionId("thread") + .put(RuntimeContext.REPLAYED_INPUT, true) + .build(); + agent.getAgentState(otherUser).contextMutable().add(calls("bob-pending")); + assertThrows( + IllegalStateException.class, + () -> + agent.call( + List.of( + calls("client"), + result("old"), + user("continue")), + otherUser) + .block()); + assertEquals(1, agent.getAgentState(otherUser).getContext().size()); + assertEquals(2, agent.getAgentState(CONTEXT).getContext().size()); + assertTrue(model.inputs.isEmpty()); + } + } +} diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/processor/AguiRequestProcessor.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/processor/AguiRequestProcessor.java index e45e2ad92d..5c6917d2bf 100644 --- a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/processor/AguiRequestProcessor.java +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/processor/AguiRequestProcessor.java @@ -163,6 +163,9 @@ public ProcessResult process(AguiRuntimeContextRequest request) { } try { + RuntimeContext effectiveRuntimeContext = + resumeCoordinator.addResumeInterrupts( + input, runtimeContext); // Determine effective input based on server-side memory RunAgentInput effectiveInput = input; if (agentResolver.hasMemory(runtimeContext)) { @@ -171,13 +174,19 @@ public ProcessResult process(AguiRuntimeContextRequest request) { + " extracting follow-up messages", threadId, runtimeContext.getUserId()); - effectiveInput = extractLatestUserMessage(input); + if (AguiUtil.asReActAgent(agent) != null) { + // The authoritative state is reloaded under the agent's + // session lock. Keep the transcript until then so pending + // results in its middle cannot be lost to a stale cache. + effectiveRuntimeContext = + RuntimeContext.builder(effectiveRuntimeContext) + .put(RuntimeContext.REPLAYED_INPUT, true) + .build(); + } else { + effectiveInput = extractLatestUserMessage(input); + } } - RuntimeContext effectiveRuntimeContext = - resumeCoordinator.addResumeInterrupts( - input, runtimeContext); - // Create adapter and run AguiAgentAdapter adapter = adapterFactory.create(agent, config); AtomicBoolean runErrorSeen = new AtomicBoolean(false); diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/processor/AguiRequestProcessorTest.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/processor/AguiRequestProcessorTest.java index c4f127b260..4616fc7902 100644 --- a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/processor/AguiRequestProcessorTest.java +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/processor/AguiRequestProcessorTest.java @@ -583,7 +583,7 @@ void processUsesCustomAdapterFactory() { @Test void processExtractsFollowUpMessagesWhenServerHasMemory() { AgentResolver resolver = mock(AgentResolver.class); - ReActAgent agent = mock(ReActAgent.class); + Agent agent = mock(Agent.class); when(resolver.resolveAgent(eq("default"), eq("thread-1"), nullable(String.class))) .thenReturn(agent); when(resolver.hasMemory(any(RuntimeContext.class))).thenReturn(true); diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/processor/AguiToolResultReplayTest.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/processor/AguiToolResultReplayTest.java new file mode 100644 index 0000000000..9abda1e763 --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/processor/AguiToolResultReplayTest.java @@ -0,0 +1,291 @@ +/* + * 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.agui.processor; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.agentscope.core.ReActAgent; +import io.agentscope.core.agent.Agent; +import io.agentscope.core.agent.RuntimeContext; +import io.agentscope.core.agui.converter.AguiMessageConverter; +import io.agentscope.core.agui.event.AguiEvent; +import io.agentscope.core.agui.model.AguiMessage; +import io.agentscope.core.agui.model.AguiResume; +import io.agentscope.core.agui.model.RunAgentInput; +import io.agentscope.core.agui.runtime.AguiRuntimeContextRequest; +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.ToolUseBlock; +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.ArrayList; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; + +/** Exercises the processor, converter, resume coordinator and real ReAct loop together. */ +class AguiToolResultReplayTest { + private static final RuntimeContext CONTEXT = + RuntimeContext.builder().userId("alice").sessionId("thread-1").build(); + + private static final class Model extends ChatModelBase { + private final boolean suspendFirst; + private final List> inputs = new ArrayList<>(); + + Model(boolean suspendFirst) { + this.suspendFirst = suspendFirst; + } + + @Override + public String getModelName() { + return "agui-replay-test"; + } + + @Override + protected Flux doStream( + List messages, List tools, GenerateOptions options) { + inputs.add(List.copyOf(messages)); + return Flux.just( + ChatResponse.builder() + .content( + suspendFirst && inputs.size() == 1 + ? calls("pending").getContent() + : List.of(TextBlock.builder().text("done").build())) + .build()); + } + } + + private static ReActAgent agent(Model model) { + Toolkit toolkit = new Toolkit(); + toolkit.registerSchema( + ToolSchema.builder() + .name("external") + .description("Execute externally") + .parameters(Map.of("type", "object", "properties", Map.of())) + .build()); + return ReActAgent.builder() + .name("asst") + .model(model) + .toolkit(toolkit) + .enablePendingToolRecovery(true) + .build(); + } + + private static Msg calls(String id) { + return Msg.builder() + .role(MsgRole.ASSISTANT) + .content(ToolUseBlock.builder().id(id).name("external").input(Map.of()).build()) + .build(); + } + + private static Msg result(String id) { + return new AguiMessageConverter().toMsg(AguiMessage.toolMessage("result-" + id, id, "ok")); + } + + private static void seed(ReActAgent agent, boolean pending) { + agent.getAgentState(CONTEXT).contextMutable().addAll(List.of(calls("old"), result("old"))); + if (pending) { + agent.getAgentState(CONTEXT).contextMutable().add(calls("pending")); + } + } + + private static AguiRequestProcessor processor(ReActAgent agent) { + return AguiRequestProcessor.builder() + .agentResolver( + new AgentResolver() { + @Override + public Agent resolveAgent(String agentId, String threadId) { + return agent; + } + + @Override + public boolean hasMemory(RuntimeContext ctx) { + return !agent.getAgentState(ctx).getContext().isEmpty(); + } + }) + .runtimeContextResolver(request -> CONTEXT) + .build(); + } + + private static RunAgentInput input(String run, List messages) { + return RunAgentInput.builder().threadId("thread-1").runId(run).messages(messages).build(); + } + + private static List run(AguiRequestProcessor processor, RunAgentInput input) { + return processor + .process(AguiRuntimeContextRequest.builder().input(input).build()) + .events() + .collectList() + .block(); + } + + private static void assertSuccess(List events) { + assertFalse( + events.stream().anyMatch(AguiEvent.RunError.class::isInstance), events.toString()); + assertInstanceOf(AguiEvent.RunStarted.class, events.get(0)); + assertInstanceOf(AguiEvent.RunFinished.class, events.get(events.size() - 1)); + } + + private static List resultIds(ReActAgent agent) { + return agent.getAgentState(CONTEXT).getContext().stream() + .flatMap(m -> m.getContentBlocks(ToolResultBlock.class).stream()) + .map(ToolResultBlock::getId) + .toList(); + } + + @Test + void recoversMidTranscriptPendingResultAndDropsConsumedTrailingResult() { + Model model = new Model(false); + try (ReActAgent agent = agent(model)) { + seed(agent, true); + List transcript = + List.of( + AguiMessage.systemMessage("system", "historical client system"), + AguiMessage.userMessage("old-user", "old prompt"), + AguiMessage.assistantMessage("assistant-one", "first local round"), + AguiMessage.toolMessage("pending-result", "pending", "valid result"), + AguiMessage.toolMessage("compacted-result", "compacted", "old history"), + AguiMessage.assistantMessage("assistant-two", "split local round"), + AguiMessage.toolMessage("old-result", "old", "replayed result"), + AguiMessage.userMessage("fresh", "continue")); + assertSuccess(run(processor(agent), input("run-1", transcript))); + assertEquals(List.of("old", "pending"), resultIds(agent)); + List seen = model.inputs.get(0); + assertTrue(seen.stream().anyMatch(m -> "continue".equals(m.getTextContent()))); + assertFalse( + seen.stream() + .anyMatch( + m -> + "old prompt".equals(m.getTextContent()) + || "historical client system" + .equals(m.getTextContent()))); + ToolResultBlock pending = + seen.stream() + .flatMap(m -> m.getContentBlocks(ToolResultBlock.class).stream()) + .filter(r -> "pending".equals(r.getId())) + .findFirst() + .orElseThrow(); + assertEquals("valid result", ((TextBlock) pending.getOutput().get(0)).getText()); + } + } + + @Test + void realSuspensionResumesWithMidTranscriptResultAndOfficialResumeWithoutDuplicates() { + Model model = new Model(true); + try (ReActAgent agent = agent(model)) { + AguiRequestProcessor processor = processor(agent); + List first = + run( + processor, + input("run-1", List.of(AguiMessage.userMessage("user", "start")))); + assertSuccess(first); + AguiEvent.RunFinished finished = (AguiEvent.RunFinished) first.get(first.size() - 1); + AguiEvent.RunFinishedInterruptOutcome outcome = + assertInstanceOf( + AguiEvent.RunFinishedInterruptOutcome.class, finished.outcome()); + String interruptId = outcome.interrupts().get(0).id(); + RunAgentInput resume = + RunAgentInput.builder() + .threadId("thread-1") + .runId("run-2") + .messages( + List.of( + AguiMessage.userMessage("user", "start"), + AguiMessage.assistantMessage("first", "tool call"), + AguiMessage.toolMessage( + "result", "pending", "browser result"), + AguiMessage.assistantMessage( + "split", "local continuation"))) + .resume( + List.of( + new AguiResume( + interruptId, + AguiResume.STATUS_RESOLVED, + "browser result"))) + .build(); + assertSuccess(run(processor, resume)); + assertEquals(List.of("pending"), resultIds(agent)); + assertEquals(2, model.inputs.size()); + assertEquals( + 1, + agent.getAgentState(CONTEXT).getContext().stream() + .filter(m -> m.getRole() == MsgRole.USER) + .count()); + } + } + + @Test + void repeatedStaleResultWithNewUserMessagesDoesNotWedgeCompletedSession() { + Model model = new Model(false); + try (ReActAgent agent = agent(model)) { + seed(agent, false); + AguiRequestProcessor processor = processor(agent); + for (int i = 0; i < 2; i++) { + assertSuccess( + run( + processor, + input( + "run-" + i, + List.of( + AguiMessage.assistantMessage("asst", "completed"), + AguiMessage.toolMessage("stale", "old", "replayed"), + AguiMessage.userMessage( + "user-" + i, "next " + i))))); + } + assertEquals(List.of("old"), resultIds(agent)); + assertEquals(2, model.inputs.size()); + } + } + + @Test + void unknownTrailingResultFailsWithoutPoisoningTheNextCorrectedRequest() { + Model model = new Model(false); + try (ReActAgent agent = agent(model)) { + seed(agent, true); + AguiRequestProcessor processor = processor(agent); + List rejected = + run( + processor, + input( + "bad", + List.of( + AguiMessage.assistantMessage("asst", "pending"), + AguiMessage.toolMessage("unknown", "unknown", "bad"), + AguiMessage.userMessage("user", "continue")))); + assertTrue(rejected.stream().anyMatch(AguiEvent.RunError.class::isInstance)); + assertEquals(List.of("old"), resultIds(agent)); + assertSuccess( + run( + processor, + input( + "good", + List.of( + AguiMessage.assistantMessage("asst", "pending"), + AguiMessage.toolMessage("valid", "pending", "good"), + AguiMessage.userMessage("user", "continue"))))); + assertEquals(List.of("old", "pending"), resultIds(agent)); + } + } +} diff --git a/docs/v2/en/integration/protocol/agui.md b/docs/v2/en/integration/protocol/agui.md index a4b748b997..4e02046475 100644 --- a/docs/v2/en/integration/protocol/agui.md +++ b/docs/v2/en/integration/protocol/agui.md @@ -335,6 +335,12 @@ For permission confirmations, `payload.approved` must be the boolean `true` to a The front end does not need to echo `metadata` in `resume[]`; it only sends `interruptId`, `status`, and `payload`. Through the Spring `AguiRequestProcessor` entry point, AgentScope Java records the latest `RUN_FINISHED.outcome.interrupts[]` server-side, validates that the next `resume[]` covers all open interrupts, and passes the originating interrupts into the adapter for conversion. +## Replayed Transcripts With Server-Side Memory + +When server-side memory is present, `AguiRequestProcessor` passes ReAct-backed agents the full client transcript with `RuntimeContext.REPLAYED_INPUT=true`. After loading the current user/session state, the agent selects messages after the last client assistant turn and any earlier tool results matching pending calls. Other earlier history is not appended again. Stateless input and non-ReAct agents retain their existing behavior. + +Previously consumed results in the selected input are removed. Unknown result IDs and duplicate submitted IDs remain errors, as does partial tool completion combined with text. A request containing only consumed results is rejected. ReActAgent also rejects orphan result IDs when no calls are pending; an initial, client-owned transcript can still contain results paired with preceding assistant tool calls. If cleaning leaves a new instruction without current tool results, `enablePendingToolRecovery=true` is still required to cancel abandoned calls with error results. Permission confirmation and the official `resume[]` contract remain required; replay handling does not approve ASKING tools. + ## Example Project See the complete example at [agentscope-examples/agui](https://github.com/agentscope-ai/agentscope-java/tree/main/agentscope-examples/agui): diff --git a/docs/v2/zh/integration/protocol/agui.md b/docs/v2/zh/integration/protocol/agui.md index 29466b7a50..fe98a689bf 100644 --- a/docs/v2/zh/integration/protocol/agui.md +++ b/docs/v2/zh/integration/protocol/agui.md @@ -334,6 +334,12 @@ AG-UI 前端可以在 `RunAgentInput.tools` 中传入工具 schema。adapter 会 前端不需要在 `resume[]` 中回传 `metadata`；只需要发送 `interruptId`、`status` 和 `payload`。通过 Spring `AguiRequestProcessor` 入口时，AgentScope Java 会在服务端记录最近一次 `RUN_FINISHED.outcome.interrupts[]`，校验下一次 `resume[]` 是否覆盖所有 open interrupts，并把原始 interrupt 传给 adapter 做恢复转换。 +## 服务端记忆与历史消息重放 + +存在服务端记忆时，`AguiRequestProcessor` 会向 ReAct 类型的 agent 传入完整客户端历史，并设置 `RuntimeContext.REPLAYED_INPUT=true`。Agent 加载当前用户、会话的最新状态后，选取客户端最后一条 assistant 之后的消息，同时找回历史中匹配当前待处理工具调用的结果；其他早期历史不再追加。无状态输入和非 ReAct agent 保持原有行为。 + +选中输入里已消费的工具结果会被移除。未知结果 ID、重复提交的 ID，以及“部分工具结果加文本”仍会报错；仅包含已消费结果的请求也会被拒绝。ReActAgent 在没有待处理调用时也会拒绝孤立的结果 ID；首次传入的客户端历史仍可包含与前序 assistant 工具调用配对的结果。清理后若只剩新指令、缺少当前工具结果，仍须启用 `enablePendingToolRecovery=true`，才能为遗留调用生成错误结果并继续处理。权限确认及官方 `resume[]` 契约继续生效，历史重放不会自动批准 ASKING 工具。 + ## 示例项目 完整示例见 [agentscope-examples/agui](https://github.com/agentscope-ai/agentscope-java/tree/main/agentscope-examples/agui)：