diff --git a/agentscope-extensions/agentscope-extensions-sandbox/agentscope-extensions-sandbox-kubernetes/src/main/java/io/agentscope/extensions/sandbox/kubernetes/KubernetesSandbox.java b/agentscope-extensions/agentscope-extensions-sandbox/agentscope-extensions-sandbox-kubernetes/src/main/java/io/agentscope/extensions/sandbox/kubernetes/KubernetesSandbox.java index 52c6b129c3..afa4108bbf 100644 --- a/agentscope-extensions/agentscope-extensions-sandbox/agentscope-extensions-sandbox-kubernetes/src/main/java/io/agentscope/extensions/sandbox/kubernetes/KubernetesSandbox.java +++ b/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 diff --git a/agentscope-extensions/agentscope-extensions-sandbox/agentscope-extensions-sandbox-kubernetes/src/test/java/io/agentscope/extensions/sandbox/kubernetes/KubernetesSandboxTest.java b/agentscope-extensions/agentscope-extensions-sandbox/agentscope-extensions-sandbox-kubernetes/src/test/java/io/agentscope/extensions/sandbox/kubernetes/KubernetesSandboxTest.java index 2141fb1aaf..f5d77043f8 100644 --- a/agentscope-extensions/agentscope-extensions-sandbox/agentscope-extensions-sandbox-kubernetes/src/test/java/io/agentscope/extensions/sandbox/kubernetes/KubernetesSandboxTest.java +++ b/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,6 +48,7 @@ class KubernetesSandboxTest { private CommandExecutor commands; private Filesystem files; + private Sandbox sdkSandbox; private KubernetesSandboxState state; private KubernetesSandbox sandbox; @@ -52,9 +56,10 @@ class KubernetesSandboxTest { 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.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.any()); + } + @Test void persistUsesExecWhenFileApiDisabled() throws Exception { state.setFileApiBaseDir(""); diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/filesystem/sandbox/PinnedSandboxFilesystem.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/filesystem/sandbox/PinnedSandboxFilesystem.java index d153755d6f..5f7f24062e 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/filesystem/sandbox/PinnedSandboxFilesystem.java +++ b/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. * *

Safe for DataAgent-style user-managed 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 STALE_SANDBOXES = new ReferenceQueue<>(); + private static final Map 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. + * + *

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. + * + *

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 uploadFiles( + RuntimeContext runtimeContext, List> files) { + mirrorGate.lock.readLock().lock(); + try { + if (mirrorGate.released || !pinnedSandbox.isRunning()) { + List failed = new ArrayList<>(files.size()); + for (Map.Entry 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 { + private final int identityHash; + + private IdentityWeakReference(Sandbox sandbox) { + super(sandbox); + this.identityHash = System.identityHashCode(sandbox); + } + + private IdentityWeakReference( + Sandbox sandbox, ReferenceQueue 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(); + } + } } diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/memory/session/SessionTree.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/memory/session/SessionTree.java index 2e15276a21..55b0504bc9 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/memory/session/SessionTree.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/memory/session/SessionTree.java @@ -18,6 +18,7 @@ import io.agentscope.core.agent.RuntimeContext; import io.agentscope.core.util.JsonUtils; import io.agentscope.harness.agent.filesystem.AbstractFilesystem; +import io.agentscope.harness.agent.filesystem.model.FileUploadResponse; import io.agentscope.harness.agent.filesystem.model.ReadResult; import io.agentscope.harness.agent.filesystem.sandbox.PinnedSandboxFilesystem; import io.agentscope.harness.agent.sandbox.Sandbox; @@ -562,9 +563,19 @@ private void scheduleSegmentMirror(List entries, long seqStart, lo String wid = writerId; MIRROR_EXECUTOR.execute( () -> { + if (skipMirrorForReleasedSandbox(mirrorFs, ref.prefix())) { + return; + } try { store.appendSegment(ref, seqStart, seqEnd, wid, payload); } catch (Exception e) { + if (isReleasedSandbox(mirrorFs)) { + log.debug( + "Skipping best-effort transcript segment mirror for {} because" + + " its sandbox has been released", + ref.prefix()); + return; + } log.warn( "Failed to append transcript segment for {}: {}", ref.prefix(), @@ -728,20 +739,62 @@ private void mirrorToFilesystem(AbstractFilesystem fs, Path file, String relativ if (relativePath == null || relativePath.isBlank()) { return; } + // The index describes the authoritative local file, not the best-effort remote mirror. + // Keep it current even when the sandbox has already been released or upload fails. + if (index != null) { + index.upsertFromLocalFile(relativePath, file); + } + if (skipMirrorForReleasedSandbox(fs, relativePath)) { + return; + } try { byte[] bytes = Files.readAllBytes(file); - fs.uploadFiles(fsRc, List.of(Map.entry(relativePath, bytes))); - // Best-effort: the local file already exists — update index with its current stats - if (index != null) { - index.upsertFromLocalFile(relativePath, file); + List uploads = + fs.uploadFiles(fsRc, List.of(Map.entry(relativePath, bytes))); + if (uploads.size() != 1 || !uploads.get(0).isSuccess()) { + String error = + uploads.size() == 1 ? uploads.get(0).error() : "missing upload response"; + if (isReleasedSandbox(fs)) { + log.debug( + "Skipping best-effort session mirror for {} because its sandbox has" + + " been released", + relativePath); + return; + } + log.warn("Failed to mirror session file {} to filesystem: {}", file, error); + return; } } catch (IOException e) { log.warn("Failed to mirror session file {} to filesystem: {}", file, e.getMessage()); } catch (RuntimeException e) { + if (isReleasedSandbox(fs)) { + log.debug( + "Skipping best-effort session mirror for {} because its sandbox has been" + + " released", + relativePath); + return; + } log.warn("Failed to mirror session file {} to filesystem: {}", file, e.getMessage()); } } + /** Returns true when an asynchronous mirror has outlived its self-managed sandbox. */ + private static boolean skipMirrorForReleasedSandbox( + AbstractFilesystem fs, String mirrorTarget) { + if (fs instanceof PinnedSandboxFilesystem pinned && !pinned.isSandboxRunning()) { + log.debug( + "Skipping best-effort session mirror for {} because its sandbox has been" + + " released", + mirrorTarget); + return true; + } + return false; + } + + private static boolean isReleasedSandbox(AbstractFilesystem fs) { + return fs instanceof PinnedSandboxFilesystem pinned && pinned.isSandboxReleased(); + } + /** * Restores {@code file} from the remote filesystem mirror when the local file is absent. * Used by {@link #syncFromLog()} to ensure the log file is available locally before reading. 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 202d5dbeae..9bcd8840b7 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,6 +16,7 @@ package io.agentscope.harness.agent.middleware; import io.agentscope.core.agent.RuntimeContext; +import io.agentscope.harness.agent.filesystem.sandbox.PinnedSandboxFilesystem; import io.agentscope.harness.agent.filesystem.sandbox.SandboxBackedFilesystem; import io.agentscope.harness.agent.sandbox.Sandbox; import io.agentscope.harness.agent.sandbox.SandboxAcquireResult; @@ -109,6 +110,7 @@ public void acquireForCall(RuntimeContext ctx) { } SandboxAcquireResult result = sandboxManager.acquire(sandboxContext, ctx); Sandbox sandbox = result.getSandbox(); + PinnedSandboxFilesystem.markSandboxAcquired(sandbox); try { sandbox.start(); // Bind the acquired sandbox per-call on this invocation's RuntimeContext rather @@ -126,6 +128,7 @@ public void acquireForCall(RuntimeContext ctx) { ctx.put(SandboxAcquireResult.class, null); filesystemProxy.clearSandboxIfCurrent(sandbox); try { + markMirrorSandboxReleased(result); sandboxManager.release(result); } catch (Exception releaseErr) { log.warn( @@ -169,10 +172,17 @@ public void releaseForCall(RuntimeContext ctx) { log.warn("[sandbox-mw] Failed to persist sandbox state: {}", e.getMessage(), e); } try { + markMirrorSandboxReleased(result); sandboxManager.release(result); } catch (Exception e) { log.warn("[sandbox-mw] Failed to release sandbox session: {}", e.getMessage(), e); } result.getLease().close(); } + + private static void markMirrorSandboxReleased(SandboxAcquireResult result) { + if (result.isSelfManaged()) { + PinnedSandboxFilesystem.markSandboxReleased(result.getSandbox()); + } + } } diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/sandbox/SandboxErrorCode.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/sandbox/SandboxErrorCode.java index d6d527082e..6ff02f6b9e 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/sandbox/SandboxErrorCode.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/sandbox/SandboxErrorCode.java @@ -32,6 +32,9 @@ public enum SandboxErrorCode { /** Failed to stop/persist the workspace store. */ WORKSPACE_STOP_ERROR, + /** An operation was attempted after the sandbox connection was closed. */ + SANDBOX_CONNECTION_CLOSED, + /** Failed to read or parse a workspace archive (tar). */ WORKSPACE_ARCHIVE_READ_ERROR, diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/transcript/ObjectStoreTranscriptStore.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/transcript/ObjectStoreTranscriptStore.java index b0ba56b6c9..a6795d9ce1 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/transcript/ObjectStoreTranscriptStore.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/transcript/ObjectStoreTranscriptStore.java @@ -18,6 +18,7 @@ import io.agentscope.core.agent.RuntimeContext; import io.agentscope.harness.agent.filesystem.AbstractFilesystem; import io.agentscope.harness.agent.filesystem.model.FileInfo; +import io.agentscope.harness.agent.filesystem.model.FileUploadResponse; import io.agentscope.harness.agent.filesystem.model.GlobResult; import io.agentscope.harness.agent.filesystem.model.ReadResult; import java.io.ByteArrayInputStream; @@ -83,7 +84,12 @@ public String appendSegment( TranscriptRef ref, long seqStart, long seqEnd, String writerId, byte[] jsonl) { String name = seqStart + "-" + seqEnd + "-" + sanitize(writerId) + ".jsonl"; String key = rootPrefix + ref.prefix() + "/events/" + name; - filesystem.uploadFiles(rc, List.of(Map.entry(key, jsonl))); + List uploads = + filesystem.uploadFiles(rc, List.of(Map.entry(key, jsonl))); + if (uploads.size() != 1 || !uploads.get(0).isSuccess()) { + String error = uploads.size() == 1 ? uploads.get(0).error() : "missing upload response"; + throw new IllegalStateException("segment upload failed for " + key + ": " + error); + } return key; } diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/transcript/TranscriptStore.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/transcript/TranscriptStore.java index f0e8facf3e..61392a7a78 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/transcript/TranscriptStore.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/transcript/TranscriptStore.java @@ -44,6 +44,8 @@ public interface TranscriptStore { * @param jsonl UTF-8 JSONL payload (one {@link * io.agentscope.harness.agent.memory.session.SessionEntry} per line) * @return storage key of the written segment + * @throws RuntimeException when the segment cannot be written completely; callers must not + * assume a returned key unless the write succeeded */ String appendSegment( TranscriptRef ref, long seqStart, long seqEnd, String writerId, byte[] jsonl); diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/filesystem/sandbox/PinnedSandboxFilesystemTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/filesystem/sandbox/PinnedSandboxFilesystemTest.java new file mode 100644 index 0000000000..d72fdbc436 --- /dev/null +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/filesystem/sandbox/PinnedSandboxFilesystemTest.java @@ -0,0 +1,179 @@ +/* + * 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.filesystem.sandbox; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.agentscope.core.agent.RuntimeContext; +import io.agentscope.harness.agent.sandbox.ExecResult; +import io.agentscope.harness.agent.sandbox.Sandbox; +import io.agentscope.harness.agent.sandbox.SandboxState; +import java.io.InputStream; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; + +class PinnedSandboxFilesystemTest { + + @Test + void releaseWaitsForActiveUploadThenRejectsLaterUploads() throws Exception { + BlockingSandbox sandbox = new BlockingSandbox(); + PinnedSandboxFilesystem.markSandboxAcquired(sandbox); + PinnedSandboxFilesystem filesystem = new PinnedSandboxFilesystem(sandbox); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future upload = + executor.submit( + () -> + filesystem.uploadFiles( + RuntimeContext.empty(), + List.of(Map.entry("session.jsonl", new byte[] {1})))); + assertTrue(sandbox.uploadStarted.await(2, TimeUnit.SECONDS)); + + Future release = + executor.submit(() -> PinnedSandboxFilesystem.markSandboxReleased(sandbox)); + TimeUnit.MILLISECONDS.sleep(100); + assertFalse(release.isDone(), "release should briefly coordinate with active upload"); + + sandbox.allowUploadToFinish.countDown(); + upload.get(2, TimeUnit.SECONDS); + release.get(2, TimeUnit.SECONDS); + + var rejected = + filesystem.uploadFiles( + RuntimeContext.empty(), + List.of(Map.entry("later.jsonl", new byte[] {2}))); + assertFalse(rejected.get(0).isSuccess()); + } finally { + sandbox.allowUploadToFinish.countDown(); + executor.shutdownNow(); + } + } + + @Test + void releaseDoesNotWaitIndefinitelyForStalledUpload() throws Exception { + BlockingSandbox sandbox = new BlockingSandbox(); + PinnedSandboxFilesystem.markSandboxAcquired(sandbox); + PinnedSandboxFilesystem filesystem = new PinnedSandboxFilesystem(sandbox); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future upload = + executor.submit( + () -> + filesystem.uploadFiles( + RuntimeContext.empty(), + List.of(Map.entry("session.jsonl", new byte[] {1})))); + assertTrue(sandbox.uploadStarted.await(2, TimeUnit.SECONDS)); + + long started = System.nanoTime(); + PinnedSandboxFilesystem.markSandboxReleased(sandbox); + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started); + + assertTrue(elapsedMillis < 2_500, "release gate wait must be bounded"); + assertFalse(upload.isDone(), "the simulated remote upload should still be stalled"); + assertFalse(filesystem.isSandboxRunning()); + } finally { + sandbox.allowUploadToFinish.countDown(); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(2, TimeUnit.SECONDS)); + } + } + + @Test + void reacquireCreatesNewGateGenerationForSameSandboxObject() { + BlockingSandbox sandbox = new BlockingSandbox(); + PinnedSandboxFilesystem oldFilesystem = new PinnedSandboxFilesystem(sandbox); + PinnedSandboxFilesystem.markSandboxReleased(sandbox); + + PinnedSandboxFilesystem.markSandboxAcquired(sandbox); + PinnedSandboxFilesystem newFilesystem = new PinnedSandboxFilesystem(sandbox); + + assertFalse(oldFilesystem.isSandboxRunning()); + assertTrue(newFilesystem.isSandboxRunning()); + } + + @Test + void distinctEqualSandboxObjectsUseIndependentGates() { + EqualSandbox first = new EqualSandbox(); + EqualSandbox second = new EqualSandbox(); + PinnedSandboxFilesystem firstFilesystem = new PinnedSandboxFilesystem(first); + PinnedSandboxFilesystem secondFilesystem = new PinnedSandboxFilesystem(second); + + PinnedSandboxFilesystem.markSandboxReleased(first); + + assertFalse(firstFilesystem.isSandboxRunning()); + assertTrue(secondFilesystem.isSandboxRunning()); + } + + private static class BlockingSandbox implements Sandbox { + private final CountDownLatch uploadStarted = new CountDownLatch(1); + private final CountDownLatch allowUploadToFinish = new CountDownLatch(1); + + @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 runtimeContext, String command, Integer timeoutSeconds) { + throw new UnsupportedOperationException(); + } + + @Override + public InputStream persistWorkspace() { + throw new UnsupportedOperationException(); + } + + @Override + public void hydrateWorkspace(InputStream archive) throws InterruptedException { + uploadStarted.countDown(); + allowUploadToFinish.await(); + } + } + + private static final class EqualSandbox extends BlockingSandbox { + @Override + public boolean equals(Object other) { + return other instanceof EqualSandbox; + } + + @Override + public int hashCode() { + return 1; + } + } +} diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/memory/session/SessionTreeMirrorTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/memory/session/SessionTreeMirrorTest.java index 8c90ed3953..2f90cd7262 100644 --- a/agentscope-harness/src/test/java/io/agentscope/harness/agent/memory/session/SessionTreeMirrorTest.java +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/memory/session/SessionTreeMirrorTest.java @@ -22,10 +22,18 @@ import io.agentscope.harness.agent.filesystem.AbstractFilesystem; import io.agentscope.harness.agent.filesystem.BakedContextFilesystem; import io.agentscope.harness.agent.filesystem.remote.store.InMemoryStore; +import io.agentscope.harness.agent.filesystem.sandbox.PinnedSandboxFilesystem; +import io.agentscope.harness.agent.filesystem.sandbox.SandboxBackedFilesystem; import io.agentscope.harness.agent.filesystem.spec.RemoteFilesystemSpec; +import io.agentscope.harness.agent.sandbox.ExecResult; +import io.agentscope.harness.agent.sandbox.Sandbox; +import io.agentscope.harness.agent.sandbox.SandboxFileTransfer; +import io.agentscope.harness.agent.sandbox.SandboxState; import io.agentscope.harness.agent.transcript.ObjectStoreTranscriptStore; import io.agentscope.harness.agent.transcript.TranscriptRef; import io.agentscope.harness.agent.transcript.TranscriptStore; +import io.agentscope.harness.agent.workspace.WorkspaceIndex; +import java.io.InputStream; import java.nio.file.Files; import java.nio.file.Path; import java.util.List; @@ -203,6 +211,62 @@ void flush_localWriteCompletesImmediately() throws Exception { "local log file must exist immediately after flush()"); } + @Test + void flush_skipsMirrorWhenPinnedSandboxWasReleased() throws Exception { + StoppedTransferSandbox sandbox = new StoppedTransferSandbox(); + PinnedSandboxFilesystem.markSandboxReleased(sandbox); + SandboxBackedFilesystem filesystem = new SandboxBackedFilesystem(); + filesystem.setSandbox(sandbox); + Path context = workspace.resolve("agents/agent-a/sessions/released.jsonl"); + + SessionTree tree = new SessionTree(context, workspace, filesystem); + tree.append(new SessionEntry.MessageEntry(null, null, null, "USER", "hello", null)); + tree.flush(); + + assertTrue(SessionTree.awaitMirrorQuiescence(5, TimeUnit.SECONDS)); + assertEquals(0, sandbox.uploadAttempts); + } + + @Test + void flush_skipsTranscriptSegmentWhenPinnedSandboxWasReleased() throws Exception { + StoppedTransferSandbox sandbox = new StoppedTransferSandbox(); + PinnedSandboxFilesystem.markSandboxReleased(sandbox); + SandboxBackedFilesystem filesystem = new SandboxBackedFilesystem(); + filesystem.setSandbox(sandbox); + Path context = workspace.resolve("agents/agent-a/sessions/released-segment.jsonl"); + + SessionTree tree = new SessionTree(context, workspace, filesystem); + tree.setTranscriptStore( + new ObjectStoreTranscriptStore(filesystem), + new TranscriptRef("tenant", "agent-a", "released-segment")); + tree.append(new SessionEntry.MessageEntry(null, null, null, "USER", "hello", null)); + tree.flush(); + + assertTrue(SessionTree.awaitMirrorQuiescence(5, TimeUnit.SECONDS)); + assertEquals(0, sandbox.uploadAttempts); + } + + @Test + void flush_updatesLocalIndexWhenPinnedSandboxWasReleased() throws Exception { + StoppedTransferSandbox sandbox = new StoppedTransferSandbox(); + PinnedSandboxFilesystem.markSandboxReleased(sandbox); + SandboxBackedFilesystem filesystem = new SandboxBackedFilesystem(); + filesystem.setSandbox(sandbox); + Path context = workspace.resolve("agents/agent-a/sessions/released-index.jsonl"); + + try (WorkspaceIndex index = WorkspaceIndex.open(workspace)) { + assertTrue(index != null, "workspace index should be available for this test"); + SessionTree tree = new SessionTree(context, workspace, filesystem, index); + tree.append(new SessionEntry.MessageEntry(null, null, null, "USER", "hello", null)); + tree.flush(); + + assertTrue(SessionTree.awaitMirrorQuiescence(5, TimeUnit.SECONDS)); + assertTrue(index.exists("agents/agent-a/sessions/released-index.jsonl")); + assertTrue(index.exists("agents/agent-a/sessions/released-index.log.jsonl")); + } + assertEquals(0, sandbox.uploadAttempts); + } + @Test void flush_withTranscriptStore_stillMirrorsCanonicalFiles() throws Exception { InMemoryStore store = new InMemoryStore(); @@ -244,4 +308,59 @@ void flush_withTranscriptStore_stillMirrorsCanonicalFiles() throws Exception { private static void awaitMirror() throws InterruptedException { TimeUnit.MILLISECONDS.sleep(300); } + + private static final class StoppedTransferSandbox implements Sandbox, SandboxFileTransfer { + + private int uploadAttempts; + + @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 runtimeContext, String command, Integer timeoutSeconds) { + throw new AssertionError("released sandbox must not execute commands"); + } + + @Override + public InputStream persistWorkspace() { + throw new AssertionError("released sandbox must not persist workspaces"); + } + + @Override + public void hydrateWorkspace(InputStream archive) { + throw new AssertionError("released sandbox must not hydrate workspaces"); + } + + @Override + public boolean supportsFileTransfer(String absolutePath) { + return true; + } + + @Override + public void uploadFile(String absolutePath, byte[] content) { + uploadAttempts++; + } + + @Override + public byte[] downloadFile(String absolutePath) { + throw new AssertionError("released sandbox must not download files"); + } + } } diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/transcript/ObjectStoreTranscriptStoreTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/transcript/ObjectStoreTranscriptStoreTest.java new file mode 100644 index 0000000000..924e54f659 --- /dev/null +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/transcript/ObjectStoreTranscriptStoreTest.java @@ -0,0 +1,56 @@ +/* + * 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.transcript; + +import static org.junit.jupiter.api.Assertions.assertThrows; + +import io.agentscope.harness.agent.filesystem.AbstractFilesystem; +import io.agentscope.harness.agent.filesystem.model.FileUploadResponse; +import java.lang.reflect.Proxy; +import java.util.List; +import org.junit.jupiter.api.Test; + +class ObjectStoreTranscriptStoreTest { + + @Test + void appendSegmentPropagatesReturnedUploadFailure() { + AbstractFilesystem filesystem = + (AbstractFilesystem) + Proxy.newProxyInstance( + AbstractFilesystem.class.getClassLoader(), + new Class[] {AbstractFilesystem.class}, + (proxy, method, args) -> { + if (method.getName().equals("uploadFiles")) { + return List.of( + FileUploadResponse.fail( + "segment", "connection closed")); + } + return null; + }); + + ObjectStoreTranscriptStore store = new ObjectStoreTranscriptStore(filesystem); + + assertThrows( + IllegalStateException.class, + () -> + store.appendSegment( + new TranscriptRef("tenant", "agent", "session"), + 0, + 0, + "writer", + new byte[] {1})); + } +}