From 2e6ef4140a826ffdddc3fa1c8486136cf97281d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B1=9F=E5=AE=9D=E5=9D=A4?= Date: Mon, 29 Jun 2026 21:18:57 +0800 Subject: [PATCH 1/5] fix(harness): fix concurrent sandbox isolation in multi-user singleton deployments Two related bugs in HarnessAgent when used as a singleton serving concurrent sessions: Bug 1: SandboxBackedFilesystem used a single volatile Sandbox field and SandboxLifecycleMiddleware used a single AtomicReference. Concurrent sessions overwrote each other's bindings, causing cross-user sandbox contamination. Fixed by replacing both with ConcurrentHashMap, keyed by sessionId so each session gets an isolated slot. SandboxAware interface updated: setSandbox/getSandbox replaced by bindSandbox/unbindSandbox/getSandbox(key). Bug 2: WorkspaceMessageBus and WorkspaceAsyncToolRegistry were constructed with the post-sandbox filesystem (SandboxBackedFilesystem) in HarnessAgent.build(). These are process-scoped infrastructure components; routing them through the sandbox filesystem causes RuntimeContext.empty() calls to fail or contaminate sandbox state. Fixed by capturing busFilesystem before the sandbox replacement and using it exclusively for bus/registry construction. Closes #1896 --- .../harness/agent/HarnessAgent.java | 14 ++- .../sandbox/SandboxBackedFilesystem.java | 38 +++--- .../SandboxLifecycleMiddleware.java | 26 ++-- .../harness/agent/sandbox/SandboxAware.java | 12 +- .../sandbox/SandboxBackedFilesystemTest.java | 115 ++++++++++++++++-- 5 files changed, 163 insertions(+), 42 deletions(-) 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 2a3b8f7b03..278dee1cce 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 @@ -2047,6 +2047,9 @@ public HarnessAgent build() { SandboxLifecycleMiddleware sandboxLifecycleMw = null; SandboxContext defaultSandboxContext = null; SandboxBackedFilesystem capturedSandboxFs = null; + // bus/registry are process-scoped infrastructure; keep them on the pre-sandbox + // filesystem so they are never routed through SandboxBackedFilesystem + AbstractFilesystem busFilesystem = filesystem; if (sandboxFilesystemSpec != null) { capturedSandboxFs = new SandboxBackedFilesystem(); filesystem = capturedSandboxFs; @@ -2098,16 +2101,17 @@ public HarnessAgent build() { // ---- MessageBus / AsyncToolRegistry: workspace defaults ---- // If not set explicitly or via DistributedStore, fall back to workspace-backed - // implementations that use the same AbstractFilesystem as the rest of the agent. - if (messageBus == null && filesystem != null) { + // implementations. Use busFilesystem (pre-sandbox) so these process-scoped components + // are never accidentally routed through SandboxBackedFilesystem. + if (messageBus == null && busFilesystem != null) { messageBus = new io.agentscope.harness.agent.bus.WorkspaceMessageBus( - filesystem, ".agentscope/bus"); + busFilesystem, ".agentscope/bus"); } - if (asyncToolRegistry == null && filesystem != null) { + if (asyncToolRegistry == null && busFilesystem != null) { asyncToolRegistry = new io.agentscope.harness.agent.bus.WorkspaceAsyncToolRegistry( - filesystem, ".agentscope/bus/async-tools"); + busFilesystem, ".agentscope/bus/async-tools"); } // ---- Middlewares ---- diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystem.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystem.java index 36903ae79e..2e7a947c91 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystem.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystem.java @@ -28,35 +28,42 @@ import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * A {@link BaseSandboxFilesystem} that delegates execution to a live {@link Sandbox}. * - *

Stable proxy created at agent build time; a fresh {@link Sandbox} is injected on each call - * via the volatile {@code sandbox} field by {@link - * io.agentscope.harness.agent.middleware.SandboxLifecycleMiddleware}. + *

Stable proxy created at agent build time. A sandbox is bound per session key on each call via + * {@link io.agentscope.harness.agent.middleware.SandboxLifecycleMiddleware}, allowing a single + * proxy instance to serve concurrent sessions without cross-user contamination. */ public class SandboxBackedFilesystem extends BaseSandboxFilesystem implements SandboxAware { private static final Logger log = LoggerFactory.getLogger(SandboxBackedFilesystem.class); private final String fsId; - private volatile Sandbox sandbox; + // per-session sandbox bindings; key is sessionId (or userId for USER isolation scope) + private final ConcurrentHashMap activeSandboxes = new ConcurrentHashMap<>(); public SandboxBackedFilesystem() { this.fsId = "sandbox-" + UUID.randomUUID().toString().substring(0, 8); } @Override - public void setSandbox(Sandbox sandbox) { - this.sandbox = sandbox; + public void bindSandbox(String sessionKey, Sandbox sandbox) { + activeSandboxes.put(sessionKey, sandbox); } @Override - public Sandbox getSandbox() { - return sandbox; + public void unbindSandbox(String sessionKey) { + activeSandboxes.remove(sessionKey); + } + + @Override + public Sandbox getSandbox(String sessionKey) { + return activeSandboxes.get(sessionKey); } @Override @@ -67,7 +74,7 @@ public String id() { @Override public ExecuteResponse execute( RuntimeContext runtimeContext, String command, Integer timeoutSeconds) { - Sandbox active = requireSandbox(); + Sandbox active = requireSandbox(runtimeContext); try { ExecResult result = active.exec(runtimeContext, command, timeoutSeconds); return new ExecuteResponse( @@ -90,7 +97,7 @@ public ExecuteResponse execute( @Override public List uploadFiles( RuntimeContext runtimeContext, List> files) { - Sandbox active = requireSandbox(); + Sandbox active = requireSandbox(runtimeContext); List results = new ArrayList<>(files.size()); for (Map.Entry file : files) { @@ -134,7 +141,7 @@ public List uploadFiles( @Override public List downloadFiles( RuntimeContext runtimeContext, List paths) { - Sandbox active = requireSandbox(); + Sandbox active = requireSandbox(runtimeContext); List results = new ArrayList<>(paths.size()); for (String path : paths) { @@ -168,11 +175,14 @@ public List downloadFiles( return results; } - private Sandbox requireSandbox() { - Sandbox s = sandbox; + private Sandbox requireSandbox(RuntimeContext rc) { + String key = rc != null ? rc.getSessionId() : null; + Sandbox s = key != null ? activeSandboxes.get(key) : null; if (s == null) { throw new SandboxException.SandboxConfigurationException( - "No active sandbox — sandbox filesystem used outside of a call context"); + "No active sandbox for session '" + + key + + "' — sandbox filesystem used outside of a call context"); } return s; } diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java index d754345c79..236d58a0bc 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java @@ -21,7 +21,7 @@ import io.agentscope.harness.agent.sandbox.SandboxAcquireResult; import io.agentscope.harness.agent.sandbox.SandboxContext; import io.agentscope.harness.agent.sandbox.SandboxManager; -import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.ConcurrentHashMap; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -53,8 +53,9 @@ public class SandboxLifecycleMiddleware implements HarnessRuntimeMiddleware { private final SandboxManager sandboxManager; private final SandboxBackedFilesystem filesystemProxy; - private final AtomicReference currentAcquireResult = - new AtomicReference<>(); + // per-session acquire results; keyed by sessionId so concurrent sessions don't interfere + private final ConcurrentHashMap acquireResults = + new ConcurrentHashMap<>(); public SandboxLifecycleMiddleware( SandboxManager sandboxManager, SandboxBackedFilesystem filesystemProxy) { @@ -77,18 +78,20 @@ public void acquireForCall(RuntimeContext ctx) { if (sandboxContext == null) { return; } + String sessionKey = ctx.getSessionId(); try { SandboxAcquireResult result = sandboxManager.acquire(sandboxContext, ctx); Sandbox sandbox = result.getSandbox(); try { sandbox.start(); - filesystemProxy.setSandbox(sandbox); - currentAcquireResult.set(result); + filesystemProxy.bindSandbox(sessionKey, sandbox); + acquireResults.put(sessionKey, result); log.debug( - "[sandbox-mw] Acquired sandbox {}", - sandbox.getState() != null ? sandbox.getState().getSessionId() : "?"); + "[sandbox-mw] Acquired sandbox {} for session {}", + sandbox.getState() != null ? sandbox.getState().getSessionId() : "?", + sessionKey); } catch (Exception e) { - filesystemProxy.setSandbox(null); + filesystemProxy.unbindSandbox(sessionKey); try { sandboxManager.release(result); } catch (Exception releaseErr) { @@ -113,11 +116,12 @@ public void acquireForCall(RuntimeContext ctx) { * @param ctx the per-call RuntimeContext (captured at acquire time) */ public void releaseForCall(RuntimeContext ctx) { - SandboxAcquireResult result = currentAcquireResult.getAndSet(null); + String sessionKey = ctx != null ? ctx.getSessionId() : null; + SandboxAcquireResult result = sessionKey != null ? acquireResults.remove(sessionKey) : null; if (result == null) { return; } - SandboxContext sandboxContext = ctx != null ? ctx.get(SandboxContext.class) : null; + SandboxContext sandboxContext = ctx.get(SandboxContext.class); try { sandboxManager.persistState(result, sandboxContext, ctx); } catch (Exception e) { @@ -129,6 +133,6 @@ public void releaseForCall(RuntimeContext ctx) { log.warn("[sandbox-mw] Failed to release sandbox session: {}", e.getMessage(), e); } result.getLease().close(); - filesystemProxy.setSandbox(null); + filesystemProxy.unbindSandbox(sessionKey); } } diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/sandbox/SandboxAware.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/sandbox/SandboxAware.java index d6904e8512..40532b133a 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/sandbox/SandboxAware.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/sandbox/SandboxAware.java @@ -21,12 +21,16 @@ * Marks a filesystem that can have its backing {@link Sandbox} injected at runtime. * *

Implemented by {@link SandboxBackedFilesystem} so {@link - * io.agentscope.harness.agent.middleware.SandboxLifecycleMiddleware} can set the active sandbox for each - * call and clear it afterward. + * io.agentscope.harness.agent.middleware.SandboxLifecycleMiddleware} can bind and unbind a sandbox + * per session key, allowing a single filesystem proxy to serve concurrent sessions safely. */ public interface SandboxAware { - void setSandbox(Sandbox sandbox); + // Binds a live sandbox to the given session key for the duration of one call. + void bindSandbox(String sessionKey, Sandbox sandbox); - Sandbox getSandbox(); + // Removes the sandbox binding for the given session key after a call completes. + void unbindSandbox(String sessionKey); + + Sandbox getSandbox(String sessionKey); } diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystemTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystemTest.java index 6e2c740ad2..6f2d79a5ab 100644 --- a/agentscope-harness/src/test/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystemTest.java +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystemTest.java @@ -17,30 +17,41 @@ import static org.junit.jupiter.api.Assertions.assertArrayEquals; 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.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import io.agentscope.core.agent.RuntimeContext; import io.agentscope.harness.agent.filesystem.model.FileDownloadResponse; import io.agentscope.harness.agent.sandbox.ExecResult; import io.agentscope.harness.agent.sandbox.Sandbox; +import io.agentscope.harness.agent.sandbox.SandboxException; import io.agentscope.harness.agent.sandbox.SandboxState; import java.io.InputStream; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.Test; class SandboxBackedFilesystemTest { - private static final RuntimeContext RT = RuntimeContext.empty(); + private static RuntimeContext rc(String sessionId) { + return RuntimeContext.builder().sessionId(sessionId).build(); + } @Test void downloadFiles_decodesWrappedBase64Output() { byte[] expected = new byte[] {1, 2, 3, 4, 5, 6}; SandboxBackedFilesystem filesystem = new SandboxBackedFilesystem(); FakeSandbox sandbox = new FakeSandbox(new ExecResult(0, "AQID\nBAUG", "", false)); - filesystem.setSandbox(sandbox); + RuntimeContext ctx = rc("session-a"); + filesystem.bindSandbox(ctx.getSessionId(), sandbox); List responses = - filesystem.downloadFiles(RT, List.of("/tmp/data.bin")); + filesystem.downloadFiles(ctx, List.of("/tmp/data.bin")); assertEquals("base64 '/tmp/data.bin'", sandbox.lastCommand); assertEquals(1, responses.size()); @@ -53,10 +64,11 @@ void downloadFiles_decodesWrappedBase64Output() { void downloadFiles_decodesEmptyPayloadWhenStdoutIsNull() { SandboxBackedFilesystem filesystem = new SandboxBackedFilesystem(); FakeSandbox sandbox = new FakeSandbox(new ExecResult(0, null, "", false)); - filesystem.setSandbox(sandbox); + RuntimeContext ctx = rc("session-a"); + filesystem.bindSandbox(ctx.getSessionId(), sandbox); List responses = - filesystem.downloadFiles(RT, List.of("/tmp/empty.bin")); + filesystem.downloadFiles(ctx, List.of("/tmp/empty.bin")); assertEquals("base64 '/tmp/empty.bin'", sandbox.lastCommand); assertEquals(1, responses.size()); @@ -69,10 +81,11 @@ void downloadFiles_decodesEmptyPayloadWhenStdoutIsNull() { void downloadFiles_returnsFailureWhenCommandFails() { SandboxBackedFilesystem filesystem = new SandboxBackedFilesystem(); FakeSandbox sandbox = new FakeSandbox(new ExecResult(1, "", "boom", false)); - filesystem.setSandbox(sandbox); + RuntimeContext ctx = rc("session-a"); + filesystem.bindSandbox(ctx.getSessionId(), sandbox); List responses = - filesystem.downloadFiles(RT, List.of("/tmp/fail.bin")); + filesystem.downloadFiles(ctx, List.of("/tmp/fail.bin")); assertEquals("base64 '/tmp/fail.bin'", sandbox.lastCommand); assertEquals(1, responses.size()); @@ -81,10 +94,96 @@ void downloadFiles_returnsFailureWhenCommandFails() { assertEquals("[stderr] boom", responses.get(0).error()); } + @Test + void requireSandbox_throwsWhenNoBindingForSession() { + SandboxBackedFilesystem filesystem = new SandboxBackedFilesystem(); + RuntimeContext ctx = rc("unknown-session"); + + assertThrows( + SandboxException.SandboxConfigurationException.class, + () -> filesystem.execute(ctx, "echo hi", null)); + } + + @Test + void bindAndUnbind_isolatesSessionsFromEachOther() { + SandboxBackedFilesystem filesystem = new SandboxBackedFilesystem(); + FakeSandbox sandboxA = new FakeSandbox(new ExecResult(0, "AAAA", "", false)); + FakeSandbox sandboxB = new FakeSandbox(new ExecResult(0, "BBBB", "", false)); + + filesystem.bindSandbox("session-a", sandboxA); + filesystem.bindSandbox("session-b", sandboxB); + + // session-a routes to sandboxA + filesystem.downloadFiles(rc("session-a"), List.of("/f")); + assertNotNull(sandboxA.lastCommand); + assertNull(sandboxB.lastCommand); + + // session-b routes to sandboxB + filesystem.downloadFiles(rc("session-b"), List.of("/g")); + assertNotNull(sandboxB.lastCommand); + + // unbind session-a; session-b still works + filesystem.unbindSandbox("session-a"); + assertThrows( + SandboxException.SandboxConfigurationException.class, + () -> filesystem.execute(rc("session-a"), "echo", null)); + filesystem.downloadFiles(rc("session-b"), List.of("/h")); // must not throw + } + + @Test + void concurrentSessionsDontCrossContaminate() throws InterruptedException { + int sessions = 8; + SandboxBackedFilesystem filesystem = new SandboxBackedFilesystem(); + FakeSandbox[] sandboxes = new FakeSandbox[sessions]; + for (int i = 0; i < sessions; i++) { + sandboxes[i] = new FakeSandbox(new ExecResult(0, "AA==", "", false)); // 1 byte + filesystem.bindSandbox("session-" + i, sandboxes[i]); + } + + CountDownLatch start = new CountDownLatch(1); + CountDownLatch done = new CountDownLatch(sessions); + ExecutorService pool = Executors.newFixedThreadPool(sessions); + String[] errors = new String[sessions]; + + for (int i = 0; i < sessions; i++) { + final int idx = i; + pool.submit( + () -> { + try { + start.await(); + RuntimeContext ctx = rc("session-" + idx); + // each session executes a uniquely named command + filesystem.execute(ctx, "cmd-" + idx, null); + // verify command reached the correct sandbox + String expected = "cmd-" + idx; + if (!expected.equals(sandboxes[idx].lastCommand)) { + errors[idx] = + "session-" + + idx + + " routed to wrong sandbox: " + + sandboxes[idx].lastCommand; + } + } catch (Exception e) { + errors[idx] = e.getMessage(); + } finally { + done.countDown(); + } + }); + } + + start.countDown(); + assertTrue(done.await(10, TimeUnit.SECONDS)); + pool.shutdown(); + + for (int i = 0; i < sessions; i++) { + assertNull(errors[i], "Error in session-" + i + ": " + errors[i]); + } + } + private static final class FakeSandbox implements Sandbox { private final ExecResult execResult; - private String lastCommand; + volatile String lastCommand; private FakeSandbox(ExecResult execResult) { this.execResult = execResult; From 9a2fa5fcb944783c366d090f374ef35ae62357cd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B1=9F=E5=AE=9D=E5=9D=A4?= Date: Wed, 8 Jul 2026 15:42:43 +0800 Subject: [PATCH 2/5] fix(harness): use composite userId/sessionId key for sandbox binding Aligns the in-memory sandbox binding key with ReActAgent.slotKey() to prevent cross-user contamination when different users share the same sessionId in multi-tenant singleton deployments. Addresses review feedback from PR #1964. --- .../sandbox/SandboxBackedFilesystem.java | 13 +++- .../SandboxLifecycleMiddleware.java | 19 +++++- .../sandbox/SandboxBackedFilesystemTest.java | 65 +++++++++++++------ 3 files changed, 73 insertions(+), 24 deletions(-) diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystem.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystem.java index 2e7a947c91..14f5519274 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystem.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystem.java @@ -44,7 +44,7 @@ public class SandboxBackedFilesystem extends BaseSandboxFilesystem implements Sa private static final Logger log = LoggerFactory.getLogger(SandboxBackedFilesystem.class); private final String fsId; - // per-session sandbox bindings; key is sessionId (or userId for USER isolation scope) + // per-call sandbox bindings; key is "userId/sessionId" (aligned with ReActAgent.slotKey) private final ConcurrentHashMap activeSandboxes = new ConcurrentHashMap<>(); public SandboxBackedFilesystem() { @@ -176,7 +176,7 @@ public List downloadFiles( } private Sandbox requireSandbox(RuntimeContext rc) { - String key = rc != null ? rc.getSessionId() : null; + String key = rc != null ? bindingKey(rc) : null; Sandbox s = key != null ? activeSandboxes.get(key) : null; if (s == null) { throw new SandboxException.SandboxConfigurationException( @@ -187,6 +187,15 @@ private Sandbox requireSandbox(RuntimeContext rc) { return s; } + private static String bindingKey(RuntimeContext rc) { + String uid = rc.getUserId(); + String sid = rc.getSessionId(); + if (sid == null || sid.isBlank()) { + return null; + } + return (uid == null || uid.isBlank() ? "__anon__" : uid) + "/" + sid; + } + private String shellSingleQuote(String s) { return "'" + s.replace("'", "'\\''") + "'"; } diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java index 236d58a0bc..639f8ca5ad 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java @@ -53,7 +53,7 @@ public class SandboxLifecycleMiddleware implements HarnessRuntimeMiddleware { private final SandboxManager sandboxManager; private final SandboxBackedFilesystem filesystemProxy; - // per-session acquire results; keyed by sessionId so concurrent sessions don't interfere + // per-call acquire results; keyed by "userId/sessionId" so concurrent users don't interfere private final ConcurrentHashMap acquireResults = new ConcurrentHashMap<>(); @@ -78,7 +78,10 @@ public void acquireForCall(RuntimeContext ctx) { if (sandboxContext == null) { return; } - String sessionKey = ctx.getSessionId(); + String sessionKey = bindingKey(ctx); + if (sessionKey == null) { + return; + } try { SandboxAcquireResult result = sandboxManager.acquire(sandboxContext, ctx); Sandbox sandbox = result.getSandbox(); @@ -116,7 +119,7 @@ public void acquireForCall(RuntimeContext ctx) { * @param ctx the per-call RuntimeContext (captured at acquire time) */ public void releaseForCall(RuntimeContext ctx) { - String sessionKey = ctx != null ? ctx.getSessionId() : null; + String sessionKey = ctx != null ? bindingKey(ctx) : null; SandboxAcquireResult result = sessionKey != null ? acquireResults.remove(sessionKey) : null; if (result == null) { return; @@ -135,4 +138,14 @@ public void releaseForCall(RuntimeContext ctx) { result.getLease().close(); filesystemProxy.unbindSandbox(sessionKey); } + + // (userId ?? "__anon__") + "/" + sessionId — aligned with ReActAgent.slotKey() + private static String bindingKey(RuntimeContext ctx) { + String uid = ctx.getUserId(); + String sid = ctx.getSessionId(); + if (sid == null || sid.isBlank()) { + return null; + } + return (uid == null || uid.isBlank() ? "__anon__" : uid) + "/" + sid; + } } diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystemTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystemTest.java index 6f2d79a5ab..58c77effc0 100644 --- a/agentscope-harness/src/test/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystemTest.java +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystemTest.java @@ -42,13 +42,24 @@ private static RuntimeContext rc(String sessionId) { return RuntimeContext.builder().sessionId(sessionId).build(); } + private static RuntimeContext rc(String userId, String sessionId) { + return RuntimeContext.builder().userId(userId).sessionId(sessionId).build(); + } + + // mirrors the binding key logic in SandboxBackedFilesystem / SandboxLifecycleMiddleware + private static String bindKey(RuntimeContext ctx) { + String uid = ctx.getUserId(); + String sid = ctx.getSessionId(); + return (uid == null || uid.isBlank() ? "__anon__" : uid) + "/" + sid; + } + @Test void downloadFiles_decodesWrappedBase64Output() { byte[] expected = new byte[] {1, 2, 3, 4, 5, 6}; SandboxBackedFilesystem filesystem = new SandboxBackedFilesystem(); FakeSandbox sandbox = new FakeSandbox(new ExecResult(0, "AQID\nBAUG", "", false)); RuntimeContext ctx = rc("session-a"); - filesystem.bindSandbox(ctx.getSessionId(), sandbox); + filesystem.bindSandbox(bindKey(ctx), sandbox); List responses = filesystem.downloadFiles(ctx, List.of("/tmp/data.bin")); @@ -65,7 +76,7 @@ void downloadFiles_decodesEmptyPayloadWhenStdoutIsNull() { SandboxBackedFilesystem filesystem = new SandboxBackedFilesystem(); FakeSandbox sandbox = new FakeSandbox(new ExecResult(0, null, "", false)); RuntimeContext ctx = rc("session-a"); - filesystem.bindSandbox(ctx.getSessionId(), sandbox); + filesystem.bindSandbox(bindKey(ctx), sandbox); List responses = filesystem.downloadFiles(ctx, List.of("/tmp/empty.bin")); @@ -82,7 +93,7 @@ void downloadFiles_returnsFailureWhenCommandFails() { SandboxBackedFilesystem filesystem = new SandboxBackedFilesystem(); FakeSandbox sandbox = new FakeSandbox(new ExecResult(1, "", "boom", false)); RuntimeContext ctx = rc("session-a"); - filesystem.bindSandbox(ctx.getSessionId(), sandbox); + filesystem.bindSandbox(bindKey(ctx), sandbox); List responses = filesystem.downloadFiles(ctx, List.of("/tmp/fail.bin")); @@ -110,24 +121,23 @@ void bindAndUnbind_isolatesSessionsFromEachOther() { FakeSandbox sandboxA = new FakeSandbox(new ExecResult(0, "AAAA", "", false)); FakeSandbox sandboxB = new FakeSandbox(new ExecResult(0, "BBBB", "", false)); - filesystem.bindSandbox("session-a", sandboxA); - filesystem.bindSandbox("session-b", sandboxB); + RuntimeContext ctxA = rc("session-a"); + RuntimeContext ctxB = rc("session-b"); + filesystem.bindSandbox(bindKey(ctxA), sandboxA); + filesystem.bindSandbox(bindKey(ctxB), sandboxB); - // session-a routes to sandboxA - filesystem.downloadFiles(rc("session-a"), List.of("/f")); + filesystem.downloadFiles(ctxA, List.of("/f")); assertNotNull(sandboxA.lastCommand); assertNull(sandboxB.lastCommand); - // session-b routes to sandboxB - filesystem.downloadFiles(rc("session-b"), List.of("/g")); + filesystem.downloadFiles(ctxB, List.of("/g")); assertNotNull(sandboxB.lastCommand); - // unbind session-a; session-b still works - filesystem.unbindSandbox("session-a"); + filesystem.unbindSandbox(bindKey(ctxA)); assertThrows( SandboxException.SandboxConfigurationException.class, - () -> filesystem.execute(rc("session-a"), "echo", null)); - filesystem.downloadFiles(rc("session-b"), List.of("/h")); // must not throw + () -> filesystem.execute(ctxA, "echo", null)); + filesystem.downloadFiles(ctxB, List.of("/h")); } @Test @@ -135,9 +145,11 @@ void concurrentSessionsDontCrossContaminate() throws InterruptedException { int sessions = 8; SandboxBackedFilesystem filesystem = new SandboxBackedFilesystem(); FakeSandbox[] sandboxes = new FakeSandbox[sessions]; + RuntimeContext[] contexts = new RuntimeContext[sessions]; for (int i = 0; i < sessions; i++) { - sandboxes[i] = new FakeSandbox(new ExecResult(0, "AA==", "", false)); // 1 byte - filesystem.bindSandbox("session-" + i, sandboxes[i]); + sandboxes[i] = new FakeSandbox(new ExecResult(0, "AA==", "", false)); + contexts[i] = rc("session-" + i); + filesystem.bindSandbox(bindKey(contexts[i]), sandboxes[i]); } CountDownLatch start = new CountDownLatch(1); @@ -151,10 +163,7 @@ void concurrentSessionsDontCrossContaminate() throws InterruptedException { () -> { try { start.await(); - RuntimeContext ctx = rc("session-" + idx); - // each session executes a uniquely named command - filesystem.execute(ctx, "cmd-" + idx, null); - // verify command reached the correct sandbox + filesystem.execute(contexts[idx], "cmd-" + idx, null); String expected = "cmd-" + idx; if (!expected.equals(sandboxes[idx].lastCommand)) { errors[idx] = @@ -180,6 +189,24 @@ void concurrentSessionsDontCrossContaminate() throws InterruptedException { } } + @Test + void sameSessionIdDifferentUsersAreIsolated() { + SandboxBackedFilesystem filesystem = new SandboxBackedFilesystem(); + FakeSandbox sandboxAlice = new FakeSandbox(new ExecResult(0, "AA==", "", false)); + FakeSandbox sandboxBob = new FakeSandbox(new ExecResult(0, "AA==", "", false)); + + RuntimeContext ctxAlice = rc("alice", "default"); + RuntimeContext ctxBob = rc("bob", "default"); + filesystem.bindSandbox(bindKey(ctxAlice), sandboxAlice); + filesystem.bindSandbox(bindKey(ctxBob), sandboxBob); + + filesystem.execute(ctxAlice, "alice-cmd", null); + filesystem.execute(ctxBob, "bob-cmd", null); + + assertEquals("alice-cmd", sandboxAlice.lastCommand); + assertEquals("bob-cmd", sandboxBob.lastCommand); + } + private static final class FakeSandbox implements Sandbox { private final ExecResult execResult; From dca488d7e7236a981c4cf9730bf47d570e717e43 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B1=9F=E5=AE=9D=E5=9D=A4?= Date: Wed, 8 Jul 2026 21:27:36 +0800 Subject: [PATCH 3/5] fix(harness): serialize same-session sandbox acquire to prevent race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit HarnessAgent.wrappedCall acquired the sandbox in Mono.using's resource supplier (subscription-time eager), which runs before ReActAgent's per-session serialization gate. Two concurrent same-session requests both ran acquireForCall(), the second overwrote the first's binding key in the ConcurrentHashMap — leaking the first sandbox and failing the second call with "No active sandbox". Fix: introduce a per-session async gate in SandboxLifecycleMiddleware (mirroring AgentBase.serializeOnKey) via serializedCall/serializedFlux. Same-session acquires now queue non-blocking; the second waits until the first's release completes before proceeding. Different sessions still run concurrently. HarnessAgent.wrappedCall/wrappedStream/wrappedStreamEvents delegate to the new methods instead of managing Mono.using/Flux.using directly. --- .../harness/agent/HarnessAgent.java | 48 +---- .../SandboxLifecycleMiddleware.java | 129 ++++++++++-- .../SandboxLifecycleMiddlewareTest.java | 196 ++++++++++++++++++ 3 files changed, 320 insertions(+), 53 deletions(-) create mode 100644 agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddlewareTest.java 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 278dee1cce..03e2c0c633 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 @@ -792,19 +792,9 @@ public Toolkit getToolkit() { private Mono wrappedCall( List msgs, RuntimeContext effective, Supplier> inner) { Mono base = - Mono.using( - () -> { - if (sandboxLifecycleMw != null) { - sandboxLifecycleMw.acquireForCall(effective); - } - return effective; - }, - eff -> inner.get(), - eff -> { - if (sandboxLifecycleMw != null) { - sandboxLifecycleMw.releaseForCall(eff); - } - }); + sandboxLifecycleMw != null + ? sandboxLifecycleMw.serializedCall(effective, inner) + : inner.get(); if (compactionHook != null) { return base.onErrorResume( e -> { @@ -823,36 +813,16 @@ private Mono wrappedCall( */ @Deprecated(since = "2.0.0", forRemoval = true) private Flux wrappedStream(RuntimeContext effective, Supplier> inner) { - return Flux.using( - () -> { - if (sandboxLifecycleMw != null) { - sandboxLifecycleMw.acquireForCall(effective); - } - return effective; - }, - eff -> inner.get(), - eff -> { - if (sandboxLifecycleMw != null) { - sandboxLifecycleMw.releaseForCall(eff); - } - }); + return sandboxLifecycleMw != null + ? sandboxLifecycleMw.serializedFlux(effective, inner) + : inner.get(); } private Flux wrappedStreamEvents( RuntimeContext effective, Supplier> inner) { - return Flux.using( - () -> { - if (sandboxLifecycleMw != null) { - sandboxLifecycleMw.acquireForCall(effective); - } - return effective; - }, - eff -> inner.get(), - eff -> { - if (sandboxLifecycleMw != null) { - sandboxLifecycleMw.releaseForCall(eff); - } - }); + return sandboxLifecycleMw != null + ? sandboxLifecycleMw.serializedFlux(effective, inner) + : inner.get(); } /** diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java index 639f8ca5ad..41db9ace4e 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java @@ -16,14 +16,19 @@ package io.agentscope.harness.agent.middleware; import io.agentscope.core.agent.RuntimeContext; +import io.agentscope.core.message.Msg; import io.agentscope.harness.agent.filesystem.sandbox.SandboxBackedFilesystem; import io.agentscope.harness.agent.sandbox.Sandbox; import io.agentscope.harness.agent.sandbox.SandboxAcquireResult; import io.agentscope.harness.agent.sandbox.SandboxContext; import io.agentscope.harness.agent.sandbox.SandboxManager; import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Supplier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Sinks; /** * Middleware that manages the sandbox session lifecycle around each agent call. @@ -53,9 +58,11 @@ public class SandboxLifecycleMiddleware implements HarnessRuntimeMiddleware { private final SandboxManager sandboxManager; private final SandboxBackedFilesystem filesystemProxy; - // per-call acquire results; keyed by "userId/sessionId" so concurrent users don't interfere private final ConcurrentHashMap acquireResults = new ConcurrentHashMap<>(); + // Per-session async serialization gate — ensures same-session acquire/release pairs + // do not overlap even when HarnessAgent.wrappedCall is subscribed concurrently. + private final ConcurrentHashMap> sandboxGates = new ConcurrentHashMap<>(); public SandboxLifecycleMiddleware( SandboxManager sandboxManager, SandboxBackedFilesystem filesystemProxy) { @@ -63,12 +70,97 @@ public SandboxLifecycleMiddleware( this.filesystemProxy = filesystemProxy; } + // ==================== Serialized call wrappers ==================== + + /** + * Wraps a {@code Mono} call with serialized sandbox acquire/release. + * + *

Same-session calls are queued so that the second subscription waits (non-blocking) + * until the first completes and releases its sandbox, preventing concurrent overwrites + * of the filesystem binding map. + */ + public Mono serializedCall(RuntimeContext ctx, Supplier> inner) { + SandboxContext sandboxContext = ctx != null ? ctx.get(SandboxContext.class) : null; + String sessionKey = ctx != null ? bindingKey(ctx) : null; + if (sessionKey == null || sandboxContext == null) { + return inner.get(); + } + return Mono.defer( + () -> { + Sinks.Empty release = Sinks.empty(); + Mono releaseMono = release.asMono(); + @SuppressWarnings("unchecked") + Mono[] prev = new Mono[1]; + sandboxGates.compute( + sessionKey, + (k, tail) -> { + prev[0] = tail == null ? Mono.empty() : tail; + return releaseMono; + }); + return prev[0].onErrorComplete() + .then( + Mono.using( + () -> { + doAcquire(ctx, sandboxContext, sessionKey); + return ctx; + }, + c -> inner.get(), + this::doRelease)) + .doFinally( + sig -> { + release.tryEmitEmpty(); + sandboxGates.remove(sessionKey, releaseMono); + }); + }); + } + /** - * Acquires the sandbox for the current call. Called from - * {@code ReActAgent.beforeAgentExecution()} to ensure the sandbox is available - * for both the {@code call()} and {@code streamEvents()} paths. + * Wraps a {@code Flux} stream with serialized sandbox acquire/release. * - * @param ctx the per-call RuntimeContext (must not be null) + * @see #serializedCall(RuntimeContext, Supplier) + */ + public Flux serializedFlux(RuntimeContext ctx, Supplier> inner) { + SandboxContext sandboxContext = ctx != null ? ctx.get(SandboxContext.class) : null; + String sessionKey = ctx != null ? bindingKey(ctx) : null; + if (sessionKey == null || sandboxContext == null) { + return inner.get(); + } + return Flux.defer( + () -> { + Sinks.Empty release = Sinks.empty(); + Mono releaseMono = release.asMono(); + @SuppressWarnings("unchecked") + Mono[] prev = new Mono[1]; + sandboxGates.compute( + sessionKey, + (k, tail) -> { + prev[0] = tail == null ? Mono.empty() : tail; + return releaseMono; + }); + return prev[0].onErrorComplete() + .thenMany( + Flux.using( + () -> { + doAcquire(ctx, sandboxContext, sessionKey); + return ctx; + }, + c -> inner.get(), + this::doRelease)) + .doFinally( + sig -> { + release.tryEmitEmpty(); + sandboxGates.remove(sessionKey, releaseMono); + }); + }); + } + + // ==================== Direct acquire/release (for tests) ==================== + + /** + * Acquires the sandbox for the current call. + * + *

Warning: this method does NOT serialize concurrent same-session calls. + * Production code should use {@link #serializedCall} or {@link #serializedFlux} instead. */ public void acquireForCall(RuntimeContext ctx) { if (ctx == null) { @@ -82,6 +174,22 @@ public void acquireForCall(RuntimeContext ctx) { if (sessionKey == null) { return; } + doAcquire(ctx, sandboxContext, sessionKey); + } + + /** + * Releases the sandbox after the current call. + * + *

Warning: this method does NOT serialize concurrent same-session calls. + * Production code should use {@link #serializedCall} or {@link #serializedFlux} instead. + */ + public void releaseForCall(RuntimeContext ctx) { + doRelease(ctx); + } + + // ==================== Internal ==================== + + private void doAcquire(RuntimeContext ctx, SandboxContext sandboxContext, String sessionKey) { try { SandboxAcquireResult result = sandboxManager.acquire(sandboxContext, ctx); Sandbox sandbox = result.getSandbox(); @@ -112,13 +220,7 @@ public void acquireForCall(RuntimeContext ctx) { } } - /** - * Releases the sandbox after the current call. Called from - * {@code ReActAgent.afterAgentExecution()} to ensure cleanup for both paths. - * - * @param ctx the per-call RuntimeContext (captured at acquire time) - */ - public void releaseForCall(RuntimeContext ctx) { + private void doRelease(RuntimeContext ctx) { String sessionKey = ctx != null ? bindingKey(ctx) : null; SandboxAcquireResult result = sessionKey != null ? acquireResults.remove(sessionKey) : null; if (result == null) { @@ -139,8 +241,7 @@ public void releaseForCall(RuntimeContext ctx) { filesystemProxy.unbindSandbox(sessionKey); } - // (userId ?? "__anon__") + "/" + sessionId — aligned with ReActAgent.slotKey() - private static String bindingKey(RuntimeContext ctx) { + static String bindingKey(RuntimeContext ctx) { String uid = ctx.getUserId(); String sid = ctx.getSessionId(); if (sid == null || sid.isBlank()) { diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddlewareTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddlewareTest.java new file mode 100644 index 0000000000..3c2848431a --- /dev/null +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddlewareTest.java @@ -0,0 +1,196 @@ +/* + * 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.harness.agent.middleware; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import io.agentscope.core.agent.RuntimeContext; +import io.agentscope.core.message.AssistantMessage; +import io.agentscope.core.message.Msg; +import io.agentscope.harness.agent.filesystem.sandbox.SandboxBackedFilesystem; +import io.agentscope.harness.agent.sandbox.ExecResult; +import io.agentscope.harness.agent.sandbox.Sandbox; +import io.agentscope.harness.agent.sandbox.SandboxAcquireResult; +import io.agentscope.harness.agent.sandbox.SandboxContext; +import io.agentscope.harness.agent.sandbox.SandboxManager; +import io.agentscope.harness.agent.sandbox.SandboxState; +import java.io.InputStream; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; + +class SandboxLifecycleMiddlewareTest { + + private static RuntimeContext ctxWithSandbox(String sessionId) { + return RuntimeContext.builder() + .sessionId(sessionId) + .put(SandboxContext.class, SandboxContext.builder().build()) + .build(); + } + + private static SandboxManager mockManager(AtomicInteger acquireCount) throws Exception { + SandboxManager manager = mock(SandboxManager.class); + when(manager.acquire(any(), any())) + .thenAnswer( + inv -> { + acquireCount.incrementAndGet(); + return SandboxAcquireResult.selfManaged(new NoopSandbox()); + }); + doNothing().when(manager).persistState(any(), any(), any()); + doNothing().when(manager).release(any()); + return manager; + } + + @Test + void serializedCall_sameSession_secondWaitsForFirstToRelease() throws Exception { + AtomicInteger acquireCount = new AtomicInteger(); + CountDownLatch firstAcquired = new CountDownLatch(1); + CountDownLatch allowFirstToComplete = new CountDownLatch(1); + AtomicBoolean secondAcquiredBeforeFirstReleased = new AtomicBoolean(false); + + SandboxManager manager = mockManager(acquireCount); + SandboxBackedFilesystem fs = new SandboxBackedFilesystem(); + SandboxLifecycleMiddleware mw = new SandboxLifecycleMiddleware(manager, fs); + + RuntimeContext ctx1 = ctxWithSandbox("shared-session"); + RuntimeContext ctx2 = ctxWithSandbox("shared-session"); + + Mono call1 = + mw.serializedCall( + ctx1, + () -> + Mono.fromCallable( + () -> { + firstAcquired.countDown(); + allowFirstToComplete.await(5, TimeUnit.SECONDS); + return new AssistantMessage("r1"); + })); + + Mono call2 = + mw.serializedCall( + ctx2, + () -> + Mono.fromCallable( + () -> { + if (acquireCount.get() < 2) { + secondAcquiredBeforeFirstReleased.set(true); + } + return new AssistantMessage("r2"); + })); + + CountDownLatch done = new CountDownLatch(2); + call1.subscribeOn(Schedulers.boundedElastic()).doFinally(s -> done.countDown()).subscribe(); + call2.subscribeOn(Schedulers.boundedElastic()).doFinally(s -> done.countDown()).subscribe(); + + assertTrue(firstAcquired.await(3, TimeUnit.SECONDS)); + assertEquals(1, acquireCount.get()); + + allowFirstToComplete.countDown(); + assertTrue(done.await(5, TimeUnit.SECONDS)); + + assertEquals(2, acquireCount.get()); + assertFalse(secondAcquiredBeforeFirstReleased.get()); + } + + @Test + void serializedCall_differentSessions_runConcurrently() throws Exception { + AtomicInteger acquireCount = new AtomicInteger(); + CountDownLatch bothAcquired = new CountDownLatch(2); + + SandboxManager manager = mockManager(acquireCount); + SandboxBackedFilesystem fs = new SandboxBackedFilesystem(); + SandboxLifecycleMiddleware mw = new SandboxLifecycleMiddleware(manager, fs); + + RuntimeContext ctxA = ctxWithSandbox("session-a"); + RuntimeContext ctxB = ctxWithSandbox("session-b"); + + CountDownLatch done = new CountDownLatch(2); + + mw.serializedCall( + ctxA, + () -> + Mono.fromCallable( + () -> { + bothAcquired.countDown(); + bothAcquired.await(3, TimeUnit.SECONDS); + return new AssistantMessage("a"); + })) + .subscribeOn(Schedulers.boundedElastic()) + .doFinally(s -> done.countDown()) + .subscribe(); + + mw.serializedCall( + ctxB, + () -> + Mono.fromCallable( + () -> { + bothAcquired.countDown(); + bothAcquired.await(3, TimeUnit.SECONDS); + return new AssistantMessage("b"); + })) + .subscribeOn(Schedulers.boundedElastic()) + .doFinally(s -> done.countDown()) + .subscribe(); + + assertTrue(bothAcquired.await(3, TimeUnit.SECONDS)); + assertTrue(done.await(5, TimeUnit.SECONDS)); + assertEquals(2, acquireCount.get()); + } + + private static class NoopSandbox implements Sandbox { + @Override + public void start() {} + + @Override + public void stop() {} + + @Override + public void close() {} + + @Override + public boolean isRunning() { + return true; + } + + @Override + public SandboxState getState() { + return null; + } + + @Override + public ExecResult exec(RuntimeContext rc, String cmd, Integer timeout) { + return new ExecResult(0, "", "", false); + } + + @Override + public InputStream persistWorkspace() { + return InputStream.nullInputStream(); + } + + @Override + public void hydrateWorkspace(InputStream archive) {} + } +} From 3f8f8aeab26dce726d554014693b2869fffada14 Mon Sep 17 00:00:00 2001 From: Buktal Date: Thu, 9 Jul 2026 14:09:53 +0800 Subject: [PATCH 4/5] refactor(sandbox): remove dead currentAcquireResult field and add Javadoc to internal methods --- .../SandboxLifecycleMiddleware.java | 35 +++++++++++-------- 1 file changed, 21 insertions(+), 14 deletions(-) diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java index d228346642..208a5a20a3 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java @@ -23,9 +23,8 @@ import io.agentscope.harness.agent.sandbox.SandboxContext; import io.agentscope.harness.agent.sandbox.SandboxManager; import java.util.concurrent.ConcurrentHashMap; -import java.util.function.Supplier; -import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; +import java.util.function.Supplier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import reactor.core.publisher.Flux; @@ -60,13 +59,9 @@ public class SandboxLifecycleMiddleware implements HarnessRuntimeMiddleware { private final SandboxManager sandboxManager; private final SandboxBackedFilesystem filesystemProxy; - private final AtomicReference currentAcquireResult = - new AtomicReference<>(); private volatile Consumer beforeStartCallback; private final ConcurrentHashMap acquireResults = new ConcurrentHashMap<>(); - // Per-session async serialization gate — ensures same-session acquire/release pairs - // do not overlap even when HarnessAgent.wrappedCall is subscribed concurrently. private final ConcurrentHashMap> sandboxGates = new ConcurrentHashMap<>(); public SandboxLifecycleMiddleware( @@ -87,9 +82,6 @@ public void setBeforeStartCallback(Consumer callback) { this.beforeStartCallback = callback; } - - // ==================== Serialized call wrappers ==================== - /** * Wraps a {@code Mono} call with serialized sandbox acquire/release. * @@ -172,9 +164,6 @@ public Flux serializedFlux(RuntimeContext ctx, Supplier> inner) { }); } - // ==================== Direct acquire/release (for tests) ==================== - - /** * Acquires the sandbox for the current call. * @@ -206,8 +195,14 @@ public void releaseForCall(RuntimeContext ctx) { doRelease(ctx); } - // ==================== Internal ==================== - + /** + * Acquires a sandbox session: runs the optional {@link #beforeStartCallback}, + * obtains a lease via {@link SandboxManager#acquire}, starts the sandbox, and + * binds it to the filesystem proxy under {@code sessionKey}. + * + *

On failure after acquire, the sandbox is released and the lease closed to + * prevent resource leaks. + */ private void doAcquire(RuntimeContext ctx, SandboxContext sandboxContext, String sessionKey) { try { Consumer cb = beforeStartCallback; @@ -251,6 +246,14 @@ private void doAcquire(RuntimeContext ctx, SandboxContext sandboxContext, String } } + /** + * Releases a previously acquired sandbox session: persists state, releases the + * sandbox via {@link SandboxManager#release}, closes the lease, and unbinds the + * filesystem proxy. + * + *

Failures during persist or release are logged but do not propagate, ensuring + * the agent call result is always delivered to the caller. + */ private void doRelease(RuntimeContext ctx) { String sessionKey = ctx != null ? bindingKey(ctx) : null; SandboxAcquireResult result = sessionKey != null ? acquireResults.remove(sessionKey) : null; @@ -272,6 +275,10 @@ private void doRelease(RuntimeContext ctx) { filesystemProxy.unbindSandbox(sessionKey); } + /** + * Computes the composite binding key for sandbox isolation: {@code "userId/sessionId"}. + * Returns {@code null} if sessionId is absent, signalling that no sandbox should be bound. + */ static String bindingKey(RuntimeContext ctx) { String uid = ctx.getUserId(); String sid = ctx.getSessionId(); From 175edadf2cf89a8b69be089ddc37ffbc4fe99515 Mon Sep 17 00:00:00 2001 From: Buktal Date: Fri, 10 Jul 2026 15:41:26 +0800 Subject: [PATCH 5/5] refactor(harness): drop same-session serialization gate and share bindingKey - Remove serializedCall/serializedFlux + sandboxGates (same-session concurrency serialization gate). Same-session concurrency is guarded at the business entry layer, not re-queued in the sandbox layer. - Restore acquireForCall/releaseForCall with Mono.using/Flux.using (aligned with main); internal acquire/release now operate per sessionKey. - Extract shared SandboxBindingKey to remove duplicated bindingKey impl in SandboxBackedFilesystem and SandboxLifecycleMiddleware. --- .../harness/agent/HarnessAgent.java | 48 +++++-- .../sandbox/SandboxBackedFilesystem.java | 14 +- .../SandboxLifecycleMiddleware.java | 121 ++---------------- .../agent/sandbox/SandboxBindingKey.java | 52 ++++++++ .../SandboxLifecycleMiddlewareTest.java | 104 ++++++--------- .../agent/sandbox/SandboxBindingKeyTest.java | 48 +++++++ 6 files changed, 194 insertions(+), 193 deletions(-) create mode 100644 agentscope-harness/src/main/java/io/agentscope/harness/agent/sandbox/SandboxBindingKey.java create mode 100644 agentscope-harness/src/test/java/io/agentscope/harness/agent/sandbox/SandboxBindingKeyTest.java 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 8f7999e2b8..89c79cf767 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 @@ -791,9 +791,19 @@ public Toolkit getToolkit() { private Mono wrappedCall( List msgs, RuntimeContext effective, Supplier> inner) { Mono base = - sandboxLifecycleMw != null - ? sandboxLifecycleMw.serializedCall(effective, inner) - : inner.get(); + Mono.using( + () -> { + if (sandboxLifecycleMw != null) { + sandboxLifecycleMw.acquireForCall(effective); + } + return effective; + }, + eff -> inner.get(), + eff -> { + if (sandboxLifecycleMw != null) { + sandboxLifecycleMw.releaseForCall(eff); + } + }); if (compactionHook != null) { return base.onErrorResume( e -> { @@ -812,16 +822,36 @@ private Mono wrappedCall( */ @Deprecated(since = "2.0.0", forRemoval = true) private Flux wrappedStream(RuntimeContext effective, Supplier> inner) { - return sandboxLifecycleMw != null - ? sandboxLifecycleMw.serializedFlux(effective, inner) - : inner.get(); + return Flux.using( + () -> { + if (sandboxLifecycleMw != null) { + sandboxLifecycleMw.acquireForCall(effective); + } + return effective; + }, + eff -> inner.get(), + eff -> { + if (sandboxLifecycleMw != null) { + sandboxLifecycleMw.releaseForCall(eff); + } + }); } private Flux wrappedStreamEvents( RuntimeContext effective, Supplier> inner) { - return sandboxLifecycleMw != null - ? sandboxLifecycleMw.serializedFlux(effective, inner) - : inner.get(); + return Flux.using( + () -> { + if (sandboxLifecycleMw != null) { + sandboxLifecycleMw.acquireForCall(effective); + } + return effective; + }, + eff -> inner.get(), + eff -> { + if (sandboxLifecycleMw != null) { + sandboxLifecycleMw.releaseForCall(eff); + } + }); } /** diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystem.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystem.java index 14f5519274..8affbbcbb4 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystem.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/filesystem/sandbox/SandboxBackedFilesystem.java @@ -22,6 +22,7 @@ import io.agentscope.harness.agent.sandbox.ExecResult; import io.agentscope.harness.agent.sandbox.Sandbox; import io.agentscope.harness.agent.sandbox.SandboxAware; +import io.agentscope.harness.agent.sandbox.SandboxBindingKey; import io.agentscope.harness.agent.sandbox.SandboxException; import java.util.ArrayList; import java.util.Base64; @@ -176,8 +177,8 @@ public List downloadFiles( } private Sandbox requireSandbox(RuntimeContext rc) { - String key = rc != null ? bindingKey(rc) : null; - Sandbox s = key != null ? activeSandboxes.get(key) : null; + String key = rc != null ? SandboxBindingKey.resolve(rc) : null; + Sandbox s = key != null ? getSandbox(key) : null; if (s == null) { throw new SandboxException.SandboxConfigurationException( "No active sandbox for session '" @@ -187,15 +188,6 @@ private Sandbox requireSandbox(RuntimeContext rc) { return s; } - private static String bindingKey(RuntimeContext rc) { - String uid = rc.getUserId(); - String sid = rc.getSessionId(); - if (sid == null || sid.isBlank()) { - return null; - } - return (uid == null || uid.isBlank() ? "__anon__" : uid) + "/" + sid; - } - private String shellSingleQuote(String s) { return "'" + s.replace("'", "'\\''") + "'"; } diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java index 208a5a20a3..56c718cb4d 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddleware.java @@ -16,20 +16,16 @@ package io.agentscope.harness.agent.middleware; import io.agentscope.core.agent.RuntimeContext; -import io.agentscope.core.message.Msg; import io.agentscope.harness.agent.filesystem.sandbox.SandboxBackedFilesystem; import io.agentscope.harness.agent.sandbox.Sandbox; import io.agentscope.harness.agent.sandbox.SandboxAcquireResult; +import io.agentscope.harness.agent.sandbox.SandboxBindingKey; import io.agentscope.harness.agent.sandbox.SandboxContext; import io.agentscope.harness.agent.sandbox.SandboxManager; import java.util.concurrent.ConcurrentHashMap; import java.util.function.Consumer; -import java.util.function.Supplier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.core.publisher.Sinks; /** * Middleware that manages the sandbox session lifecycle around each agent call. @@ -62,7 +58,6 @@ public class SandboxLifecycleMiddleware implements HarnessRuntimeMiddleware { private volatile Consumer beforeStartCallback; private final ConcurrentHashMap acquireResults = new ConcurrentHashMap<>(); - private final ConcurrentHashMap> sandboxGates = new ConcurrentHashMap<>(); public SandboxLifecycleMiddleware( SandboxManager sandboxManager, SandboxBackedFilesystem filesystemProxy) { @@ -83,92 +78,13 @@ public void setBeforeStartCallback(Consumer callback) { } /** - * Wraps a {@code Mono} call with serialized sandbox acquire/release. + * Acquires the sandbox for the current call, binding it under the call's session key. * - *

Same-session calls are queued so that the second subscription waits (non-blocking) - * until the first completes and releases its sandbox, preventing concurrent overwrites - * of the filesystem binding map. - */ - public Mono serializedCall(RuntimeContext ctx, Supplier> inner) { - SandboxContext sandboxContext = ctx != null ? ctx.get(SandboxContext.class) : null; - String sessionKey = ctx != null ? bindingKey(ctx) : null; - if (sessionKey == null || sandboxContext == null) { - return inner.get(); - } - return Mono.defer( - () -> { - Sinks.Empty release = Sinks.empty(); - Mono releaseMono = release.asMono(); - @SuppressWarnings("unchecked") - Mono[] prev = new Mono[1]; - sandboxGates.compute( - sessionKey, - (k, tail) -> { - prev[0] = tail == null ? Mono.empty() : tail; - return releaseMono; - }); - return prev[0].onErrorComplete() - .then( - Mono.using( - () -> { - doAcquire(ctx, sandboxContext, sessionKey); - return ctx; - }, - c -> inner.get(), - this::doRelease)) - .doFinally( - sig -> { - release.tryEmitEmpty(); - sandboxGates.remove(sessionKey, releaseMono); - }); - }); - } - - /** - * Wraps a {@code Flux} stream with serialized sandbox acquire/release. - * - * @see #serializedCall(RuntimeContext, Supplier) - */ - public Flux serializedFlux(RuntimeContext ctx, Supplier> inner) { - SandboxContext sandboxContext = ctx != null ? ctx.get(SandboxContext.class) : null; - String sessionKey = ctx != null ? bindingKey(ctx) : null; - if (sessionKey == null || sandboxContext == null) { - return inner.get(); - } - return Flux.defer( - () -> { - Sinks.Empty release = Sinks.empty(); - Mono releaseMono = release.asMono(); - @SuppressWarnings("unchecked") - Mono[] prev = new Mono[1]; - sandboxGates.compute( - sessionKey, - (k, tail) -> { - prev[0] = tail == null ? Mono.empty() : tail; - return releaseMono; - }); - return prev[0].onErrorComplete() - .thenMany( - Flux.using( - () -> { - doAcquire(ctx, sandboxContext, sessionKey); - return ctx; - }, - c -> inner.get(), - this::doRelease)) - .doFinally( - sig -> { - release.tryEmitEmpty(); - sandboxGates.remove(sessionKey, releaseMono); - }); - }); - } - - /** - * Acquires the sandbox for the current call. - * - *

Warning: this method does NOT serialize concurrent same-session calls. - * Production code should use {@link #serializedCall} or {@link #serializedFlux} instead. + *

Intended to be wrapped by the caller in a {@code Mono.using}/{@code Flux.using} so that + * {@link #releaseForCall} is always invoked on termination (complete/error/cancel). This + * method does not serialize same-session concurrency; callers must ensure per-session + * non-concurrency at the entry point, or configure a distributed {@code SandboxExecutionGuard} + * whose lease serializes by {@code SandboxIsolationKey}. */ public void acquireForCall(RuntimeContext ctx) { if (ctx == null) { @@ -178,7 +94,7 @@ public void acquireForCall(RuntimeContext ctx) { if (sandboxContext == null) { return; } - String sessionKey = bindingKey(ctx); + String sessionKey = SandboxBindingKey.resolve(ctx); if (sessionKey == null) { return; } @@ -186,10 +102,8 @@ public void acquireForCall(RuntimeContext ctx) { } /** - * Releases the sandbox after the current call. - * - *

Warning: this method does NOT serialize concurrent same-session calls. - * Production code should use {@link #serializedCall} or {@link #serializedFlux} instead. + * Releases the sandbox acquired by {@link #acquireForCall}. Intended as the cleanup step of + * a {@code Mono.using}/{@code Flux.using} wrapper; failures are logged but do not propagate. */ public void releaseForCall(RuntimeContext ctx) { doRelease(ctx); @@ -255,7 +169,7 @@ private void doAcquire(RuntimeContext ctx, SandboxContext sandboxContext, String * the agent call result is always delivered to the caller. */ private void doRelease(RuntimeContext ctx) { - String sessionKey = ctx != null ? bindingKey(ctx) : null; + String sessionKey = ctx != null ? SandboxBindingKey.resolve(ctx) : null; SandboxAcquireResult result = sessionKey != null ? acquireResults.remove(sessionKey) : null; if (result == null) { return; @@ -274,17 +188,4 @@ private void doRelease(RuntimeContext ctx) { result.getLease().close(); filesystemProxy.unbindSandbox(sessionKey); } - - /** - * Computes the composite binding key for sandbox isolation: {@code "userId/sessionId"}. - * Returns {@code null} if sessionId is absent, signalling that no sandbox should be bound. - */ - static String bindingKey(RuntimeContext ctx) { - String uid = ctx.getUserId(); - String sid = ctx.getSessionId(); - if (sid == null || sid.isBlank()) { - return null; - } - return (uid == null || uid.isBlank() ? "__anon__" : uid) + "/" + sid; - } } diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/sandbox/SandboxBindingKey.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/sandbox/SandboxBindingKey.java new file mode 100644 index 0000000000..b523b8b776 --- /dev/null +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/sandbox/SandboxBindingKey.java @@ -0,0 +1,52 @@ +/* + * 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.harness.agent.sandbox; + +import io.agentscope.core.agent.RuntimeContext; + +/** + * Resolves the per-call sandbox binding key ({@code "userId/sessionId"}) used to slot + * live {@link Sandbox} instances in a shared filesystem proxy. + * + *

Distinct from {@link SandboxIsolationKey}, which governs the coarser acquire / lease / + * persist isolation granularity (scope-based). This key is the per-session handle under which + * {@code SandboxLifecycleMiddleware} binds a sandbox for the duration of one call. + */ +public final class SandboxBindingKey { + + private static final String ANONYMOUS_USER = "__anon__"; + + private SandboxBindingKey() {} + + /** + * Resolves {@code "userId/sessionId"} from the given context. + * + * @param ctx runtime context; may be {@code null} + * @return the binding key, or {@code null} when {@code sessionId} is absent (signalling + * that no sandbox should be bound for this call) + */ + public static String resolve(RuntimeContext ctx) { + if (ctx == null) { + return null; + } + String sid = ctx.getSessionId(); + if (sid == null || sid.isBlank()) { + return null; + } + String uid = ctx.getUserId(); + return (uid == null || uid.isBlank() ? ANONYMOUS_USER : uid) + "/" + sid; + } +} diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddlewareTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddlewareTest.java index 3c2848431a..2da5d53d0c 100644 --- a/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddlewareTest.java +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddlewareTest.java @@ -16,27 +16,28 @@ package io.agentscope.harness.agent.middleware; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; +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 static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doNothing; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import io.agentscope.core.agent.RuntimeContext; import io.agentscope.core.message.AssistantMessage; -import io.agentscope.core.message.Msg; import io.agentscope.harness.agent.filesystem.sandbox.SandboxBackedFilesystem; import io.agentscope.harness.agent.sandbox.ExecResult; import io.agentscope.harness.agent.sandbox.Sandbox; import io.agentscope.harness.agent.sandbox.SandboxAcquireResult; +import io.agentscope.harness.agent.sandbox.SandboxBindingKey; import io.agentscope.harness.agent.sandbox.SandboxContext; import io.agentscope.harness.agent.sandbox.SandboxManager; import io.agentscope.harness.agent.sandbox.SandboxState; import java.io.InputStream; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; import reactor.core.publisher.Mono; @@ -65,58 +66,7 @@ private static SandboxManager mockManager(AtomicInteger acquireCount) throws Exc } @Test - void serializedCall_sameSession_secondWaitsForFirstToRelease() throws Exception { - AtomicInteger acquireCount = new AtomicInteger(); - CountDownLatch firstAcquired = new CountDownLatch(1); - CountDownLatch allowFirstToComplete = new CountDownLatch(1); - AtomicBoolean secondAcquiredBeforeFirstReleased = new AtomicBoolean(false); - - SandboxManager manager = mockManager(acquireCount); - SandboxBackedFilesystem fs = new SandboxBackedFilesystem(); - SandboxLifecycleMiddleware mw = new SandboxLifecycleMiddleware(manager, fs); - - RuntimeContext ctx1 = ctxWithSandbox("shared-session"); - RuntimeContext ctx2 = ctxWithSandbox("shared-session"); - - Mono call1 = - mw.serializedCall( - ctx1, - () -> - Mono.fromCallable( - () -> { - firstAcquired.countDown(); - allowFirstToComplete.await(5, TimeUnit.SECONDS); - return new AssistantMessage("r1"); - })); - - Mono call2 = - mw.serializedCall( - ctx2, - () -> - Mono.fromCallable( - () -> { - if (acquireCount.get() < 2) { - secondAcquiredBeforeFirstReleased.set(true); - } - return new AssistantMessage("r2"); - })); - - CountDownLatch done = new CountDownLatch(2); - call1.subscribeOn(Schedulers.boundedElastic()).doFinally(s -> done.countDown()).subscribe(); - call2.subscribeOn(Schedulers.boundedElastic()).doFinally(s -> done.countDown()).subscribe(); - - assertTrue(firstAcquired.await(3, TimeUnit.SECONDS)); - assertEquals(1, acquireCount.get()); - - allowFirstToComplete.countDown(); - assertTrue(done.await(5, TimeUnit.SECONDS)); - - assertEquals(2, acquireCount.get()); - assertFalse(secondAcquiredBeforeFirstReleased.get()); - } - - @Test - void serializedCall_differentSessions_runConcurrently() throws Exception { + void acquireForCall_differentSessions_runConcurrently() throws Exception { AtomicInteger acquireCount = new AtomicInteger(); CountDownLatch bothAcquired = new CountDownLatch(2); @@ -129,28 +79,36 @@ void serializedCall_differentSessions_runConcurrently() throws Exception { CountDownLatch done = new CountDownLatch(2); - mw.serializedCall( - ctxA, - () -> + Mono.using( + () -> { + mw.acquireForCall(ctxA); + return ctxA; + }, + ctx -> Mono.fromCallable( () -> { bothAcquired.countDown(); bothAcquired.await(3, TimeUnit.SECONDS); return new AssistantMessage("a"); - })) + }), + ctx -> mw.releaseForCall(ctx)) .subscribeOn(Schedulers.boundedElastic()) .doFinally(s -> done.countDown()) .subscribe(); - mw.serializedCall( - ctxB, - () -> + Mono.using( + () -> { + mw.acquireForCall(ctxB); + return ctxB; + }, + ctx -> Mono.fromCallable( () -> { bothAcquired.countDown(); bothAcquired.await(3, TimeUnit.SECONDS); return new AssistantMessage("b"); - })) + }), + ctx -> mw.releaseForCall(ctx)) .subscribeOn(Schedulers.boundedElastic()) .doFinally(s -> done.countDown()) .subscribe(); @@ -160,6 +118,26 @@ void serializedCall_differentSessions_runConcurrently() throws Exception { assertEquals(2, acquireCount.get()); } + @Test + void acquireForCall_releaseForCall_bindsAndUnbindsSandbox() throws Exception { + AtomicInteger acquireCount = new AtomicInteger(); + SandboxManager manager = mockManager(acquireCount); + SandboxBackedFilesystem fs = new SandboxBackedFilesystem(); + SandboxLifecycleMiddleware mw = new SandboxLifecycleMiddleware(manager, fs); + + RuntimeContext ctx = ctxWithSandbox("solo-session"); + String expectedKey = SandboxBindingKey.resolve(ctx); + + mw.acquireForCall(ctx); + assertEquals(1, acquireCount.get()); + assertNotNull(fs.getSandbox(expectedKey)); + + mw.releaseForCall(ctx); + verify(manager).release(any()); + // sandbox is unbound after release + assertNull(fs.getSandbox(expectedKey)); + } + private static class NoopSandbox implements Sandbox { @Override public void start() {} diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/sandbox/SandboxBindingKeyTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/sandbox/SandboxBindingKeyTest.java new file mode 100644 index 0000000000..7da211440d --- /dev/null +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/sandbox/SandboxBindingKeyTest.java @@ -0,0 +1,48 @@ +/* + * 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.harness.agent.sandbox; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +import io.agentscope.core.agent.RuntimeContext; +import org.junit.jupiter.api.Test; + +class SandboxBindingKeyTest { + + @Test + void resolve_nullContext_returnsNull() { + assertNull(SandboxBindingKey.resolve(null)); + } + + @Test + void resolve_sessionIdAbsent_returnsNull() { + RuntimeContext ctx = RuntimeContext.builder().userId("u1").build(); + assertNull(SandboxBindingKey.resolve(ctx)); + } + + @Test + void resolve_userIdAbsent_usesAnonymous() { + RuntimeContext ctx = RuntimeContext.builder().sessionId("s1").build(); + assertEquals("__anon__/s1", SandboxBindingKey.resolve(ctx)); + } + + @Test + void resolve_bothPresent_compositeKey() { + RuntimeContext ctx = RuntimeContext.builder().userId("u1").sessionId("s1").build(); + assertEquals("u1/s1", SandboxBindingKey.resolve(ctx)); + } +}