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 e75bb35b8c..4c0beb416a 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 @@ -2779,6 +2779,7 @@ public HarnessAgent build() { visibilityFilter, stager, shellPolicy); + skillMiddleware.isolationScope(fsIsolationScope); inner.middleware(skillMiddleware); // Harness owns both the live and frozen repository paths. diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/HarnessSkillMiddleware.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/HarnessSkillMiddleware.java index 94f20cb425..bc4c3c0132 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/HarnessSkillMiddleware.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/HarnessSkillMiddleware.java @@ -21,6 +21,7 @@ import io.agentscope.core.skill.SkillFilter; import io.agentscope.core.skill.repository.AgentSkillRepository; import io.agentscope.core.tool.Toolkit; +import io.agentscope.harness.agent.IsolationScope; import io.agentscope.harness.agent.skill.LazyResourceCapable; import io.agentscope.harness.agent.skill.RuntimeContextSkillRepository; import io.agentscope.harness.agent.skill.SkillResources; @@ -81,6 +82,36 @@ public class HarnessSkillMiddleware implements HarnessRuntimeMiddleware { private final SkillRuntime runtime; private final Map sourceNamespaces; private final Map frozenSkills; + private IsolationScope isolationScope; + + /** + * Per-call cache scope, mirroring the identity {@link IsolationScope} already applies to + * runtime data. Calls that share a scope share a {@code .skills-cache} subtree, and only + * those calls can sweep it — which is what makes one call's visible-skill white-list + * authoritative for everything the sweep can reach. + */ + private String scopeKeyFor(RuntimeContext ctx) { + IsolationScope scope = isolationScope != null ? isolationScope : IsolationScope.USER; + return switch (scope) { + case USER -> { + String uid = ctx != null ? ctx.getUserId() : null; + if (uid != null && !uid.isBlank()) { + yield uid; + } + // Mirrors IsolationScope.USER's documented fall back to the session identity. + // null means "no identity to key on" and is distinct from an identity that + // happens to be spelled like the stager's shared bucket. + String sid = ctx != null ? ctx.getSessionId() : null; + yield sid != null && !sid.isBlank() ? sid : null; + } + case SESSION -> { + String sid = ctx != null ? ctx.getSessionId() : null; + yield sid != null && !sid.isBlank() ? sid : null; + } + // The workspace is already per-agent, so these need no further separation. + case AGENT, GLOBAL -> null; + }; + } public HarnessSkillMiddleware(List repositories, Toolkit toolkit) { this(repositories, toolkit, null, null, null, ShellPathPolicy.noShell()); @@ -169,6 +200,7 @@ private HarnessSkillMiddleware( this.stager = stager; this.shellPathPolicy = shellPathPolicy != null ? shellPathPolicy : ShellPathPolicy.noShell(); + this.isolationScope = IsolationScope.USER; this.runtime = new SkillRuntime(); // Pre-resolve source namespaces once at build time. The compose order is fixed for // the lifetime of the middleware, so this is safe and avoids repeated work per call. @@ -186,6 +218,15 @@ public SkillRuntime runtime() { return runtime; } + /** + * Overrides the isolation dimension used to separate {@code .skills-cache} subtrees. + * Defaults to {@link IsolationScope#USER}, matching the default for runtime data. + */ + public HarnessSkillMiddleware isolationScope(IsolationScope scope) { + this.isolationScope = scope; + return this; + } + /** Whether repository enumeration is frozen to the construction-time snapshot. */ public boolean isFrozen() { return frozenSkills != null; @@ -213,7 +254,7 @@ public void prestageMarketplaceSkills(RuntimeContext ctx) { List visible = applyVisibility(merged.values(), ctx); List enabled = applySkillFilter(visible, effectiveFilter(ctx)); if (!enabled.isEmpty()) { - stager.stage(enabled, sourceNamespaces); + stager.stage(enabled, sourceNamespaces, scopeKeyFor(ctx)); } } @@ -239,7 +280,9 @@ public Mono onSystemPrompt(Agent agent, RuntimeContext ctx, String curre } Map staged = - stager != null ? stager.stage(enabled, sourceNamespaces) : Map.of(); + stager != null + ? stager.stage(enabled, sourceNamespaces, scopeKeyFor(ctx)) + : Map.of(); List entries = new ArrayList<>(enabled.size()); for (RepoBound bound : enabled) { diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/skill/runtime/MarketplaceStager.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/skill/runtime/MarketplaceStager.java index 151428935b..4755bb2d04 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/skill/runtime/MarketplaceStager.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/skill/runtime/MarketplaceStager.java @@ -19,12 +19,19 @@ import io.agentscope.core.skill.repository.AgentSkillRepository; import io.agentscope.harness.agent.skill.WorkspaceSkillRepository; import java.io.IOException; +import java.io.UncheckedIOException; import java.nio.charset.StandardCharsets; +import java.nio.file.FileVisitResult; import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.SimpleFileVisitor; +import java.nio.file.attribute.BasicFileAttributes; +import java.nio.file.attribute.FileTime; import java.nio.file.attribute.PosixFilePermission; import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; +import java.time.Duration; +import java.time.Instant; import java.util.ArrayList; import java.util.Base64; import java.util.EnumSet; @@ -54,6 +61,14 @@ * directories that should remain under {@code .skills-cache}, materialises any files whose * SHA-256 has changed, and deletes orphan directories not present in the white-list. * + *

The cache is shared. One stager instance serves every concurrent call against a + * workspace root, and on a shared volume other replicas write into the same tree. A white-list + * built from one call's visible skills therefore says nothing about what other in-flight calls + * still need: with per-user skill visibility, call B's "orphan" is call A's live + * {@code filesRoot}. Orphan GC is consequently gated on {@link #DEFAULT_ORPHAN_GRACE} — a + * directory is deleted only after sitting untouched for that long, and every retained directory + * is touched on each pass — and every traversal tolerates entries deleted underneath it. + * *

Workspace-native skills (those produced by {@link WorkspaceSkillRepository}) are NOT * staged: they already live under {@code /skills/} (or are produced lazily from the * sandbox-backed filesystem) and projection covers them through the regular {@code skills} @@ -67,10 +82,32 @@ public final class MarketplaceStager { public static final String CACHE_DIR = ".skills-cache"; public static final String GLOBAL_NAMESPACE = "_global"; + /** + * How long an orphan must sit untouched before it may be deleted. The window has to outlast + * the longest call that could still shell out to a staged script, because a staged path is + * handed to the model in the system prompt and used much later in the call. Leaving a stale + * directory costs a few KB; deleting a live one breaks another user's call, so the default + * errs long. + */ + public static final Duration DEFAULT_ORPHAN_GRACE = Duration.ofMinutes(30); + + /** Segment used when the caller has no isolation identity to key on. */ + public static final String SHARED_SCOPE = "_shared"; + private final Path workspaceRoot; + private final Duration orphanGrace; public MarketplaceStager(Path workspaceRoot) { + this(workspaceRoot, DEFAULT_ORPHAN_GRACE); + } + + /** Overload for callers (and tests) that need a non-default orphan grace window. */ + public MarketplaceStager(Path workspaceRoot, Duration orphanGrace) { this.workspaceRoot = workspaceRoot; + this.orphanGrace = + orphanGrace != null && !orphanGrace.isNegative() + ? orphanGrace + : DEFAULT_ORPHAN_GRACE; } /** @@ -93,6 +130,22 @@ public MarketplaceStager(Path workspaceRoot) { */ public Map stage( List visible, Map sourceNs) { + // null means "no identity to key on", which is distinct from an identity that + // happens to be spelled like the shared bucket. + return stage(visible, sourceNs, null); + } + + /** + * Stages into this call's isolation scope. Every directory GC may remove lives under + * {@code .skills-cache//}, and only calls that share a scope share that subtree — + * so a white-list built from one call's visible skills is authoritative for everything it + * can reach, which is the property the flat layout never had. + * + * @param scope per-call isolation key (typically {@code userId} or {@code sessionId}); + * blank collapses to {@value #SHARED_SCOPE} + */ + public Map stage( + List visible, Map sourceNs, String scope) { Map roots = new HashMap<>(visible.size()); if (workspaceRoot == null) { // No host workspace available (rare; e.g. classpath-only build). Skip staging @@ -108,9 +161,57 @@ public Map stage( return roots; } - Path cacheRoot = workspaceRoot.resolve(CACHE_DIR); + Path scopeRoot = workspaceRoot.resolve(CACHE_DIR).resolve(scopeSegment(scope)); Set retained = new HashSet<>(); + stageAll(visible, sourceNs, scopeRoot, retained, roots); + + try { + garbageCollectOrphans(scopeRoot, retained); + } catch (Exception e) { + // Cache hygiene is best-effort and must never fail the agent call — the same + // fallback the per-skill materialisation applies. + log.warn("Orphan GC under {} skipped: {}", scopeRoot, e.getMessage()); + } + return roots; + } + + /** Blank scopes collapse to one shared segment so GC always has exactly one subtree. */ + private static final int MAX_SCOPE_SEGMENT = 64; + + /** + * Maps a caller-supplied identity to one path segment, injectively. Sanitising alone would + * not do: {@code alice@corp.com} and {@code alice#corp.com} both flatten to + * {@code alice_corp.com}, and two identities sharing a subtree is exactly what the scope + * exists to prevent. Anything that is not already a distinct, filesystem-safe segment keeps + * a readable prefix and is disambiguated by a digest of the original. + */ + private static String scopeSegment(String scope) { + if (scope == null || scope.isBlank()) { + return SHARED_SCOPE; + } + String safe = scope.replaceAll("[^A-Za-z0-9._-]", "_"); + boolean lossless = + safe.equals(scope) + && !safe.equals(SHARED_SCOPE) + && safe.length() <= MAX_SCOPE_SEGMENT + // Windows rejects a trailing dot or space in a path component. + && !safe.endsWith("."); + if (lossless) { + return safe; + } + String digest = sha256(scope.getBytes(StandardCharsets.UTF_8)).substring(0, 12); + int keep = Math.min(safe.length(), MAX_SCOPE_SEGMENT - digest.length() - 1); + return safe.substring(0, Math.max(keep, 0)) + "-" + digest; + } + + /** Materialises every eligible input under this call's scope subtree. */ + private void stageAll( + List visible, + Map sourceNs, + Path scopeRoot, + Set retained, + Map roots) { for (RepoBound bound : visible) { AgentSkill skill = bound.skill(); String name = skill.getName(); @@ -132,19 +233,20 @@ public Map stage( } } - Path stagedDir = cacheRoot.resolve(ns).resolve(name); + Path stagedDir = scopeRoot.resolve(ns).resolve(name); try { materializeIfChanged(stagedDir, skill.getResources()); + // Mark as live before GC runs: this is what stops a concurrent call — or + // another replica sharing the volume — from treating it as an orphan. + touch(stagedDir); retained.add(stagedDir); - roots.put(name, new StageResult.Cached(ns, name)); + roots.put( + name, new StageResult.Cached(scopeRoot.getFileName().toString(), ns, name)); } catch (Exception e) { log.warn("Failed to stage skill '{}' (source-ns={}): {}", name, ns, e.getMessage()); roots.put(name, StageResult.NONE); } } - - garbageCollectOrphans(cacheRoot, retained); - return roots; } /** Convenience for callers that don't care about return values. */ @@ -153,12 +255,13 @@ public void invalidateAll() { return; } Path cacheRoot = workspaceRoot.resolve(CACHE_DIR); - if (Files.isDirectory(cacheRoot)) { - try { - deleteRecursively(cacheRoot); - } catch (IOException e) { - log.warn("Failed to clear {}: {}", cacheRoot, e.getMessage()); - } + if (!Files.isDirectory(cacheRoot)) { + return; + } + try { + deleteRecursively(cacheRoot); + } catch (IOException | RuntimeException e) { + log.warn("Failed to clear {}: {}", cacheRoot, e.getMessage()); } } @@ -273,7 +376,7 @@ private void removeUnexpected(Path stagedDir, Set expected) { .filter(p -> !expected.contains(p.normalize())) .forEach(toDelete::add); for (Path p : toDelete) { - Files.deleteIfExists(p); + deleteQuietly(p); } // Prune now-empty subdirectories left after file removal. try (var dirStream = Files.walk(stagedDir)) { @@ -287,12 +390,14 @@ private void removeUnexpected(Path stagedDir, Set expected) { for (Path d : dirs) { try (var probe = Files.list(d)) { if (probe.findAny().isEmpty()) { - Files.deleteIfExists(d); + deleteQuietly(d); } } } } - } catch (IOException e) { + } catch (IOException | UncheckedIOException e) { + // Files.walk / Files.list wrap mid-iteration IO errors (an entry deleted by a + // concurrent call) in UncheckedIOException, which is NOT an IOException. log.debug("Cleanup of {} failed: {}", stagedDir, e.getMessage()); } } @@ -301,6 +406,9 @@ private void garbageCollectOrphans(Path cacheRoot, Set retained) { if (!Files.isDirectory(cacheRoot)) { return; } + // `retained` reflects ONE call's visible skills; the cache is shared. Only delete + // entries that no call has staged for a full grace window. + Instant cutoff = Instant.now().minus(orphanGrace); // Two-level layout: // try (var nsStream = Files.list(cacheRoot)) { List nsDirs = new ArrayList<>(); @@ -310,35 +418,83 @@ private void garbageCollectOrphans(Path cacheRoot, Set retained) { List skillDirs = new ArrayList<>(); skillStream.filter(Files::isDirectory).forEach(skillDirs::add); for (Path skillDir : skillDirs) { - if (!retained.contains(skillDir)) { - deleteRecursively(skillDir); + if (retained.contains(skillDir) || !isStale(skillDir, cutoff)) { + continue; } + deleteRecursively(skillDir); } } // Clean up empty namespace dir. try (var probe = Files.list(nsDir)) { if (probe.findAny().isEmpty()) { - Files.deleteIfExists(nsDir); + deleteQuietly(nsDir); } } } - } catch (IOException e) { + } catch (IOException | UncheckedIOException e) { log.debug("Orphan GC under {} failed: {}", cacheRoot, e.getMessage()); } } + /** Refreshes the mtime GC reads, so a live directory is never mistaken for an orphan. */ + private static void touch(Path dir) { + try { + Files.setLastModifiedTime(dir, FileTime.from(Instant.now())); + } catch (IOException | RuntimeException e) { + log.debug("Failed to touch {}: {}", dir, e.getMessage()); + } + } + + /** Unreadable mtime means we cannot prove the entry is dead, so we keep it. */ + private static boolean isStale(Path dir, Instant cutoff) { + try { + return Files.getLastModifiedTime(dir).toInstant().isBefore(cutoff); + } catch (IOException | RuntimeException e) { + log.debug("Cannot read mtime of {}; keeping it: {}", dir, e.getMessage()); + return false; + } + } + + /** Delete one entry, tolerating a concurrent call having deleted or replaced it already. */ + private static void deleteQuietly(Path p) { + try { + Files.deleteIfExists(p); + } catch (IOException | RuntimeException e) { + log.debug("Failed to delete {}: {}", p, e.getMessage()); + } + } + + /** + * Recursive delete that tolerates the tree changing underneath it. Uses + * {@link Files#walkFileTree} rather than {@link Files#walk}: the lazy stream aborts the + * whole traversal with an {@link UncheckedIOException} the moment an entry vanishes, while + * the visitor reports it through {@code visitFileFailed} and we simply carry on. + */ private void deleteRecursively(Path root) throws IOException { if (!Files.exists(root)) { return; } - try (var stream = Files.walk(root)) { - List all = new ArrayList<>(); - stream.forEach(all::add); - all.sort((a, b) -> b.getNameCount() - a.getNameCount()); - for (Path p : all) { - Files.deleteIfExists(p); - } - } + Files.walkFileTree( + root, + new SimpleFileVisitor() { + @Override + public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) { + deleteQuietly(file); + return FileVisitResult.CONTINUE; + } + + @Override + public FileVisitResult visitFileFailed(Path file, IOException exc) { + // Gone or unreadable — nothing left for us to remove. + return FileVisitResult.CONTINUE; + } + + @Override + public FileVisitResult postVisitDirectory(Path dir, IOException exc) { + deleteQuietly(dir); + return FileVisitResult.CONTINUE; + } + }); } private static byte[] decode(String content) { @@ -430,6 +586,7 @@ record None() implements StageResult {} record WorkspaceNative() implements StageResult {} /** Skill staged under {@code .skills-cache///}. */ - record Cached(String sourceNamespace, String skillName) implements StageResult {} + record Cached(String scopeSegment, String sourceNamespace, String skillName) + implements StageResult {} } } diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/skill/runtime/ShellPathPolicy.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/skill/runtime/ShellPathPolicy.java index 6120ae608f..a2a3d9f6db 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/skill/runtime/ShellPathPolicy.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/skill/runtime/ShellPathPolicy.java @@ -105,7 +105,7 @@ public String resolve(String skillName, StageResult stage) { return joinSkills(skillName); } if (stage instanceof StageResult.Cached cached) { - return joinCache(cached.sourceNamespace(), cached.skillName()); + return joinCache(cached.scopeSegment(), cached.sourceNamespace(), cached.skillName()); } return null; } @@ -124,7 +124,7 @@ private String joinSkills(String skillName) { }; } - private String joinCache(String sourceNs, String skillName) { + private String joinCache(String scopeSegment, String sourceNs, String skillName) { return switch (mode) { case SANDBOX -> escapeSpaces( @@ -132,6 +132,8 @@ private String joinCache(String sourceNs, String skillName) { + "/" + MarketplaceStager.CACHE_DIR + "/" + + scopeSegment + + "/" + sourceNs + "/" + skillName); @@ -139,6 +141,7 @@ private String joinCache(String sourceNs, String skillName) { escapeSpaces( workspaceRoot .resolve(MarketplaceStager.CACHE_DIR) + .resolve(scopeSegment) .resolve(sourceNs) .resolve(skillName) .toAbsolutePath() diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddlewareCallbackTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddlewareCallbackTest.java index ef052147a5..a09e0a77c0 100644 --- a/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddlewareCallbackTest.java +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/SandboxLifecycleMiddlewareCallbackTest.java @@ -70,7 +70,8 @@ void prestageMarketplaceSkillsMaterialisesSkillsCache() throws IOException { middleware.prestageMarketplaceSkills(RuntimeContext.empty()); assertTrue(Files.isDirectory(cacheDir), ".skills-cache should be created by prestage"); - Path stagedScript = cacheDir.resolve("test-db").resolve("db-tool").resolve("run.sh"); + Path stagedScript = + cacheDir.resolve("_shared").resolve("test-db").resolve("db-tool").resolve("run.sh"); assertTrue(Files.exists(stagedScript), "run.sh should be staged"); String content = Files.readString(stagedScript); assertEquals("#!/bin/bash\necho hello", content); @@ -97,7 +98,7 @@ void prestageIsIdempotent() { middleware.prestageMarketplaceSkills(RuntimeContext.empty()); middleware.prestageMarketplaceSkills(RuntimeContext.empty()); - Path staged = tempWorkspace.resolve(".skills-cache/src/idempotent-skill/data.txt"); + Path staged = tempWorkspace.resolve(".skills-cache/_shared/src/idempotent-skill/data.txt"); assertTrue(Files.exists(staged)); } @@ -137,7 +138,7 @@ void prestageCalledViaCallbackMaterialisesCache() { java.util.function.Consumer callback = skillMw::prestageMarketplaceSkills; callback.accept(RuntimeContext.empty()); - Path staged = tempWorkspace.resolve(".skills-cache/cb-src/callback-skill/tool.py"); + Path staged = tempWorkspace.resolve(".skills-cache/_shared/cb-src/callback-skill/tool.py"); assertTrue( Files.exists(staged), ".skills-cache should be populated by the callback before sandbox.start()"); diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/skill/runtime/MarketplaceStagerOrphanGcTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/skill/runtime/MarketplaceStagerOrphanGcTest.java new file mode 100644 index 0000000000..871d184047 --- /dev/null +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/skill/runtime/MarketplaceStagerOrphanGcTest.java @@ -0,0 +1,340 @@ +/* + * 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.skill.runtime; + +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 io.agentscope.core.skill.AgentSkill; +import io.agentscope.core.skill.repository.AgentSkillRepository; +import io.agentscope.core.skill.repository.AgentSkillRepositoryInfo; +import io.agentscope.harness.agent.skill.runtime.MarketplaceStager.RepoBound; +import java.io.IOException; +import java.nio.file.FileSystems; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.attribute.FileTime; +import java.nio.file.attribute.PosixFilePermissions; +import java.time.Duration; +import java.time.Instant; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Assumptions; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * Orphan-GC behaviour of {@link MarketplaceStager} under a shared cache root. + * + *

{@code stage()} rebuilds its retain-list from one call's visible skills, but the + * {@code .skills-cache} tree is shared by every concurrent call against the workspace (and by + * other replicas on a shared volume). With per-user skill visibility, call B's "orphan" is call + * A's live {@code filesRoot}, so GC must (a) leave recently staged directories alone and (b) + * never abort a call when the tree changes underneath a traversal. + */ +class MarketplaceStagerOrphanGcTest { + + private static final String NS = "market"; + + @Test + @DisplayName("A skill dropped from the visible set survives while it is still fresh") + void freshOrphanIsKept(@TempDir Path workspace) { + StubRepo repo = new StubRepo(NS); + MarketplaceStager stager = new MarketplaceStager(workspace); + + stager.stage(List.of(bound("alpha", repo), bound("beta", repo)), namespaces(repo)); + // Second call sees only "alpha" — e.g. a different user's visibility filter. + stager.stage(List.of(bound("alpha", repo)), namespaces(repo)); + + assertTrue( + Files.isDirectory(skillDir(workspace, "beta")), + "a directory staged seconds ago must not be GC'd on another call's behalf"); + } + + @Test + @DisplayName("An orphan untouched past the grace window is reclaimed") + void staleOrphanIsReclaimed(@TempDir Path workspace) throws IOException { + StubRepo repo = new StubRepo(NS); + MarketplaceStager stager = new MarketplaceStager(workspace, Duration.ofHours(6)); + + stager.stage(List.of(bound("alpha", repo), bound("beta", repo)), namespaces(repo)); + Files.setLastModifiedTime( + skillDir(workspace, "beta"), + FileTime.from(Instant.now().minus(Duration.ofDays(7)))); + + stager.stage(List.of(bound("alpha", repo)), namespaces(repo)); + + assertFalse(Files.exists(skillDir(workspace, "beta")), "stale orphan should be reclaimed"); + assertTrue( + Files.isDirectory(skillDir(workspace, "alpha")), "retained skill must survive GC"); + } + + @Test + @DisplayName("Concurrent calls with different visible sets never fail staging") + void concurrentCallsWithDifferentVisibleSetsDoNotThrow(@TempDir Path workspace) + throws InterruptedException { + StubRepo repo = new StubRepo(NS); + // Zero grace maximises the pressure: every pass deletes the other caller's tree while + // that caller is still walking it. + MarketplaceStager stager = new MarketplaceStager(workspace, Duration.ZERO); + Map ns = namespaces(repo); + + List both = List.of(bound("alpha", repo), bound("beta", repo)); + List alphaOnly = List.of(bound("alpha", repo)); + + AtomicInteger failures = new AtomicInteger(); + AtomicReference firstFailure = new AtomicReference<>(); + CountDownLatch start = new CountDownLatch(1); + CountDownLatch finished = new CountDownLatch(2); + + for (List visible : List.of(both, alphaOnly)) { + Thread t = + new Thread( + () -> { + try { + start.await(); + for (int i = 0; i < 150; i++) { + stager.stage(visible, ns); + } + } catch (Throwable e) { + failures.incrementAndGet(); + firstFailure.compareAndSet(null, e); + } finally { + finished.countDown(); + } + }); + t.setDaemon(true); + t.start(); + } + start.countDown(); + assertTrue(finished.await(60, TimeUnit.SECONDS), "staging threads should finish"); + + assertEquals( + 0, + failures.get(), + () -> + "stage() must tolerate a concurrent call mutating the shared cache, but" + + " threw " + + firstFailure.get()); + } + + @Test + @DisplayName("An unreadable entry under the cache degrades GC, not the call") + void unreadableEntryDoesNotFailStage(@TempDir Path workspace) throws IOException { + Assumptions.assumeTrue( + FileSystems.getDefault().supportedFileAttributeViews().contains("posix"), + "requires POSIX permissions"); + StubRepo repo = new StubRepo(NS); + MarketplaceStager stager = new MarketplaceStager(workspace, Duration.ZERO); + + stager.stage(List.of(bound("alpha", repo)), namespaces(repo)); + Path sub = skillDir(workspace, "alpha").resolve("scripts"); + assertTrue(Files.isDirectory(sub), "fixture should have staged a subdirectory"); + Files.setPosixFilePermissions(sub, Set.of()); + try { + // "alpha" is now an orphan, and walking it fails part-way through. + stager.stage(List.of(bound("beta", repo)), namespaces(repo)); + } finally { + Files.setPosixFilePermissions(sub, PosixFilePermissions.fromString("rwxr-xr-x")); + } + } + + @Test + @DisplayName("invalidateAll clears the cache and does not create one that was absent") + void invalidateAllPurgesWithoutCreating(@TempDir Path workspace) { + StubRepo repo = new StubRepo(NS); + MarketplaceStager stager = new MarketplaceStager(workspace); + + // Absent cache: an explicit purge must stay a no-op rather than create the tree. + stager.invalidateAll(); + assertFalse( + Files.exists(workspace.resolve(MarketplaceStager.CACHE_DIR)), + "invalidateAll must not create the cache it was asked to clear"); + + // Populated cache: the purge must actually remove it. This is the documented escape + // hatch for mounts where automatic GC is disabled, so it must not share that gate. + stager.stage(List.of(bound("alpha", repo)), namespaces(repo)); + assertTrue(Files.isDirectory(skillDir(workspace, "alpha")), "fixture should have staged"); + + stager.invalidateAll(); + assertFalse( + Files.exists(workspace.resolve(MarketplaceStager.CACHE_DIR)), + "invalidateAll must clear the whole cache"); + } + + @Test + @DisplayName("A sweep in one scope cannot reach another scope's tree, even when aged") + void anotherScopeIsUnreachableBySweep(@TempDir Path workspace) throws IOException { + StubRepo repo = new StubRepo(NS); + MarketplaceStager stager = new MarketplaceStager(workspace); + Map ns = namespaces(repo); + + stager.stage(List.of(bound("alpha", repo), bound("beta", repo)), ns, "bob"); + Path bobBeta = skillDir(workspace, "bob", "beta"); + assertTrue(Files.isDirectory(bobBeta), "fixture should have staged beta for bob"); + + // Age it well past the grace window: staleness is the one thing that would make a + // sweep delete it, so if scoping works this still survives. + age(bobBeta); + + // Alice never sees "beta". Under the old flat layout her sweep deleted it; her sweep + // now runs against .skills-cache/alice and cannot address bob's tree at all. + for (int i = 0; i < 5; i++) { + stager.stage(List.of(bound("alpha", repo)), ns, "alice"); + } + + assertTrue(Files.isDirectory(bobBeta), "another scope's sweep must not delete this tree"); + assertEquals( + 9, fileCount(bobBeta), "bob's staged files must be untouched by alice's sweeps"); + } + + @Test + @DisplayName("Identities that sanitise alike still get separate subtrees") + void collidingScopeNamesDoNotShareASubtree(@TempDir Path workspace) { + StubRepo repo = new StubRepo(NS); + MarketplaceStager stager = new MarketplaceStager(workspace); + Map ns = namespaces(repo); + + // Both flatten to "alice_corp.com" under a plain character substitution. + stager.stage(List.of(bound("alpha", repo)), ns, "alice@corp.com"); + stager.stage(List.of(bound("beta", repo)), ns, "alice#corp.com"); + + Path cacheRoot = workspace.resolve(MarketplaceStager.CACHE_DIR); + try (var scopes = Files.list(cacheRoot)) { + assertEquals( + 2, + scopes.count(), + "two identities must never be mapped onto one scope subtree"); + } catch (IOException e) { + throw new java.io.UncheckedIOException(e); + } + } + + // ========================================================================= + // Fixtures + // ========================================================================= + + private static Path skillDir(Path workspace, String name) { + return skillDir(workspace, MarketplaceStager.SHARED_SCOPE, name); + } + + private static Path skillDir(Path workspace, String scope, String name) { + return workspace + .resolve(MarketplaceStager.CACHE_DIR) + .resolve(scope) + .resolve(NS) + .resolve(name); + } + + private static Map namespaces(AgentSkillRepository repo) { + return MarketplaceStager.resolveSourceNamespaces(List.of(repo)); + } + + /** A skill with a subdirectory — the shape that made the original traversal abort. */ + private static RepoBound bound(String name, AgentSkillRepository repo) { + return bound(name, repo, 8); + } + + private static RepoBound bound(String name, AgentSkillRepository repo, int scriptCount) { + Map resources = new LinkedHashMap<>(); + resources.put("SKILL.md", "# " + name + "\n"); + for (int i = 0; i < scriptCount; i++) { + resources.put("scripts/run" + i + ".sh", "#!/bin/sh\necho " + name + "\n"); + } + return new RepoBound(new AgentSkill(name, "desc", "# " + name + "\n", resources, NS), repo); + } + + private static long fileCount(Path dir) { + if (!Files.isDirectory(dir)) { + return 0; + } + try (var walk = Files.walk(dir)) { + return walk.filter(Files::isRegularFile).count(); + } catch (IOException | RuntimeException e) { + return -1; + } + } + + private static void age(Path dir) throws IOException { + Files.setLastModifiedTime(dir, FileTime.from(Instant.now().minus(Duration.ofDays(7)))); + } + + /** Minimal repository stub: the stager only reads {@link #getSource()}. */ + private static final class StubRepo implements AgentSkillRepository { + + private final String source; + + StubRepo(String source) { + this.source = source; + } + + @Override + public AgentSkill getSkill(String name) { + return null; + } + + @Override + public List getAllSkillNames() { + return List.of(); + } + + @Override + public List getAllSkills() { + return List.of(); + } + + @Override + public boolean save(List skills, boolean force) { + return false; + } + + @Override + public boolean delete(String skillName) { + return false; + } + + @Override + public boolean skillExists(String skillName) { + return false; + } + + @Override + public AgentSkillRepositoryInfo getRepositoryInfo() { + return new AgentSkillRepositoryInfo(source, "", false); + } + + @Override + public String getSource() { + return source; + } + + @Override + public void setWriteable(boolean writeable) {} + + @Override + public boolean isWriteable() { + return false; + } + } +} diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/skill/runtime/SharedWorkspaceSkillStagingE2ETest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/skill/runtime/SharedWorkspaceSkillStagingE2ETest.java new file mode 100644 index 0000000000..136159dd92 --- /dev/null +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/skill/runtime/SharedWorkspaceSkillStagingE2ETest.java @@ -0,0 +1,346 @@ +/* + * 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.skill.runtime; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.agentscope.core.agent.RuntimeContext; +import io.agentscope.core.message.ContentBlock; +import io.agentscope.core.message.Msg; +import io.agentscope.core.message.TextBlock; +import io.agentscope.core.message.UserMessage; +import io.agentscope.core.model.ChatResponse; +import io.agentscope.core.model.GenerateOptions; +import io.agentscope.core.model.Model; +import io.agentscope.core.model.ToolSchema; +import io.agentscope.core.skill.AgentSkill; +import io.agentscope.core.skill.repository.AgentSkillRepository; +import io.agentscope.core.skill.repository.AgentSkillRepositoryInfo; +import io.agentscope.core.tool.Toolkit; +import io.agentscope.harness.agent.HarnessAgent; +import io.agentscope.harness.agent.filesystem.local.LocalFilesystem; +import io.agentscope.harness.agent.middleware.HarnessSkillMiddleware; +import io.agentscope.harness.agent.skill.curator.SkillVisibilityFilter; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import reactor.core.publisher.Flux; + +/** + * End-to-end coverage for skill staging when several callers share one workspace root. + * + *

Reproduces the shape reported against 2.0.1: concurrent calls whose visible skill sets + * differ used to make one caller's orphan GC delete the directory another caller had just + * staged — aborting that call with an {@code UncheckedIOException} out of + * {@code HarnessSkillMiddleware.onSystemPrompt}, before the first model round. + * + *

Both tests drive the real production path (middleware → {@link MarketplaceStager} → disk); + * the model is a stub because the failure happens strictly before any model call. + */ +class SharedWorkspaceSkillStagingE2ETest { + + // [^<]* on purpose: the prompt's legend also mentions the tag, and a + // DOTALL wildcard would splice that mention onto the first real closing tag. + private static final Pattern FILES_ROOT = Pattern.compile("([^<]*)"); + + @Test + @DisplayName("E2E: two agents sharing one workspace both complete their calls") + void concurrentAgentsSharingWorkspaceBothComplete(@TempDir Path shared) throws Exception { + // Agent A sees both skills, agent B sees only one — B's retain-list makes "beta" an + // orphan from B's point of view while A is still staging it. Both agents are closed + // afterwards: a HarnessAgent owns maintenance threads, and leaking them into the shared + // surefire JVM starves later timing-sensitive tests. + try (HarnessAgent agentA = agent("agent-a", shared, repo("src-a", "alpha", "beta")); + HarnessAgent agentB = agent("agent-b", shared, repo("src-b", "alpha"))) { + + List errors = runConcurrently(50, agentA, agentB); + + assertEquals(List.of(), errors, () -> "no call should fail, but got: " + errors); + assertTrue( + Files.isDirectory(shared.resolve(MarketplaceStager.CACHE_DIR)), + "the shared cache should exist after the calls"); + } + } + + @Test + @DisplayName("E2E: per-user visibility — the prompt never points at a deleted directory") + void perUserVisibilityKeepsEachUsersFilesOnDisk(@TempDir Path shared) throws Exception { + HarnessSkillMiddleware middleware = perUserMiddleware(shared); + + List errors = new ArrayList<>(); + List missing = new ArrayList<>(); + AtomicInteger advertisedCount = new AtomicInteger(); + CountDownLatch start = new CountDownLatch(1); + CountDownLatch done = new CountDownLatch(2); + for (String user : List.of("alice", "bob")) { + Thread t = + new Thread( + () -> { + RuntimeContext ctx = + RuntimeContext.builder() + .sessionId("s-" + user) + .userId(user) + .build(); + try { + start.await(); + for (int i = 0; i < 60; i++) { + String prompt = + middleware.onSystemPrompt(null, ctx, "").block(); + // Checked HERE, not after the run: the point is that the + // path handed to the model is valid at the moment it is + // handed over. Checking at the end would only observe + // whichever caller happened to stage last. + assertNotNull(prompt); + Matcher m = FILES_ROOT.matcher(prompt); + int seen = 0; + while (m.find()) { + Path advertised = + Paths.get( + m.group(1).trim().replace("\\ ", " ")); + if (!Files.isDirectory(advertised)) { + synchronized (missing) { + missing.add(advertised.toString()); + } + } + seen++; + } + advertisedCount.addAndGet(seen); + } + } catch (Throwable e) { + synchronized (errors) { + errors.add(e); + } + } finally { + done.countDown(); + } + }); + t.setDaemon(true); + t.start(); + } + start.countDown(); + assertTrue(done.await(60, TimeUnit.SECONDS), "both users should finish"); + + assertEquals(List.of(), errors, () -> "onSystemPrompt must not fail, but got: " + errors); + assertEquals( + List.of(), + missing, + () -> "prompt advertised files-roots that were not on disk: " + missing); + assertTrue(advertisedCount.get() > 0, "fixture should have advertised a files-root"); + } + + @Test + @DisplayName("E2E: one user's call must not delete a skill staged for another user") + void oneUsersCallDoesNotDeleteAnotherUsersSkill(@TempDir Path shared) { + HarnessSkillMiddleware middleware = perUserMiddleware(shared); + RuntimeContext bob = RuntimeContext.builder().sessionId("s-bob").userId("bob").build(); + RuntimeContext alice = + RuntimeContext.builder().sessionId("s-alice").userId("alice").build(); + + // Bob can see "beta", so his call stages it and his prompt points the model at it. + String bobPrompt = middleware.onSystemPrompt(null, bob, "").block(); + assertNotNull(bobPrompt); + assertTrue(bobPrompt.contains("beta"), "bob should see beta"); + // Scoped by userId: bob's subtree is one alice's call cannot reach at all, which is + // what turns "must not delete" from a timing property into a structural one. + Path betaDir = + shared.resolve(MarketplaceStager.CACHE_DIR) + .resolve("bob") + .resolve("market") + .resolve("beta"); + assertTrue(Files.isDirectory(betaDir), "beta should be staged for bob"); + + // Alice cannot see "beta". Her orphan GC must not reclaim what bob is still using. + middleware.onSystemPrompt(null, alice, "").block(); + + assertTrue( + Files.isDirectory(betaDir), + "alice's call deleted the directory bob's prompt still points at: " + betaDir); + } + + // ========================================================================= + // Fixtures + // ========================================================================= + + /** "beta" is visible to bob only; "alpha" to everyone — the report's per-user setup. */ + private static HarnessSkillMiddleware perUserMiddleware(Path workspace) { + SkillVisibilityFilter perUser = + (all, ctx) -> + all.stream() + .filter( + s -> + !"beta".equals(s.getName()) + || "bob".equals(ctx.getUserId())) + .toList(); + return new HarnessSkillMiddleware( + List.of(repo("market", "alpha", "beta")), + new Toolkit(), + null, + perUser, + new MarketplaceStager(workspace), + ShellPathPolicy.localWithShell(workspace)); + } + + private static List runConcurrently(int rounds, HarnessAgent... agents) + throws InterruptedException { + List errors = new ArrayList<>(); + CountDownLatch start = new CountDownLatch(1); + CountDownLatch done = new CountDownLatch(agents.length); + for (int i = 0; i < agents.length; i++) { + HarnessAgent agent = agents[i]; + String user = "user-" + i; + Thread t = + new Thread( + () -> { + try { + start.await(); + for (int r = 0; r < rounds; r++) { + RuntimeContext ctx = + RuntimeContext.builder() + .sessionId(user + "-" + r) + .userId(user) + .build(); + agent.call(new UserMessage("hi"), ctx).block(); + } + } catch (Throwable e) { + synchronized (errors) { + errors.add(e); + } + } finally { + done.countDown(); + } + }); + t.setDaemon(true); + t.start(); + } + start.countDown(); + assertTrue(done.await(120, TimeUnit.SECONDS), "agent calls should finish"); + return errors; + } + + private static HarnessAgent agent(String name, Path workspace, AgentSkillRepository repo) { + return HarnessAgent.builder() + .name(name) + .model(new StubModel()) + .workspace(workspace) + .abstractFilesystem(new LocalFilesystem(workspace)) + .skillRepository(repo) + .build(); + } + + /** Skills carry a subdirectory — the shape whose traversal used to abort. */ + private static AgentSkillRepository repo(String source, String... names) { + List skills = new ArrayList<>(); + for (String name : names) { + Map resources = new LinkedHashMap<>(); + resources.put("SKILL.md", "# " + name + "\n"); + for (int i = 0; i < 6; i++) { + resources.put("scripts/run" + i + ".sh", "#!/bin/sh\necho " + name + "\n"); + } + skills.add(new AgentSkill(name, "desc", "# " + name + "\n", resources, source)); + } + return new StubRepo(skills, source); + } + + private static final class StubModel implements Model { + + @Override + public Flux stream( + List messages, List tools, GenerateOptions options) { + return Flux.just( + ChatResponse.builder() + .content(List.of(TextBlock.builder().text("ok").build())) + .build()); + } + + @Override + public String getModelName() { + return "stub-model"; + } + } + + private static final class StubRepo implements AgentSkillRepository { + + private final List skills; + private final String source; + + StubRepo(List skills, String source) { + this.skills = skills; + this.source = source; + } + + @Override + public AgentSkill getSkill(String name) { + return skills.stream().filter(s -> s.getName().equals(name)).findFirst().orElse(null); + } + + @Override + public List getAllSkillNames() { + return skills.stream().map(AgentSkill::getName).toList(); + } + + @Override + public List getAllSkills() { + return skills; + } + + @Override + public boolean save(List skills, boolean force) { + return false; + } + + @Override + public boolean delete(String skillName) { + return false; + } + + @Override + public boolean skillExists(String skillName) { + return skills.stream().anyMatch(s -> s.getName().equals(skillName)); + } + + @Override + public AgentSkillRepositoryInfo getRepositoryInfo() { + return new AgentSkillRepositoryInfo(source, "", false); + } + + @Override + public String getSource() { + return source; + } + + @Override + public void setWriteable(boolean writeable) {} + + @Override + public boolean isWriteable() { + return false; + } + } +} diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/skill/runtime/SkillRuntimeTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/skill/runtime/SkillRuntimeTest.java index fd522e9c2c..ebe5327ea1 100644 --- a/agentscope-harness/src/test/java/io/agentscope/harness/agent/skill/runtime/SkillRuntimeTest.java +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/skill/runtime/SkillRuntimeTest.java @@ -107,8 +107,9 @@ void sandboxResolveEscapesSpacesInSkillName() { @Test void sandboxResolveEscapesSpacesInCachedSkill() { ShellPathPolicy policy = ShellPathPolicy.sandbox(); - String result = policy.resolve("ignored", new StageResult.Cached("ns", "my skill")); - assertEquals("/workspace/.skills-cache/ns/my\\ skill", result); + String result = + policy.resolve("ignored", new StageResult.Cached("alice", "ns", "my skill")); + assertEquals("/workspace/.skills-cache/alice/ns/my\\ skill", result); } @Test @@ -129,7 +130,7 @@ void localWithShellResolveEscapesSpaces() { void noShellAlwaysReturnsNull() { ShellPathPolicy policy = ShellPathPolicy.noShell(); assertNull(policy.resolve("any name", new StageResult.WorkspaceNative())); - assertNull(policy.resolve("any name", new StageResult.Cached("ns", "name"))); + assertNull(policy.resolve("any name", new StageResult.Cached("alice", "ns", "name"))); assertNull(policy.resolve("any name", StageResult.NONE)); assertNull(policy.resolve("any name", null)); }