fix(sandbox): avoid session mirrors after sandbox release

agentscope-extensions/agentscope-extensions-sandbox/agentscope-extensions-sandbox-kubernetes/src/main/java/io/agentscope/extensions/sandbox/kubernetes/KubernetesSandbox.java

@@ -88,6 +88,7 @@ public void shutdown() throws Exception {
     @Override
     protected ExecResult doExec(RuntimeContext runtimeContext, String command, int timeoutSeconds)
             throws Exception {
+        requireActiveConnection();
         String wrapped = "cd " + shellQuote(k8sState.getWorkspaceRoot()) + " && (" + command + ")";
         ExecutionResult result =
                 sdkSandbox.commands().run(wrapped, Duration.ofSeconds(Math.max(timeoutSeconds, 1)));
@@ -105,6 +106,7 @@ protected ExecResult doExec(RuntimeContext runtimeContext, String command, int t
 
     @Override
     protected InputStream doPersistWorkspace() throws Exception {
+        requireActiveConnection();
         String root = k8sState.getWorkspaceRoot();
         StringBuilder tarArgs = new StringBuilder();
         // The temp archive may live inside the workspace when the file API is rooted there;
@@ -163,6 +165,7 @@ private InputStream persistViaExec(String root, String tarArgs) throws Exception
 
     @Override
     protected void doHydrateWorkspace(InputStream archive) throws Exception {
+        requireActiveConnection();
         String root = k8sState.getWorkspaceRoot();
         sdkSandbox.commands().run("mkdir -p " + shellQuote(root));
 
@@ -262,6 +265,7 @@ public boolean supportsFileTransfer(String absolutePath) {
 
     @Override
     public void uploadFile(String absolutePath, byte[] content) throws Exception {
+        requireActiveConnection();
         String rel = requireFileApiRelative(absolutePath);
         int slash = absolutePath.lastIndexOf('/');
         if (slash > 0) {
@@ -281,9 +285,18 @@ public void uploadFile(String absolutePath, byte[] content) throws Exception {
 
     @Override
     public byte[] downloadFile(String absolutePath) throws Exception {
+        requireActiveConnection();
         return sdkSandbox.files().read(requireFileApiRelative(absolutePath));
     }
 
+    private void requireActiveConnection() {
+        if (!sdkSandbox.isActive()) {
+            throw new SandboxException.SandboxRuntimeException(
+                    SandboxErrorCode.SANDBOX_CONNECTION_CLOSED,
+                    "Kubernetes sandbox connection has been closed");
+        }
+    }
+
     /**
      * Maps an absolute sandbox path to a file-API-relative path, or null when the file API
      * is disabled, the path lies outside the file API base dir, or it contains traversal

agentscope-extensions/agentscope-extensions-sandbox/agentscope-extensions-sandbox-kubernetes/src/test/java/io/agentscope/extensions/sandbox/kubernetes/KubernetesSandboxTest.java

@@ -17,6 +17,7 @@
 
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
@@ -33,6 +34,8 @@
 import io.agentscope.extensions.sandbox.kubernetes.client.Sandbox;
 import io.agentscope.extensions.sandbox.kubernetes.client.model.ExecutionResult;
 import io.agentscope.harness.agent.sandbox.ExecResult;
+import io.agentscope.harness.agent.sandbox.SandboxErrorCode;
+import io.agentscope.harness.agent.sandbox.SandboxException;
 import java.io.ByteArrayInputStream;
 import java.io.InputStream;
 import java.time.Duration;
@@ -45,16 +48,18 @@ class KubernetesSandboxTest {
 
     private CommandExecutor commands;
     private Filesystem files;
+    private Sandbox sdkSandbox;
     private KubernetesSandboxState state;
     private KubernetesSandbox sandbox;
 
     @BeforeEach
     void setUp() {
         commands = mock(CommandExecutor.class);
         files = mock(Filesystem.class);
-        Sandbox sdkSandbox = mock(Sandbox.class);
+        sdkSandbox = mock(Sandbox.class);
         when(sdkSandbox.commands()).thenReturn(commands);
         when(sdkSandbox.files()).thenReturn(files);
+        when(sdkSandbox.isActive()).thenReturn(true);
 
         state = new KubernetesSandboxState();
         state.setSessionId("session-1");
@@ -194,6 +199,50 @@ void downloadFileReadsRelativePath() throws Exception {
         assertArrayEquals(content, sandbox.downloadFile("/workspace/out/report.pdf"));
     }
 
+    @Test
+    void fileTransferFailsClearlyWhenConnectionIsClosed() {
+        when(sdkSandbox.isActive()).thenReturn(false);
+
+        SandboxException uploadFailure =
+                assertThrows(
+                        SandboxException.class,
+                        () -> sandbox.uploadFile("/workspace/src/Foo.java", new byte[] {1}));
+        SandboxException downloadFailure =
+                assertThrows(
+                        SandboxException.class,
+                        () -> sandbox.downloadFile("/workspace/src/Foo.java"));
+
+        assertEquals("Kubernetes sandbox connection has been closed", uploadFailure.getMessage());
+        assertEquals("Kubernetes sandbox connection has been closed", downloadFailure.getMessage());
+        assertEquals(SandboxErrorCode.SANDBOX_CONNECTION_CLOSED, uploadFailure.getErrorCode());
+        assertEquals(SandboxErrorCode.SANDBOX_CONNECTION_CLOSED, downloadFailure.getErrorCode());
+        verify(commands, never()).run(anyString());
+        verify(files, never()).read(anyString());
+        verify(files, never()).write(anyString(), org.mockito.ArgumentMatchers.<byte[]>any());
+    }
+
+    @Test
+    void operationalEntryPointsFailClearlyWhenConnectionIsClosed() {
+        when(sdkSandbox.isActive()).thenReturn(false);
+
+        SandboxException execFailure =
+                assertThrows(SandboxException.class, () -> sandbox.doExec(null, "echo hi", 30));
+        SandboxException persistFailure =
+                assertThrows(SandboxException.class, sandbox::doPersistWorkspace);
+        SandboxException hydrateFailure =
+                assertThrows(
+                        SandboxException.class,
+                        () -> sandbox.doHydrateWorkspace(new ByteArrayInputStream(new byte[] {1})));
+
+        assertEquals(SandboxErrorCode.SANDBOX_CONNECTION_CLOSED, execFailure.getErrorCode());
+        assertEquals(SandboxErrorCode.SANDBOX_CONNECTION_CLOSED, persistFailure.getErrorCode());
+        assertEquals(SandboxErrorCode.SANDBOX_CONNECTION_CLOSED, hydrateFailure.getErrorCode());
+        verify(commands, never()).run(anyString());
+        verify(commands, never()).run(anyString(), any(Duration.class));
+        verify(files, never()).read(anyString());
+        verify(files, never()).write(anyString(), org.mockito.ArgumentMatchers.<byte[]>any());
+    }
+
     @Test
     void persistUsesExecWhenFileApiDisabled() throws Exception {
         state.setFileApiBaseDir("");

agentscope-harness/src/main/java/io/agentscope/harness/agent/filesystem/sandbox/PinnedSandboxFilesystem.java

@@ -15,8 +15,20 @@
  */
 package io.agentscope.harness.agent.filesystem.sandbox;
 
+import io.agentscope.core.agent.RuntimeContext;
+import io.agentscope.harness.agent.filesystem.model.FileUploadResponse;
 import io.agentscope.harness.agent.sandbox.Sandbox;
+import java.lang.ref.ReferenceQueue;
+import java.lang.ref.WeakReference;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * {@link SandboxBackedFilesystem} that holds a fixed {@link Sandbox} reference for the lifetime of
@@ -28,18 +40,171 @@
  * must not unpin this mirror filesystem.
  *
  * <p>Safe for DataAgent-style <em>user-managed</em> sandboxes that stay alive across
- * acquire/release. Self-managed sandboxes that stop on release may still fail if the async upload
- * races past shutdown.
+ * acquire/release. Self-managed sandbox release takes an exclusive gate before shutdown, so an
+ * upload either completes before release or is skipped without touching released resources.
  */
 public final class PinnedSandboxFilesystem extends SandboxBackedFilesystem {
 
+    private static final Logger log = LoggerFactory.getLogger(PinnedSandboxFilesystem.class);
+    private static final long RELEASE_GATE_TIMEOUT_MILLIS = 1_000;
+    private static final ReferenceQueue<Sandbox> STALE_SANDBOXES = new ReferenceQueue<>();
+    private static final Map<IdentityWeakReference, MirrorGate> MIRROR_GATES = new HashMap<>();
+
+    private final Sandbox pinnedSandbox;
+    private final MirrorGate mirrorGate;
+
     public PinnedSandboxFilesystem(Sandbox sandbox) {
-        Objects.requireNonNull(sandbox, "sandbox");
+        this.pinnedSandbox = Objects.requireNonNull(sandbox, "sandbox");
+        this.mirrorGate = gateFor(sandbox);
         super.setSandbox(sandbox);
     }
 
+    /**
+     * Starts a fresh mirror-gate generation for a newly acquired sandbox.
+     *
+     * <p>A manager may reuse the same {@link Sandbox} object for a later call. Replacing rather
+     * than reopening the old gate keeps mirrors from the previous call permanently released while
+     * ensuring they cannot attach themselves to the new call's lifecycle.
+     */
+    public static void markSandboxAcquired(Sandbox sandbox) {
+        if (sandbox == null) {
+            return;
+        }
+        synchronized (MIRROR_GATES) {
+            expungeStaleGates();
+            MIRROR_GATES.put(new IdentityWeakReference(sandbox, STALE_SANDBOXES), new MirrorGate());
+        }
+    }
+
+    /** Prevents new mirror uploads and briefly waits for an in-flight upload before shutdown. */
+    public static void markSandboxReleased(Sandbox sandbox) {
+        if (sandbox != null) {
+            gateFor(sandbox).markReleased();
+        }
+    }
+
+    /**
+     * Whether the sandbox pinned for an asynchronous mirror is still running.
+     *
+     * <p>A self-managed sandbox is stopped as part of call release. In that case a session
+     * mirror is already best-effort and must not attempt a transfer against released resources.
+     */
+    public boolean isSandboxRunning() {
+        mirrorGate.lock.readLock().lock();
+        try {
+            return !mirrorGate.released && pinnedSandbox.isRunning();
+        } finally {
+            mirrorGate.lock.readLock().unlock();
+        }
+    }
+
+    /** Returns whether this pinned filesystem's call generation has been released. */
+    public boolean isSandboxReleased() {
+        return mirrorGate.released;
+    }
+
+    @Override
+    public List<FileUploadResponse> uploadFiles(
+            RuntimeContext runtimeContext, List<Map.Entry<String, byte[]>> files) {
+        mirrorGate.lock.readLock().lock();
+        try {
+            if (mirrorGate.released || !pinnedSandbox.isRunning()) {
+                List<FileUploadResponse> failed = new ArrayList<>(files.size());
+                for (Map.Entry<String, byte[]> file : files) {
+                    failed.add(FileUploadResponse.fail(file.getKey(), "Sandbox has been released"));
+                }
+                return failed;
+            }
+            return super.uploadFiles(runtimeContext, files);
+        } finally {
+            mirrorGate.lock.readLock().unlock();
+        }
+    }
+
     @Override
     public synchronized void clearSandboxIfCurrent(Sandbox expected) {
         // Keep the pin for out-of-call mirror uploads.
     }
+
+    private static MirrorGate gateFor(Sandbox sandbox) {
+        synchronized (MIRROR_GATES) {
+            expungeStaleGates();
+            IdentityWeakReference lookup = new IdentityWeakReference(sandbox);
+            MirrorGate gate = MIRROR_GATES.get(lookup);
+            if (gate == null) {
+                gate = new MirrorGate();
+                MIRROR_GATES.put(new IdentityWeakReference(sandbox, STALE_SANDBOXES), gate);
+            }
+            return gate;
+        }
+    }
+
+    private static void expungeStaleGates() {
+        IdentityWeakReference stale;
+        while ((stale = (IdentityWeakReference) STALE_SANDBOXES.poll()) != null) {
+            MIRROR_GATES.remove(stale);
+        }
+    }
+
+    private static final class MirrorGate {
+        private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+        private volatile boolean released;
+
+        private void markReleased() {
+            // Publish the latch before waiting so new uploads are rejected even if an existing
+            // remote transfer is wedged. Release remains bounded because mirrors are best-effort.
+            released = true;
+            boolean acquired = false;
+            try {
+                acquired =
+                        lock.writeLock()
+                                .tryLock(RELEASE_GATE_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
+                if (!acquired) {
+                    log.warn(
+                            "Mirror upload still in flight after {} ms; releasing sandbox",
+                            RELEASE_GATE_TIMEOUT_MILLIS);
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                log.warn("Interrupted while waiting for an in-flight sandbox mirror upload");
+            } finally {
+                if (acquired) {
+                    lock.writeLock().unlock();
+                }
+            }
+        }
+    }
+
+    /** Weak key whose equality follows object identity rather than {@code equals/hashCode}. */
+    private static final class IdentityWeakReference extends WeakReference<Sandbox> {
+        private final int identityHash;
+
+        private IdentityWeakReference(Sandbox sandbox) {
+            super(sandbox);
+            this.identityHash = System.identityHashCode(sandbox);
+        }
+
+        private IdentityWeakReference(
+                Sandbox sandbox, ReferenceQueue<? super Sandbox> referenceQueue) {
+            super(sandbox, referenceQueue);
+            this.identityHash = System.identityHashCode(sandbox);
+        }
+
+        @Override
+        public int hashCode() {
+            return identityHash;
+        }
+
+        @Override
+        public boolean equals(Object other) {
+            if (this == other) {
+                return true;
+            }
+            if (!(other instanceof IdentityWeakReference that)) {
+                return false;
+            }
+            Sandbox sandbox = get();
+            return sandbox != null && sandbox == that.get();
+        }
+    }
 }