fix(issue #2528): index active RuntimeContext by runId for concurrent multi-session support

agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java

@@ -195,12 +195,13 @@
  *     .build()).block();
  * }</pre>
  *
- * <p><b>Thread Safety:</b> {@code ReActAgent} is <em>not</em> thread-safe. A single instance
- * processes exactly one {@code call()} at a time; a concurrent invocation on the same instance
- * throws {@link IllegalStateException}. For web services or other concurrent scenarios, create
- * one agent instance per request via a factory method. {@link io.agentscope.core.model.Model},
- * {@link io.agentscope.core.tool.Toolkit} (as a template — {@code build()} deep-copies it), and
- * {@link io.agentscope.core.state.AgentStateStore} are all safe to share across instances.
+ * <p><b>Thread Safety:</b> A single instance supports concurrent {@code call()} invocations on
+ * <em>distinct</em> {@code (userId, sessionId)} slots (calls sharing a slot are serialized FIFO).
+ * Per-call state — including the active {@link RuntimeContext} — is tracked in a runId-indexed
+ * registry (see {@link #getRuntimeContext(String)}), so concurrent sessions no longer overwrite
+ * each other. {@link io.agentscope.core.model.Model}, {@link io.agentscope.core.tool.Toolkit} (as a
+ * template — {@code build()} deep-copies it), and {@link io.agentscope.core.state.AgentStateStore}
+ * are all safe to share across instances.
  */
 @SuppressWarnings("deprecation")
 public class ReActAgent extends AgentBase implements AutoCloseable {
@@ -264,9 +265,6 @@ public class ReActAgent extends AgentBase implements AutoCloseable {
 
     // ==================== 2.0 Core Fields ====================
 
-    /** Active per-call RuntimeContext, set during call lifecycle only. */
-    private volatile RuntimeContext activeRc;
-
     /** Cache of state per {@code (userId, sessionId)} slot key. */
     private final ConcurrentHashMap<String, AgentState> stateCache = new ConcurrentHashMap<>();
 
@@ -546,7 +544,6 @@ protected Object beforeAgentExecution(List<Msg> msgs, RuntimeContext rc) {
         // the active session's state via rc.getAgentState() (call-scoped, concurrency-safe)
         // rather than agent.getAgentState() (not call-scoped under concurrency).
         ctx.setAgentState(scope.state);
-        this.activeRc = ctx;
         bindRuntimeContextToHooks(ctx);
         // Seed per-call state onto the active execution scope. The system message is initialised
         // by consumeSystemMsgAfterPreCall; the event sink (if any) is bound in doCall() from the
@@ -560,8 +557,7 @@ protected Object beforeAgentExecution(List<Msg> msgs, RuntimeContext rc) {
 
     @Override
     protected Mono<Msg> seedSystemMsg(Object callExectution) {
-        RuntimeContext rc =
-                callExectution instanceof CallExecution ce ? ce.rc : getRuntimeContext();
+        RuntimeContext rc = callExectution instanceof CallExecution ce ? ce.rc : null;
         String base = sysPrompt != null ? sysPrompt.trim() : "";
         return applySystemPromptMiddlewares(base, rc)
                 .filter(prompt -> !prompt.isEmpty())
@@ -618,12 +614,6 @@ protected void consumeSystemMsgAfterPreCall(Msg systemMsg, Object callScope) {
         syncToolkitToState(ce.state);
     }
 
-    @Override
-    protected void afterAgentExecution() {
-        this.activeRc = null;
-        unbindRuntimeContextFromHooks();
-    }
-
     private RuntimeContext buildMergedRuntimeContext(RuntimeContext run) {
         if (run == null) {
             if (toolExecutionContext != null) {
@@ -850,7 +840,8 @@ private Flux<AgentEvent> buildAgentStream(
 
                                     // Call runLifecycle directly — NOT call() — to avoid the
                                     // onAgent chain being applied a second time.
-                                    Mono<Msg> lifecycle = runLifecycle(input.msgs(), doCallFn);
+                                    Mono<Msg> lifecycle =
+                                            runLifecycle(input.msgs(), context, doCallFn);
                                     if (context != null) {
                                         lifecycle =
                                                 lifecycle.contextWrite(
@@ -1016,7 +1007,8 @@ protected Mono<Msg> doCall(List<Msg> msgs) {
                                 .ifPresent(ae -> scope.externalEventEmitter = ae);
                     }
                     return scope.doCallInner(msgs)
-                            .flatMap(result -> saveStateToSession(scope).thenReturn(result));
+                            .flatMap(result -> saveStateToSession(scope).thenReturn(result))
+                            .doFinally(s -> activeContexts.remove(scope.runId));
                 });
     }
 
@@ -1119,7 +1111,8 @@ private Mono<Msg> doNativeStructuredCall(List<Msg> msgs, Map<String, Object> jso
                                             ctx.remove(ctx.size() - 1);
                                         }
                                         scope.nativeResponseFormat = null;
-                                    });
+                                    })
+                            .doFinally(s -> activeContexts.remove(scope.runId));
                 });
     }
 
@@ -1171,7 +1164,8 @@ private Mono<Msg> doFallbackStructuredCall(List<Msg> msgs, Map<String, Object> j
                                             scope.state.contextMutable().add(out);
                                         }
                                         return saveStateToSession(scope).thenReturn(out);
-                                    });
+                                    })
+                            .doFinally(s -> activeContexts.remove(scope.runId));
                 });
     }
 
@@ -1438,6 +1432,9 @@ final class CallExecution {
         PermissionEngine permissionEngine;
         String slotKey;
 
+        /** Per-call runId, captured at registration for reliable cleanup via doFinally. */
+        String runId;
+
         /**
          * Per-call system message, propagated across PreCallEvent → PreReasoningEvent /
          * PreSummaryEvent. Owned by a single logical execution: seeded to {@code null} at call
@@ -3648,11 +3645,6 @@ public AgentState getAgentState() {
         return getAgentState(null, defaultSessionId);
     }
 
-    @Override
-    public RuntimeContext getRuntimeContext() {
-        return activeRc;
-    }
-
     /**
      * Returns the {@link AgentState} for the session identified by the given {@link RuntimeContext}.
      *

agentscope-core/src/main/java/io/agentscope/core/agent/Agent.java

@@ -111,4 +111,28 @@ default io.agentscope.core.state.AgentState getAgentState() {
     default Toolkit getToolkit() {
         return null;
     }
+
+    /**
+     * Returns the active {@link RuntimeContext} for the given runId, or {@code null} if no such
+     * call is currently in flight. Use {@link RuntimeContext#getRunId()} to obtain the key.
+     *
+     * <p>The base implementation returns {@code null}; agents that track per-call contexts (e.g.
+     * {@code ReActAgent}) override this to look up the runId in their active-context registry.
+     *
+     * @param runId the per-call run id
+     * @return the matching in-flight context, or {@code null}
+     */
+    default RuntimeContext getRuntimeContext(String runId) {
+        return null;
+    }
+
+    /**
+     * Returns a read-only snapshot of all currently in-flight {@link RuntimeContext}s on this agent
+     * instance. The base implementation returns an empty list.
+     *
+     * @return unmodifiable list of active contexts (may be empty)
+     */
+    default java.util.List<RuntimeContext> getActiveRuntimeContexts() {
+        return java.util.List.of();
+    }
 }

agentscope-core/src/main/java/io/agentscope/core/agent/AgentBase.java

@@ -104,6 +104,14 @@ public abstract class AgentBase implements Agent {
     private final CopyOnWriteArrayList<RuntimeContextAware> runtimeContextAwareHooks =
             new CopyOnWriteArrayList<>();
 
+    /**
+     * Active per-call {@link RuntimeContext}s, indexed by {@link RuntimeContext#getRunId()}.
+     * Subclasses that manage call lifecycles (e.g. {@code ReActAgent}) register each call's RC at
+     * call entry and remove it on termination, enabling concurrent multi-session support.
+     */
+    protected final ConcurrentHashMap<String, RuntimeContext> activeContexts =
+            new ConcurrentHashMap<>();
+
     /**
      * Per-key call serialization tails. Each entry holds the completion signal of the most recently
      * enqueued call for that key; the next call for the same key chains after it, so calls sharing a
@@ -199,7 +207,7 @@ public final Mono<Msg> call(List<Msg> msgs) {
      *
      * <p>The default implementation attaches {@code context} to the Reactor Context (when
      * non-null) and delegates straight to {@link #runLifecycle}. Subclasses that override this
-     * method must eventually invoke {@code runLifecycle(msgs, doCallFn)} to run the standard
+     * method must eventually invoke {@code runLifecycle(msgs, context, doCallFn)} to run the standard
      * lifecycle (shutdown guard, serialization gate, pre/post hooks, tracing).
      *
      * @param msgs     input messages
@@ -210,7 +218,7 @@ public final Mono<Msg> call(List<Msg> msgs) {
      */
     protected Mono<Msg> callInternal(
             List<Msg> msgs, RuntimeContext context, Function<List<Msg>, Mono<Msg>> doCallFn) {
-        Mono<Msg> lifecycle = runLifecycle(msgs, doCallFn);
+        Mono<Msg> lifecycle = runLifecycle(msgs, context, doCallFn);
         return context == null
                 ? lifecycle
                 : lifecycle.contextWrite(c -> c.put(RUNTIME_CONTEXT_KEY, context));
@@ -250,24 +258,30 @@ protected Mono<Msg> callInternal(
      * that scope on the Reactor Context, and run the preCall → doCall → postCall chain with error
      * handling, releasing execution on terminate.
      */
-    protected Mono<Msg> runLifecycle(List<Msg> msgs, Function<List<Msg>, Mono<Msg>> doCallFn) {
+    protected Mono<Msg> runLifecycle(
+            List<Msg> msgs, RuntimeContext context, Function<List<Msg>, Mono<Msg>> doCallFn) {
         return Mono.using(
-                this::acquireExecution,
+                () -> acquireExecution(context),
                 resource ->
                         Mono.deferContextual(
                                 cv -> {
-                                    RuntimeContext rc =
-                                            (RuntimeContext)
-                                                    cv.getOrDefault(RUNTIME_CONTEXT_KEY, null);
+                                    final RuntimeContext rc =
+                                            context != null ? context : RuntimeContext.empty();
                                     // Track this call as a distinct shutdown request (keyed by its
                                     // own requestId, not the shared agent id) so concurrent calls
-                                    // on
-                                    // one instance are interrupted/saved/unregistered
+                                    // on one instance are interrupted/saved/unregistered
                                     // independently.
                                     String requestId =
                                             GracefulShutdownManager.getInstance()
                                                     .registerRequest(this);
                                     Object gateKey = callSerializationKey(rc);
+                                    // Register the caller-supplied RC (if any) so
+                                    // getRuntimeContext(runId) resolves during the call. Removal
+                                    // is in releaseExecution (Mono.using cleanup, guaranteed on
+                                    // complete / error / cancel) — both use the same context.
+                                    if (context != null) {
+                                        activeContexts.put(context.getRunId(), context);
+                                    }
                                     // Build the per-call lifecycle lazily so it only runs once the
                                     // serialization gate (if any) admits this call:
                                     // beforeAgentExecution resolves/loads the session slot and must
@@ -293,7 +307,7 @@ protected Mono<Msg> runLifecycle(List<Msg> msgs, Function<List<Msg>, Mono<Msg>>
                                                             GracefulShutdownManager.getInstance()
                                                                     .unregisterRequest(requestId));
                                 }),
-                this::releaseExecution,
+                resource -> releaseExecution(resource, context),
                 true);
     }
 
@@ -469,19 +483,20 @@ private InterruptContext createInterruptContext() {
     }
 
     /**
-     * Acquire execution resources for a {@code call()} invocation.
-     * Used as the {@code resourceSupplier} in {@link Mono#using} to guarantee that
-     * {@link #releaseExecution} is always called on completion, error, or cancellation.
-     *
-     * @return this agent instance
+     * Acquire execution resources for a {@code call()} invocation. Used as the
+     * {@code resourceSupplier} in {@link Mono#using} to guarantee that {@link #releaseExecution}
+     * is always called on completion, error, or cancellation.
      */
-    private AgentBase acquireExecution() {
+    private AgentBase acquireExecution(RuntimeContext context) {
         GracefulShutdownManager.getInstance().ensureAcceptingRequests();
         return this;
     }
 
-    private void releaseExecution(AgentBase resource) {
-        afterAgentExecution();
+    private void releaseExecution(AgentBase resource, RuntimeContext context) {
+        if (context != null) {
+            activeContexts.remove(context.getRunId());
+        }
+        afterAgentExecution(context);
     }
 
     /**
@@ -560,15 +575,26 @@ public AgentState getAgentState() {
     }
 
     /**
-     * Returns the current per-call {@link RuntimeContext}, or {@code null} when the agent keeps no
-     * per-call scope. The base implementation returns {@code null}; agents with per-call state
-     * (e.g. {@code ReActAgent}) override this to return their active call scope's context. Because
-     * the value is sourced from the agent's most-recently-activated scope, under concurrent calls
-     * on one instance this reflects the latest call — middlewares/tools that need their own call's
-     * context should read it from the per-subscription {@link RuntimeContext} they are handed.
+     * Returns the active {@link RuntimeContext} for the given runId, or {@code null} if no such
+     * call is currently in flight. Looks up the runId in {@link #activeContexts}.
+     *
+     * @param runId the per-call run id (from {@link RuntimeContext#getRunId()})
+     * @return the matching in-flight context, or {@code null}
      */
-    public RuntimeContext getRuntimeContext() {
-        return null;
+    @Override
+    public RuntimeContext getRuntimeContext(String runId) {
+        return runId != null ? activeContexts.get(runId) : null;
+    }
+
+    /**
+     * Returns a read-only snapshot of all currently in-flight {@link RuntimeContext}s on this agent
+     * instance.
+     *
+     * @return unmodifiable list of active contexts (may be empty)
+     */
+    @Override
+    public List<RuntimeContext> getActiveRuntimeContexts() {
+        return List.copyOf(activeContexts.values());
     }
 
     /**
@@ -590,15 +616,25 @@ protected Object beforeAgentExecution(List<Msg> msgs, RuntimeContext rc) {
     }
 
     /**
-     * Invoked in {@code Mono.using} cleanup, before clearing the running state. Pairs with {@link
-     * #beforeAgentExecution(List, RuntimeContext)}. The default is a no-op.
+     * Invoked when a {@code call()} terminates (complete, error, or cancel), paired with {@link
+     * #beforeAgentExecution(List, RuntimeContext)}. Removes the call's RuntimeContext from
+     * {@link #activeContexts} and unbinds hooks.
+     *
+     * @param rc the per-call {@link RuntimeContext} (never {@code null}; resolved to empty when the
+     *     caller supplied none)
      */
-    protected void afterAgentExecution() {}
+    protected void afterAgentExecution(RuntimeContext rc) {
+        if (rc != null) {
+            activeContexts.remove(rc.getRunId());
+        }
+        unbindRuntimeContextFromHooks();
+    }
 
     /**
      * Pushes {@code ctx} to all {@link RuntimeContextAware} hooks registered for this agent. The
-     * per-call {@link RuntimeContext} itself is no longer stored on a shared instance field; it
-     * lives on the agent's per-call scope (see {@link #getRuntimeContext()}).
+     * per-call {@link RuntimeContext} is delivered through this setter, the middleware {@code ctx}
+     * parameter, and {@code ToolCallParam.getRuntimeContext()} — it is not stored on a shared
+     * instance field.
      */
     protected void bindRuntimeContextToHooks(RuntimeContext ctx) {
         for (RuntimeContextAware h : runtimeContextAwareHooks) {