fix(agui): suppress resolved historical interrupts on connect replay (#2911)

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.
  *
+ * <p>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 {

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<AguiSnapshotStore> snapshotStoreProvider;
+    private final AguiSnapshotHydrator hydrator = new AguiSnapshotHydrator();
     private final AguiEventEncoder encoder = new AguiEventEncoder();
 
     public CopilotKitRuntimeService(
-            AguiAgentRegistry aguiAgentRegistry, AgentEventAguiReplayer eventReplayer) {
+            AguiAgentRegistry aguiAgentRegistry,
+            ObjectProvider<AguiSnapshotStore> 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.
      *
-     * <p>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.
+     * <p>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<ServerSentEvent<String>> connect(RunAgentInput input) {
         String threadId = input.getThreadId();
         String runId = input.getRunId();
-        List<AguiEvent> 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<AguiEvent> emptyHandshake(String threadId, String runId) {
-        List<AguiEvent> 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<AguiEvent> frames = hydrator.hydrate(snapshot, threadId, runId);
+        return Flux.fromIterable(frames).map(this::sse);
     }
 
     private ServerSentEvent<String> sse(AguiEvent event) {

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.
  *
- * <p>AG-UI frames themselves are not stored — connect replay re-projects through converters.
+ * <p>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.
  *
  * <p>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).

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:

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).

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:

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<AguiEventEnricher> 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<AguiEventEnricher> buildEventEnrichers(
+            Builder builder, SnapshotRecordingEnricher recorder) {
+        List<AguiEventEnricher> 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.
+     *
+     * <p>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<AguiEventEnricher> 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<AguiEventEnricher> 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.
          *