diff --git a/agentscope-examples/agentscope-copilotkit/src/main/java/io/agentscope/examples/copilotkit/service/AgentEventAguiReplayer.java b/agentscope-examples/agentscope-copilotkit/src/main/java/io/agentscope/examples/copilotkit/service/AgentEventAguiReplayer.java index 793118797a..87a0d7408a 100644 --- a/agentscope-examples/agentscope-copilotkit/src/main/java/io/agentscope/examples/copilotkit/service/AgentEventAguiReplayer.java +++ b/agentscope-examples/agentscope-copilotkit/src/main/java/io/agentscope/examples/copilotkit/service/AgentEventAguiReplayer.java @@ -39,6 +39,10 @@ * and resets workbench snapshot baselines per historical run so {@code STATE_SNAPSHOT} / * {@code STATE_DELTA} projection matches a fresh conversion. * + *

Presentation replay for reconnect (resolved-interrupt suppression and dangling tool-call + * synthesis) has moved to the framework presentation snapshot store; this replayer now serves the + * {@code /threads/{id}/events} inspect API, which is legitimately an event log rather than + * presentation state. */ @Component public final class AgentEventAguiReplayer { diff --git a/agentscope-examples/agentscope-copilotkit/src/main/java/io/agentscope/examples/copilotkit/service/CopilotKitRuntimeService.java b/agentscope-examples/agentscope-copilotkit/src/main/java/io/agentscope/examples/copilotkit/service/CopilotKitRuntimeService.java index a39db89094..cc054ae7ae 100644 --- a/agentscope-examples/agentscope-copilotkit/src/main/java/io/agentscope/examples/copilotkit/service/CopilotKitRuntimeService.java +++ b/agentscope-examples/agentscope-copilotkit/src/main/java/io/agentscope/examples/copilotkit/service/CopilotKitRuntimeService.java @@ -19,22 +19,24 @@ import io.agentscope.core.agui.event.AguiEvent; import io.agentscope.core.agui.model.RunAgentInput; import io.agentscope.core.agui.registry.AguiAgentRegistry; +import io.agentscope.core.agui.store.AguiSnapshotHydrator; +import io.agentscope.core.agui.store.AguiSnapshotStore; +import io.agentscope.core.agui.store.AguiThreadSnapshot; import io.agentscope.examples.copilotkit.model.CopilotKitModels.AgentInfo; import io.agentscope.examples.copilotkit.model.CopilotKitModels.InfoResponse; import io.agentscope.examples.copilotkit.model.CopilotKitModels.Intelligence; import io.agentscope.examples.copilotkit.model.CopilotKitModels.ThreadEndpoints; -import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.http.codec.ServerSentEvent; import org.springframework.stereotype.Service; import reactor.core.publisher.Flux; /** - * CopilotKit Runtime info and multi-route connect handshake with AgentEvent replay. - * - * + * CopilotKit Runtime info and multi-route connect hydrate backed by the framework presentation + * snapshot store. */ @Service public final class CopilotKitRuntimeService { @@ -47,13 +49,15 @@ public final class CopilotKitRuntimeService { "humanInTheLoop", true); private final AguiAgentRegistry aguiAgentRegistry; - private final AgentEventAguiReplayer eventReplayer; + private final ObjectProvider snapshotStoreProvider; + private final AguiSnapshotHydrator hydrator = new AguiSnapshotHydrator(); private final AguiEventEncoder encoder = new AguiEventEncoder(); public CopilotKitRuntimeService( - AguiAgentRegistry aguiAgentRegistry, AgentEventAguiReplayer eventReplayer) { + AguiAgentRegistry aguiAgentRegistry, + ObjectProvider snapshotStoreProvider) { this.aguiAgentRegistry = aguiAgentRegistry; - this.eventReplayer = eventReplayer; + this.snapshotStoreProvider = snapshotStoreProvider; } public InfoResponse info() { @@ -105,29 +109,22 @@ private AgentInfo resolveAgentInfo(String agentId) { } /** - * AG-UI connect: replay persisted AgentEvents through converters, or emit an empty handshake. + * AG-UI connect: rebuild the visible conversation from the framework presentation snapshot + * store. * - *

History is stored as AgentScope {@code AgentEvent}s. On connect they are projected to - * AG-UI frames with the same converter registry used by {@code /run}, so CopilotKit can - * restore the conversation. Without history a minimal - * {@code RUN_STARTED → MESSAGES_SNAPSHOT([]) → RUN_FINISHED} handshake is returned. + *

Read-only: it looks up the stored snapshot for the thread and delegates to + * {@link AguiSnapshotHydrator}. When the snapshot store is disabled (or the thread has no + * history) the hydrator returns the minimal {@code RUN_STARTED → MESSAGES_SNAPSHOT([]) → + * RUN_FINISHED} handshake. Only the trailing unresolved interrupt is ever replayed, so a + * resolved historical interrupt can never reappear. */ public Flux> connect(RunAgentInput input) { String threadId = input.getThreadId(); String runId = input.getRunId(); - List history = eventReplayer.replay(threadId, input); - if (history.isEmpty()) { - return Flux.fromIterable(emptyHandshake(threadId, runId)).map(this::sse); - } - return Flux.fromIterable(history).map(this::sse); - } - - private List emptyHandshake(String threadId, String runId) { - List events = new ArrayList<>(3); - events.add(new AguiEvent.RunStarted(threadId, runId)); - events.add(new AguiEvent.MessagesSnapshot(threadId, runId, List.of())); - events.add(new AguiEvent.RunFinished(threadId, runId)); - return events; + AguiSnapshotStore store = snapshotStoreProvider.getIfAvailable(); + AguiThreadSnapshot snapshot = store != null ? store.find(threadId).orElse(null) : null; + List frames = hydrator.hydrate(snapshot, threadId, runId); + return Flux.fromIterable(frames).map(this::sse); } private ServerSentEvent sse(AguiEvent event) { diff --git a/agentscope-examples/agentscope-copilotkit/src/main/java/io/agentscope/examples/copilotkit/service/PersistingAgentEventEnricher.java b/agentscope-examples/agentscope-copilotkit/src/main/java/io/agentscope/examples/copilotkit/service/PersistingAgentEventEnricher.java index 392a13266c..591d1e61c5 100644 --- a/agentscope-examples/agentscope-copilotkit/src/main/java/io/agentscope/examples/copilotkit/service/PersistingAgentEventEnricher.java +++ b/agentscope-examples/agentscope-copilotkit/src/main/java/io/agentscope/examples/copilotkit/service/PersistingAgentEventEnricher.java @@ -29,7 +29,10 @@ /** * Persists the source {@link AgentEvent} while the AG-UI adapter projects it. * - *

AG-UI frames themselves are not stored — connect replay re-projects through converters. + *

AG-UI frames themselves are not stored. Presentation replay for reconnect has moved to the + * framework presentation snapshot store ({@link io.agentscope.core.agui.store.AguiSnapshotStore}); + * this enricher now serves only the {@code /threads/{id}/events} inspect API, which is + * legitimately an event log rather than presentation state. * *

On {@link AgentStartEvent}, also stores the run's input messages so reconnect can rebuild * {@code RUN_STARTED.input.messages} (how CopilotKit restores user / tool turns). diff --git a/agentscope-examples/agentscope-copilotkit/src/main/resources/application.yml b/agentscope-examples/agentscope-copilotkit/src/main/resources/application.yml index 91a3df343d..ea6d6394db 100644 --- a/agentscope-examples/agentscope-copilotkit/src/main/resources/application.yml +++ b/agentscope-examples/agentscope-copilotkit/src/main/resources/application.yml @@ -54,6 +54,10 @@ agentscope: max-thread-sessions: 1000 session-timeout-minutes: 30 enable-reasoning: true + # Presentation snapshot store: enables POST /agui/connect hydrate so reconnecting clients + # rebuild the visible conversation without re-running the agent. + snapshot-store-enabled: true + snapshot-max-threads: 1000 # Logging logging: diff --git a/agentscope-examples/agui/README.md b/agentscope-examples/agui/README.md new file mode 100644 index 0000000000..e49e4f1aa5 --- /dev/null +++ b/agentscope-examples/agui/README.md @@ -0,0 +1,49 @@ +# AG-UI Example + +A minimal Spring Boot WebFlux application exposing AgentScope agents over the AG-UI protocol. + +## Run + +```bash +mvn -q -pl agentscope-examples/agui spring-boot:run +``` + +The server listens on `http://localhost:8080` and exposes `POST /agui/run` (and `/agui/run/{agentId}` +when path routing is enabled). + +## Presentation Snapshot Hydrate + +This example enables the AG-UI presentation snapshot store: + +```yaml +agentscope: + agui: + snapshot-store-enabled: true + snapshot-max-threads: 1000 +``` + +With the store enabled, a reconnecting client can rebuild the visible conversation **without +re-running the agent** by calling the read-only hydrate endpoint `POST /agui/connect`. + +Try it: + +1. Run the agent once against `/agui/run` with a `threadId`: + + ```bash + curl -N http://localhost:8080/agui/run \ + -H 'Content-Type: application/json' \ + -d '{"threadId":"demo-1","runId":"run-1","messages":[{"id":"m1","role":"user","content":"hello"}]}' + ``` + +2. Replay the same `threadId` against `/agui/connect` and observe a `MESSAGES_SNAPSHOT` restoring + the conversation with **no model call**: + + ```bash + curl -N http://localhost:8080/agui/connect \ + -H 'Content-Type: application/json' \ + -d '{"threadId":"demo-1","runId":"connect-1"}' + ``` + +The hydrate response is strictly read-only: it never mutates the agent, the snapshot store, or the +resume coordinator, and only the **trailing unresolved** interrupt is ever replayed (so a resolved +historical interrupt cannot reappear). diff --git a/agentscope-examples/agui/src/main/resources/application.yml b/agentscope-examples/agui/src/main/resources/application.yml index 2d55e3ac89..700b9a1131 100644 --- a/agentscope-examples/agui/src/main/resources/application.yml +++ b/agentscope-examples/agui/src/main/resources/application.yml @@ -52,6 +52,10 @@ agentscope: max-thread-sessions: 1000 session-timeout-minutes: 30 enable-reasoning: true + # Presentation snapshot store: enables POST /agui/connect hydrate so reconnecting clients + # rebuild the visible conversation without re-running the agent. + snapshot-store-enabled: true + snapshot-max-threads: 1000 # Logging logging: diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/adapter/AguiAdapterConfig.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/adapter/AguiAdapterConfig.java index cb9211274f..21e2425bf9 100644 --- a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/adapter/AguiAdapterConfig.java +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/adapter/AguiAdapterConfig.java @@ -19,6 +19,8 @@ import io.agentscope.core.agui.adapter.strategy.AguiEventEnricher; import io.agentscope.core.agui.adapter.strategy.BaseEventPropertiesEnricher; import io.agentscope.core.agui.model.ToolMergeMode; +import io.agentscope.core.agui.store.AguiSnapshotStore; +import io.agentscope.core.agui.store.SnapshotRecordingEnricher; import java.time.Duration; import java.util.ArrayList; import java.util.List; @@ -45,6 +47,9 @@ public class AguiAdapterConfig { private final List eventEnrichers; private final boolean baseEventPropertiesEnricherEnabled; private final boolean emitSubagentEventsAsNative; + private final boolean snapshotStoreEnabled; + private final AguiSnapshotStore snapshotStore; + private final SnapshotRecordingEnricher snapshotRecorder; private AguiAdapterConfig(Builder builder) { this.toolMergeMode = builder.toolMergeMode; @@ -56,9 +61,34 @@ private AguiAdapterConfig(Builder builder) { this.runTimeout = builder.runTimeout; this.defaultAgentId = builder.defaultAgentId; this.eventConverters = List.copyOf(builder.eventConverters); - this.eventEnrichers = buildEventEnrichers(builder); + SnapshotRecordingEnricher recorder = recorderFrom(builder); + this.eventEnrichers = buildEventEnrichers(builder, recorder); this.baseEventPropertiesEnricherEnabled = builder.baseEventPropertiesEnricherEnabled; this.emitSubagentEventsAsNative = builder.emitSubagentEventsAsNative; + this.snapshotStoreEnabled = builder.snapshotStoreEnabled; + this.snapshotStore = builder.snapshotStore; + this.snapshotRecorder = recorder; + } + + private static List buildEventEnrichers( + Builder builder, SnapshotRecordingEnricher recorder) { + List enrichers = new ArrayList<>(); + if (builder.baseEventPropertiesEnricherEnabled) { + enrichers.add(new BaseEventPropertiesEnricher()); + } + enrichers.addAll(builder.eventEnrichers); + if (recorder != null) { + // Appended last so it observes fully enriched frames from all converters. + enrichers.add(recorder); + } + return List.copyOf(enrichers); + } + + private static SnapshotRecordingEnricher recorderFrom(Builder builder) { + if (!builder.snapshotStoreEnabled || builder.snapshotStore == null) { + return null; + } + return new SnapshotRecordingEnricher(builder.snapshotStore); } /** @@ -183,6 +213,40 @@ public boolean isEmitSubagentEventsAsNative() { return emitSubagentEventsAsNative; } + /** + * Check whether the AG-UI presentation snapshot store is enabled. + * + *

When {@code true} and {@link #getSnapshotStore()} is set, a {@link + * SnapshotRecordingEnricher} is appended last in the enricher chain so reconnecting clients + * can rebuild the visible conversation via {@code POST {path-prefix}/connect}. Default is + * {@code false} so existing clients stay byte-identical. + * + * @return true if the snapshot store is enabled + */ + public boolean isSnapshotStoreEnabled() { + return snapshotStoreEnabled; + } + + /** + * Get the configured presentation snapshot store, or null when disabled. + * + * @return the snapshot store, or null + */ + public AguiSnapshotStore getSnapshotStore() { + return snapshotStore; + } + + /** + * Get the recording enricher appended to the enricher chain, or null when the snapshot store is + * disabled. This is the single instance shared with the chain, so callers (e.g. the request + * processor's flush safety net) target the same accumulator that is recording the live stream. + * + * @return the snapshot recording enricher, or null + */ + public SnapshotRecordingEnricher getSnapshotRecorder() { + return snapshotRecorder; + } + /** * Creates a new builder for AguiAdapterConfig. * @@ -207,6 +271,10 @@ private static List buildEventEnrichers(Builder builder) { enrichers.add(new BaseEventPropertiesEnricher()); } enrichers.addAll(builder.eventEnrichers); + if (builder.snapshotStoreEnabled && builder.snapshotStore != null) { + // Appended last so it observes fully enriched frames from all converters. + enrichers.add(new SnapshotRecordingEnricher(builder.snapshotStore)); + } return List.copyOf(enrichers); } @@ -227,6 +295,8 @@ public static class Builder { private final List eventEnrichers = new ArrayList<>(); private boolean baseEventPropertiesEnricherEnabled = false; private boolean emitSubagentEventsAsNative = false; + private boolean snapshotStoreEnabled = false; + private AguiSnapshotStore snapshotStore; /** * Set the tool merge mode. @@ -408,6 +478,29 @@ public Builder emitSubagentEventsAsNative(boolean emitSubagentEventsAsNative) { return this; } + /** + * Enable the AG-UI presentation snapshot store so a {@link SnapshotRecordingEnricher} is + * appended last in the enricher chain. + * + * @param snapshotStoreEnabled true to enable + * @return This builder + */ + public Builder snapshotStoreEnabled(boolean snapshotStoreEnabled) { + this.snapshotStoreEnabled = snapshotStoreEnabled; + return this; + } + + /** + * Set the presentation snapshot store used to record and hydrate thread state. + * + * @param snapshotStore the store, or null + * @return This builder + */ + public Builder snapshotStore(AguiSnapshotStore snapshotStore) { + this.snapshotStore = snapshotStore; + return this; + } + /** * Build the configuration. * diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/converter/AguiActivityConverter.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/converter/AguiActivityConverter.java new file mode 100644 index 0000000000..a4dae8ac4b --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/converter/AguiActivityConverter.java @@ -0,0 +1,89 @@ +/* + * 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.converter; + +import io.agentscope.core.agui.event.AguiEvent; +import io.agentscope.core.agui.event.AguiEvent.JsonPatchOperation; +import java.util.List; +import java.util.Map; + +/** + * Converter for AG-UI activity events, mirroring {@link AguiStateConverter}. + * + *

Creates {@link AguiEvent.ActivitySnapshot} and {@link AguiEvent.ActivityDelta} events from + * before/after activity content maps. Delta computation reuses {@link AguiJsonDiff} so activity + * deltas are byte-compatible with state deltas. + */ +public class AguiActivityConverter { + + /** + * Create an {@link AguiEvent.ActivitySnapshot} event. + * + * @param threadId the thread id + * @param runId the run id + * @param messageId the message id the activity is attached to + * @param activityType the activity type + * @param content the activity content + * @param replace whether the snapshot replaces prior content for this (messageId, activityType) + * @return the activity snapshot event + */ + public AguiEvent.ActivitySnapshot createSnapshot( + String threadId, + String runId, + String messageId, + String activityType, + Map content, + boolean replace) { + return new AguiEvent.ActivitySnapshot( + threadId, runId, messageId, activityType, content, replace); + } + + /** + * Create an {@link AguiEvent.ActivityDelta} event by comparing before and after content. + * + * @param threadId the thread id + * @param runId the run id + * @param messageId the message id the activity is attached to + * @param activityType the activity type + * @param before the activity content before changes + * @param after the activity content after changes + * @return the activity delta event, or null if there are no changes + */ + public AguiEvent.ActivityDelta createDelta( + String threadId, + String runId, + String messageId, + String activityType, + Map before, + Map after) { + List operations = AguiJsonDiff.computeDelta(before, after, ""); + if (operations.isEmpty()) { + return null; + } + return new AguiEvent.ActivityDelta(threadId, runId, messageId, activityType, operations); + } + + /** + * Check if there are any differences between two activity content maps. + * + * @param before the content before changes + * @param after the content after changes + * @return true if there are differences + */ + public boolean hasChanges(Map before, Map after) { + return AguiJsonDiff.hasChanges(before, after); + } +} diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/converter/AguiJsonDiff.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/converter/AguiJsonDiff.java new file mode 100644 index 0000000000..2439cd526b --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/converter/AguiJsonDiff.java @@ -0,0 +1,111 @@ +/* + * 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.converter; + +import io.agentscope.core.agui.event.AguiEvent.JsonPatchOperation; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; + +/** + * Shared RFC 6902 diff helpers used by {@link AguiStateConverter} and + * {@link AguiActivityConverter}. + * + *

Emits only {@code add} / {@code remove} / {@code replace} operations with RFC 6901 escaping + * so the result is symmetric with {@link io.agentscope.core.agui.store.AguiJsonPatch#apply}. + */ +final class AguiJsonDiff { + + private AguiJsonDiff() {} + + /** + * Compute the JSON Patch operations needed to transform {@code before} into {@code after}. + * + * @param before the state before changes, may be null + * @param after the state after changes, may be null + * @param basePath the base JSON Pointer path + * @return list of patch operations, never null + */ + @SuppressWarnings("unchecked") + static List computeDelta( + Map before, Map after, String basePath) { + List operations = new ArrayList<>(); + + if (before == null && after == null) { + return operations; + } + if (before == null) { + before = Map.of(); + } + if (after == null) { + after = Map.of(); + } + + Set allKeys = new HashSet<>(); + allKeys.addAll(before.keySet()); + allKeys.addAll(after.keySet()); + + for (String key : allKeys) { + String path = basePath + "/" + escapeJsonPointer(key); + Object beforeValue = before.get(key); + Object afterValue = after.get(key); + + if (!before.containsKey(key)) { + operations.add(JsonPatchOperation.add(path, afterValue)); + } else if (!after.containsKey(key)) { + operations.add(JsonPatchOperation.remove(path)); + } else if (!Objects.equals(beforeValue, afterValue)) { + if (beforeValue instanceof Map && afterValue instanceof Map) { + operations.addAll( + computeDelta( + (Map) beforeValue, + (Map) afterValue, + path)); + } else { + operations.add(JsonPatchOperation.replace(path, afterValue)); + } + } + } + + return operations; + } + + /** + * Check whether two states differ. + * + * @param before the state before changes, may be null + * @param after the state after changes, may be null + * @return true if there are differences + */ + static boolean hasChanges(Map before, Map after) { + return !computeDelta(before, after, "").isEmpty(); + } + + /** + * Escape a string for use in a JSON Pointer (RFC 6901). + * + *

Per RFC 6901, {@code ~} must be escaped as {@code ~0} and {@code /} as {@code ~1}. + * + * @param value the string to escape + * @return the escaped string + */ + static String escapeJsonPointer(String value) { + return value.replace("~", "~0").replace("/", "~1"); + } +} diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/converter/AguiStateConverter.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/converter/AguiStateConverter.java index 65c3abe32c..fe7c2c98ae 100644 --- a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/converter/AguiStateConverter.java +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/converter/AguiStateConverter.java @@ -17,12 +17,8 @@ import io.agentscope.core.agui.event.AguiEvent; import io.agentscope.core.agui.event.AguiEvent.JsonPatchOperation; -import java.util.ArrayList; -import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.Objects; -import java.util.Set; /** * Converter for state management in the AG-UI protocol. @@ -59,7 +55,7 @@ public AguiEvent.StateSnapshot createSnapshot( */ public AguiEvent.StateDelta createDelta( Map before, Map after, String threadId, String runId) { - List operations = computeDelta(before, after, ""); + List operations = AguiJsonDiff.computeDelta(before, after, ""); if (operations.isEmpty()) { return null; // No changes @@ -76,76 +72,6 @@ public AguiEvent.StateDelta createDelta( * @return true if there are differences */ public boolean hasChanges(Map before, Map after) { - return !computeDelta(before, after, "").isEmpty(); - } - - /** - * Compute the JSON Patch operations needed to transform "before" into "after". - * - * @param before The state before changes - * @param after The state after changes - * @param basePath The base JSON Pointer path - * @return List of JsonPatchOperations - */ - @SuppressWarnings("unchecked") - private List computeDelta( - Map before, Map after, String basePath) { - List operations = new ArrayList<>(); - - if (before == null && after == null) { - return operations; - } - - if (before == null) { - before = Map.of(); - } - - if (after == null) { - after = Map.of(); - } - - Set allKeys = new HashSet<>(); - allKeys.addAll(before.keySet()); - allKeys.addAll(after.keySet()); - - for (String key : allKeys) { - String path = basePath + "/" + escapeJsonPointer(key); - Object beforeValue = before.get(key); - Object afterValue = after.get(key); - - if (!before.containsKey(key)) { - // Key was added - operations.add(JsonPatchOperation.add(path, afterValue)); - } else if (!after.containsKey(key)) { - // Key was removed - operations.add(JsonPatchOperation.remove(path)); - } else if (!Objects.equals(beforeValue, afterValue)) { - // Value changed - if (beforeValue instanceof Map && afterValue instanceof Map) { - // Recurse into nested maps - operations.addAll( - computeDelta( - (Map) beforeValue, - (Map) afterValue, - path)); - } else { - // Replace value - operations.add(JsonPatchOperation.replace(path, afterValue)); - } - } - } - - return operations; - } - - /** - * Escape a string for use in a JSON Pointer (RFC 6901). - * - * @param value The string to escape - * @return The escaped string - */ - private String escapeJsonPointer(String value) { - // Per RFC 6901, ~ must be escaped as ~0 and / as ~1 - return value.replace("~", "~0").replace("/", "~1"); + return AguiJsonDiff.hasChanges(before, after); } } 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..68ab3fc9ca 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 @@ -27,6 +27,10 @@ import io.agentscope.core.agui.model.RunAgentInput; import io.agentscope.core.agui.runtime.AguiRuntimeContextRequest; import io.agentscope.core.agui.runtime.AguiRuntimeContextResolver; +import io.agentscope.core.agui.store.AguiSnapshotHydrator; +import io.agentscope.core.agui.store.AguiSnapshotStore; +import io.agentscope.core.agui.store.AguiThreadSnapshot; +import io.agentscope.core.agui.store.SnapshotRecordingEnricher; import java.util.ArrayList; import java.util.List; import java.util.Objects; @@ -72,6 +76,8 @@ public class AguiRequestProcessor { private final AguiAgentAdapterFactory adapterFactory; private final AguiResumeCoordinator resumeCoordinator; private final AguiRuntimeContextResolver runtimeContextResolver; + private final AguiSnapshotStore snapshotStore; + private final AguiSnapshotHydrator snapshotHydrator; private AguiRequestProcessor(Builder builder) { this.agentResolver = @@ -83,6 +89,8 @@ private AguiRequestProcessor(Builder builder) { : AguiAgentAdapterFactory.defaultFactory(); this.resumeCoordinator = new AguiResumeCoordinator(); this.runtimeContextResolver = builder.runtimeContextResolver; + this.snapshotStore = builder.snapshotStore; + this.snapshotHydrator = new AguiSnapshotHydrator(); } /** @@ -162,6 +170,12 @@ public ProcessResult process(AguiRuntimeContextRequest request) { config.isEmitRunFinishedAfterError())); } + // A new run starts: drop any trailing interrupt from a prior run so a + // resolved historical interrupt can never reappear on reconnect. + if (snapshotStore != null) { + snapshotStore.clearPendingInterrupts(threadId); + } + try { // Determine effective input based on server-side memory RunAgentInput effectiveInput = input; @@ -197,17 +211,59 @@ public ProcessResult process(AguiRuntimeContextRequest request) { runErrorSeen.get()); }) .doFinally( - signalType -> - resumeCoordinator.finishRun( - threadId, runId)); + signalType -> { + resumeCoordinator.finishRun(threadId, runId); + // Safety net for RUN_ERROR paths that bypass + // the enricher (e.g. adapter-produced errors). + flushSnapshotRecorder(threadId, runId); + }); } catch (Throwable error) { resumeCoordinator.finishRun(threadId, runId); + flushSnapshotRecorder(threadId, runId); return processorErrorEvents(input, error); } }); return new ProcessResult(agent, events, runtimeContext); } + /** + * Rehydrate the visible conversation for a thread from the presentation snapshot store. + * + *

Strictly read-only: resolves {@code threadId} / {@code runId} from the request input, + * looks up the stored snapshot, and delegates to {@link AguiSnapshotHydrator}. No agent is + * resolved, the resume coordinator is not mutated, and no adapter is created. When no store + * is configured (or the thread has no history) the minimal three-frame handshake is returned. + * + * @param request the AG-UI request context carrying input + * @return a flux of read-only hydrate frames + */ + public Flux hydrate(AguiRuntimeContextRequest request) { + Objects.requireNonNull(request, "request cannot be null"); + RunAgentInput input = request.getInput(); + String threadId = input != null ? input.getThreadId() : null; + String runId = input != null ? input.getRunId() : null; + if (threadId == null) { + threadId = "unknown"; + } + if (runId == null) { + runId = "unknown"; + } + AguiThreadSnapshot snapshot = + snapshotStore != null ? snapshotStore.find(threadId).orElse(null) : null; + return Flux.fromIterable(snapshotHydrator.hydrate(snapshot, threadId, runId)); + } + + private void flushSnapshotRecorder(String threadId, String runId) { + SnapshotRecordingEnricher recorder = snapshotRecorder(); + if (recorder != null) { + recorder.flush(threadId, runId); + } + } + + private SnapshotRecordingEnricher snapshotRecorder() { + return config != null ? config.getSnapshotRecorder() : null; + } + private Flux processorErrorEvents(RunAgentInput input, Throwable error) { String errorMessage = error.getMessage() != null ? error.getMessage() : error.getClass().getSimpleName(); @@ -363,6 +419,7 @@ public static class Builder { private AguiAdapterConfig config; private AguiAgentAdapterFactory adapterFactory; private AguiRuntimeContextResolver runtimeContextResolver; + private AguiSnapshotStore snapshotStore; /** * Set the agent resolver. @@ -409,6 +466,19 @@ public Builder runtimeContextResolver(AguiRuntimeContextResolver runtimeContextR return this; } + /** + * Set the presentation snapshot store used to hydrate reconnects and clear trailing + * interrupts. Optional; when null, {@link #hydrate} returns the empty handshake and no + * snapshots are recorded. + * + * @param snapshotStore the snapshot store, or null + * @return This builder + */ + public Builder snapshotStore(AguiSnapshotStore snapshotStore) { + this.snapshotStore = snapshotStore; + return this; + } + /** * Build the processor. * diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/AguiJsonPatch.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/AguiJsonPatch.java new file mode 100644 index 0000000000..7db6de6c5d --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/AguiJsonPatch.java @@ -0,0 +1,250 @@ +/* + * 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.store; + +import io.agentscope.core.agui.event.AguiEvent.JsonPatchOperation; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Minimal RFC 6902 patch applier for the AG-UI presentation snapshot store. + * + *

Supports {@code add} / {@code replace} / {@code remove}; unknown operations are ignored and + * logged at debug. JSON Pointer tokens are split on {@code /}, unescaping {@code ~1} to {@code /} + * and then {@code ~0} to {@code ~} (order matters). Nested {@link Map} values are traversed; + * {@link List} index segments and the {@code -} append token are supported. + * + *

This is the inverse of {@link io.agentscope.core.agui.converter.AguiJsonDiff}, which emits + * exactly {@code add} / {@code remove} / {@code replace} with the same escaping, so a delta + * computed from one state always applies cleanly onto a matching state. + */ +final class AguiJsonPatch { + + private static final Logger logger = LoggerFactory.getLogger(AguiJsonPatch.class); + + private AguiJsonPatch() {} + + /** + * Apply a list of patch operations to a (defensively copied) state map. + * + * @param target the state to patch, may be null + * @param ops the patch operations + * @return a new map with the operations applied; never mutates the input + */ + static Map apply(Map target, List ops) { + Map result = deepCopyMap(target); + if (ops == null || ops.isEmpty()) { + return result; + } + for (JsonPatchOperation op : ops) { + switch (op.op()) { + case "add" -> applyAdd(result, op.path(), op.value()); + case "replace" -> applyReplace(result, op.path(), op.value()); + case "remove" -> applyRemove(result, op.path()); + default -> + logger.debug( + "Ignoring unknown JSON Patch op '{}' at {}", op.op(), op.path()); + } + } + return result; + } + + private static void applyAdd(Map root, String pointer, Object value) { + List tokens = parsePointer(pointer); + if (tokens.isEmpty()) { + return; + } + Object parent = navigateParent(root, tokens); + if (parent == null) { + logger.debug("Cannot add to missing parent at {}", pointer); + return; + } + String last = tokens.get(tokens.size() - 1); + Object coerced = deepCopyValue(value); + if (parent instanceof Map) { + asMap(parent).put(last, coerced); + } else if (parent instanceof List) { + List list = asList(parent); + if ("-".equals(last)) { + list.add(coerced); + } else { + int index = parseIndex(last); + if (index < 0) { + logger.debug("Cannot add to non-numeric list index '{}' at {}", last, pointer); + return; + } + if (index >= list.size()) { + list.add(coerced); + } else { + list.add(index, coerced); + } + } + } + } + + private static void applyReplace(Map root, String pointer, Object value) { + List tokens = parsePointer(pointer); + if (tokens.isEmpty()) { + return; + } + Object parent = navigateParent(root, tokens); + if (parent == null) { + logger.debug("Cannot replace missing parent at {}", pointer); + return; + } + String last = tokens.get(tokens.size() - 1); + Object coerced = deepCopyValue(value); + if (parent instanceof Map) { + if (asMap(parent).containsKey(last)) { + asMap(parent).put(last, coerced); + } else { + logger.debug("Cannot replace missing key '{}' at {}", last, pointer); + } + } else if (parent instanceof List) { + List list = asList(parent); + int index = parseIndex(last); + if (index < 0 || index >= list.size()) { + logger.debug("Cannot replace list index '{}' at {}", last, pointer); + return; + } + list.set(index, coerced); + } + } + + private static void applyRemove(Map root, String pointer) { + List tokens = parsePointer(pointer); + if (tokens.isEmpty()) { + return; + } + Object parent = navigateParent(root, tokens); + if (parent == null) { + logger.debug("Cannot remove from missing parent at {}", pointer); + return; + } + String last = tokens.get(tokens.size() - 1); + if (parent instanceof Map) { + asMap(parent).remove(last); + } else if (parent instanceof List) { + List list = asList(parent); + int index = parseIndex(last); + if (index >= 0 && index < list.size()) { + list.remove(index); + } + } + } + + private static Object navigateParent(Object root, List tokens) { + Object current = root; + for (int i = 0; i < tokens.size() - 1; i++) { + current = descend(current, tokens.get(i)); + if (current == null) { + return null; + } + } + return current; + } + + @SuppressWarnings("unchecked") + private static Object descend(Object current, String token) { + if (current instanceof Map) { + return ((Map) current).get(token); + } + if (current instanceof List) { + List list = (List) current; + int index = parseIndex(token); + if (index >= 0 && index < list.size()) { + return list.get(index); + } + } + return null; + } + + private static List parsePointer(String pointer) { + if (pointer == null || pointer.isEmpty()) { + return List.of(); + } + if (!pointer.startsWith("/")) { + return List.of(unescape(pointer)); + } + String body = pointer.substring(1); + if (body.isEmpty()) { + return List.of(); + } + String[] parts = body.split("/", -1); + List tokens = new ArrayList<>(parts.length); + for (String part : parts) { + tokens.add(unescape(part)); + } + return tokens; + } + + /** Unescape a JSON Pointer reference token: {@code ~1} to {@code /} then {@code ~0} to {@code ~}. */ + private static String unescape(String token) { + return token.replace("~1", "/").replace("~0", "~"); + } + + private static int parseIndex(String token) { + if (token == null || token.isEmpty()) { + return -1; + } + try { + return Integer.parseInt(token); + } catch (NumberFormatException e) { + return -1; + } + } + + @SuppressWarnings("unchecked") + private static Map asMap(Object value) { + return (Map) value; + } + + @SuppressWarnings("unchecked") + private static List asList(Object value) { + return (List) value; + } + + @SuppressWarnings("unchecked") + private static Map deepCopyMap(Map source) { + Map copy = new LinkedHashMap<>(); + if (source == null) { + return copy; + } + for (Map.Entry entry : source.entrySet()) { + copy.put(entry.getKey(), deepCopyValue(entry.getValue())); + } + return copy; + } + + @SuppressWarnings("unchecked") + private static Object deepCopyValue(Object value) { + if (value instanceof Map) { + return deepCopyMap((Map) value); + } + if (value instanceof List) { + List copy = new ArrayList<>(((List) value).size()); + for (Object element : (List) value) { + copy.add(deepCopyValue(element)); + } + return copy; + } + return value; + } +} diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/AguiSnapshotAccumulator.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/AguiSnapshotAccumulator.java new file mode 100644 index 0000000000..88e73ed567 --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/AguiSnapshotAccumulator.java @@ -0,0 +1,414 @@ +/* + * 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.store; + +import io.agentscope.core.agui.event.AguiEvent; +import io.agentscope.core.agui.event.AguiEvent.JsonPatchOperation; +import io.agentscope.core.agui.model.AguiFunctionCall; +import io.agentscope.core.agui.model.AguiMessage; +import io.agentscope.core.agui.model.AguiToolCall; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; + +/** + * Per-{@code threadId:runId} mutable folder that consumes outbound {@link AguiEvent}s and produces + * a materialized {@link AguiThreadSnapshot}. + * + *

Seeded from {@link AguiSnapshotStore#find(String)} so messages / state / activities from + * earlier runs survive. This is where the former CopilotKit replay workarounds + * (resolved-interrupt suppression and dangling-tool-call synthesis) are absorbed: the snapshot + * only ever retains the trailing unresolved interrupt, and dangling tool calls are closed with a + * synthetic empty result so the browser never renders a stuck spinner — except for the tool that + * belongs to an open interrupt, which is left pending. + */ +final class AguiSnapshotAccumulator { + + private final String threadId; + private final String runId; + + /** Ordered message transcript, keyed by message id so upserts preserve position. */ + private final LinkedHashMap messages = new LinkedHashMap<>(); + + /** Folded AG-UI state. */ + private Map state = new LinkedHashMap<>(); + + /** Activity frames keyed by (messageId \0 activityType), insertion-ordered. */ + private final LinkedHashMap activities = + new LinkedHashMap<>(); + + /** Assistant message builders keyed by message id, for text/tool-call folding. */ + private final LinkedHashMap assistantBuilders = new LinkedHashMap<>(); + + /** Tool call frames keyed by tool call id. */ + private final LinkedHashMap toolCallFrames = new LinkedHashMap<>(); + + /** Message id of the assistant turn currently receiving tool calls. */ + private String currentAssistantMessageId; + + private AguiEvent.RunFinishedOutcome pendingOutcome; + + AguiSnapshotAccumulator(String threadId, String runId, AguiThreadSnapshot seed) { + this.threadId = threadId; + this.runId = runId; + if (seed != null) { + for (AguiMessage message : seed.messages()) { + if (message != null && message.getId() != null) { + messages.put(message.getId(), message); + } + } + state.putAll(seed.state()); + for (AguiThreadSnapshot.ActivityFrame frame : seed.activities()) { + activities.put(activityKey(frame.messageId(), frame.activityType()), frame); + } + } + } + + /** + * Consume one outbound AG-UI event and fold it into the accumulator. + * + * @param event the event to consume + */ + void consume(AguiEvent event) { + if (event == null) { + return; + } + if (event instanceof AguiEvent.RunStarted runStarted) { + captureInputMessages(runStarted); + } else if (event instanceof AguiEvent.TextMessageStart start) { + startAssistantText(start.messageId(), start.role()); + } else if (event instanceof AguiEvent.TextMessageContent content) { + appendAssistantText(content.messageId(), content.delta()); + } else if (event instanceof AguiEvent.TextMessageEnd end) { + endAssistantText(end.messageId()); + } else if (event instanceof AguiEvent.TextMessageChunk chunk) { + handleTextMessageChunk(chunk); + } else if (event instanceof AguiEvent.ToolCallStart toolCallStart) { + startToolCall(toolCallStart.toolCallId(), toolCallStart.toolCallName(), null); + } else if (event instanceof AguiEvent.ToolCallArgs args) { + appendToolCallArgs(args.toolCallId(), args.delta()); + } else if (event instanceof AguiEvent.ToolCallChunk toolCallChunk) { + handleToolCallChunk(toolCallChunk); + } else if (event instanceof AguiEvent.ToolCallResult result) { + recordToolCallResult(result); + } else if (event instanceof AguiEvent.StateSnapshot stateSnapshot) { + state = new LinkedHashMap<>(stateSnapshot.snapshot()); + } else if (event instanceof AguiEvent.StateDelta stateDelta) { + state = AguiJsonPatch.apply(state, stateDelta.delta()); + } else if (event instanceof AguiEvent.ActivitySnapshot activitySnapshot) { + applyActivitySnapshot(activitySnapshot); + } else if (event instanceof AguiEvent.ActivityDelta activityDelta) { + applyActivityDelta(activityDelta); + } else if (event instanceof AguiEvent.RunFinished runFinished) { + if (runFinished.outcome() instanceof AguiEvent.RunFinishedInterruptOutcome interrupt) { + pendingOutcome = interrupt; + } else { + pendingOutcome = null; + } + } else if (event instanceof AguiEvent.RunError) { + pendingOutcome = null; + } + } + + /** + * Materialize the accumulated state into an immutable snapshot. + * + *

This is a read-only view of the current accumulated state: it builds a fresh + * transcript without mutating the accumulator, so it is safe to call repeatedly and even after + * additional events have been consumed. + * + * @return the materialized snapshot + */ + AguiThreadSnapshot materialize() { + // 1. Assistant messages carry accumulated tool calls. Work on a fresh copy so the live + // transcript stays untouched and materialize stays idempotent. + LinkedHashMap transcript = new LinkedHashMap<>(messages); + for (AssistantBuilder builder : assistantBuilders.values()) { + List calls = + builder.toolCallIds.stream() + .map(toolCallFrames::get) + .filter(java.util.Objects::nonNull) + .map( + frame -> + new AguiToolCall( + frame.toolCallId, + new AguiFunctionCall( + frame.name != null + ? frame.name + : "unknown", + frame.args.toString()))) + .toList(); + transcript.put( + builder.messageId, + AguiMessage.textMessage( + builder.messageId, builder.role, builder.text.toString(), calls, null)); + } + + // 2. Resolved tool calls already appended their tool result messages (on result). + + // 3. Dangling tool calls get a synthetic empty result, unless the tool belongs to an open + // interrupt (then it genuinely still awaits the user and is left open). + Set openToolCallIds = openInterruptToolCallIds(); + for (ToolCallFrame frame : toolCallFrames.values()) { + if (frame.resolved) { + continue; + } + if (openToolCallIds.contains(frame.toolCallId)) { + continue; + } + String resultId = syntheticResultId(frame.toolCallId); + transcript.putIfAbsent( + resultId, AguiMessage.toolMessage(resultId, frame.toolCallId, null)); + } + + return new AguiThreadSnapshot( + threadId, + List.copyOf(transcript.values()), + state, + List.copyOf(activities.values()), + pendingOutcome, + runId, + System.currentTimeMillis()); + } + + private void captureInputMessages(AguiEvent.RunStarted runStarted) { + AguiEvent.RunStarted start = runStarted; + if (start.input() == null || start.input().getMessages() == null) { + return; + } + for (AguiMessage message : start.input().getMessages()) { + if (message == null || message.getId() == null) { + continue; + } + messages.putIfAbsent(message.getId(), message); + } + } + + private void startAssistantText(String messageId, String role) { + if (messageId == null) { + return; + } + currentAssistantMessageId = messageId; + AssistantBuilder builder = + assistantBuilders.computeIfAbsent( + messageId, ignored -> new AssistantBuilder(messageId)); + builder.role = role != null ? role : "assistant"; + // Ensure a placeholder exists in the transcript so ordering is preserved. + messages.putIfAbsent( + messageId, AguiMessage.textMessage(messageId, builder.role, null, List.of(), null)); + } + + private void appendAssistantText(String messageId, String delta) { + if (messageId == null || delta == null) { + return; + } + currentAssistantMessageId = messageId; + AssistantBuilder builder = + assistantBuilders.computeIfAbsent( + messageId, ignored -> new AssistantBuilder(messageId)); + builder.text.append(delta); + messages.put( + messageId, + AguiMessage.textMessage( + messageId, builder.role, builder.text.toString(), List.of(), null)); + } + + private void endAssistantText(String messageId) { + if (messageId == null) { + return; + } + AssistantBuilder builder = assistantBuilders.get(messageId); + if (builder != null) { + messages.put( + messageId, + AguiMessage.textMessage( + messageId, builder.role, builder.text.toString(), List.of(), null)); + } + currentAssistantMessageId = messageId; + } + + private void handleTextMessageChunk(AguiEvent.TextMessageChunk chunk) { + if (chunk.messageId() == null) { + return; + } + startAssistantText(chunk.messageId(), chunk.role() != null ? chunk.role() : "assistant"); + if (chunk.delta() != null) { + appendAssistantText(chunk.messageId(), chunk.delta()); + } + } + + private void startToolCall(String toolCallId, String toolCallName, String parentMessageId) { + if (toolCallId == null) { + return; + } + ToolCallFrame frame = + toolCallFrames.computeIfAbsent( + toolCallId, ignored -> new ToolCallFrame(toolCallId)); + if (toolCallName != null && !toolCallName.isBlank()) { + frame.name = toolCallName; + } + linkToolCallToAssistant(frame, parentMessageId); + } + + private void appendToolCallArgs(String toolCallId, String delta) { + if (toolCallId == null || delta == null) { + return; + } + ToolCallFrame frame = toolCallFrames.get(toolCallId); + if (frame != null) { + frame.args.append(delta); + } + } + + private void handleToolCallChunk(AguiEvent.ToolCallChunk chunk) { + if (chunk.toolCallId() == null) { + return; + } + // Chunk mode carries an explicit parentMessageId, so prefer it over the heuristic. + startToolCall(chunk.toolCallId(), chunk.toolCallName(), chunk.parentMessageId()); + if (chunk.delta() != null) { + appendToolCallArgs(chunk.toolCallId(), chunk.delta()); + } + } + + private void recordToolCallResult(AguiEvent.ToolCallResult result) { + ToolCallFrame frame = toolCallFrames.get(result.toolCallId()); + if (frame == null) { + frame = new ToolCallFrame(result.toolCallId()); + toolCallFrames.put(result.toolCallId(), frame); + } + frame.resultContent = result.content(); + frame.resultMessageId = result.messageId(); + frame.resolved = true; + String resultId = + result.messageId() != null + ? result.messageId() + : syntheticResultId(result.toolCallId()); + messages.putIfAbsent( + resultId, AguiMessage.toolMessage(resultId, result.toolCallId(), result.content())); + } + + private void linkToolCallToAssistant(ToolCallFrame frame, String parentMessageId) { + // Prefer an explicit parent (chunk mode) over the heuristic last-text-message owner. + String owner = parentMessageId != null ? parentMessageId : currentAssistantMessageId; + if (owner == null) { + owner = "assistant:" + UUID.randomUUID(); + currentAssistantMessageId = owner; + } + AssistantBuilder builder = assistantBuilders.computeIfAbsent(owner, AssistantBuilder::new); + // Ensure a placeholder exists in the transcript so ordering is preserved even when the + // parent's TextMessageStart has not (or will not) arrive. + messages.putIfAbsent( + owner, AguiMessage.textMessage(owner, builder.role, null, List.of(), null)); + builder.addToolCallId(frame.toolCallId); + } + + private void applyActivitySnapshot(AguiEvent.ActivitySnapshot snapshot) { + String key = activityKey(snapshot.messageId(), snapshot.activityType()); + boolean replace = snapshot.replace() == null || snapshot.replace(); + if (replace) { + activities.put( + key, + new AguiThreadSnapshot.ActivityFrame( + snapshot.messageId(), snapshot.activityType(), snapshot.content())); + } else { + AguiThreadSnapshot.ActivityFrame existing = activities.get(key); + Map merged = + new LinkedHashMap<>(existing != null ? existing.content() : Map.of()); + merged.putAll(snapshot.content()); + activities.put( + key, + new AguiThreadSnapshot.ActivityFrame( + snapshot.messageId(), snapshot.activityType(), merged)); + } + } + + private void applyActivityDelta(AguiEvent.ActivityDelta delta) { + String key = activityKey(delta.messageId(), delta.activityType()); + AguiThreadSnapshot.ActivityFrame existing = activities.get(key); + Map content = + new LinkedHashMap<>(existing != null ? existing.content() : Map.of()); + List patch = delta.patch(); + if (patch != null && !patch.isEmpty()) { + content = AguiJsonPatch.apply(content, patch); + } + activities.put( + key, + new AguiThreadSnapshot.ActivityFrame( + delta.messageId(), delta.activityType(), content)); + } + + private Set openInterruptToolCallIds() { + if (!(pendingOutcome instanceof AguiEvent.RunFinishedInterruptOutcome interrupt)) { + return Set.of(); + } + List interrupts = interrupt.interrupts(); + if (interrupts == null || interrupts.isEmpty()) { + return Set.of(); + } + Set ids = new HashSet<>(); + for (AguiEvent.Interrupt i : interrupts) { + if (i.toolCallId() != null) { + ids.add(i.toolCallId()); + } + } + return ids; + } + + private static String activityKey(String messageId, String activityType) { + return messageId + "\u0000" + activityType; + } + + private static String syntheticResultId(String toolCallId) { + return toolCallId + ":result"; + } + + /** Mutable builder for an assistant message's text and linked tool calls. */ + private static final class AssistantBuilder { + final String messageId; + String role = "assistant"; + final StringBuilder text = new StringBuilder(); + final List toolCallIds = new ArrayList<>(); + + AssistantBuilder(String messageId) { + this.messageId = messageId; + } + + void addToolCallId(String toolCallId) { + if (!toolCallIds.contains(toolCallId)) { + toolCallIds.add(toolCallId); + } + } + } + + /** Mutable holder for a tool call's streamed args and result. */ + private static final class ToolCallFrame { + final String toolCallId; + String name; + final StringBuilder args = new StringBuilder(); + String resultContent; + String resultMessageId; + boolean resolved; + + ToolCallFrame(String toolCallId) { + this.toolCallId = toolCallId; + } + } +} diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/AguiSnapshotHydrator.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/AguiSnapshotHydrator.java new file mode 100644 index 0000000000..278732cb36 --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/AguiSnapshotHydrator.java @@ -0,0 +1,88 @@ +/* + * 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.store; + +import io.agentscope.core.agui.event.AguiEvent; +import io.agentscope.core.agui.model.AguiMessage; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +/** + * Replays a stored {@link AguiThreadSnapshot} as a read-only sequence of AG-UI frames. + * + *

Frames are emitted in this exact order: + * + * 

{@code
+ * RUN_STARTED(threadId, runId)
+ * MESSAGES_SNAPSHOT(messages)
+ * STATE_SNAPSHOT(state)                     // omitted when state is empty
+ * ACTIVITY_SNAPSHOT(...) per ActivityFrame  // omitted when none
+ * RUN_FINISHED(result=null, outcome=pendingOutcome)
+ * }
+ * + *

A null {@code pendingOutcome} serializes as a plain successful run — exactly the shape the + * former CopilotKit empty-handshake forced via post-processing. An empty or missing snapshot + * produces the minimal three-frame handshake ({@code RUN_STARTED} → {@code MESSAGES_SNAPSHOT([])} + * → {@code RUN_FINISHED}). + * + *

Hydration is strictly read-only: it never mutates the snapshot store, the agent state store, + * or the resume coordinator. + */ +public final class AguiSnapshotHydrator { + + /** Create a new stateless hydrator. */ + public AguiSnapshotHydrator() {} + + /** + * Build the hydrate frame sequence for a snapshot. + * + * @param snapshot the stored snapshot, or null when the thread has no history + * @param threadId the thread id for the emitted frames + * @param runId the run id for the emitted frames + * @return the ordered AG-UI frames + */ + public List hydrate(AguiThreadSnapshot snapshot, String threadId, String runId) { + List events = new ArrayList<>(); + events.add(new AguiEvent.RunStarted(threadId, runId)); + + List messages = + snapshot != null && snapshot.messages() != null ? snapshot.messages() : List.of(); + events.add(new AguiEvent.MessagesSnapshot(threadId, runId, messages)); + + Map state = snapshot != null ? snapshot.state() : Map.of(); + if (state != null && !state.isEmpty()) { + events.add(new AguiEvent.StateSnapshot(threadId, runId, state)); + } + + if (snapshot != null && snapshot.activities() != null) { + for (AguiThreadSnapshot.ActivityFrame frame : snapshot.activities()) { + events.add( + new AguiEvent.ActivitySnapshot( + threadId, + runId, + frame.messageId(), + frame.activityType(), + frame.content(), + true)); + } + } + + AguiEvent.RunFinishedOutcome outcome = snapshot != null ? snapshot.pendingOutcome() : null; + events.add(new AguiEvent.RunFinished(threadId, runId, null, outcome)); + return events; + } +} diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/AguiSnapshotStore.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/AguiSnapshotStore.java new file mode 100644 index 0000000000..bacdb56cdc --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/AguiSnapshotStore.java @@ -0,0 +1,72 @@ +/* + * 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.store; + +import java.util.Optional; + +/** + * Presentation-state store for AG-UI threads. + * + *

Holds the derived presentation state (materialized messages / state / activity) so a + * reconnecting client can rebuild the visible conversation without re-running the agent. This is + * presentation-only data: it is safe to lose, and it is not a source of truth for + * human-in-the-loop interrupts. Authoritative agent state and the live HITL contract are owned by + * the agent state store and the resume coordinator; hydrate is strictly read-only and never + * mutates either. + * + *

Because the store only ever retains the trailing unresolved interrupt, a resolved + * historical interrupt cannot be revived on reconnect — that failure mode is removed at the data + * model rather than filtered after the fact. + */ +public interface AguiSnapshotStore { + + /** + * Persist a materialized snapshot for a thread. + * + * @param snapshot the snapshot to store + */ + void save(AguiThreadSnapshot snapshot); + + /** + * Look up the snapshot for a thread. + * + * @param threadId the thread id + * @return the snapshot, or empty if none is stored + */ + Optional find(String threadId); + + /** + * Delete the snapshot for a thread. + * + * @param threadId the thread id + */ + void delete(String threadId); + + /** + * Drop the trailing interrupt outcome for a thread, if any. + * + *

Called when a new run starts so a previously-unresolved interrupt cannot reappear on + * reconnect. Implementations that retain only the trailing interrupt can satisfy this with a + * read-modify-write via {@link AguiThreadSnapshot#withoutPendingOutcome()}. + * + * @param threadId the thread id + */ + default void clearPendingInterrupts(String threadId) { + find(threadId) + .filter(snapshot -> snapshot.pendingOutcome() != null) + .ifPresent(snapshot -> save(snapshot.withoutPendingOutcome())); + } +} diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/AguiThreadSnapshot.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/AguiThreadSnapshot.java new file mode 100644 index 0000000000..bdf2e7a959 --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/AguiThreadSnapshot.java @@ -0,0 +1,110 @@ +/* + * 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.store; + +import io.agentscope.core.agui.event.AguiEvent; +import io.agentscope.core.agui.model.AguiMessage; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * Immutable materialized presentation state for one AG-UI thread. + * + *

A snapshot captures what a reconnecting client should draw: the message transcript, the + * folded state, any activity frames, and the trailing run id. It is presentation-only: + * derived from the live event stream, safe to lose, and not a source of truth for human-in-the-loop + * contracts (the agent state store and resume coordinator own those). + * + * @param threadId the AG-UI thread id + * @param messages the materialized conversation messages + * @param state the folded AG-UI state + * @param activities the materialized activity frames + * @param pendingOutcome the trailing unresolved run outcome, or null when the run completed + * normally; only the trailing interrupt is ever retained + * @param lastRunId the run id that produced this snapshot + * @param updatedAt the wall-clock millis when the snapshot was materialized + */ +public record AguiThreadSnapshot( + String threadId, + List messages, + Map state, + List activities, + AguiEvent.RunFinishedOutcome pendingOutcome, + String lastRunId, + long updatedAt) { + + /** + * One materialized activity frame, keyed by {@code (messageId, activityType)}. + * + * @param messageId the message id the activity is attached to + * @param activityType the activity type + * @param content the activity content + */ + public record ActivityFrame( + String messageId, String activityType, Map content) { + public ActivityFrame { + content = + content != null + ? Collections.unmodifiableMap(new LinkedHashMap<>(content)) + : Collections.emptyMap(); + } + } + + /** + * Canonical constructor with defensive copies, matching the style of + * {@link AguiEvent.MessagesSnapshot}. + */ + public AguiThreadSnapshot { + messages = + messages != null + ? Collections.unmodifiableList(new ArrayList<>(messages)) + : Collections.emptyList(); + state = + state != null + ? Collections.unmodifiableMap(new LinkedHashMap<>(state)) + : Collections.emptyMap(); + activities = + activities != null + ? Collections.unmodifiableList(new ArrayList<>(activities)) + : Collections.emptyList(); + } + + /** + * Create an empty snapshot for a thread. + * + * @param threadId the thread id + * @return an empty snapshot with no messages, state, activities or pending outcome + */ + public static AguiThreadSnapshot empty(String threadId) { + return new AguiThreadSnapshot(threadId, List.of(), Map.of(), List.of(), null, null, 0L); + } + + /** + * Return a copy of this snapshot with the trailing interrupt outcome cleared. + * + *

Used when a new run starts so a previously-unresolved interrupt cannot reappear on + * reconnect. + * + * @return a snapshot with a null pending outcome + */ + public AguiThreadSnapshot withoutPendingOutcome() { + return new AguiThreadSnapshot( + threadId, messages, state, activities, null, lastRunId, updatedAt); + } +} diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/InMemoryAguiSnapshotStore.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/InMemoryAguiSnapshotStore.java new file mode 100644 index 0000000000..0488708070 --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/InMemoryAguiSnapshotStore.java @@ -0,0 +1,90 @@ +/* + * 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.store; + +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +/** + * In-memory {@link AguiSnapshotStore} backed by a {@link ConcurrentHashMap}. + * + *

Retains at most {@code maxThreads} snapshots; when the limit is exceeded the snapshot with the + * oldest {@code updatedAt} is evicted in a single pass (no external dependencies). + */ +public final class InMemoryAguiSnapshotStore implements AguiSnapshotStore { + + private final int maxThreads; + private final ConcurrentMap snapshots = new ConcurrentHashMap<>(); + + /** Create a store with the default capacity of 1000 threads. */ + public InMemoryAguiSnapshotStore() { + this(1000); + } + + /** + * Create a store with a fixed capacity. + * + * @param maxThreads the maximum number of threads to retain + */ + public InMemoryAguiSnapshotStore(int maxThreads) { + if (maxThreads <= 0) { + throw new IllegalArgumentException("maxThreads must be positive"); + } + this.maxThreads = maxThreads; + } + + @Override + public void save(AguiThreadSnapshot snapshot) { + if (snapshot == null || snapshot.threadId() == null) { + return; + } + snapshots.put(snapshot.threadId(), snapshot); + evictIfOverCapacity(); + } + + @Override + public Optional find(String threadId) { + if (threadId == null) { + return Optional.empty(); + } + return Optional.ofNullable(snapshots.get(threadId)); + } + + @Override + public void delete(String threadId) { + if (threadId != null) { + snapshots.remove(threadId); + } + } + + private void evictIfOverCapacity() { + while (snapshots.size() > maxThreads) { + Map.Entry oldest = null; + for (Map.Entry entry : snapshots.entrySet()) { + if (oldest == null + || entry.getValue().updatedAt() < oldest.getValue().updatedAt()) { + oldest = entry; + } + } + if (oldest == null) { + break; + } + snapshots.remove(oldest.getKey(), oldest.getValue()); + } + } +} diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/SnapshotRecordingEnricher.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/SnapshotRecordingEnricher.java new file mode 100644 index 0000000000..dc46de4f34 --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/main/java/io/agentscope/core/agui/store/SnapshotRecordingEnricher.java @@ -0,0 +1,116 @@ +/* + * 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.store; + +import io.agentscope.core.agui.adapter.strategy.AguiEventEnricher; +import io.agentscope.core.agui.adapter.strategy.AguiStreamContext; +import io.agentscope.core.agui.event.AguiEvent; +import io.agentscope.core.event.AgentEvent; +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +/** + * Records outbound AG-UI events into an {@link AguiSnapshotStore}. + * + *

Appended last in the enricher chain so it observes fully enriched frames. Feeds every + * frame to a per-{@code threadId:runId} {@link AguiSnapshotAccumulator}; on a terminal + * {@link AguiEvent.RunFinished} / {@link AguiEvent.RunError} it materializes and persists the + * snapshot, then drops the accumulator. Because the enricher runs on both the + * {@code AgentEventConverterRegistry.convert()} path and the framework + * {@code enrich(null, finishPendingEvents, ctx)} path, it sees frames from all converters. + * + *

{@link #flush(String, String)} is a safety net for streams that terminate without emitting a + * terminal frame (e.g. {@code RUN_ERROR} produced directly by the adapter, which bypasses the + * enricher). + */ +public final class SnapshotRecordingEnricher implements AguiEventEnricher { + + private final AguiSnapshotStore store; + private final ConcurrentMap accumulators = + new ConcurrentHashMap<>(); + + /** Create a recording enricher backed by the given store. */ + public SnapshotRecordingEnricher(AguiSnapshotStore store) { + this.store = store; + } + + /** The store this enricher records into. */ + public AguiSnapshotStore getStore() { + return store; + } + + @Override + public List enrich( + AgentEvent source, List events, AguiStreamContext context) { + if (store == null || context == null || events == null || events.isEmpty()) { + return events; + } + String threadId = context.getThreadId(); + String runId = context.getRunId(); + if (threadId == null || runId == null) { + return events; + } + String key = accumulatorKey(threadId, runId); + AguiSnapshotAccumulator accumulator = + accumulators.computeIfAbsent( + key, + ignored -> + new AguiSnapshotAccumulator( + threadId, runId, store.find(threadId).orElse(null))); + for (AguiEvent event : events) { + accumulator.consume(event); + } + if (isTerminal(events)) { + store.save(accumulator.materialize()); + accumulators.remove(key); + } + return events; + } + + /** + * Flush and persist the accumulator for a run, if one is still in flight. + * + *

Called as a safety net when a stream terminates without a terminal frame reaching the + * enricher (e.g. an adapter-produced {@code RUN_ERROR}). Safe to call after a normal + * terminal frame: the accumulator will already have been removed and this is a no-op. + * + * @param threadId the thread id + * @param runId the run id + */ + public void flush(String threadId, String runId) { + if (store == null || threadId == null || runId == null) { + return; + } + AguiSnapshotAccumulator accumulator = accumulators.remove(accumulatorKey(threadId, runId)); + if (accumulator != null) { + store.save(accumulator.materialize()); + } + } + + private static boolean isTerminal(List events) { + for (AguiEvent event : events) { + if (event instanceof AguiEvent.RunFinished || event instanceof AguiEvent.RunError) { + return true; + } + } + return false; + } + + private static String accumulatorKey(String threadId, String runId) { + return threadId + ":" + runId; + } +} diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/adapter/AguiAdapterConfigTest.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/adapter/AguiAdapterConfigTest.java index ee1403cf4a..345e9e9190 100644 --- a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/adapter/AguiAdapterConfigTest.java +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/adapter/AguiAdapterConfigTest.java @@ -26,6 +26,9 @@ import io.agentscope.core.agui.adapter.strategy.AguiEventEnricher; import io.agentscope.core.agui.adapter.strategy.BaseEventPropertiesEnricher; import io.agentscope.core.agui.model.ToolMergeMode; +import io.agentscope.core.agui.store.AguiSnapshotStore; +import io.agentscope.core.agui.store.InMemoryAguiSnapshotStore; +import io.agentscope.core.agui.store.SnapshotRecordingEnricher; import io.agentscope.core.event.AgentEvent; import java.time.Duration; import java.util.Collections; @@ -325,6 +328,49 @@ void testBaseEventPropertiesEnricherDisabledByDefault() { assertTrue(config.getEventEnrichers().isEmpty()); } + @Test + void testSnapshotStoreDisabledByDefault() { + AguiAdapterConfig config = AguiAdapterConfig.builder().build(); + + assertFalse(config.isSnapshotStoreEnabled()); + assertNull(config.getSnapshotStore()); + // No SnapshotRecordingEnricher in the chain by default. + assertTrue( + config.getEventEnrichers().stream() + .noneMatch(e -> e instanceof SnapshotRecordingEnricher)); + } + + @Test + void testSnapshotRecordingEnricherAppendedLastWhenEnabled() { + AguiEventEnricher custom = (source, events, context) -> events; + AguiSnapshotStore store = new InMemoryAguiSnapshotStore(); + AguiAdapterConfig config = + AguiAdapterConfig.builder() + .baseEventPropertiesEnricherEnabled(true) + .addEventEnricher(custom) + .snapshotStoreEnabled(true) + .snapshotStore(store) + .build(); + + assertTrue(config.isSnapshotStoreEnabled()); + assertSame(store, config.getSnapshotStore()); + + // Base enricher, then custom, then the recorder last. + assertEquals(3, config.getEventEnrichers().size()); + assertTrue(config.getEventEnrichers().get(0) instanceof BaseEventPropertiesEnricher); + assertSame(custom, config.getEventEnrichers().get(1)); + assertTrue(config.getEventEnrichers().get(2) instanceof SnapshotRecordingEnricher); + } + + @Test + void testSnapshotStoreEnabledWithoutStoreAddsNoEnricher() { + AguiAdapterConfig config = AguiAdapterConfig.builder().snapshotStoreEnabled(true).build(); + + assertTrue(config.isSnapshotStoreEnabled()); + assertNull(config.getSnapshotStore()); + assertTrue(config.getEventEnrichers().isEmpty()); + } + private AgentEventConverter noopConverter() { return new AgentEventConverter() { @Override diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/converter/AguiActivityConverterTest.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/converter/AguiActivityConverterTest.java new file mode 100644 index 0000000000..6dd683f2bf --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/converter/AguiActivityConverterTest.java @@ -0,0 +1,127 @@ +/* + * 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.converter; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.agentscope.core.agui.event.AguiEvent; +import java.util.HashMap; +import java.util.Map; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for {@link AguiActivityConverter}, mirroring {@link AguiStateConverterTest}. + */ +class AguiActivityConverterTest { + + private AguiActivityConverter converter; + + @BeforeEach + void setUp() { + converter = new AguiActivityConverter(); + } + + @Test + void testCreateSnapshot() { + Map content = Map.of("step", 1, "label", "running"); + + AguiEvent.ActivitySnapshot snapshot = + converter.createSnapshot("t1", "r1", "m1", "progress", content, true); + + assertEquals("t1", snapshot.getThreadId()); + assertEquals("r1", snapshot.getRunId()); + assertEquals("m1", snapshot.messageId()); + assertEquals("progress", snapshot.activityType()); + assertEquals(1, snapshot.content().get("step")); + assertEquals(true, snapshot.replace()); + } + + @Test + void testCreateDeltaForAddedKey() { + AguiEvent.ActivityDelta delta = + converter.createDelta( + "t1", "r1", "m1", "progress", new HashMap<>(), Map.of("newKey", "v")); + + assertNotNull(delta); + assertEquals(1, delta.patch().size()); + assertEquals("add", delta.patch().get(0).op()); + assertEquals("/newKey", delta.patch().get(0).path()); + } + + @Test + void testCreateDeltaForRemovedKey() { + AguiEvent.ActivityDelta delta = + converter.createDelta( + "t1", "r1", "m1", "progress", Map.of("oldKey", "v"), new HashMap<>()); + + assertNotNull(delta); + assertEquals(1, delta.patch().size()); + assertEquals("remove", delta.patch().get(0).op()); + } + + @Test + void testCreateDeltaForReplacedValue() { + AguiEvent.ActivityDelta delta = + converter.createDelta( + "t1", "r1", "m1", "progress", Map.of("key", "old"), Map.of("key", "new")); + + assertNotNull(delta); + assertEquals("replace", delta.patch().get(0).op()); + assertEquals("new", delta.patch().get(0).value()); + } + + @Test + void testCreateDeltaReturnsNullForNoChanges() { + AguiEvent.ActivityDelta delta = + converter.createDelta( + "t1", "r1", "m1", "progress", Map.of("k", "v"), Map.of("k", "v")); + + assertNull(delta); + } + + @Test + void testHasChanges() { + assertTrue(converter.hasChanges(Map.of("k", "a"), Map.of("k", "b"))); + assertFalse(converter.hasChanges(Map.of("k", "a"), Map.of("k", "a"))); + assertFalse(converter.hasChanges(null, null)); + } + + @Test + void testCreateDeltaNested() { + Map nestedBefore = new HashMap<>(); + nestedBefore.put("inner", "old"); + Map before = new HashMap<>(); + before.put("nested", nestedBefore); + + Map nestedAfter = new HashMap<>(); + nestedAfter.put("inner", "new"); + Map after = new HashMap<>(); + after.put("nested", nestedAfter); + + AguiEvent.ActivityDelta delta = + converter.createDelta("t1", "r1", "m1", "progress", before, after); + + assertNotNull(delta); + assertEquals(1, delta.patch().size()); + assertEquals("/nested/inner", delta.patch().get(0).path()); + assertEquals("replace", delta.patch().get(0).op()); + } +} 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..83a7edb12a 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 @@ -40,7 +40,10 @@ 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.agui.store.AguiThreadSnapshot; +import io.agentscope.core.agui.store.InMemoryAguiSnapshotStore; import io.agentscope.core.event.AgentEndEvent; +import io.agentscope.core.event.AgentStartEvent; import io.agentscope.core.message.Msg; import io.agentscope.core.message.ToolResultBlock; import java.util.List; @@ -729,6 +732,122 @@ public Flux run(RunAgentInput input, RuntimeContext runtimeContext) { } } + @Test + void hydrateWithNoStoreReturnsEmptyHandshake() { + AguiRequestProcessor processor = + AguiRequestProcessor.builder().agentResolver(mock(AgentResolver.class)).build(); + + List events = processor.hydrate(request(input("run-1"))).collectList().block(); + + assertNotNull(events); + assertEquals(3, events.size()); + assertEquals(AguiEventType.RUN_STARTED, events.get(0).getType()); + assertInstanceOf(AguiEvent.MessagesSnapshot.class, events.get(1)); + assertEquals(AguiEventType.RUN_FINISHED, events.get(2).getType()); + } + + @Test + void hydrateWithSnapshotReplaysMessages() { + InMemoryAguiSnapshotStore store = new InMemoryAguiSnapshotStore(); + store.save( + new AguiThreadSnapshot( + "thread-1", + List.of(AguiMessage.userMessage("u1", "hi")), + Map.of(), + List.of(), + null, + "run-0", + 1L)); + AguiRequestProcessor processor = + AguiRequestProcessor.builder() + .agentResolver(mock(AgentResolver.class)) + .snapshotStore(store) + .build(); + + List events = processor.hydrate(request(input("run-1"))).collectList().block(); + + assertNotNull(events); + AguiEvent.MessagesSnapshot snapshot = + assertInstanceOf(AguiEvent.MessagesSnapshot.class, events.get(1)); + assertEquals(1, snapshot.messages().size()); + assertEquals("u1", snapshot.messages().get(0).getId()); + } + + @Test + void beginRunClearsPendingInterruptFromSnapshotStore() { + InMemoryAguiSnapshotStore store = new InMemoryAguiSnapshotStore(); + store.save( + new AguiThreadSnapshot( + "thread-1", + List.of(AguiMessage.userMessage("u1", "hi")), + Map.of(), + List.of(), + new AguiEvent.RunFinishedInterruptOutcome(List.of(interrupt("i1", "tc1"))), + "run-0", + 1L)); + AgentResolver resolver = mock(AgentResolver.class); + ReActAgent agent = mock(ReActAgent.class); + when(resolver.resolveAgent(eq("default"), eq("thread-1"), nullable(String.class))) + .thenReturn(agent); + when(resolver.hasMemory(any(RuntimeContext.class))).thenReturn(false); + when(agent.streamEvents(anyList(), any(RuntimeContext.class))) + .thenReturn(Flux.just(new AgentEndEvent("ok"))); + AguiRequestProcessor processor = + AguiRequestProcessor.builder() + .agentResolver(resolver) + .runtimeContextResolver(request -> null) + .snapshotStore(store) + .build(); + + processor.process(request(input("run-1"))).events().collectList().block(); + + AguiThreadSnapshot snapshot = store.find("thread-1").orElseThrow(); + assertEquals(null, snapshot.pendingOutcome()); + } + + @Test + void processRecordsSnapshotReplayedByHydrate() { + InMemoryAguiSnapshotStore store = new InMemoryAguiSnapshotStore(); + AguiAdapterConfig config = + AguiAdapterConfig.builder().snapshotStoreEnabled(true).snapshotStore(store).build(); + AgentResolver resolver = mock(AgentResolver.class); + ReActAgent agent = mock(ReActAgent.class); + when(resolver.resolveAgent(eq("default"), eq("thread-1"), nullable(String.class))) + .thenReturn(agent); + when(resolver.hasMemory(any(RuntimeContext.class))).thenReturn(false); + when(agent.streamEvents(anyList(), any(RuntimeContext.class))) + .thenReturn( + Flux.just( + new AgentStartEvent("thread-1", "reply-1", "assistant"), + new AgentEndEvent("ok"))); + AguiRequestProcessor processor = + AguiRequestProcessor.builder() + .agentResolver(resolver) + .runtimeContextResolver(request -> null) + .config(config) + .snapshotStore(store) + .build(); + + processor.process(request(input("run-1"))).events().collectList().block(); + + // The framework recorded a presentation snapshot for the thread. + AguiThreadSnapshot snapshot = store.find("thread-1").orElseThrow(); + assertEquals("run-1", snapshot.lastRunId()); + // The user turn from RUN_STARTED.input survived. + assertEquals(1, snapshot.messages().size()); + assertEquals("msg-1", snapshot.messages().get(0).getId()); + + // Hydrate replays a read-only frame sequence from the recorded snapshot. + List frames = processor.hydrate(request(input("run-2"))).collectList().block(); + assertNotNull(frames); + assertEquals(AguiEventType.RUN_STARTED, frames.get(0).getType()); + assertInstanceOf(AguiEvent.MessagesSnapshot.class, frames.get(1)); + AguiEvent.MessagesSnapshot messages = (AguiEvent.MessagesSnapshot) frames.get(1); + assertEquals(1, messages.messages().size()); + assertEquals("msg-1", messages.messages().get(0).getId()); + assertEquals(AguiEventType.RUN_FINISHED, frames.get(frames.size() - 1).getType()); + } + private static void assertResumeContractErrorLifecycle(List events) { assertEquals( List.of(AguiEventType.RUN_STARTED, AguiEventType.RUN_ERROR), diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/store/AguiJsonPatchTest.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/store/AguiJsonPatchTest.java new file mode 100644 index 0000000000..1b7efef0ec --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/store/AguiJsonPatchTest.java @@ -0,0 +1,182 @@ +/* + * 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.store; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.agentscope.core.agui.event.AguiEvent.JsonPatchOperation; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for the package-visible {@link AguiJsonPatch} RFC 6902 applier. + */ +class AguiJsonPatchTest { + + @Test + void addReplacesMissingKey() { + Map target = new java.util.LinkedHashMap<>(); + Map result = + AguiJsonPatch.apply(target, List.of(JsonPatchOperation.add("/key", "value"))); + + assertEquals("value", result.get("key")); + } + + @Test + void replaceOverwritesExistingValue() { + Map target = new java.util.LinkedHashMap<>(); + target.put("key", "old"); + + Map result = + AguiJsonPatch.apply(target, List.of(JsonPatchOperation.replace("/key", "new"))); + + assertEquals("new", result.get("key")); + } + + @Test + void removeDeletesKey() { + Map target = new java.util.LinkedHashMap<>(); + target.put("key", "value"); + + Map result = + AguiJsonPatch.apply(target, List.of(JsonPatchOperation.remove("/key"))); + + assertTrue(!result.containsKey("key")); + } + + @Test + void nestedPointersTraverseMaps() { + Map inner = new java.util.LinkedHashMap<>(); + inner.put("inner", "old"); + Map target = new java.util.LinkedHashMap<>(); + target.put("nested", inner); + + Map result = + AguiJsonPatch.apply( + target, List.of(JsonPatchOperation.replace("/nested/inner", "new"))); + + @SuppressWarnings("unchecked") + Map nested = (Map) result.get("nested"); + assertEquals("new", nested.get("inner")); + } + + @Test + void tildeIsUnescaped() { + Map target = new java.util.LinkedHashMap<>(); + + Map result = + AguiJsonPatch.apply(target, List.of(JsonPatchOperation.add("/key~0with", "value"))); + + assertEquals("value", result.get("key~with")); + } + + @Test + void slashIsUnescaped() { + Map target = new java.util.LinkedHashMap<>(); + + Map result = + AguiJsonPatch.apply(target, List.of(JsonPatchOperation.add("/key~1with", "value"))); + + assertEquals("value", result.get("key/with")); + } + + @Test + void addToListWithAppendToken() { + Map target = new java.util.LinkedHashMap<>(); + target.put("items", new java.util.ArrayList<>(List.of(1, 2, 3))); + + Map result = + AguiJsonPatch.apply(target, List.of(JsonPatchOperation.add("/items/-", 4))); + + assertEquals(List.of(1, 2, 3, 4), result.get("items")); + } + + @Test + void addToListAtIndexInserts() { + Map target = new java.util.LinkedHashMap<>(); + target.put("items", new java.util.ArrayList<>(List.of(1, 2, 3))); + + Map result = + AguiJsonPatch.apply(target, List.of(JsonPatchOperation.add("/items/1", 9))); + + assertEquals(List.of(1, 9, 2, 3), result.get("items")); + } + + @Test + void replaceListItem() { + Map target = new java.util.LinkedHashMap<>(); + target.put("items", new java.util.ArrayList<>(List.of(1, 2, 3))); + + Map result = + AguiJsonPatch.apply(target, List.of(JsonPatchOperation.replace("/items/0", 7))); + + assertEquals(List.of(7, 2, 3), result.get("items")); + } + + @Test + void removeListItem() { + Map target = new java.util.LinkedHashMap<>(); + target.put("items", new java.util.ArrayList<>(List.of(1, 2, 3))); + + Map result = + AguiJsonPatch.apply(target, List.of(JsonPatchOperation.remove("/items/1"))); + + assertEquals(List.of(1, 3), result.get("items")); + } + + @Test + void unknownOpIsIgnored() { + Map target = new java.util.LinkedHashMap<>(); + target.put("key", "value"); + + Map result = + AguiJsonPatch.apply( + target, List.of(new JsonPatchOperation("move", "/key", null, "/other"))); + + assertEquals("value", result.get("key")); + } + + @Test + void doesNotMutateInput() { + Map target = new java.util.LinkedHashMap<>(); + target.put("key", "old"); + + Map result = + AguiJsonPatch.apply(target, List.of(JsonPatchOperation.replace("/key", "new"))); + + assertNotSame(target, result); + assertEquals("old", target.get("key")); + } + + @Test + void multipleOpsApplyInOrder() { + Map target = new java.util.LinkedHashMap<>(); + + Map result = + AguiJsonPatch.apply( + target, + List.of( + JsonPatchOperation.add("/a", 1), + JsonPatchOperation.add("/b", 2), + JsonPatchOperation.remove("/a"))); + + assertTrue(!result.containsKey("a")); + assertEquals(2, result.get("b")); + } +} diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/store/AguiSnapshotAccumulatorTest.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/store/AguiSnapshotAccumulatorTest.java new file mode 100644 index 0000000000..f222c8fd7d --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/store/AguiSnapshotAccumulatorTest.java @@ -0,0 +1,307 @@ +/* + * 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.store; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.agentscope.core.agui.event.AguiEvent; +import io.agentscope.core.agui.event.AguiEvent.JsonPatchOperation; +import io.agentscope.core.agui.model.AguiMessage; +import io.agentscope.core.agui.model.AguiToolCall; +import io.agentscope.core.agui.model.RunAgentInput; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for the package-visible {@link AguiSnapshotAccumulator}, which absorbs the former + * CopilotKit replay workarounds. + */ +class AguiSnapshotAccumulatorTest { + + private static final String THREAD = "thread-1"; + private static final String RUN = "run-1"; + + private static AguiSnapshotAccumulator newAccumulator(AguiThreadSnapshot seed) { + return new AguiSnapshotAccumulator(THREAD, RUN, seed); + } + + private static AguiThreadSnapshot materialize( + AguiSnapshotAccumulator acc, AguiEvent... events) { + for (AguiEvent event : events) { + acc.consume(event); + } + return acc.materialize(); + } + + @Test + void textFoldingProducesAssistantMessage() { + AguiThreadSnapshot snapshot = + materialize( + newAccumulator(null), + new AguiEvent.TextMessageStart(THREAD, RUN, "m1", "assistant"), + new AguiEvent.TextMessageContent(THREAD, RUN, "m1", "Hello "), + new AguiEvent.TextMessageContent(THREAD, RUN, "m1", "world"), + new AguiEvent.TextMessageEnd(THREAD, RUN, "m1"), + new AguiEvent.RunFinished(THREAD, RUN)); + + assertEquals(1, snapshot.messages().size()); + AguiMessage assistant = snapshot.messages().get(0); + assertEquals("m1", assistant.getId()); + assertEquals("assistant", assistant.getRole()); + assertEquals("Hello world", assistant.getTextContent()); + assertNull(snapshot.pendingOutcome()); + } + + @Test + void toolCallWithArgsAndResultAttachesToAssistant() { + AguiThreadSnapshot snapshot = + materialize( + newAccumulator(null), + new AguiEvent.TextMessageStart(THREAD, RUN, "m1", "assistant"), + new AguiEvent.TextMessageEnd(THREAD, RUN, "m1"), + new AguiEvent.ToolCallStart(THREAD, RUN, "tc1", "search"), + new AguiEvent.ToolCallArgs(THREAD, RUN, "tc1", "{\"q\":\"hi\"}"), + new AguiEvent.ToolCallEnd(THREAD, RUN, "tc1"), + new AguiEvent.ToolCallResult( + THREAD, RUN, "tc1", "result-body", "tool", "r1"), + new AguiEvent.RunFinished(THREAD, RUN)); + + // assistant message + tool result message + assertEquals(2, snapshot.messages().size()); + AguiMessage assistant = snapshot.messages().get(0); + assertEquals(1, assistant.getToolCalls().size()); + AguiToolCall call = assistant.getToolCalls().get(0); + assertEquals("tc1", call.getId()); + assertEquals("search", call.getFunction().getName()); + assertEquals("{\"q\":\"hi\"}", call.getFunction().getArguments()); + + AguiMessage tool = snapshot.messages().get(1); + assertEquals("tool", tool.getRole()); + assertEquals("tc1", tool.getToolCallId()); + assertEquals("result-body", tool.getTextContent()); + } + + @Test + void danglingToolCallGetsSyntheticResult() { + AguiThreadSnapshot snapshot = + materialize( + newAccumulator(null), + new AguiEvent.ToolCallStart(THREAD, RUN, "tc1", "search"), + new AguiEvent.ToolCallArgs(THREAD, RUN, "tc1", "{}"), + new AguiEvent.RunFinished(THREAD, RUN)); + + // One synthesized assistant message (holding the tool call) + one synthetic tool result. + assertEquals(2, snapshot.messages().size()); + AguiMessage tool = snapshot.messages().get(1); + assertEquals("tool", tool.getRole()); + assertEquals("tc1", tool.getToolCallId()); + assertEquals("tc1:result", tool.getId()); + assertNull(snapshot.pendingOutcome()); + } + + @Test + void danglingToolOfOpenInterruptIsLeftOpen() { + AguiEvent.RunFinishedOutcome interrupt = + new AguiEvent.RunFinishedInterruptOutcome( + List.of( + new AguiEvent.Interrupt( + "i1", "tool_call", "approve?", "tc1", null, null, null))); + AguiThreadSnapshot snapshot = + materialize( + newAccumulator(null), + new AguiEvent.ToolCallStart(THREAD, RUN, "tc1", "search"), + new AguiEvent.ToolCallArgs(THREAD, RUN, "tc1", "{}"), + new AguiEvent.RunFinished(THREAD, RUN, null, interrupt)); + + // The open interrupt is retained, and its tool is NOT given a synthetic result. + assertNotNull(snapshot.pendingOutcome()); + assertEquals(1, snapshot.messages().size()); + AguiMessage assistant = snapshot.messages().get(0); + assertEquals(1, assistant.getToolCalls().size()); + assertEquals("tc1", assistant.getToolCalls().get(0).getId()); + for (AguiMessage message : snapshot.messages()) { + assertFalse("tc1:result".equals(message.getId())); + } + } + + @Test + void stateSnapshotThenDeltaFoldsToFinalState() { + AguiThreadSnapshot snapshot = + materialize( + newAccumulator(null), + new AguiEvent.StateSnapshot( + THREAD, RUN, Map.of("count", 1, "name", "alice")), + new AguiEvent.StateDelta( + THREAD, + RUN, + List.of( + JsonPatchOperation.replace("/count", 2), + JsonPatchOperation.add("active", true))), + new AguiEvent.RunFinished(THREAD, RUN)); + + assertEquals(2, snapshot.state().get("count")); + assertEquals("alice", snapshot.state().get("name")); + assertEquals(true, snapshot.state().get("active")); + } + + @Test + void activitySnapshotAndDeltaAreFolded() { + AguiThreadSnapshot snapshot = + materialize( + newAccumulator(null), + new AguiEvent.ActivitySnapshot( + THREAD, RUN, "m1", "progress", Map.of("step", 1), true), + new AguiEvent.ActivityDelta( + THREAD, + RUN, + "m1", + "progress", + List.of(JsonPatchOperation.replace("/step", 2))), + new AguiEvent.RunFinished(THREAD, RUN)); + + assertEquals(1, snapshot.activities().size()); + AguiThreadSnapshot.ActivityFrame frame = snapshot.activities().get(0); + assertEquals("m1", frame.messageId()); + assertEquals("progress", frame.activityType()); + assertEquals(2, frame.content().get("step")); + } + + @Test + void trailingInterruptIsKept() { + AguiEvent.RunFinishedOutcome interrupt = + new AguiEvent.RunFinishedInterruptOutcome( + List.of( + new AguiEvent.Interrupt( + "i1", "tool_call", "msg", "tc1", null, null, null))); + AguiThreadSnapshot snapshot = + materialize( + newAccumulator(null), + new AguiEvent.RunFinished(THREAD, RUN, null, interrupt)); + + assertTrue(snapshot.pendingOutcome() instanceof AguiEvent.RunFinishedInterruptOutcome); + } + + @Test + void danglingToolsOfMultiInterruptOutcomeAreAllLeftOpen() { + AguiEvent.RunFinishedOutcome interrupt = + new AguiEvent.RunFinishedInterruptOutcome( + List.of( + new AguiEvent.Interrupt( + "i1", "tool_call", "msg", "tc1", null, null, null), + new AguiEvent.Interrupt( + "i2", "tool_call", "msg", "tc2", null, null, null))); + AguiThreadSnapshot snapshot = + materialize( + newAccumulator(null), + new AguiEvent.ToolCallStart(THREAD, RUN, "tc1", "search"), + new AguiEvent.ToolCallArgs(THREAD, RUN, "tc1", "{}"), + new AguiEvent.ToolCallStart(THREAD, RUN, "tc2", "write"), + new AguiEvent.ToolCallArgs(THREAD, RUN, "tc2", "{}"), + new AguiEvent.RunFinished(THREAD, RUN, null, interrupt)); + + assertNotNull(snapshot.pendingOutcome()); + // No synthetic results for either open-interrupt tool — both genuinely await the user. + for (AguiMessage message : snapshot.messages()) { + assertFalse("tc1:result".equals(message.getId())); + assertFalse("tc2:result".equals(message.getId())); + } + } + + @Test + void successRunClearsPriorInterruptInSameAccumulator() { + AguiEvent.RunFinishedOutcome interrupt = + new AguiEvent.RunFinishedInterruptOutcome( + List.of( + new AguiEvent.Interrupt( + "i1", "tool_call", "msg", "tc1", null, null, null))); + AguiSnapshotAccumulator acc = newAccumulator(null); + acc.consume(new AguiEvent.RunFinished(THREAD, RUN, null, interrupt)); + assertNotNull(acc.materialize().pendingOutcome()); + + acc.consume(new AguiEvent.RunFinished(THREAD, RUN)); + assertNull(acc.materialize().pendingOutcome()); + } + + @Test + void seedsFromExistingSnapshotButDoesNotCarryInterrupt() { + AguiEvent.RunFinishedOutcome interrupt = + new AguiEvent.RunFinishedInterruptOutcome( + List.of( + new AguiEvent.Interrupt( + "i1", "tool_call", "msg", "tc1", null, null, null))); + AguiThreadSnapshot seed = + new AguiThreadSnapshot( + THREAD, + List.of(AguiMessage.userMessage("u1", "hi")), + Map.of("k", "v"), + List.of( + new AguiThreadSnapshot.ActivityFrame( + "m0", "progress", Map.of("step", 1))), + interrupt, + "run-0", + 1L); + + AguiThreadSnapshot snapshot = + materialize(newAccumulator(seed), new AguiEvent.RunFinished(THREAD, RUN)); + + // Prior messages / state / activities survive, but the prior interrupt does not. + assertEquals("hi", snapshot.messages().get(0).getTextContent()); + assertEquals("v", snapshot.state().get("k")); + assertEquals(1, snapshot.activities().size()); + assertNull(snapshot.pendingOutcome()); + } + + @Test + void runStartedCapturesInputMessagesDeDuped() { + RunAgentInput input = + RunAgentInput.builder() + .threadId(THREAD) + .runId(RUN) + .messages(List.of(AguiMessage.userMessage("u1", "hi"))) + .build(); + AguiThreadSnapshot snapshot = + materialize( + newAccumulator(null), + new AguiEvent.RunStarted(THREAD, RUN, null, input), + new AguiEvent.TextMessageStart(THREAD, RUN, "m1", "assistant"), + new AguiEvent.TextMessageEnd(THREAD, RUN, "m1"), + new AguiEvent.RunFinished(THREAD, RUN)); + + // u1 from input + m1 assistant + assertEquals(2, snapshot.messages().size()); + assertEquals("u1", snapshot.messages().get(0).getId()); + } + + @Test + void runErrorClearsPendingOutcome() { + AguiSnapshotAccumulator acc = newAccumulator(null); + AguiEvent.RunFinishedOutcome interrupt = + new AguiEvent.RunFinishedInterruptOutcome( + List.of( + new AguiEvent.Interrupt( + "i1", "tool_call", "msg", "tc1", null, null, null))); + acc.consume(new AguiEvent.RunFinished(THREAD, RUN, null, interrupt)); + assertNotNull(acc.materialize().pendingOutcome()); + + acc.consume(new AguiEvent.RunError(THREAD, RUN, "boom", "ERR")); + assertNull(acc.materialize().pendingOutcome()); + } +} diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/store/AguiSnapshotHydratorTest.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/store/AguiSnapshotHydratorTest.java new file mode 100644 index 0000000000..514ed4157f --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/store/AguiSnapshotHydratorTest.java @@ -0,0 +1,117 @@ +/* + * 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.store; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.agentscope.core.agui.event.AguiEvent; +import io.agentscope.core.agui.event.AguiEventType; +import io.agentscope.core.agui.model.AguiMessage; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for {@link AguiSnapshotHydrator}. + */ +class AguiSnapshotHydratorTest { + + private final AguiSnapshotHydrator hydrator = new AguiSnapshotHydrator(); + + @Test + void emptySnapshotProducesThreeFrameHandshake() { + List frames = hydrator.hydrate(null, "t1", "r1"); + + assertEquals(3, frames.size()); + assertEquals(AguiEventType.RUN_STARTED, frames.get(0).getType()); + assertTrue(frames.get(1) instanceof AguiEvent.MessagesSnapshot); + assertTrue(frames.get(2) instanceof AguiEvent.RunFinished); + assertEquals("t1", frames.get(0).getThreadId()); + assertEquals("r1", frames.get(0).getRunId()); + } + + @Test + void snapshotProducesFullFrameOrder() { + AguiThreadSnapshot snapshot = + new AguiThreadSnapshot( + "t1", + List.of(AguiMessage.userMessage("u1", "hi")), + Map.of("count", 2), + List.of( + new AguiThreadSnapshot.ActivityFrame( + "u1", "progress", Map.of("step", 1))), + null, + "r0", + 1L); + + List frames = hydrator.hydrate(snapshot, "t1", "r1"); + + // RUN_STARTED, MESSAGES_SNAPSHOT, STATE_SNAPSHOT, ACTIVITY_SNAPSHOT, RUN_FINISHED + assertEquals(5, frames.size()); + assertEquals(AguiEventType.RUN_STARTED, frames.get(0).getType()); + assertTrue(frames.get(1) instanceof AguiEvent.MessagesSnapshot); + assertTrue(frames.get(2) instanceof AguiEvent.StateSnapshot); + assertTrue(frames.get(3) instanceof AguiEvent.ActivitySnapshot); + assertTrue(frames.get(4) instanceof AguiEvent.RunFinished); + } + + @Test + void stateOmittedWhenEmpty() { + AguiThreadSnapshot snapshot = + new AguiThreadSnapshot( + "t1", + List.of(AguiMessage.userMessage("u1", "hi")), + Map.of(), + List.of(), + null, + "r0", + 1L); + + List frames = hydrator.hydrate(snapshot, "t1", "r1"); + + // RUN_STARTED, MESSAGES_SNAPSHOT, RUN_FINISHED (no state, no activities) + assertEquals(3, frames.size()); + assertTrue(frames.get(1) instanceof AguiEvent.MessagesSnapshot); + assertTrue(frames.get(2) instanceof AguiEvent.RunFinished); + } + + @Test + void pendingOutcomePropagatesToRunFinished() { + AguiEvent.RunFinishedOutcome outcome = + new AguiEvent.RunFinishedInterruptOutcome( + List.of( + new AguiEvent.Interrupt( + "i1", "tool_call", "msg", "tc1", null, null, null))); + AguiThreadSnapshot snapshot = + new AguiThreadSnapshot("t1", List.of(), Map.of(), List.of(), outcome, "r0", 1L); + + List frames = hydrator.hydrate(snapshot, "t1", "r1"); + AguiEvent.RunFinished finished = (AguiEvent.RunFinished) frames.get(frames.size() - 1); + assertEquals(outcome, finished.outcome()); + } + + @Test + void nullPendingOutcomeSerializesAsSuccess() { + AguiThreadSnapshot snapshot = + new AguiThreadSnapshot("t1", List.of(), Map.of(), List.of(), null, "r0", 1L); + + List frames = hydrator.hydrate(snapshot, "t1", "r1"); + AguiEvent.RunFinished finished = (AguiEvent.RunFinished) frames.get(frames.size() - 1); + assertNull(finished.outcome()); + } +} diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/store/InMemoryAguiSnapshotStoreTest.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/store/InMemoryAguiSnapshotStoreTest.java new file mode 100644 index 0000000000..923c3ab112 --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/store/InMemoryAguiSnapshotStoreTest.java @@ -0,0 +1,111 @@ +/* + * 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.store; + +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.agui.event.AguiEvent; +import io.agentscope.core.agui.model.AguiMessage; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for {@link InMemoryAguiSnapshotStore}. + */ +class InMemoryAguiSnapshotStoreTest { + + private static AguiThreadSnapshot snapshot(String threadId, long updatedAt) { + return new AguiThreadSnapshot( + threadId, + List.of(AguiMessage.userMessage(threadId, "hi")), + Map.of(), + List.of(), + null, + "run-1", + updatedAt); + } + + private static AguiThreadSnapshot snapshotWithInterrupt(String threadId, long updatedAt) { + return new AguiThreadSnapshot( + threadId, + List.of(AguiMessage.userMessage(threadId, "hi")), + Map.of(), + List.of(), + new AguiEvent.RunFinishedInterruptOutcome( + List.of( + new AguiEvent.Interrupt( + "i1", "tool_call", "msg", "tc1", null, null, null))), + "run-1", + updatedAt); + } + + @Test + void saveFindDelete() { + InMemoryAguiSnapshotStore store = new InMemoryAguiSnapshotStore(); + store.save(snapshot("t1", 1L)); + + Optional found = store.find("t1"); + assertTrue(found.isPresent()); + assertEquals("t1", found.get().threadId()); + + store.delete("t1"); + assertTrue(store.find("t1").isEmpty()); + } + + @Test + void clearPendingInterruptsDropsTrailingInterrupt() { + InMemoryAguiSnapshotStore store = new InMemoryAguiSnapshotStore(); + store.save(snapshotWithInterrupt("t1", 1L)); + + store.clearPendingInterrupts("t1"); + + AguiThreadSnapshot found = store.find("t1").orElseThrow(); + assertEquals(null, found.pendingOutcome()); + // Messages are preserved. + assertEquals(1, found.messages().size()); + } + + @Test + void clearPendingInterruptsNoOpWhenNoInterrupt() { + InMemoryAguiSnapshotStore store = new InMemoryAguiSnapshotStore(); + store.save(snapshot("t1", 1L)); + // Should not throw and should not alter the snapshot. + store.clearPendingInterrupts("t1"); + assertEquals(null, store.find("t1").orElseThrow().pendingOutcome()); + } + + @Test + void overflowEvictsOldest() { + InMemoryAguiSnapshotStore store = new InMemoryAguiSnapshotStore(2); + store.save(snapshot("t1", 1L)); + store.save(snapshot("t2", 2L)); + store.save(snapshot("t3", 3L)); + + assertFalse(store.find("t1").isPresent()); // oldest evicted + assertTrue(store.find("t2").isPresent()); + assertTrue(store.find("t3").isPresent()); + } + + @Test + void rejectsNonPositiveCapacity() { + assertThrows(IllegalArgumentException.class, () -> new InMemoryAguiSnapshotStore(0)); + } +} diff --git a/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/store/SnapshotRecordingEnricherTest.java b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/store/SnapshotRecordingEnricherTest.java new file mode 100644 index 0000000000..f949f70b33 --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui/src/test/java/io/agentscope/core/agui/store/SnapshotRecordingEnricherTest.java @@ -0,0 +1,105 @@ +/* + * 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.store; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.agentscope.core.agui.adapter.AguiAdapterConfig; +import io.agentscope.core.agui.adapter.strategy.AguiStreamContext; +import io.agentscope.core.agui.event.AguiEvent; +import java.util.List; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for {@link SnapshotRecordingEnricher}. + */ +class SnapshotRecordingEnricherTest { + + private static AguiStreamContext context() { + return new AguiStreamContext("t1", "r1", AguiAdapterConfig.defaultConfig()); + } + + @Test + void savesSnapshotOnTerminalFrame() { + InMemoryAguiSnapshotStore store = new InMemoryAguiSnapshotStore(); + SnapshotRecordingEnricher enricher = new SnapshotRecordingEnricher(store); + + List events = + List.of( + new AguiEvent.TextMessageStart("t1", "r1", "m1", "assistant"), + new AguiEvent.TextMessageContent("t1", "r1", "m1", "hi"), + new AguiEvent.TextMessageEnd("t1", "r1", "m1"), + new AguiEvent.RunFinished("t1", "r1")); + + List result = enricher.enrich(null, events, context()); + + assertEquals(events, result); // events returned unmodified + assertTrue(store.find("t1").isPresent()); + AguiThreadSnapshot snapshot = store.find("t1").orElseThrow(); + assertEquals(1, snapshot.messages().size()); + assertEquals("hi", snapshot.messages().get(0).getTextContent()); + assertEquals("r1", snapshot.lastRunId()); + } + + @Test + void flushSavesOnAbnormalTermination() { + InMemoryAguiSnapshotStore store = new InMemoryAguiSnapshotStore(); + SnapshotRecordingEnricher enricher = new SnapshotRecordingEnricher(store); + + // Non-terminal events only — simulates a stream that died without a terminal frame. + enricher.enrich( + null, + List.of( + new AguiEvent.TextMessageStart("t1", "r1", "m1", "assistant"), + new AguiEvent.TextMessageContent("t1", "r1", "m1", "partial")), + context()); + + assertTrue(store.find("t1").isEmpty()); // not saved yet + + enricher.flush("t1", "r1"); + + assertTrue(store.find("t1").isPresent()); + assertEquals("partial", store.find("t1").orElseThrow().messages().get(0).getTextContent()); + } + + @Test + void flushAfterTerminalIsNoOp() { + InMemoryAguiSnapshotStore store = new InMemoryAguiSnapshotStore(); + SnapshotRecordingEnricher enricher = new SnapshotRecordingEnricher(store); + + enricher.enrich( + null, + List.of( + new AguiEvent.TextMessageStart("t1", "r1", "m1", "assistant"), + new AguiEvent.TextMessageEnd("t1", "r1", "m1"), + new AguiEvent.RunFinished("t1", "r1")), + context()); + + // Already saved on the terminal frame; flush must not throw or duplicate. + enricher.flush("t1", "r1"); + + assertTrue(store.find("t1").isPresent()); + assertEquals(1, store.find("t1").orElseThrow().messages().size()); + } + + @Test + void nullStoreIsNoOp() { + SnapshotRecordingEnricher enricher = new SnapshotRecordingEnricher(null); + List events = List.of(new AguiEvent.RunFinished("t1", "r1")); + assertEquals(events, enricher.enrich(null, events, context())); + } +} diff --git a/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/common/AguiProperties.java b/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/common/AguiProperties.java index b263e5b101..95f01838f4 100644 --- a/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/common/AguiProperties.java +++ b/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/common/AguiProperties.java @@ -41,6 +41,8 @@ * emit-token-usage: false * emit-run-finished-after-error: false * interrupt-on-disconnect: true + * snapshot-store-enabled: false + * snapshot-max-threads: 1000 * */ @ConfigurationProperties(prefix = "agentscope.agui") @@ -129,6 +131,20 @@ public class AguiProperties { /** Whether to interrupt the agent when the client disconnects. */ private boolean interruptOnDisconnect = true; + /** + * Whether the AG-UI presentation snapshot store is enabled. When enabled, an in-memory + * {@code AguiSnapshotStore} bean is created and the {@code POST {path-prefix}/connect} hydrate + * route is registered so reconnecting clients can rebuild the visible conversation without + * re-running the agent. Default is {@code false} to keep existing clients byte-identical. + */ + private boolean snapshotStoreEnabled = false; + + /** + * Maximum number of threads retained by the in-memory snapshot store. Only used when + * {@link #isSnapshotStoreEnabled()} is true. + */ + private int snapshotMaxThreads = 1000; + public String getPathPrefix() { return pathPrefix; } @@ -272,4 +288,20 @@ public boolean isInterruptOnDisconnect() { public void setInterruptOnDisconnect(boolean interruptOnDisconnect) { this.interruptOnDisconnect = interruptOnDisconnect; } + + public boolean isSnapshotStoreEnabled() { + return snapshotStoreEnabled; + } + + public void setSnapshotStoreEnabled(boolean snapshotStoreEnabled) { + this.snapshotStoreEnabled = snapshotStoreEnabled; + } + + public int getSnapshotMaxThreads() { + return snapshotMaxThreads; + } + + public void setSnapshotMaxThreads(int snapshotMaxThreads) { + this.snapshotMaxThreads = snapshotMaxThreads; + } } diff --git a/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/mvc/AgentscopeAguiMvcAutoConfiguration.java b/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/mvc/AgentscopeAguiMvcAutoConfiguration.java index 03668fc9da..61ca08220e 100644 --- a/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/mvc/AgentscopeAguiMvcAutoConfiguration.java +++ b/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/mvc/AgentscopeAguiMvcAutoConfiguration.java @@ -23,6 +23,8 @@ import io.agentscope.core.agui.registry.AguiAgentRegistry; import io.agentscope.core.agui.runtime.AguiRequestBodyParser; import io.agentscope.core.agui.runtime.AguiRuntimeContextResolver; +import io.agentscope.core.agui.store.AguiSnapshotStore; +import io.agentscope.core.agui.store.InMemoryAguiSnapshotStore; import io.agentscope.spring.boot.agui.common.AguiProperties; import io.agentscope.spring.boot.agui.common.ThreadSessionManager; import org.springframework.beans.factory.ObjectProvider; @@ -30,6 +32,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; @@ -83,6 +86,22 @@ public AguiRequestBodyParser aguiRequestBodyParser() { return new AguiRequestBodyParser(); } + /** + * Creates the in-memory AG-UI presentation snapshot store when the snapshot store is enabled. + * + * @param props The configuration properties + * @return A new in-memory snapshot store + */ + @Bean + @ConditionalOnMissingBean + @ConditionalOnProperty( + prefix = "agentscope.agui", + name = "snapshot-store-enabled", + havingValue = "true") + public AguiSnapshotStore aguiSnapshotStore(AguiProperties props) { + return new InMemoryAguiSnapshotStore(props.getSnapshotMaxThreads()); + } + /** * Creates the AG-UI MVC controller bean. * @@ -100,7 +119,9 @@ public AguiMvcController aguiMvcController( ObjectProvider eventConvertersProvider, ObjectProvider eventEnrichersProvider, ObjectProvider runtimeContextResolverProvider, - ObjectProvider adapterFactoryProvider) { + ObjectProvider adapterFactoryProvider, + ObjectProvider snapshotStoreProvider) { + AguiSnapshotStore snapshotStore = snapshotStoreProvider.getIfAvailable(); AguiAdapterConfig config = AguiAdapterConfig.builder() .toolMergeMode(props.getDefaultToolMergeMode()) @@ -113,6 +134,8 @@ public AguiMvcController aguiMvcController( .defaultAgentId(props.getDefaultAgentId()) .eventConverters(eventConvertersProvider.orderedStream().toList()) .eventEnrichers(eventEnrichersProvider.orderedStream().toList()) + .snapshotStoreEnabled(props.isSnapshotStoreEnabled()) + .snapshotStore(snapshotStore) .build(); return AguiMvcController.builder() @@ -124,6 +147,7 @@ public AguiMvcController aguiMvcController( .interruptOnDisconnect(props.isInterruptOnDisconnect()) .runtimeContextResolver(runtimeContextResolverProvider.getIfAvailable()) .adapterFactory(adapterFactoryProvider.getIfAvailable()) + .snapshotStore(snapshotStore) .config(config) .build(); } @@ -147,4 +171,23 @@ public AguiRestController aguiRestController( props.isEnablePathRouting(), requestBodyParser); } + + /** + * Creates the MVC {@code /connect} hydrate controller, only when the snapshot store is enabled, + * so the MVC route mirrors the WebFlux {@code RouterFunction} gating. + * + * @param aguiMvcController The AG-UI MVC controller + * @param requestBodyParser The parser used to decode request bodies + * @return A new AguiConnectController + */ + @Bean + @ConditionalOnMissingBean + @ConditionalOnProperty( + prefix = "agentscope.agui", + name = "snapshot-store-enabled", + havingValue = "true") + public AguiConnectController aguiConnectController( + AguiMvcController aguiMvcController, AguiRequestBodyParser requestBodyParser) { + return new AguiConnectController(aguiMvcController, requestBodyParser); + } } diff --git a/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/mvc/AguiConnectController.java b/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/mvc/AguiConnectController.java new file mode 100644 index 0000000000..e4d52326e4 --- /dev/null +++ b/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/mvc/AguiConnectController.java @@ -0,0 +1,101 @@ +/* + * 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.spring.boot.agui.mvc; + +import io.agentscope.core.agui.encoder.AguiEventEncoder; +import io.agentscope.core.agui.event.AguiEvent; +import io.agentscope.core.agui.model.RunAgentInput; +import io.agentscope.core.agui.runtime.AguiRequestBodyParser; +import io.agentscope.core.util.JsonException; +import jakarta.servlet.http.HttpServletRequest; +import java.util.Map; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.ExceptionHandler; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +/** + * MVC REST controller for the AG-UI {@code /connect} hydrate route. + * + *

Registered as a conditional bean only when the presentation snapshot store is enabled, so the + * MVC {@code POST {path-prefix}/connect} route mirrors the WebFlux {@code RouterFunction} gating: + * absent (404) when the store is off, present when on. When off, hydrate has nothing to read and + * clients should not rely on the endpoint. + * + *

Delegates to {@link AguiMvcController#handleConnect} for the read-only SSE stream. + */ +@RestController +public class AguiConnectController { + + private final AguiMvcController aguiMvcController; + private final AguiRequestBodyParser requestBodyParser; + private final AguiEventEncoder encoder = new AguiEventEncoder(); + + /** + * Creates a new AguiConnectController. + * + * @param aguiMvcController The AG-UI MVC controller + * @param requestBodyParser The parser used to decode request bodies + */ + public AguiConnectController( + AguiMvcController aguiMvcController, AguiRequestBodyParser requestBodyParser) { + this.aguiMvcController = aguiMvcController; + this.requestBodyParser = requestBodyParser; + } + + /** + * Handle an AG-UI {@code /connect} hydrate request. + * + * @param body The raw run agent input JSON (threadId / runId identify the snapshot) + * @param request The native servlet request (may carry headers for the runtime context) + * @return An SseEmitter for the read-only hydrate SSE stream + */ + @PostMapping( + value = "${agentscope.agui.path-prefix:/agui}/connect", + consumes = MediaType.APPLICATION_JSON_VALUE, + produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public SseEmitter connect(@RequestBody String body, HttpServletRequest request) { + RunAgentInput input = requestBodyParser.parse(body); + return aguiMvcController.handleConnect(input, request); + } + + /** + * Return HTTP 400 for AG-UI request body parsing failures. + * + * @param error the JSON parse failure + * @return an SSE-compatible bad request response + */ + @ExceptionHandler(JsonException.class) + public ResponseEntity handleParseError(JsonException error) { + String errorEvent = + encoder.encodeToJson( + new AguiEvent.Raw( + "unknown", + "unknown", + Map.of( + "error", + "Failed to parse request: " + error.getMessage()))) + .trim(); + String finishEvent = + encoder.encodeToJson(new AguiEvent.RunFinished("unknown", "unknown")).trim(); + return ResponseEntity.badRequest() + .contentType(MediaType.TEXT_EVENT_STREAM) + .body("data: " + errorEvent + "\n\n" + "data: " + finishEvent + "\n\n"); + } +} diff --git a/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/mvc/AguiMvcController.java b/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/mvc/AguiMvcController.java index 1558151f08..547815001a 100644 --- a/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/mvc/AguiMvcController.java +++ b/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/mvc/AguiMvcController.java @@ -25,6 +25,7 @@ import io.agentscope.core.agui.registry.AguiAgentRegistry; import io.agentscope.core.agui.runtime.AguiRuntimeContextRequest; import io.agentscope.core.agui.runtime.AguiRuntimeContextResolver; +import io.agentscope.core.agui.store.AguiSnapshotStore; import io.agentscope.spring.boot.agui.common.DefaultAgentResolver; import io.agentscope.spring.boot.agui.common.ThreadSessionManager; import jakarta.servlet.http.HttpServletRequest; @@ -41,6 +42,7 @@ import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import reactor.core.Disposable; import reactor.core.publisher.BaseSubscriber; +import reactor.core.publisher.Flux; /** * MVC controller for AG-UI protocol requests. @@ -94,6 +96,7 @@ private AguiMvcController(Builder builder) { : AguiAdapterConfig.defaultConfig()) .adapterFactory(builder.adapterFactory) .runtimeContextResolver(builder.runtimeContextResolver) + .snapshotStore(builder.snapshotStore) .build(); this.encoder = new AguiEventEncoder(); this.agentIdHeader = @@ -157,6 +160,76 @@ public SseEmitter handleWithAgentId( return handleInternal(input, headerAgentId, pathAgentId, request); } + /** + * Handle an AG-UI {@code /connect} hydrate request. + * + *

Returns a read-only SSE stream reconstructed from the presentation snapshot store. No + * agent is invoked and there is no cancel-time interrupt, because hydrate has no live agent. + * + * @param input The run agent input (threadId / runId identify the snapshot) + * @param request The native servlet request (may be null) + * @return An SseEmitter for the hydrate SSE stream + */ + public SseEmitter handleConnect(RunAgentInput input, HttpServletRequest request) { + SseEmitter emitter = new SseEmitter(sseTimeout); + String threadId = input.getThreadId(); + String runId = input.getRunId(); + executorService.submit( + () -> { + try { + Flux events = + processor.hydrate( + runtimeContextRequest(input, null, null, request)); + BaseSubscriber subscription = + new BaseSubscriber<>() { + @Override + protected void hookOnNext(AguiEvent event) { + sendEvent(emitter, event); + } + + @Override + protected void hookOnError(Throwable error) { + logger.error( + "Error during AG-UI hydrate: {}", + error.getMessage()); + sendErrorAndComplete( + emitter, threadId, runId, error.getMessage()); + } + + @Override + protected void hookOnComplete() { + try { + emitter.complete(); + } catch (Exception e) { + logger.debug( + "Error completing emitter: {}", e.getMessage()); + } + } + }; + emitter.onCompletion( + () -> logger.debug("SSE hydrate completed for run {}", runId)); + emitter.onTimeout( + () -> { + subscription.dispose(); + logger.debug("SSE hydrate timed out for run {}", runId); + }); + emitter.onError( + (ex) -> { + subscription.dispose(); + logger.debug( + "SSE hydrate error for run {}: {}", + runId, + ex.getMessage()); + }); + events.subscribe(subscription); + } catch (Exception e) { + logger.error("Error processing AG-UI connect: {}", e.getMessage()); + sendErrorAndComplete(emitter, threadId, runId, e.getMessage()); + } + }); + return emitter; + } + private SseEmitter handleInternal( RunAgentInput input, String headerAgentId, @@ -360,6 +433,7 @@ public static class Builder { private boolean interruptOnDisconnect = true; private AguiRuntimeContextResolver runtimeContextResolver; private AguiAgentAdapterFactory adapterFactory; + private AguiSnapshotStore snapshotStore; /** * Set the agent registry. @@ -460,6 +534,18 @@ public Builder adapterFactory(AguiAgentAdapterFactory adapterFactory) { return this; } + /** + * Set the presentation snapshot store used for {@code /connect} hydrate and trailing + * interrupt clearing. Optional; only effective when the snapshot store is enabled. + * + * @param snapshotStore the snapshot store, or null + * @return This builder + */ + public Builder snapshotStore(AguiSnapshotStore snapshotStore) { + this.snapshotStore = snapshotStore; + return this; + } + /** * Build the controller. * diff --git a/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/webflux/AgentscopeAguiWebFluxAutoConfiguration.java b/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/webflux/AgentscopeAguiWebFluxAutoConfiguration.java index d5f17713a6..dcce3f1eb7 100644 --- a/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/webflux/AgentscopeAguiWebFluxAutoConfiguration.java +++ b/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/webflux/AgentscopeAguiWebFluxAutoConfiguration.java @@ -23,6 +23,8 @@ import io.agentscope.core.agui.registry.AguiAgentRegistry; import io.agentscope.core.agui.runtime.AguiRequestBodyParser; import io.agentscope.core.agui.runtime.AguiRuntimeContextResolver; +import io.agentscope.core.agui.store.AguiSnapshotStore; +import io.agentscope.core.agui.store.InMemoryAguiSnapshotStore; import io.agentscope.spring.boot.agui.common.AguiProperties; import io.agentscope.spring.boot.agui.common.ThreadSessionManager; import org.springframework.beans.factory.ObjectProvider; @@ -30,6 +32,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; @@ -86,6 +89,22 @@ public AguiRequestBodyParser aguiRequestBodyParser() { return new AguiRequestBodyParser(); } + /** + * Creates the in-memory AG-UI presentation snapshot store when the snapshot store is enabled. + * + * @param props The configuration properties + * @return A new in-memory snapshot store + */ + @Bean + @ConditionalOnMissingBean + @ConditionalOnProperty( + prefix = "agentscope.agui", + name = "snapshot-store-enabled", + havingValue = "true") + public AguiSnapshotStore aguiSnapshotStore(AguiProperties props) { + return new InMemoryAguiSnapshotStore(props.getSnapshotMaxThreads()); + } + /** * Creates the AG-UI WebFlux handler bean. * @@ -104,7 +123,9 @@ public AguiWebFluxHandler aguiWebFluxHandler( ObjectProvider eventEnrichersProvider, ObjectProvider runtimeContextResolverProvider, ObjectProvider adapterFactoryProvider, + ObjectProvider snapshotStoreProvider, AguiRequestBodyParser requestBodyParser) { + AguiSnapshotStore snapshotStore = snapshotStoreProvider.getIfAvailable(); AguiAdapterConfig config = AguiAdapterConfig.builder() .toolMergeMode(props.getDefaultToolMergeMode()) @@ -117,6 +138,8 @@ public AguiWebFluxHandler aguiWebFluxHandler( .defaultAgentId(props.getDefaultAgentId()) .eventConverters(eventConvertersProvider.orderedStream().toList()) .eventEnrichers(eventEnrichersProvider.orderedStream().toList()) + .snapshotStoreEnabled(props.isSnapshotStoreEnabled()) + .snapshotStore(snapshotStore) .build(); return AguiWebFluxHandler.builder() @@ -128,6 +151,7 @@ public AguiWebFluxHandler aguiWebFluxHandler( .runtimeContextResolver(runtimeContextResolverProvider.getIfAvailable()) .adapterFactory(adapterFactoryProvider.getIfAvailable()) .requestBodyParser(requestBodyParser) + .snapshotStore(snapshotStore) .config(config) .build(); } @@ -158,6 +182,11 @@ public RouterFunction aguiRoutes( props.getPathPrefix() + "/run/{agentId}", handler::handleWithAgentId); } + // Register the read-only hydrate route when the snapshot store is enabled. + if (props.isSnapshotStoreEnabled()) { + routerBuilder.POST(props.getPathPrefix() + "/connect", handler::handleConnect); + } + return routerBuilder.build(); } } diff --git a/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/webflux/AguiWebFluxHandler.java b/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/webflux/AguiWebFluxHandler.java index dbf06d8ec2..5cfb9aae87 100644 --- a/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/webflux/AguiWebFluxHandler.java +++ b/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/main/java/io/agentscope/spring/boot/agui/webflux/AguiWebFluxHandler.java @@ -26,6 +26,7 @@ import io.agentscope.core.agui.runtime.AguiRequestBodyParser; import io.agentscope.core.agui.runtime.AguiRuntimeContextRequest; import io.agentscope.core.agui.runtime.AguiRuntimeContextResolver; +import io.agentscope.core.agui.store.AguiSnapshotStore; import io.agentscope.spring.boot.agui.common.DefaultAgentResolver; import io.agentscope.spring.boot.agui.common.ThreadSessionManager; import java.util.LinkedHashMap; @@ -97,6 +98,7 @@ private AguiWebFluxHandler(Builder builder) { : AguiAdapterConfig.defaultConfig()) .adapterFactory(builder.adapterFactory) .runtimeContextResolver(builder.runtimeContextResolver) + .snapshotStore(builder.snapshotStore) .build(); this.encoder = new AguiEventEncoder(); this.requestBodyParser = @@ -141,6 +143,37 @@ public Mono handleWithAgentId(ServerRequest request) { .onErrorResume(this::handleParseError); } + /** + * Handle an AG-UI {@code /connect} hydrate request. + * + *

Parses the request body as {@link RunAgentInput} and returns a read-only SSE stream of + * frames reconstructed from the presentation snapshot store. No agent is invoked and there is + * no cancel-time interrupt, because hydrate has no live agent. + * + * @param request The server request + * @return A Mono containing the server response with the hydrate SSE stream + */ + public Mono handleConnect(ServerRequest request) { + return request.bodyToMono(String.class) + .map(requestBodyParser::parse) + .flatMap(input -> hydrateInput(input, request)) + .onErrorResume(this::handleParseError); + } + + private Mono hydrateInput(RunAgentInput input, ServerRequest request) { + Flux events = + processor.hydrate(runtimeContextRequest(input, null, null, request)); + Flux> sseStream = + events.map( + event -> + ServerSentEvent.builder() + .data(encoder.encodeToJson(event).trim()) + .build()); + return ServerResponse.ok() + .contentType(MediaType.TEXT_EVENT_STREAM) + .body(sseStream, ServerSentEvent.class); + } + private Mono processInput( RunAgentInput input, ServerRequest request, String pathAgentId) { String threadId = input.getThreadId(); @@ -292,6 +325,7 @@ public static class Builder { private AguiRuntimeContextResolver runtimeContextResolver; private AguiAgentAdapterFactory adapterFactory; private AguiRequestBodyParser requestBodyParser; + private AguiSnapshotStore snapshotStore; /** * Set the agent registry. @@ -392,6 +426,18 @@ public Builder requestBodyParser(AguiRequestBodyParser requestBodyParser) { return this; } + /** + * Set the presentation snapshot store used for {@code /connect} hydrate and trailing + * interrupt clearing. Optional; only effective when the snapshot store is enabled. + * + * @param snapshotStore the snapshot store, or null + * @return This builder + */ + public Builder snapshotStore(AguiSnapshotStore snapshotStore) { + this.snapshotStore = snapshotStore; + return this; + } + /** * Build the handler. * diff --git a/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/test/java/io/agentscope/spring/boot/agui/common/AguiAdapterConfigAutoConfigurationTest.java b/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/test/java/io/agentscope/spring/boot/agui/common/AguiAdapterConfigAutoConfigurationTest.java index 28655acddd..a6b65cb016 100644 --- a/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/test/java/io/agentscope/spring/boot/agui/common/AguiAdapterConfigAutoConfigurationTest.java +++ b/agentscope-extensions/agentscope-spring-boot-starters/agentscope-agui-spring-boot-starter/src/test/java/io/agentscope/spring/boot/agui/common/AguiAdapterConfigAutoConfigurationTest.java @@ -15,9 +15,12 @@ */ package io.agentscope.spring.boot.agui.common; +import static org.junit.Assert.assertNotNull; 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.assertIterableEquals; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -32,6 +35,7 @@ import io.agentscope.core.agui.registry.AguiAgentRegistry; import io.agentscope.core.agui.runtime.AguiRuntimeContextRequest; import io.agentscope.core.agui.runtime.AguiRuntimeContextResolver; +import io.agentscope.core.agui.store.AguiSnapshotStore; import io.agentscope.core.event.AgentEndEvent; import io.agentscope.core.event.AgentEvent; import io.agentscope.core.event.AgentStartEvent; @@ -383,6 +387,74 @@ void testRuntimeContextRequestHelpers() { assertThrowsUnsupportedOperation(() -> request.getQueryParams().put("x", List.of("y"))); } + @Test + @DisplayName("Should create AguiSnapshotStore bean when enabled (MVC parity)") + void testMvcSnapshotStoreBeanCreatedWhenEnabled() { + mvcContextRunner + .withPropertyValues("agentscope.agui.snapshot-store-enabled=true") + .run( + ctx -> { + assertEquals(1, ctx.getBeansOfType(AguiSnapshotStore.class).size()); + AguiAdapterConfig config = + mvcConfig(ctx.getBean(AguiMvcController.class)); + assertTrue(config.isSnapshotStoreEnabled()); + assertNotNull(config.getSnapshotStore()); + }); + } + + @Test + @DisplayName("Should not create AguiSnapshotStore bean by default (WebFlux)") + void testWebFluxSnapshotStoreBeanAbsentByDefault() { + webFluxContextRunner.run( + ctx -> { + assertEquals(0, ctx.getBeansOfType(AguiSnapshotStore.class).size()); + AguiAdapterConfig config = webFluxConfig(ctx.getBean(AguiWebFluxHandler.class)); + assertFalse(config.isSnapshotStoreEnabled()); + }); + } + + @Test + @DisplayName("Should create AguiSnapshotStore bean when enabled (WebFlux)") + void testWebFluxSnapshotStoreBeanCreatedWhenEnabled() { + webFluxContextRunner + .withPropertyValues("agentscope.agui.snapshot-store-enabled=true") + .run( + ctx -> { + assertEquals(1, ctx.getBeansOfType(AguiSnapshotStore.class).size()); + AguiAdapterConfig config = + webFluxConfig(ctx.getBean(AguiWebFluxHandler.class)); + assertTrue(config.isSnapshotStoreEnabled()); + assertNotNull(config.getSnapshotStore()); + }); + } + + @Test + @DisplayName("Should wire snapshot store into the handler only when enabled") + void testWebFluxConnectWiringPresentOnlyWhenEnabled() { + // Disabled by default: the handler's processor has no snapshot store, so /connect + // returns the empty handshake and no snapshot is recorded. + webFluxContextRunner.run( + ctx -> { + Object processor = + ReflectionTestUtils.getField( + ctx.getBean(AguiWebFluxHandler.class), "processor"); + assertNull(ReflectionTestUtils.getField(processor, "snapshotStore")); + }); + + // Enabled: the handler's processor carries the snapshot store backing /connect hydrate. + webFluxContextRunner + .withPropertyValues("agentscope.agui.snapshot-store-enabled=true") + .run( + ctx -> { + Object processor = + ReflectionTestUtils.getField( + ctx.getBean(AguiWebFluxHandler.class), "processor"); + Object store = ReflectionTestUtils.getField(processor, "snapshotStore"); + assertNotNull(store); + assertInstanceOf(AguiSnapshotStore.class, store); + }); + } + private static AguiAdapterConfig mvcConfig(AguiMvcController controller) { Object processor = ReflectionTestUtils.getField(controller, "processor"); return (AguiAdapterConfig) ReflectionTestUtils.getField(processor, "config"); diff --git a/docs/v2/en/integration/protocol/agui.md b/docs/v2/en/integration/protocol/agui.md index 750c72362c..5d624129bc 100644 --- a/docs/v2/en/integration/protocol/agui.md +++ b/docs/v2/en/integration/protocol/agui.md @@ -221,6 +221,9 @@ agentscope: emit-run-finished-after-error: false server-side-memory: false interrupt-on-disconnect: true + # Presentation snapshot store (both off by default) + snapshot-store-enabled: false + snapshot-max-threads: 1000 ``` `interrupt-on-disconnect` controls whether an Agent run is interrupted when the MVC/WebFlux SSE @@ -333,6 +336,59 @@ 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. +## Presentation Snapshot Store + +A reconnecting client needs to rebuild the visible conversation without re-running the agent. AgentScope Java provides a framework-level **presentation snapshot store** that materializes the AG-UI frames a browser should draw right now, and a read-only `POST {path-prefix}/connect` hydrate route that replays them. + +### Responsibility boundary + +| Component | Role | Mutated by hydrate? | +| --- | --- | --- | +| `AguiSnapshotStore` | Presentation state — derived, replayable, safe to lose. Answers "what should the browser draw right now". | No (read-only) | +| `AgentStateStore` (core) + `AguiResumeCoordinator` | Authoritative state — agent context and the live human-in-the-loop contract. | No (read-only) | + +Because the snapshot only ever retains the **trailing unresolved** interrupt, a resolved historical interrupt cannot be revived on reconnect — the failure mode is removed at the data model instead of being filtered after the fact. + +### `/connect` frame contract + +`POST {path-prefix}/connect` accepts a `RunAgentInput` (only `threadId` / `runId` are used) and returns a read-only SSE stream. Hydrate is strictly read-only: it resolves no agent, mutates no resume coordinator, and creates no adapter. Frames are emitted in this exact order: + +``` +RUN_STARTED(threadId, runId) +MESSAGES_SNAPSHOT(messages) +STATE_SNAPSHOT(state) // omitted when state is empty +ACTIVITY_SNAPSHOT(...) per ActivityFrame // omitted when none +RUN_FINISHED(result=null, outcome=pendingOutcome) +``` + +A null `pendingOutcome` serializes as a plain successful run. An empty or missing snapshot produces the minimal three-frame handshake (`RUN_STARTED` → `MESSAGES_SNAPSHOT([])` → `RUN_FINISHED`). + +Only the trailing unresolved interrupt is replayed; when a new run starts the store drops any trailing interrupt, so a resolved interrupt can never reappear. + +### Configuration + +```yaml +agentscope: + agui: + snapshot-store-enabled: false # set true to enable POST {path-prefix}/connect + snapshot-max-threads: 1000 # in-memory store capacity +``` + +Both keys are off by default, so existing clients stay byte-identical until you opt in. When enabled, the starter creates an in-memory `AguiSnapshotStore` bean and registers the `/connect` route (WebFlux `RouterFunction` / MVC `@PostMapping`). + +### Custom stores + +To persist snapshots externally (for example in Redis), implement `AguiSnapshotStore` and expose it as a bean — the starter's `@ConditionalOnMissingBean` defers to yours: + +```java +@Bean +AguiSnapshotStore aguiSnapshotStore(RedisTemplate redis) { + return new RedisAguiSnapshotStore(redis); +} +``` + +The store API is intentionally tiny: `save`, `find`, `delete`, and a `clearPendingInterrupts` default that drops the trailing interrupt. Snapshots are immutable records, so a custom store can serialize them with any Jackson-compatible codec. + ## 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 2101561b55..bfe75f95c0 100644 --- a/docs/v2/zh/integration/protocol/agui.md +++ b/docs/v2/zh/integration/protocol/agui.md @@ -221,6 +221,9 @@ agentscope: emit-run-finished-after-error: false server-side-memory: false interrupt-on-disconnect: true + # 展示快照存储（默认均关闭） + snapshot-store-enabled: false + snapshot-max-threads: 1000 ``` `interrupt-on-disconnect` 用于控制 MVC/WebFlux 的 SSE 连接关闭、超时或发送事件失败时是否中断 @@ -332,6 +335,59 @@ AG-UI 前端可以在 `RunAgentInput.tools` 中传入工具 schema。adapter 会 前端不需要在 `resume[]` 中回传 `metadata`；只需要发送 `interruptId`、`status` 和 `payload`。通过 Spring `AguiRequestProcessor` 入口时，AgentScope Java 会在服务端记录最近一次 `RUN_FINISHED.outcome.interrupts[]`，校验下一次 `resume[]` 是否覆盖所有 open interrupts，并把原始 interrupt 传给 adapter 做恢复转换。 +## 展示快照存储 + +重连的客户端需要在不重新运行 agent 的前提下重建可见会话。AgentScope Java 提供了框架级的**展示快照存储**，物化浏览器当前应渲染的 AG-UI 帧，并通过只读的 `POST {path-prefix}/connect` 水合路由重放它们。 + +### 职责边界 + +| 组件 | 角色 | 水合是否修改？ | +| --- | --- | --- | +| `AguiSnapshotStore` | 展示状态——派生、可重放、可丢失。回答“浏览器现在应渲染什么”。 | 否（只读） | +| `AgentStateStore`（核心）+ `AguiResumeCoordinator` | 权威状态——agent 上下文与活跃的人机交互契约。 | 否（只读） | + +由于快照只保留**最后一个未解决的 interrupt**，已解决的历史 interrupt 不可能在重连时复活——这一故障模式在数据模型层被消除，而非事后过滤。 + +### `/connect` 帧契约 + +`POST {path-prefix}/connect` 接收 `RunAgentInput`（仅使用 `threadId` / `runId`），返回只读 SSE 流。水合是严格只读的：不解析 agent、不修改 resume coordinator、不创建 adapter。帧按如下顺序发送： + +``` +RUN_STARTED(threadId, runId) +MESSAGES_SNAPSHOT(messages) +STATE_SNAPSHOT(state) // state 为空时省略 +ACTIVITY_SNAPSHOT(...) 每个 ActivityFrame // 没有时省略 +RUN_FINISHED(result=null, outcome=pendingOutcome) +``` + +`pendingOutcome` 为 null 时序列化为普通成功运行。空或缺失的快照返回最小三帧握手（`RUN_STARTED` → `MESSAGES_SNAPSHOT([])` → `RUN_FINISHED`）。 + +只重放最后一个未解决的 interrupt；新 run 开始时存储会丢弃残留 interrupt，因此已解决的 interrupt 永不会再次出现。 + +### 配置 + +```yaml +agentscope: + agui: + snapshot-store-enabled: false # 设为 true 以启用 POST {path-prefix}/connect + snapshot-max-threads: 1000 # 内存存储容量 +``` + +两个键默认关闭，因此在显式开启前现有客户端保持字节级一致。启用后，starter 会创建内存版 `AguiSnapshotStore` bean 并注册 `/connect` 路由（WebFlux `RouterFunction` / MVC `@PostMapping`）。 + +### 自定义存储 + +如需将快照持久化到外部（例如 Redis），实现 `AguiSnapshotStore` 并暴露为 bean——starter 的 `@ConditionalOnMissingBean` 会优先使用你提供的： + +```java +@Bean +AguiSnapshotStore aguiSnapshotStore(RedisTemplate redis) { + return new RedisAguiSnapshotStore(redis); +} +``` + +存储 API 刻意保持极简：`save`、`find`、`delete`，以及一个丢弃残留 interrupt 的 `clearPendingInterrupts` 默认实现。快照是不可变 record，自定义存储可用任意 Jackson 兼容 codec 序列化。 + ## 示例项目 完整示例见 [agentscope-examples/agui](https://github.com/agentscope-ai/agentscope-java/tree/main/agentscope-examples/agui)：