feat(backends): implement tier-aware virtual mount resolution for ski… (#236)
* feat(backends): implement tier-aware virtual mount resolution for skills and memories * test: add end-to-end test for workspace tier shadowing global tier in CustomSandboxBackend * feat(backends): enhance virtual mount resolution for skills and memories with tier paths and quoting * fix(tests): update Python command in virtual mount resolution tests to use python3
This commit is contained in:
Xi Zhang
+119
-6
@@ -19,6 +19,12 @@ from deepagents.backends.protocol import (
|
||||
WriteResult,
|
||||
)
|
||||
|
||||
from . import paths
|
||||
|
||||
# Reproduced here to dodge a circular import from .EvoScientist (the canonical
|
||||
# SKILLS_DIR constant).
|
||||
_BUILTIN_SKILLS_DIR = Path(__file__).parent / "skills"
|
||||
|
||||
# System path prefixes that should never appear in virtual paths.
|
||||
# If the agent hallucinates an absolute system path, we block it.
|
||||
_SYSTEM_PATH_PREFIXES = (
|
||||
@@ -128,12 +134,36 @@ def _collect_executable_positions(command: str) -> set[int]:
|
||||
return offsets
|
||||
|
||||
|
||||
def _extract_all_paths(command: str) -> list[str]:
|
||||
def _is_under_allowed_prefix(path: str, allow_prefixes: tuple[str, ...]) -> bool:
|
||||
"""True if *path* equals a prefix or is a strict descendant.
|
||||
|
||||
Boundary-aware: ``str.startswith`` alone would let ``/A/skills_evil``
|
||||
match the prefix ``/A/skills`` — anchoring on ``/`` blocks neighbour
|
||||
directories that merely share a name prefix.
|
||||
"""
|
||||
for prefix in allow_prefixes:
|
||||
normalized = prefix.rstrip("/")
|
||||
# Skip empty/root prefixes: they'd reduce the check to startswith("/")
|
||||
# and admit every absolute path, silently disabling the allowlist.
|
||||
if not normalized:
|
||||
continue
|
||||
if path == normalized or path.startswith(normalized + "/"):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _extract_all_paths(
|
||||
command: str,
|
||||
allow_prefixes: tuple[str, ...] = (),
|
||||
) -> list[str]:
|
||||
"""Extract potential file paths from a command, including inside quoted strings.
|
||||
|
||||
Scans both shell tokens and string literals (single/double quoted) to find
|
||||
paths that start with system prefixes like /Users/, /etc/, /tmp/, etc.
|
||||
Skips paths in executable position (command name) and pip install targets.
|
||||
|
||||
Paths matched by ``allow_prefixes`` (via ``_is_under_allowed_prefix``)
|
||||
are dropped.
|
||||
"""
|
||||
exe_offsets = _collect_executable_positions(command)
|
||||
paths: list[str] = []
|
||||
@@ -149,14 +179,25 @@ def _extract_all_paths(command: str) -> list[str]:
|
||||
# Skip paths that land at an executable-position offset
|
||||
if m.start(1) in exe_offsets:
|
||||
continue
|
||||
paths.append(m.group(1))
|
||||
extracted = m.group(1)
|
||||
if _is_under_allowed_prefix(extracted, allow_prefixes):
|
||||
continue
|
||||
paths.append(extracted)
|
||||
return paths
|
||||
|
||||
|
||||
def validate_command(command: str) -> str | None:
|
||||
def validate_command(
|
||||
command: str,
|
||||
allow_prefixes: tuple[str, ...] = (),
|
||||
) -> str | None:
|
||||
"""
|
||||
Validate a shell command for safety.
|
||||
|
||||
Args:
|
||||
command: Shell command string.
|
||||
allow_prefixes: Absolute path prefixes exempt from the system-path
|
||||
block list (matching rules in ``_is_under_allowed_prefix``).
|
||||
|
||||
Returns:
|
||||
None if command is safe, error message string if blocked.
|
||||
"""
|
||||
@@ -187,7 +228,7 @@ def validate_command(command: str) -> str | None:
|
||||
|
||||
# Check for absolute system paths (including inside quoted strings).
|
||||
# This catches attacks like: python -c "os.remove('/Users/foo/file')"
|
||||
escaped_paths = _extract_all_paths(command)
|
||||
escaped_paths = _extract_all_paths(command, allow_prefixes=allow_prefixes)
|
||||
if escaped_paths:
|
||||
path_sample = escaped_paths[0]
|
||||
return (
|
||||
@@ -199,6 +240,61 @@ def validate_command(command: str) -> str | None:
|
||||
return None
|
||||
|
||||
|
||||
def _subpath_under_mount(token: str, mount: str) -> str | None:
|
||||
"""Return the subpath of *token* under *mount*, or ``None`` if not under it.
|
||||
|
||||
Bare ``mount`` and ``mount + "/"`` both return ``""`` so the caller can
|
||||
join uniformly (``Path(tier) / ""`` is the tier itself).
|
||||
"""
|
||||
if token == mount or token == mount + "/":
|
||||
return ""
|
||||
prefix = mount + "/"
|
||||
if token.startswith(prefix):
|
||||
return token[len(prefix) :]
|
||||
return None
|
||||
|
||||
|
||||
def _skills_tier_paths() -> tuple[Path, Path | None, Path]:
|
||||
"""``(USER, GLOBAL or None, BUILTIN)`` — the tier priority chain that
|
||||
``MergedSkillsBackend._backends()`` honors. Single source of truth so
|
||||
the resolver and the backend can't silently drift out of order.
|
||||
"""
|
||||
return (paths.USER_SKILLS_DIR, paths.GLOBAL_SKILLS_DIR, _BUILTIN_SKILLS_DIR)
|
||||
|
||||
|
||||
def _resolve_virtual_mount_path(token: str) -> str | None:
|
||||
"""Resolve a virtual mount token to a shell-safe token, or ``None`` when
|
||||
*token* is not a registered virtual mount.
|
||||
|
||||
For ``/skills/...``: walks ``_skills_tier_paths()`` priority (USER →
|
||||
GLOBAL → BUILTIN), returning ``shlex.quote`` of the first tier where the
|
||||
path exists. On miss, returns a workspace-relative ``./skills/<rel>``
|
||||
form — agent typed a virtual path, so the shell error should reference a
|
||||
location they recognise (`USER_SKILLS_DIR` defaults to
|
||||
``WORKSPACE_ROOT / "skills"``, which is also where ``MergedSkillsBackend``
|
||||
would write a new skill).
|
||||
|
||||
For ``/memories/...``: single tier (``paths.MEMORIES_DIR``), always
|
||||
absolute and ``shlex.quote``-wrapped. Memories live outside the
|
||||
workspace, so a relative form would point at an unrelated location.
|
||||
"""
|
||||
rel = _subpath_under_mount(token, "/skills")
|
||||
if rel is not None:
|
||||
for tier in _skills_tier_paths():
|
||||
if tier is None:
|
||||
continue
|
||||
candidate = Path(tier) / rel
|
||||
if candidate.exists():
|
||||
return shlex.quote(str(candidate))
|
||||
return shlex.quote("./skills/" + rel if rel else "./skills")
|
||||
|
||||
rel = _subpath_under_mount(token, "/memories")
|
||||
if rel is not None:
|
||||
return shlex.quote(str(Path(paths.MEMORIES_DIR) / rel))
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def convert_virtual_paths_in_command(
|
||||
command: str,
|
||||
workspace_name: str | None = None,
|
||||
@@ -209,6 +305,11 @@ def convert_virtual_paths_in_command(
|
||||
Also auto-corrects hallucinated system absolute paths that reference the
|
||||
workspace directory (e.g. ``/Users/.../myproject/file.py`` → ``./file.py``).
|
||||
|
||||
Tier-aware mounts (``/skills/...``, ``/memories/...``) are expanded to
|
||||
absolute paths via ``_resolve_virtual_mount_path``. Callers that pass
|
||||
the result through ``validate_command`` MUST whitelist the tier roots
|
||||
via ``allow_prefixes`` to avoid false-positive system-path blocks.
|
||||
|
||||
Args:
|
||||
command: Original command.
|
||||
workspace_name: Basename of the workspace directory (e.g. ``"workspace"``,
|
||||
@@ -232,6 +333,10 @@ def convert_virtual_paths_in_command(
|
||||
if "://" in command[max(0, match.start() - 10) : match.end() + 10]:
|
||||
return path
|
||||
|
||||
resolved = _resolve_virtual_mount_path(path)
|
||||
if resolved is not None:
|
||||
return resolved
|
||||
|
||||
# Fix hallucinated system absolute paths that reference the workspace.
|
||||
# E.g. /Users/user/.../myproject/file.py → ./file.py
|
||||
# This mirrors _resolve_path() logic but for shell command strings.
|
||||
@@ -513,8 +618,16 @@ class CustomSandboxBackend(LocalShellBackend):
|
||||
workspace_name=Path(str(self.cwd)).name,
|
||||
)
|
||||
|
||||
# Validate command safety (after path sanitization)
|
||||
error = validate_command(command)
|
||||
# USER_SKILLS_DIR must be in the allowlist: the workspace-literal
|
||||
# replace above runs BEFORE the resolver, so any absolute path the
|
||||
# resolver later injects reaches validate_command unstripped.
|
||||
allow_prefixes = (
|
||||
str(paths.USER_SKILLS_DIR),
|
||||
str(paths.GLOBAL_SKILLS_DIR),
|
||||
str(paths.MEMORIES_DIR),
|
||||
str(_BUILTIN_SKILLS_DIR),
|
||||
)
|
||||
error = validate_command(command, allow_prefixes=allow_prefixes)
|
||||
if error:
|
||||
return ExecuteResponse(
|
||||
output=error,
|
||||
|
||||
@@ -1,10 +1,13 @@
|
||||
"""Tests for EvoScientist/backends.py — validate_command, path conversion, resolve_path."""
|
||||
|
||||
import re
|
||||
import shlex
|
||||
from pathlib import Path
|
||||
|
||||
from EvoScientist import backends, paths
|
||||
from EvoScientist.backends import (
|
||||
CustomSandboxBackend,
|
||||
MergedSkillsBackend,
|
||||
convert_virtual_paths_in_command,
|
||||
validate_command,
|
||||
)
|
||||
@@ -150,6 +153,430 @@ class TestConvertVirtualPaths:
|
||||
assert result == "cat ./tmp/somefile"
|
||||
|
||||
|
||||
# === tier-aware virtual mounts (/skills/, /memories/) ===
|
||||
|
||||
|
||||
class TestVirtualMountResolution:
|
||||
"""``convert_virtual_paths_in_command`` must resolve ``/skills/...`` and
|
||||
``/memories/...`` against the same tier priority chain used by
|
||||
``MergedSkillsBackend``, not blindly rewrite them as ``./skills/...``.
|
||||
"""
|
||||
|
||||
def _setup_tiers(self, monkeypatch, tmp_path):
|
||||
"""Create three skills tiers + a memories dir under tmp_path and
|
||||
monkeypatch the path constants to point at them. Returns the tier
|
||||
directories so tests can populate them.
|
||||
"""
|
||||
user_dir = tmp_path / "ws_skills"
|
||||
global_dir = tmp_path / "global_skills"
|
||||
builtin_dir = tmp_path / "builtin_skills"
|
||||
memories_dir = tmp_path / "memories"
|
||||
for d in (user_dir, global_dir, builtin_dir, memories_dir):
|
||||
d.mkdir()
|
||||
monkeypatch.setattr(paths, "USER_SKILLS_DIR", user_dir)
|
||||
monkeypatch.setattr(paths, "GLOBAL_SKILLS_DIR", global_dir)
|
||||
monkeypatch.setattr(paths, "MEMORIES_DIR", memories_dir)
|
||||
monkeypatch.setattr(backends, "_BUILTIN_SKILLS_DIR", builtin_dir)
|
||||
return user_dir, global_dir, builtin_dir, memories_dir
|
||||
|
||||
def test_skills_path_resolves_to_workspace_tier_when_present(
|
||||
self, monkeypatch, tmp_path
|
||||
):
|
||||
user_dir, global_dir, _, _ = self._setup_tiers(monkeypatch, tmp_path)
|
||||
(user_dir / "hello").mkdir()
|
||||
(user_dir / "hello" / "main.py").write_text("print('ws')")
|
||||
(global_dir / "hello").mkdir()
|
||||
(global_dir / "hello" / "main.py").write_text("print('global')")
|
||||
result = convert_virtual_paths_in_command("python /skills/hello/main.py")
|
||||
assert result == f"python {user_dir / 'hello' / 'main.py'}"
|
||||
|
||||
def test_skills_path_resolves_to_global_tier_when_workspace_missing(
|
||||
self, monkeypatch, tmp_path
|
||||
):
|
||||
_, global_dir, _, _ = self._setup_tiers(monkeypatch, tmp_path)
|
||||
(global_dir / "hello").mkdir()
|
||||
(global_dir / "hello" / "main.py").write_text("print('global')")
|
||||
result = convert_virtual_paths_in_command("python /skills/hello/main.py")
|
||||
assert result == f"python {global_dir / 'hello' / 'main.py'}"
|
||||
|
||||
def test_skills_path_resolves_to_builtin_tier_when_higher_missing(
|
||||
self, monkeypatch, tmp_path
|
||||
):
|
||||
_, _, builtin_dir, _ = self._setup_tiers(monkeypatch, tmp_path)
|
||||
(builtin_dir / "find-skills").mkdir()
|
||||
(builtin_dir / "find-skills" / "tool.py").write_text("print('builtin')")
|
||||
result = convert_virtual_paths_in_command("python /skills/find-skills/tool.py")
|
||||
assert result == f"python {builtin_dir / 'find-skills' / 'tool.py'}"
|
||||
|
||||
def test_skills_path_unresolvable_falls_back_to_workspace_relative(
|
||||
self, monkeypatch, tmp_path
|
||||
):
|
||||
"""Fallback returns a workspace-relative ``./skills/<rel>`` shape, not
|
||||
an absolute path. The agent typed a virtual mount, so its shell error
|
||||
should reference a location it recognises (the workspace tier is also
|
||||
where MergedSkillsBackend.write would land a new skill).
|
||||
"""
|
||||
self._setup_tiers(monkeypatch, tmp_path)
|
||||
result = convert_virtual_paths_in_command(
|
||||
"python /skills/never-installed/foo.py"
|
||||
)
|
||||
assert result == "python ./skills/never-installed/foo.py"
|
||||
|
||||
def test_memories_path_substitutes_absolute_memories_dir(
|
||||
self, monkeypatch, tmp_path
|
||||
):
|
||||
_, _, _, memories_dir = self._setup_tiers(monkeypatch, tmp_path)
|
||||
result = convert_virtual_paths_in_command("cat /memories/note.md")
|
||||
assert result == f"cat {memories_dir / 'note.md'}"
|
||||
|
||||
def test_skills_bare_root_resolves_to_user_skills_dir(self, monkeypatch, tmp_path):
|
||||
"""Bare /skills and /skills/ (no subpath) resolve to USER_SKILLS_DIR;
|
||||
mirrors the existing `/` → `.` rule but for the mount root.
|
||||
"""
|
||||
user_dir, _, _, _ = self._setup_tiers(monkeypatch, tmp_path)
|
||||
assert convert_virtual_paths_in_command("ls /skills") == f"ls {user_dir}"
|
||||
assert convert_virtual_paths_in_command("ls /skills/") == f"ls {user_dir}"
|
||||
|
||||
def test_skills_prefix_not_overmatched(self, monkeypatch, tmp_path):
|
||||
"""Paths starting with /skills but not /skills/ (e.g. /skillset/foo)
|
||||
must fall through to the existing workspace-relative branch.
|
||||
"""
|
||||
self._setup_tiers(monkeypatch, tmp_path)
|
||||
assert (
|
||||
convert_virtual_paths_in_command("cat /skillset/foo")
|
||||
== "cat ./skillset/foo"
|
||||
)
|
||||
# Same defense for /memories prefix.
|
||||
assert (
|
||||
convert_virtual_paths_in_command("cat /memoriesfoo") == "cat ./memoriesfoo"
|
||||
)
|
||||
|
||||
def test_validate_command_allows_resolved_skills_absolute_path(self, tmp_path):
|
||||
"""An absolute path whose prefix is in ``allow_prefixes`` must NOT be
|
||||
flagged as a system path. This is what lets execute() forward
|
||||
tier-resolved /skills/ expansions to the shell.
|
||||
"""
|
||||
global_dir = tmp_path / "global_skills"
|
||||
global_dir.mkdir()
|
||||
command = f"python {global_dir / 'hello' / 'main.py'}"
|
||||
assert validate_command(command, allow_prefixes=(str(global_dir),)) is None
|
||||
|
||||
def test_validate_command_still_blocks_unrelated_system_path(self, tmp_path):
|
||||
"""The allowlist must NOT weaken the block list for arbitrary system
|
||||
paths — only the whitelisted prefixes are exempted.
|
||||
"""
|
||||
global_dir = tmp_path / "global_skills"
|
||||
global_dir.mkdir()
|
||||
result = validate_command(
|
||||
"cat /etc/passwd",
|
||||
allow_prefixes=(str(global_dir),),
|
||||
)
|
||||
assert result is not None
|
||||
assert "blocked" in result.lower()
|
||||
|
||||
def test_validate_command_prefix_boundary_not_bypassed(self):
|
||||
"""Allowlist matching must be directory-boundary-aware: a neighbour
|
||||
directory sharing a string prefix (``..._evil``, ``...BACKDOOR``)
|
||||
must NOT be admitted because its name happens to start with an
|
||||
allowed prefix substring. Regression guard for the ``startswith``
|
||||
bypass flagged by code review.
|
||||
|
||||
Paths are hardcoded under ``/tmp`` rather than via ``tmp_path``
|
||||
because ``_extract_all_paths``'s regex only matches paths whose
|
||||
first component is a known system prefix (``/Users``, ``/tmp``,
|
||||
``/var``, …) — on macOS ``tmp_path`` resolves to
|
||||
``/private/var/folders/…`` which the negative lookbehind rejects
|
||||
(the ``v`` in ``/var`` is preceded by ``e`` in ``private``).
|
||||
``validate_command`` is a pure string check, so no real
|
||||
filesystem entries are required.
|
||||
"""
|
||||
allowed = "/tmp/evosci_skills_test_prefix"
|
||||
evil_path = "/tmp/evosci_skills_test_prefix_evil/secret.txt"
|
||||
result = validate_command(
|
||||
f"cat {evil_path}",
|
||||
allow_prefixes=(allowed,),
|
||||
)
|
||||
assert result is not None
|
||||
assert "blocked" in result.lower()
|
||||
# Sanity check: a real descendant of the allowed prefix still passes,
|
||||
# so we're testing boundary semantics, not a blanket block.
|
||||
legit_path = "/tmp/evosci_skills_test_prefix/real/file.txt"
|
||||
assert (
|
||||
validate_command(
|
||||
f"cat {legit_path}",
|
||||
allow_prefixes=(allowed,),
|
||||
)
|
||||
is None
|
||||
)
|
||||
|
||||
def test_validate_command_prefix_with_trailing_slash_normalized(self):
|
||||
"""An allowlist entry that already has a trailing slash should behave
|
||||
identically to the no-trailing-slash form — both reject the
|
||||
neighbour-directory bypass AND admit legitimate descendants /
|
||||
exact-match paths.
|
||||
"""
|
||||
allowed_with_slash = "/tmp/evosci_skills_test_prefix/"
|
||||
evil_path = "/tmp/evosci_skills_test_prefix_evil/x"
|
||||
legit_descendant = "/tmp/evosci_skills_test_prefix/ok/file.txt"
|
||||
exact_match = "/tmp/evosci_skills_test_prefix"
|
||||
assert (
|
||||
validate_command(
|
||||
f"cat {evil_path}",
|
||||
allow_prefixes=(allowed_with_slash,),
|
||||
)
|
||||
is not None
|
||||
)
|
||||
assert (
|
||||
validate_command(
|
||||
f"cat {legit_descendant}",
|
||||
allow_prefixes=(allowed_with_slash,),
|
||||
)
|
||||
is None
|
||||
)
|
||||
assert (
|
||||
validate_command(
|
||||
f"cat {exact_match}",
|
||||
allow_prefixes=(allowed_with_slash,),
|
||||
)
|
||||
is None
|
||||
)
|
||||
|
||||
def test_validate_command_empty_prefix_does_not_disable_allowlist(self):
|
||||
"""An empty or root-only entry in ``allow_prefixes`` must NOT silently
|
||||
admit every absolute path. Regression guard for the empty/root-prefix
|
||||
gap flagged by code review — without the ``if not normalized: continue``
|
||||
guard, ``"".rstrip("/") + "/"`` collapses to ``"/"`` and admits any
|
||||
absolute path via ``startswith("/")``.
|
||||
"""
|
||||
for trivial in ("", "/"):
|
||||
assert (
|
||||
validate_command(
|
||||
"cat /etc/passwd",
|
||||
allow_prefixes=(trivial,),
|
||||
)
|
||||
is not None
|
||||
)
|
||||
|
||||
def test_skills_resolver_quotes_path_when_tier_dir_has_whitespace(
|
||||
self, monkeypatch, tmp_path
|
||||
):
|
||||
"""When a tier directory itself sits under a path with whitespace
|
||||
(the realistic case: user home like ``/Users/Foo Bar/.evoscientist/skills``),
|
||||
the resolver must shell-quote its absolute output so the shell parses
|
||||
the command argument as a single token.
|
||||
|
||||
NOTE: the input virtual path is kept clean (no whitespace in the
|
||||
skill name). Input-side whitespace is truncated by the regex in
|
||||
``convert_virtual_paths_in_command`` before the resolver fires —
|
||||
out of scope for this PR (would require a quote-aware path regex).
|
||||
"""
|
||||
spacey_root = tmp_path / "Foo Bar"
|
||||
spacey_root.mkdir()
|
||||
user_dir = spacey_root / "ws_skills"
|
||||
user_dir.mkdir()
|
||||
for d in ("global_skills", "memories", "builtin_skills"):
|
||||
(spacey_root / d).mkdir()
|
||||
(user_dir / "hello").mkdir()
|
||||
(user_dir / "hello" / "main.py").write_text("print('ok')")
|
||||
|
||||
monkeypatch.setattr(paths, "USER_SKILLS_DIR", user_dir)
|
||||
monkeypatch.setattr(paths, "GLOBAL_SKILLS_DIR", spacey_root / "global_skills")
|
||||
monkeypatch.setattr(paths, "MEMORIES_DIR", spacey_root / "memories")
|
||||
monkeypatch.setattr(
|
||||
backends, "_BUILTIN_SKILLS_DIR", spacey_root / "builtin_skills"
|
||||
)
|
||||
|
||||
result = convert_virtual_paths_in_command("python /skills/hello/main.py")
|
||||
|
||||
tokens = shlex.split(result)
|
||||
assert tokens[0] == "python"
|
||||
assert tokens[1] == str(user_dir / "hello" / "main.py")
|
||||
|
||||
def test_memories_resolver_quotes_path_when_memories_dir_has_whitespace(
|
||||
self, monkeypatch, tmp_path
|
||||
):
|
||||
"""Memories live outside the workspace, so the relative-form rewrite
|
||||
Fix 3 applies to skills does NOT apply here. The resolver still must
|
||||
shell-quote its absolute output for whitespace safety.
|
||||
"""
|
||||
spacey = tmp_path / "Foo Bar" / "memories"
|
||||
spacey.mkdir(parents=True)
|
||||
monkeypatch.setattr(paths, "MEMORIES_DIR", spacey)
|
||||
|
||||
result = convert_virtual_paths_in_command("cat /memories/note.md")
|
||||
|
||||
tokens = shlex.split(result)
|
||||
assert tokens[0] == "cat"
|
||||
assert tokens[1] == str(spacey / "note.md")
|
||||
|
||||
def test_skills_tier_paths_matches_merged_backend_priority(
|
||||
self, monkeypatch, tmp_path
|
||||
):
|
||||
"""Drift detector: ``_skills_tier_paths()`` must list tiers in the
|
||||
same priority order ``MergedSkillsBackend._backends()`` walks. If
|
||||
either side reorders without the other, the resolver and backend
|
||||
will disagree on which tier owns a file. Verified by populating the
|
||||
same path in all three tiers with distinct content and asserting
|
||||
both reach the same (highest-priority) tier.
|
||||
"""
|
||||
user_dir, global_dir, builtin_dir, _ = self._setup_tiers(monkeypatch, tmp_path)
|
||||
for tier_dir, tag in (
|
||||
(user_dir, "USER"),
|
||||
(global_dir, "GLOBAL"),
|
||||
(builtin_dir, "BUILTIN"),
|
||||
):
|
||||
(tier_dir / "probe").mkdir()
|
||||
(tier_dir / "probe" / "main.txt").write_text(tag)
|
||||
|
||||
# Build MergedSkillsBackend wired via _skills_tier_paths positions —
|
||||
# the test FAILS if helper return order doesn't align with the
|
||||
# constructor's tier-arg semantics.
|
||||
user, global_, builtin = backends._skills_tier_paths()
|
||||
mb = MergedSkillsBackend(
|
||||
primary_dir=str(user),
|
||||
secondary_dir=str(builtin),
|
||||
global_dir=str(global_) if global_ is not None else None,
|
||||
)
|
||||
# MergedSkillsBackend.read returns content from the highest-priority
|
||||
# tier that has the file (USER per the assumed alignment).
|
||||
backend_content = mb.read("/probe/main.txt")
|
||||
text = (
|
||||
backend_content
|
||||
if isinstance(backend_content, str)
|
||||
else getattr(backend_content, "content", str(backend_content))
|
||||
)
|
||||
assert "USER" in text
|
||||
|
||||
# Resolver also returns the USER tier path (the highest-priority hit).
|
||||
resolved = backends._resolve_virtual_mount_path("/skills/probe/main.txt")
|
||||
assert str(user_dir / "probe" / "main.txt") in resolved
|
||||
|
||||
# Remove USER tier file; both should fall through to GLOBAL together.
|
||||
(user_dir / "probe" / "main.txt").unlink()
|
||||
backend_content = mb.read("/probe/main.txt")
|
||||
text = (
|
||||
backend_content
|
||||
if isinstance(backend_content, str)
|
||||
else getattr(backend_content, "content", str(backend_content))
|
||||
)
|
||||
assert "GLOBAL" in text
|
||||
resolved = backends._resolve_virtual_mount_path("/skills/probe/main.txt")
|
||||
assert str(global_dir / "probe" / "main.txt") in resolved
|
||||
|
||||
def test_skills_tier_paths_helper_returns_canonical_order(self):
|
||||
"""Pin the helper's slot order so calling code (constructor wiring,
|
||||
tests like the alignment one above) can rely on it.
|
||||
"""
|
||||
result = backends._skills_tier_paths()
|
||||
assert len(result) == 3
|
||||
assert result[0] == paths.USER_SKILLS_DIR
|
||||
assert result[1] == paths.GLOBAL_SKILLS_DIR
|
||||
assert result[2] == backends._BUILTIN_SKILLS_DIR
|
||||
|
||||
def test_execute_e2e_workspace_tier_skill(self, monkeypatch, tmp_path):
|
||||
"""End-to-end: a skill in the workspace tier (USER_SKILLS_DIR) must
|
||||
execute successfully. Regression guard: USER_SKILLS_DIR must be in
|
||||
execute()'s allow_prefixes — the workspace-literal replace at the
|
||||
top of execute() runs BEFORE convert_virtual_paths_in_command, so
|
||||
any absolute path the resolver subsequently injects reaches
|
||||
validate_command unstripped and would trip the system-path block
|
||||
list without an explicit allowlist entry.
|
||||
"""
|
||||
workspace = tmp_path / "ws"
|
||||
workspace.mkdir()
|
||||
user_dir = workspace / "skills"
|
||||
user_dir.mkdir()
|
||||
global_dir = tmp_path / "global_skills"
|
||||
global_dir.mkdir()
|
||||
memories_dir = tmp_path / "memories"
|
||||
memories_dir.mkdir()
|
||||
builtin_dir = tmp_path / "builtin_skills"
|
||||
builtin_dir.mkdir()
|
||||
# Skill lives ONLY in the workspace tier.
|
||||
(user_dir / "hello-ws").mkdir()
|
||||
(user_dir / "hello-ws" / "main.py").write_text(
|
||||
"print('workspace-tier-fix-works')"
|
||||
)
|
||||
|
||||
monkeypatch.setattr(paths, "USER_SKILLS_DIR", user_dir)
|
||||
monkeypatch.setattr(paths, "GLOBAL_SKILLS_DIR", global_dir)
|
||||
monkeypatch.setattr(paths, "MEMORIES_DIR", memories_dir)
|
||||
monkeypatch.setattr(backends, "_BUILTIN_SKILLS_DIR", builtin_dir)
|
||||
|
||||
backend = CustomSandboxBackend(root_dir=str(workspace), virtual_mode=True)
|
||||
resp = backend.execute("python3 /skills/hello-ws/main.py")
|
||||
assert resp.exit_code == 0, resp.output
|
||||
assert "workspace-tier-fix-works" in resp.output
|
||||
|
||||
def test_execute_e2e_workspace_tier_shadows_global(self, monkeypatch, tmp_path):
|
||||
"""End-to-end: when the same skill exists in BOTH workspace and global
|
||||
tiers, the workspace version must shadow the global one when invoked
|
||||
via ``CustomSandboxBackend.execute``. Mirrors
|
||||
``MergedSkillsBackend``'s priority (USER > GLOBAL > BUILTIN) at the
|
||||
full-pipeline level, complementing the unit-level priority check in
|
||||
``test_skills_path_resolves_to_workspace_tier_when_present``.
|
||||
"""
|
||||
workspace = tmp_path / "ws"
|
||||
workspace.mkdir()
|
||||
user_dir = workspace / "skills"
|
||||
user_dir.mkdir()
|
||||
global_dir = tmp_path / "global_skills"
|
||||
global_dir.mkdir()
|
||||
memories_dir = tmp_path / "memories"
|
||||
memories_dir.mkdir()
|
||||
builtin_dir = tmp_path / "builtin_skills"
|
||||
builtin_dir.mkdir()
|
||||
# Same skill name in both tiers, different outputs.
|
||||
(user_dir / "shadow-test").mkdir()
|
||||
(user_dir / "shadow-test" / "main.py").write_text(
|
||||
"print('WORKSPACE_TIER_WINS')"
|
||||
)
|
||||
(global_dir / "shadow-test").mkdir()
|
||||
(global_dir / "shadow-test" / "main.py").write_text("print('GLOBAL_TIER_LOST')")
|
||||
|
||||
monkeypatch.setattr(paths, "USER_SKILLS_DIR", user_dir)
|
||||
monkeypatch.setattr(paths, "GLOBAL_SKILLS_DIR", global_dir)
|
||||
monkeypatch.setattr(paths, "MEMORIES_DIR", memories_dir)
|
||||
monkeypatch.setattr(backends, "_BUILTIN_SKILLS_DIR", builtin_dir)
|
||||
|
||||
backend = CustomSandboxBackend(root_dir=str(workspace), virtual_mode=True)
|
||||
resp = backend.execute("python3 /skills/shadow-test/main.py")
|
||||
assert resp.exit_code == 0, resp.output
|
||||
assert "WORKSPACE_TIER_WINS" in resp.output
|
||||
assert "GLOBAL_TIER_LOST" not in resp.output
|
||||
|
||||
def test_execute_e2e_global_tier_skill(self, monkeypatch, tmp_path):
|
||||
"""End-to-end: a skill that exists ONLY in the global tier (workspace
|
||||
does not have a copy) must execute successfully via
|
||||
``CustomSandboxBackend.execute``. This is the exact bug fixed.
|
||||
"""
|
||||
workspace = tmp_path / "ws"
|
||||
workspace.mkdir()
|
||||
user_dir = workspace / "skills"
|
||||
user_dir.mkdir()
|
||||
global_dir = tmp_path / "global_skills"
|
||||
global_dir.mkdir()
|
||||
memories_dir = tmp_path / "memories"
|
||||
memories_dir.mkdir()
|
||||
builtin_dir = tmp_path / "builtin_skills"
|
||||
builtin_dir.mkdir()
|
||||
# The skill lives ONLY in global, NOT in workspace.
|
||||
(global_dir / "hello-e2e").mkdir()
|
||||
(global_dir / "hello-e2e" / "main.py").write_text(
|
||||
"print('global-tier-fix-works')"
|
||||
)
|
||||
|
||||
monkeypatch.setattr(paths, "USER_SKILLS_DIR", user_dir)
|
||||
monkeypatch.setattr(paths, "GLOBAL_SKILLS_DIR", global_dir)
|
||||
monkeypatch.setattr(paths, "MEMORIES_DIR", memories_dir)
|
||||
monkeypatch.setattr(backends, "_BUILTIN_SKILLS_DIR", builtin_dir)
|
||||
|
||||
backend = CustomSandboxBackend(root_dir=str(workspace), virtual_mode=True)
|
||||
resp = backend.execute("python3 /skills/hello-e2e/main.py")
|
||||
assert resp.exit_code == 0, resp.output
|
||||
assert "global-tier-fix-works" in resp.output
|
||||
|
||||
|
||||
# === CustomSandboxBackend._resolve_path ===
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user