[Bug]:streamEvents cancellation does not guarantee AgentState checkpoint for resumable sessions

[CyberXavier]

Describe the bug

HarnessAgent.streamEvents(...) / ReActAgent.streamEvents(...) does not appear to guarantee an AgentStateStore checkpoint when the stream is cancelled or interrupted externally, for example when an SSE client disconnects, an HTTP request times out, or the upstream subscription is disposed.

This makes it difficult to build reliable resumable conversations on top of AgentScope-Java. During a long-running turn involving tool calls, subagents, or multiple reasoning/acting iterations, the user may have already received partial visible output and the agent may have updated in-memory AgentState. However, if the stream is cancelled before normal completion, those updates may not be persisted to RedisDistributedStore / AgentStateStore.

From reading the code, AgentState is saved on normal completion via saveStateToSession(...), and some controlled paths such as RequestStopEvent or graceful shutdown can persist state. But plain stream cancellation seems to dispose the lifecycle subscription without a guaranteed best-effort AgentState save.

To Reproduce

1. Build a HarnessAgent or ReActAgent with an external AgentStateStore, for example via RedisDistributedStore.

   DistributedStore store = RedisDistributedStore.fromJedis(jedis, "test:");
   
   HarnessAgent agent = HarnessAgent.builder()
       .name("test-agent")
       .model(model)
       .distributedStore(store)
       .build();

2. Start a long-running streaming invocation. The turn should involve a model response, tool call, subagent call, or any operation that takes long enough to cancel midway.

   RuntimeContext ctx = RuntimeContext.builder()
       .userId("user-1")
       .sessionId("session-1")
       .build();
   
   Disposable disposable = agent.streamEvents(
           new UserMessage("Run a long task that uses tools or subagents."),
           ctx)
       .subscribe(
           event -> System.out.println(event.getType()),
           Throwable::printStackTrace,
           () -> System.out.println("completed"));

3. Cancel the subscription before the stream completes.

   Thread.sleep(30_000);
   disposable.dispose();

4. Load the state from the configured AgentStateStore.

   Optional<AgentState> state = store.agentStateStore()
       .get("user-1", "session-1", "agent_state", AgentState.class);

5. Observe that the persisted state may not include the latest partial reasoning/tool/subagent context that existed in the in-memory AgentState before cancellation.

Expected behavior

When a streamEvents(...) subscription is cancelled, AgentScope-Java should provide a reliable best-effort checkpoint mechanism for the current session's AgentState.

Ideally:

• On stream cancellation, timeout, or external disposal, the active call-scoped AgentState should be saved to the configured AgentStateStore if one exists.
• This should happen without requiring normal completion of the agent turn.
• The behavior should be documented clearly, especially for production SSE/WebFlux use cases.
• If saving on every cancellation is too expensive or not always desired, AgentScope-Java could expose an official middleware/checkpoint hook or configurable policy.

This would allow applications to resume conversations more reliably after browser disconnects, gateway timeouts, or user-initiated cancellation.

Error messages

There is no explicit error message. The issue is that cancellation can silently lose in-memory AgentState updates that were not yet persisted.

Environment (please complete the following information):

• AgentScope-Java Version: 2.0

Additional context

This matters for production chat/SSE applications.

A typical scenario:

1. A user starts a long-running turn.
2. The agent spends several minutes reasoning, calling tools, or delegating to subagents.
3. The user has already seen partial streamed output.
4. At minute 9, the browser disconnects, the gateway times out, or the user cancels the request.
5. The application wants to restore both:
   • the user-visible partial conversation, stored in the application database;
   • the internal AgentScope state, stored in AgentStateStore, so the next turn can continue with correct tool/subagent/plan context.

Application-level databases can store the user-visible partial text, but they cannot fully reconstruct internal AgentScope runtime state such as AgentState.context, tool_context, tasks_context, plan_mode_context, pending tool states, or subagent registrations.

So the application needs AgentScope-Java to either:

• checkpoint AgentState on cancellation; or
• provide an official high-performance middleware/checkpoint extension point for durable streaming sessions.