diff --git a/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java b/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java index 562f1f9ed6..a80fc1c08d 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java +++ b/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java @@ -195,12 +195,13 @@ * .build()).block(); * } * - *

Thread Safety: {@code ReActAgent} is not 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. + *

Thread Safety: A single instance supports concurrent {@code call()} invocations on + * distinct {@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 stateCache = new ConcurrentHashMap<>(); @@ -546,7 +544,6 @@ protected Object beforeAgentExecution(List 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 msgs, RuntimeContext rc) { @Override protected Mono 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 buildAgentStream( // Call runLifecycle directly — NOT call() — to avoid the // onAgent chain being applied a second time. - Mono lifecycle = runLifecycle(input.msgs(), doCallFn); + Mono lifecycle = + runLifecycle(input.msgs(), context, doCallFn); if (context != null) { lifecycle = lifecycle.contextWrite( @@ -1016,7 +1007,8 @@ protected Mono doCall(List 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 doNativeStructuredCall(List msgs, Map jso ctx.remove(ctx.size() - 1); } scope.nativeResponseFormat = null; - }); + }) + .doFinally(s -> activeContexts.remove(scope.runId)); }); } @@ -1171,7 +1164,8 @@ private Mono doFallbackStructuredCall(List msgs, Map 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}. * diff --git a/agentscope-core/src/main/java/io/agentscope/core/agent/Agent.java b/agentscope-core/src/main/java/io/agentscope/core/agent/Agent.java index 438d1a32f7..d8ae771c78 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/agent/Agent.java +++ b/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. + * + *

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 getActiveRuntimeContexts() { + return java.util.List.of(); + } } diff --git a/agentscope-core/src/main/java/io/agentscope/core/agent/AgentBase.java b/agentscope-core/src/main/java/io/agentscope/core/agent/AgentBase.java index a6b0248b6d..06cdd165a3 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/agent/AgentBase.java +++ b/agentscope-core/src/main/java/io/agentscope/core/agent/AgentBase.java @@ -104,6 +104,14 @@ public abstract class AgentBase implements Agent { private final CopyOnWriteArrayList 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 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 call(List msgs) { * *

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 call(List msgs) { */ protected Mono callInternal( List msgs, RuntimeContext context, Function, Mono> doCallFn) { - Mono lifecycle = runLifecycle(msgs, doCallFn); + Mono lifecycle = runLifecycle(msgs, context, doCallFn); return context == null ? lifecycle : lifecycle.contextWrite(c -> c.put(RUNTIME_CONTEXT_KEY, context)); @@ -250,24 +258,30 @@ protected Mono callInternal( * that scope on the Reactor Context, and run the preCall → doCall → postCall chain with error * handling, releasing execution on terminate. */ - protected Mono runLifecycle(List msgs, Function, Mono> doCallFn) { + protected Mono runLifecycle( + List msgs, RuntimeContext context, Function, Mono> 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 runLifecycle(List msgs, Function, Mono> 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 getActiveRuntimeContexts() { + return List.copyOf(activeContexts.values()); } /** @@ -590,15 +616,25 @@ protected Object beforeAgentExecution(List 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) { diff --git a/agentscope-core/src/main/java/io/agentscope/core/agent/RuntimeContext.java b/agentscope-core/src/main/java/io/agentscope/core/agent/RuntimeContext.java index 69c1a71f7d..78b94b5e0c 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/agent/RuntimeContext.java +++ b/agentscope-core/src/main/java/io/agentscope/core/agent/RuntimeContext.java @@ -20,6 +20,7 @@ import io.agentscope.core.tool.ToolExecutionContext; import java.util.HashMap; import java.util.Map; +import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -37,6 +38,13 @@ public class RuntimeContext { private final String sessionId; private final String userId; + /** + * Stable per-call correlation id. Always non-null: auto-generated by the {@link Builder} when + * not explicitly supplied. Use this to correlate logs, events, and traces for a single + * {@code call()} invocation, and as the key for active-context registries. + */ + private final String runId; + /** * Call-scoped {@link AgentState} for the active {@code (userId, sessionId)} slot. Set once at * call entry by the agent and read by middlewares / tools that need the live conversational @@ -59,6 +67,7 @@ public class RuntimeContext { private RuntimeContext(Builder builder) { this.sessionId = builder.sessionId; this.userId = builder.userId; + this.runId = builder.runId; this.stringAttributes = new ConcurrentHashMap<>(); this.typedAttributes = new ConcurrentHashMap<>(); this.toolExecutionContext = builder.toolExecutionContext; @@ -98,6 +107,16 @@ public String getUserId() { return userId; } + /** + * Returns the stable per-call correlation id. Always non-null: auto-generated by the builder + * when not explicitly supplied. + * + * @return the run id (never {@code null}) + */ + public String getRunId() { + return runId; + } + /** * Returns the call-scoped {@link AgentState} for this run, or {@code null} when accessed * outside of an active {@code call()}. Prefer this over {@code agent.getAgentState()} from @@ -325,6 +344,7 @@ public static Builder builder() { public static class Builder { private String sessionId; private String userId; + private String runId; private Map stringExtras; private final Map, Map> typedValues = new HashMap<>(); private ToolExecutionContext toolExecutionContext; @@ -340,6 +360,18 @@ public Builder userId(String userId) { return this; } + /** + * Sets an explicit run id. When not supplied (or blank), {@link #build()} auto-generates a + * unique id so that {@link RuntimeContext#getRunId()} is always non-null. + * + * @param runId the run id, or {@code null} to auto-generate + * @return this builder + */ + public Builder runId(String runId) { + this.runId = runId; + return this; + } + public Builder agentState(AgentState agentState) { this.agentState = agentState; return this; @@ -382,6 +414,7 @@ public Builder from(RuntimeContext source) { } this.sessionId = source.sessionId; this.userId = source.userId; + this.runId = source.runId; this.agentState = source.agentState; this.toolExecutionContext = source.toolExecutionContext; if (!source.stringAttributes.isEmpty()) { @@ -407,6 +440,9 @@ public Builder toolExecutionContext(ToolExecutionContext toolExecutionContext) { } public RuntimeContext build() { + if (runId == null || runId.isBlank()) { + runId = UUID.randomUUID().toString().replace("-", ""); + } return new RuntimeContext(this); } } diff --git a/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentActiveContextRegistryTest.java b/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentActiveContextRegistryTest.java new file mode 100644 index 0000000000..2cabc814bd --- /dev/null +++ b/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentActiveContextRegistryTest.java @@ -0,0 +1,212 @@ +/* + * 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.agent; + +import static org.junit.jupiter.api.Assertions.assertEquals; +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.ReActAgent; +import io.agentscope.core.agent.test.MockModel; +import io.agentscope.core.agent.test.TestConstants; +import io.agentscope.core.agent.test.TestUtils; +import io.agentscope.core.hook.Hook; +import io.agentscope.core.hook.HookEvent; +import io.agentscope.core.hook.PreReasoningEvent; +import io.agentscope.core.message.Msg; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import reactor.core.Disposable; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; + +/** + * Verifies that {@link ReActAgent} tracks active per-call {@link RuntimeContext}s in a + * runId-indexed registry, so concurrent calls on different sessions no longer overwrite each + * other's context. + */ +@DisplayName("ReActAgent active-context registry") +class ReActAgentActiveContextRegistryTest { + + @Test + @DisplayName("getRuntimeContext(runId) returns null after the call completes") + void runtimeContextRemovedAfterCall() throws Exception { + MockModel model = new MockModel("done"); + ReActAgent agent = buildAgent(model, null); + + RuntimeContext rc = RuntimeContext.builder().userId("u1").sessionId("s1").build(); + String runId = rc.getRunId(); + + // Before the call, no active context for this runId. + assertNull(agent.getRuntimeContext(runId)); + + Msg result = agent.call(List.of(TestUtils.createUserMessage("user", "hi")), rc).block(); + + assertNotNull(result); + // After the call completes, the runId slot is removed. + awaitRegistryDrained(agent, 5, TimeUnit.SECONDS); + assertNull(agent.getRuntimeContext(runId)); + } + + @Test + @DisplayName("getActiveRuntimeContexts() is empty when no call is in flight") + void activeContextsEmptyWhenIdle() { + ReActAgent agent = buildAgent(new MockModel("ok"), null); + assertTrue(agent.getActiveRuntimeContexts().isEmpty()); + } + + @Test + @DisplayName("concurrent calls on different sessions do not overwrite each other") + void concurrentCallsIsolated() throws Exception { + CountDownLatch enteredReasoning = new CountDownLatch(2); + CountDownLatch release = new CountDownLatch(1); + + Hook blockingHook = + new Hook() { + @Override + public Mono onEvent(T event) { + if (event instanceof PreReasoningEvent) { + enteredReasoning.countDown(); + return Mono.fromRunnable( + () -> { + try { + assertTrue(release.await(10, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }) + .then(Mono.just(event)); + } + return Mono.just(event); + } + }; + + MockModel model = new MockModel("done"); + ReActAgent agent = buildAgent(model, blockingHook); + + RuntimeContext rcAlice = + RuntimeContext.builder().userId("alice").sessionId("alice-session").build(); + RuntimeContext rcBob = + RuntimeContext.builder().userId("bob").sessionId("bob-session").build(); + String runIdAlice = rcAlice.getRunId(); + String runIdBob = rcBob.getRunId(); + + // Fire both calls on separate threads so they run concurrently (different slots). + Mono callA = + agent.call(List.of(TestUtils.createUserMessage("alice", "hi")), rcAlice) + .subscribeOn(Schedulers.boundedElastic()); + Mono callB = + agent.call(List.of(TestUtils.createUserMessage("bob", "hi")), rcBob) + .subscribeOn(Schedulers.boundedElastic()); + + CountDownLatch doneA = new CountDownLatch(1); + CountDownLatch doneB = new CountDownLatch(1); + AtomicReference resultA = new AtomicReference<>(); + AtomicReference resultB = new AtomicReference<>(); + AtomicReference errorA = new AtomicReference<>(); + AtomicReference errorB = new AtomicReference<>(); + + Disposable subA = + callA.subscribe( + msg -> { + resultA.set(msg); + doneA.countDown(); + }, + e -> { + errorA.set(e); + doneA.countDown(); + }); + Disposable subB = + callB.subscribe( + msg -> { + resultB.set(msg); + doneB.countDown(); + }, + e -> { + errorB.set(e); + doneB.countDown(); + }); + + // Wait until both calls have entered the reasoning phase (RCs registered). + assertTrue(enteredReasoning.await(30, TimeUnit.SECONDS)); + + // Both calls in-flight: registry holds exactly the two active contexts. + assertEquals(2, agent.getActiveRuntimeContexts().size()); + // Each runId resolves to the correct session — no cross-contamination. + assertEquals("alice-session", agent.getRuntimeContext(runIdAlice).getSessionId()); + assertEquals("bob-session", agent.getRuntimeContext(runIdBob).getSessionId()); + + // Release both calls and wait for completion. + release.countDown(); + assertTrue(doneA.await(30, TimeUnit.SECONDS)); + assertTrue(doneB.await(30, TimeUnit.SECONDS)); + if (errorA.get() != null) { + throw new AssertionError("callA failed", errorA.get()); + } + if (errorB.get() != null) { + throw new AssertionError("callB failed", errorB.get()); + } + assertNotNull(resultA.get()); + assertNotNull(resultB.get()); + + // After both complete, registry is empty. The doFinally cleanup fires after the + // terminal signal propagates downstream, so poll briefly for the registry to drain. + awaitRegistryDrained(agent, 5, TimeUnit.SECONDS); + assertNull(agent.getRuntimeContext(runIdAlice)); + assertNull(agent.getRuntimeContext(runIdBob)); + assertTrue(agent.getActiveRuntimeContexts().isEmpty()); + } + + /** + * Polls until the agent's active-context registry becomes empty, or fails the timeout. + * Needed because {@code doFinally} fires after the terminal signal propagates downstream. + */ + private static void awaitRegistryDrained(ReActAgent agent, long timeout, TimeUnit unit) + throws InterruptedException { + long deadlineNanos = System.nanoTime() + unit.toNanos(timeout); + while (!agent.getActiveRuntimeContexts().isEmpty()) { + if (System.nanoTime() > deadlineNanos) { + throw new AssertionError( + "Active-context registry not drained within " + + timeout + + " " + + unit + + "; still contains: " + + agent.getActiveRuntimeContexts().size()); + } + Thread.sleep(10); + } + } + + // ==================== Helpers ==================== + + private static ReActAgent buildAgent(MockModel model, Hook hook) { + ReActAgent.Builder b = + ReActAgent.builder() + .name(TestConstants.TEST_REACT_AGENT_NAME) + .sysPrompt(TestConstants.DEFAULT_SYS_PROMPT) + .model(model); + if (hook != null) { + b.hook(hook); + } + return b.build(); + } +} diff --git a/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentRuntimeContextTest.java b/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentRuntimeContextTest.java index 81c2fad263..3ee5941810 100644 --- a/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentRuntimeContextTest.java +++ b/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentRuntimeContextTest.java @@ -239,8 +239,7 @@ public Mono onEvent(T event) { return Mono.defer( () -> { if (preCount.getAndIncrement() == 0) { - AgentBase a = (AgentBase) ((PreReasoningEvent) event).getAgent(); - RuntimeContext rc = a.getRuntimeContext(); + RuntimeContext rc = fromSetter.get(); assertNotNull(rc); assertEquals("per-call-uid", rc.getUserId()); assertEquals("from-initial-put", rc.get(SharedPojo.class).value); diff --git a/agentscope-core/src/test/java/io/agentscope/core/agent/RuntimeContextTest.java b/agentscope-core/src/test/java/io/agentscope/core/agent/RuntimeContextTest.java index fb23b4b41f..f01dafb583 100644 --- a/agentscope-core/src/test/java/io/agentscope/core/agent/RuntimeContextTest.java +++ b/agentscope-core/src/test/java/io/agentscope/core/agent/RuntimeContextTest.java @@ -17,6 +17,8 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; @@ -65,6 +67,32 @@ void empty_isMutable() { assertEquals("v", ctx.get("k")); } + @Test + @DisplayName("runId is always non-null, even via empty()") + void runIdAlwaysPresent() { + assertNotNull(RuntimeContext.empty().getRunId()); + assertNotNull(RuntimeContext.builder().build().getRunId()); + } + + @Test + @DisplayName("two distinct contexts have distinct runIds") + void runIdUnique() { + assertNotEquals(RuntimeContext.empty().getRunId(), RuntimeContext.empty().getRunId()); + } + + @Test + @DisplayName("builder().runId(x).build() preserves the explicit value") + void runIdExplicitPreserved() { + assertEquals("my-run-123", RuntimeContext.builder().runId("my-run-123").build().getRunId()); + } + + @Test + @DisplayName("builder(source) copies runId from source") + void runIdCopiedFromSource() { + RuntimeContext src = RuntimeContext.builder().runId("src-id").build(); + assertEquals("src-id", RuntimeContext.builder(src).build().getRunId()); + } + @Test @DisplayName("builder sets session fields and string extras") void builderSessionAndStringExtras() { diff --git a/agentscope-examples/agents/agentscope-paw/src/main/java/io/agentscope/claw2/web/toolbus/ToolNotificationMiddleware.java b/agentscope-examples/agents/agentscope-paw/src/main/java/io/agentscope/claw2/web/toolbus/ToolNotificationMiddleware.java index c955b685d7..1ded4b0957 100644 --- a/agentscope-examples/agents/agentscope-paw/src/main/java/io/agentscope/claw2/web/toolbus/ToolNotificationMiddleware.java +++ b/agentscope-examples/agents/agentscope-paw/src/main/java/io/agentscope/claw2/web/toolbus/ToolNotificationMiddleware.java @@ -15,14 +15,12 @@ */ package io.agentscope.claw2.web.toolbus; -import io.agentscope.core.ReActAgent; import io.agentscope.core.agent.Agent; import io.agentscope.core.agent.RuntimeContext; import io.agentscope.core.event.AgentEvent; import io.agentscope.core.message.ToolUseBlock; import io.agentscope.core.middleware.ActingInput; import io.agentscope.core.middleware.MiddlewareBase; -import io.agentscope.harness.agent.HarnessAgent; import java.util.LinkedHashMap; import java.util.Map; import java.util.function.Function; @@ -36,8 +34,7 @@ * {@link io.agentscope.claw2.web.api.ChatController}. * *

The session key is derived from {@link RuntimeContext#getSessionId()} (falling back to - * {@link RuntimeContext#getUserId()}) on the in-flight call, accessed via the agent's - * {@code getRuntimeContext()}. + * {@link RuntimeContext#getUserId()}) on the per-call context passed to {@code onActing}. */ public class ToolNotificationMiddleware implements MiddlewareBase { @@ -55,7 +52,7 @@ public Flux onActing( RuntimeContext ctx, ActingInput input, Function> next) { - String sessionKey = resolveSessionKey(agent); + String sessionKey = resolveSessionKey(ctx); if (sessionKey != null && input.toolCalls() != null) { for (ToolUseBlock tu : input.toolCalls()) { Map inputData = new LinkedHashMap<>(); @@ -80,13 +77,7 @@ public Flux onActing( return next.apply(input); } - private static String resolveSessionKey(Agent agent) { - RuntimeContext ctx = null; - if (agent instanceof HarnessAgent h) { - ctx = h.getRuntimeContext(); - } else if (agent instanceof ReActAgent r) { - ctx = r.getRuntimeContext(); - } + private static String resolveSessionKey(RuntimeContext ctx) { if (ctx == null) return null; if (ctx.getSessionId() != null) { return ctx.getSessionId(); diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java index 38e3e8f201..fac3057435 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java @@ -285,9 +285,15 @@ public Mono runCuratorOnce() { /** * Promote a draft skill from {@code skills/_drafts/} to the live skills root via the * configured {@link SkillPromotionGate}. + * + * @param name the draft skill name + * @param reviewerId the reviewer identity + * @param runId the per-call run id (from {@link RuntimeContext#getRunId()}) used to resolve the + * active session's namespace; pass {@code null} when not inside a call */ - public Mono promoteSkill(String name, String reviewerId) { - return promoteSkill(name, reviewerId, getRuntimeContext()); + public Mono promoteSkill( + String name, String reviewerId, String runId) { + return promoteSkill(name, reviewerId, getRuntimeContext(runId)); } /** @@ -426,8 +432,27 @@ public int getMaxIters() { return delegate.getMaxIters(); } - public RuntimeContext getRuntimeContext() { - return delegate.getRuntimeContext(); + /** + * Returns the active {@link RuntimeContext} for the given runId, or {@code null} if no such + * call is currently in flight on the underlying {@link io.agentscope.core.ReActAgent}. + * + * @param runId the per-call run id (from {@link RuntimeContext#getRunId()}) + * @return the matching in-flight context, or {@code null} + */ + @Override + public RuntimeContext getRuntimeContext(String runId) { + return delegate.getRuntimeContext(runId); + } + + /** + * Returns a read-only snapshot of all currently in-flight {@link RuntimeContext}s on the + * underlying agent instance. + * + * @return unmodifiable list of active contexts (may be empty) + */ + @Override + public List getActiveRuntimeContexts() { + return delegate.getActiveRuntimeContexts(); } public AgentStateStore getStateStore() { @@ -2367,7 +2392,7 @@ public HarnessAgent build() { Supplier currentRcSupplier = () -> { ReActAgent self = selfRef.get(); - RuntimeContext rc = self != null ? self.getRuntimeContext() : null; + RuntimeContext rc = self != null ? self.getRuntimeContext(null) : null; return rc != null ? rc : RuntimeContext.empty(); }; List orderedSkillRepos =