Repository navigation
Conversation
|
cc @dragondyt |
…ntscope-ai#2920) Framework-level presentation-state store so reconnecting clients rebuild the visible conversation without re-running the agent; collapses the agentscope-ai#2911 copilotkit workaround into the data model (only the trailing unresolved interrupt is ever retained, so a resolved historical interrupt cannot reappear on reconnect). - store/: AguiJsonPatch (RFC 6902 applier), AguiThreadSnapshot, AguiSnapshotStore + InMemoryAguiSnapshotStore, AguiSnapshotAccumulator (absorbs resolved-interrupt suppression + dangling-tool synthesis), SnapshotRecordingEnricher (last observer, flush safety net), AguiSnapshotHydrator (read-only replay; empty -> 3-frame handshake) - AguiActivityConverter + shared AguiJsonDiff helper (mirrors state converter) - AguiAdapterConfig snapshot-store-enabled/snapshotStore (default off); AguiRequestProcessor.hydrate(), clearPendingInterrupts at beginRun, flush in doFinally - starters: gated AguiSnapshotStore bean + POST {path-prefix}/connect route (WebFlux RouterFunction + MVC conditional AguiConnectController) - copilotkit: migrate /connect to framework hydrator, delete suppressResolvedInterrupts/danglingToolCalls + obsolete test, enable store - docs (en/zh): Presentation Snapshot Store section + YAML sample - tests: store package, activity converter, config/processor/starter wiring and an end-to-end process-records -> hydrate-replays case
|
The #2911 reconnect bug is clear, but this PR now includes two different changes. |
…2911-copilotkit-replay-interrupt
oss-maintainer
left a comment
oss-maintainer
left a comment
There was a problem hiding this comment.
Summary
Replaces CopilotKit's event-log replay on connect with a framework-level AG-UI presentation snapshot store (AguiSnapshotStore / AguiSnapshotAccumulator / AguiSnapshotHydrator), so a resolved historical interrupt cannot be revived on reconnect and dangling tool calls are closed with a synthetic result. The direction is sound and the 9 new test classes are well above this repo's average. The findings below are about concurrency, unbounded state, and scope rather than about the fix being wrong — leaving as COMMENT for maintainer judgement.
Findings
- [Warning]
AguiSnapshotStore.java:68— defaultclearPendingInterruptsis a non-atomic read-modify-write that can resurrect the trailing interrupt it is meant to clear. - [Warning]
SnapshotRecordingEnricher.java:44—accumulatorsis unbounded and a leaked entry retains the whole thread transcript, not just one run. - [Warning]
SnapshotRecordingEnricher.java:76—AguiSnapshotAccumulatoruses plainLinkedHashMaps while the enricher is documented as reachable from two conversion paths. - [Warning]
AguiRequestProcessor.java:246— the"unknown"threadId fallback becomes a shared store key for reconnect. - [Info]
InMemoryAguiSnapshotStore.java:77— full-map rescan (plus retries) on every save once at capacity. - [Info]
CopilotKitRuntimeService.java:125— with the defaultsnapshot-store-enabled=false,connectnow returns an empty handshake where it used to replay persisted history. - [Info]
AguiConnectController.java:88— parser message echoed to the client, SSE frames assembled by hand.
Scope notes (no anchor line)
- Title/description vs. implementation. The description still presents this as a CopilotKit fix that post-processes
replay()viasuppressResolvedInterrupts(...)and adds anAgentEventAguiReplayerTest— neither exists in the diff (replay now serves only/threads/{id}/events, and the suppression moved into the framework accumulator). Refreshing the description would keep the rationale readable in history. The CopilotKit-side regression test for #2911 is also gone; the new store tests cover the mechanism, but not the reported user-visible flow. - API surface. This adds a public SPI (
AguiSnapshotStoreplus 6 new public types), two starter properties, and a newPOST {path-prefix}/connectroute in both MVC and WebFlux — a considerably wider blast radius thanfix(agui)suggests. Per the repo's review focus ("changes to core interfaces cascade to harness, distribution and all extensions"), splitting the framework store from the CopilotKit fix, or relabelling asfeat(agui), would give the interface addition its own review. - Durability. The only implementation is in-memory and per-node, so hydrate cannot serve a reconnect after a restart or behind horizontal scaling (a stated goal of
agentscope-distribution). "Presentation-only, safe to lose" is a fair framing; a sentence indocs/v2/{en,zh}/integration/protocol/agui.mdon the multi-instance limitation would set expectations for anyone writing a durable implementation.
Verification asked
mvn -pl agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agui test plus both affected starters (agentscope-agui-spring-boot-starter) — the description reports only the CopilotKit module run.
Automated review by github-manager-bot
| * @param threadId the thread id | ||
| */ | ||
| default void clearPendingInterrupts(String threadId) { | ||
| find(threadId) |
There was a problem hiding this comment.
clearPendingInterrupts is a non-atomic read-modify-write (find -> filter -> save). If a run for the same thread reaches its terminal frame concurrently, the older snapshot written here can overwrite the freshly materialized one, which resurrects exactly the trailing RunFinishedInterruptOutcome this PR exists to remove. An atomic SPI (compute(threadId, UnaryOperator<AguiThreadSnapshot>), which ConcurrentHashMap gives for free) would close the window; at minimum the interface javadoc should state that implementations must serialize this transition.
|
|
||
| private final AguiSnapshotStore store; | ||
| private final ConcurrentMap<String, AguiSnapshotAccumulator> accumulators = | ||
| new ConcurrentHashMap<>(); |
There was a problem hiding this comment.
This map is bounded only by traffic: an entry is dropped on a terminal frame or by an explicit flush(threadId, runId). A run that ends before either path is reached (transport error outside the enricher, a converter path that never emits the terminal frame, a cancellation that skips the doFinally hook) leaks its accumulator forever — and because each accumulator is seeded from store.find(threadId), the leaked object holds the whole thread transcript, not just one run. Could you bound it (max in-flight entries, or a TTL sweep alongside snapshotMaxThreads)?
| threadId, runId, store.find(threadId).orElse(null))); | ||
| for (AguiEvent event : events) { | ||
| accumulator.consume(event); | ||
| } |
There was a problem hiding this comment.
AguiSnapshotAccumulator is built on plain LinkedHashMaps with no synchronization, and the class javadoc says the enricher is reached from both AgentEventConverterRegistry.convert() and the framework enrich(null, finishPendingEvents, ctx) path. If those two can run on different threads for the same threadId:runId, consume races on the same instance (lost deltas, or ConcurrentModificationException during materialize). If single-threaded delivery is guaranteed by the caller, please say so here; otherwise a synchronized (accumulator) around consume/materialize is cheap insurance.
|
|
||
| private void evictIfOverCapacity() { | ||
| while (snapshots.size() > maxThreads) { | ||
| Map.Entry<String, AguiThreadSnapshot> oldest = null; |
There was a problem hiding this comment.
Once at capacity every save(...) scans the whole map to find the oldest entry, and the conditional remove(key, value) below can fail when a concurrent writer replaced that entry, so the while loop rescans. At the default snapshotMaxThreads=1000 that is a 1000-entry scan (plus retries) per completed run, on the request path. Sampling a few entries, keeping a separate ordering index, or evicting only when a size check wins would all be cheaper.
| String threadId = input != null ? input.getThreadId() : null; | ||
| String runId = input != null ? input.getRunId() : null; | ||
| if (threadId == null) { | ||
| threadId = "unknown"; |
There was a problem hiding this comment.
Substituting the literal "unknown" for a missing threadId and then querying the store with it makes every request that omits threadId share one bucket; if any run is ever recorded under that same fallback key, a later threadId-less connect from a different client hydrates someone else's conversation. Hydrate is read-only and has no agent to resolve, so rejecting the request (or returning the empty handshake without touching the store) is safer than inventing a key.
| events.add(new AguiEvent.RunFinished(threadId, runId)); | ||
| return events; | ||
| AguiSnapshotStore store = snapshotStoreProvider.getIfAvailable(); | ||
| AguiThreadSnapshot snapshot = store != null ? store.find(threadId).orElse(null) : null; |
There was a problem hiding this comment.
With the new default (snapshot-store-enabled=false) store is null here, so connect always answers with the three-frame empty handshake where the previous implementation replayed persisted AgentEvents. That is intentional, but it is a user-visible behaviour change for anyone upgrading an existing deployment: reconnect silently loses history instead of restoring it. Worth a line in the docs next to the property, and a one-time startup log when connect is registered without a store.
| public ResponseEntity<String> handleParseError(JsonException error) { | ||
| String errorEvent = | ||
| encoder.encodeToJson( | ||
| new AguiEvent.Raw( |
There was a problem hiding this comment.
Two small points on this error path: error.getMessage() is echoed straight to the caller (Jackson messages can carry payload fragments and internal type names — a generic client message plus a server-side log is safer), and the SSE frames are assembled by hand, so a message containing a newline breaks frame framing. Reusing encoder.encodeToJson framing for both events, as the happy path does, keeps the error response consistent.
|
This PR currently conflicts with git fetch upstream
git checkout fix/2911-copilotkit-replay-interrupt
git rebase upstream/main
# resolve conflicts, then:
git push --force-with-leaseOnce conflicts are resolved, ping @oss-maintainer for a re-review if the head also moves. This is a one-time reminder. Automated notification by github-manager-bot |
|
@chang6666 这块还有继续推进吗 |
Fix #2911
Connect replay (/agent/{agentId}/connect) re-projects stored AgentEvents through the same converters, so historical RequireUserConfirmEvents that a later run already resolved were converted back into RunFinishedInterruptOutcome, re-activating the frontend interrupt form and leaving the suspended tool rendered as Running instead of Done.
Changes
Verification