m4
8376f56ab4
Adds native sandbox execution runtime, dynamic review middleware, and workspace file handling, with supporting stream events, prompt, and scope registry changes plus architecture docs.
268 lines
8.5 KiB
Python
268 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from deepagents.backends.protocol import ExecuteResponse
|
|
|
|
from EvoScientist.native_sandbox import (
|
|
NativeSandboxExecutor,
|
|
NativeSandboxUnavailable,
|
|
NativeWorkspaceBackend,
|
|
)
|
|
from EvoScientist.workspace_files import (
|
|
RootedWorkspace,
|
|
ScopedFilesystemBackend,
|
|
WorkspacePathError,
|
|
normalize_workspace_path,
|
|
)
|
|
from EvoScientist.workspace_scope import (
|
|
DeferredScopedBackend,
|
|
_RuntimeScopeConfig,
|
|
conversation_files_dir,
|
|
delete_conversation_scope,
|
|
provision_conversation_scope,
|
|
)
|
|
|
|
|
|
def test_normalize_workspace_path_is_strict():
|
|
assert normalize_workspace_path("/workspace") == ()
|
|
assert normalize_workspace_path("/workspace/reports/a.txt") == (
|
|
"reports",
|
|
"a.txt",
|
|
)
|
|
for invalid in (
|
|
"/etc/passwd",
|
|
"/workspace/../secret",
|
|
"/workspace/a//b",
|
|
"/workspace/./a",
|
|
"workspace/a",
|
|
"/workspace/a\\b",
|
|
):
|
|
with pytest.raises(WorkspacePathError):
|
|
normalize_workspace_path(invalid)
|
|
|
|
|
|
def test_scoped_filesystem_backend_complete_round_trip(tmp_path: Path):
|
|
backend = ScopedFilesystemBackend(tmp_path)
|
|
|
|
assert backend.write("/workspace/notes/a.txt", "alpha\nbeta\n").error is None
|
|
assert backend.read("/workspace/notes/a.txt").file_data == {
|
|
"content": "alpha\nbeta\n",
|
|
"encoding": "utf-8",
|
|
}
|
|
edit = backend.edit("/workspace/notes/a.txt", "beta", "gamma")
|
|
assert edit.error is None
|
|
assert edit.occurrences == 1
|
|
|
|
listing = backend.ls("/workspace/notes")
|
|
assert [item["path"] for item in listing.entries or []] == [
|
|
"/workspace/notes/a.txt"
|
|
]
|
|
assert [item["path"] for item in backend.glob("**/*.txt").matches or []] == [
|
|
"/workspace/notes/a.txt"
|
|
]
|
|
assert backend.grep("gamma").matches == [
|
|
{"path": "/workspace/notes/a.txt", "line": 2, "text": "gamma"}
|
|
]
|
|
|
|
upload = backend.upload_files([("/workspace/data.bin", b"\x00\x01")])[0]
|
|
assert upload.error is None
|
|
download = backend.download_files(["/workspace/data.bin"])[0]
|
|
assert download.error is None
|
|
assert download.content == b"\x00\x01"
|
|
|
|
|
|
def test_root_lists_workspace_namespace(tmp_path: Path):
|
|
backend = ScopedFilesystemBackend(tmp_path)
|
|
assert backend.ls("/").entries == [
|
|
{"path": "/workspace/", "is_dir": True, "size": 0}
|
|
]
|
|
|
|
|
|
def test_docx_and_unknown_binary_read_with_base64_contract(tmp_path: Path):
|
|
backend = ScopedFilesystemBackend(tmp_path)
|
|
docx = b"PK\x03\x04\x00word/document.xml"
|
|
unknown = b"custom\x00binary"
|
|
backend.upload_files(
|
|
[
|
|
("/workspace/input.docx", docx),
|
|
("/workspace/payload.custom", unknown),
|
|
]
|
|
)
|
|
|
|
assert backend.read("/workspace/input.docx").file_data == {
|
|
"content": base64.standard_b64encode(docx).decode("ascii"),
|
|
"encoding": "base64",
|
|
}
|
|
assert backend.read("/workspace/payload.custom").file_data == {
|
|
"content": base64.standard_b64encode(unknown).decode("ascii"),
|
|
"encoding": "base64",
|
|
}
|
|
|
|
|
|
def test_symlink_targets_and_parents_are_rejected(tmp_path: Path):
|
|
outside = tmp_path.parent / "outside-secret.txt"
|
|
outside.write_text("secret", encoding="utf-8")
|
|
(tmp_path / "leak.txt").symlink_to(outside)
|
|
(tmp_path / "escape").symlink_to(tmp_path.parent, target_is_directory=True)
|
|
backend = ScopedFilesystemBackend(tmp_path)
|
|
|
|
assert backend.read("/workspace/leak.txt").error
|
|
assert backend.write("/workspace/escape/new.txt", "nope").error
|
|
assert backend.download_files(["/workspace/leak.txt"])[0].error == "invalid_path"
|
|
assert backend.ls("/workspace").entries == []
|
|
|
|
|
|
def test_rooted_workspace_returns_open_verified_handle(tmp_path: Path):
|
|
workspace = RootedWorkspace(tmp_path)
|
|
workspace.replace_file("/workspace/result.txt", b"result")
|
|
handle = workspace.open_binary("/workspace/result.txt")
|
|
os.unlink(tmp_path / "result.txt")
|
|
try:
|
|
assert handle.read() == b"result"
|
|
finally:
|
|
handle.close()
|
|
|
|
|
|
def test_rooted_workspace_entry_and_recursive_delete(tmp_path: Path):
|
|
workspace = RootedWorkspace(tmp_path)
|
|
workspace.replace_file("/workspace/report/assets/app.js", b"app()")
|
|
|
|
assert workspace.entry("/workspace/report").is_dir
|
|
assert workspace.entry("/workspace/report/assets/app.js").size == 5
|
|
|
|
workspace.delete("/workspace/report")
|
|
|
|
with pytest.raises(FileNotFoundError):
|
|
workspace.entry("/workspace/report")
|
|
|
|
|
|
def test_native_backend_delegates_validated_command_to_executor(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
files = tmp_path / "files"
|
|
runtime = tmp_path / "runtime"
|
|
files.mkdir()
|
|
runtime.mkdir()
|
|
captured: dict[str, object] = {}
|
|
|
|
def fake_execute(
|
|
self,
|
|
command,
|
|
*,
|
|
timeout=None,
|
|
skip_readiness_check=False,
|
|
cancel_event=None,
|
|
):
|
|
captured.update(command=command, timeout=timeout)
|
|
return ExecuteResponse(output="ok\n", exit_code=0, truncated=False)
|
|
|
|
monkeypatch.setattr(NativeSandboxExecutor, "execute", fake_execute)
|
|
|
|
backend = NativeWorkspaceBackend(files, runtime, timeout=30)
|
|
result = backend.execute("cat probe.txt", timeout=12)
|
|
assert result.exit_code == 0
|
|
assert captured == {"command": "cat probe.txt", "timeout": 12}
|
|
|
|
blocked = backend.execute("sudo cat probe.txt")
|
|
assert blocked.exit_code == 1
|
|
assert "blocked" in blocked.output.lower()
|
|
|
|
invalid = backend.execute("printf 'bad\x00command'")
|
|
assert invalid.exit_code == 1
|
|
|
|
|
|
def test_native_backend_fails_closed_when_executor_is_unavailable(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
files = tmp_path / "files"
|
|
runtime = tmp_path / "runtime"
|
|
files.mkdir()
|
|
runtime.mkdir()
|
|
|
|
def unavailable(*args, **kwargs):
|
|
raise NativeSandboxUnavailable("private deployment detail")
|
|
|
|
monkeypatch.setattr(NativeSandboxExecutor, "execute", unavailable)
|
|
result = NativeWorkspaceBackend(files, runtime, timeout=30).execute("echo ok")
|
|
|
|
assert result.exit_code == 125
|
|
assert "private deployment detail" not in result.output
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deferred_backend_preserves_async_executor_cancellation_path(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
):
|
|
proxy = DeferredScopedBackend(
|
|
_RuntimeScopeConfig(
|
|
scope_id="00000000-0000-0000-0000-000000000001",
|
|
owner_id="00000000-0000-0000-0000-000000000002",
|
|
thread_id="thread",
|
|
deployment_id="deployment",
|
|
),
|
|
dangerous=False,
|
|
)
|
|
captured: dict[str, object] = {}
|
|
|
|
class Delegate:
|
|
async def aexecute(self, command, *, timeout=None):
|
|
captured.update(command=command, timeout=timeout)
|
|
return ExecuteResponse(output="async", exit_code=0, truncated=False)
|
|
|
|
monkeypatch.setattr(proxy, "_delegate", lambda: Delegate())
|
|
result = await proxy.aexecute("sleep 1", timeout=7)
|
|
|
|
assert result.output == "async"
|
|
assert captured == {"command": "sleep 1", "timeout": 7}
|
|
|
|
|
|
def test_scope_delete_is_idempotent_and_removes_directory(tmp_path: Path):
|
|
thread_id = "0f88db64-720e-4f88-ac92-ea9a76b45596"
|
|
deployment_id = "test-workspace-delete"
|
|
record = provision_conversation_scope(
|
|
thread_id,
|
|
deployment_id=deployment_id,
|
|
workspace_root=tmp_path,
|
|
)
|
|
scope_root = tmp_path / ".evoscientist" / "conversations" / record.scope_id
|
|
(scope_root / "files" / "result.txt").write_text("done", encoding="utf-8")
|
|
|
|
deleted = delete_conversation_scope(
|
|
thread_id,
|
|
deployment_id=deployment_id,
|
|
workspace_root=tmp_path,
|
|
)
|
|
assert deleted.state == "deleted"
|
|
assert not scope_root.exists()
|
|
assert (
|
|
delete_conversation_scope(
|
|
thread_id,
|
|
deployment_id=deployment_id,
|
|
workspace_root=tmp_path,
|
|
).state
|
|
== "deleted"
|
|
)
|
|
|
|
|
|
def test_deleted_scope_can_be_reprovisioned_for_create_retry(tmp_path: Path):
|
|
thread_id = "a224305c-32ae-43c5-bdae-76b52912fb37"
|
|
deployment_id = "test-workspace-retry"
|
|
original = provision_conversation_scope(
|
|
thread_id, deployment_id=deployment_id, workspace_root=tmp_path
|
|
)
|
|
delete_conversation_scope(
|
|
thread_id, deployment_id=deployment_id, workspace_root=tmp_path
|
|
)
|
|
|
|
retried = provision_conversation_scope(
|
|
thread_id, deployment_id=deployment_id, workspace_root=tmp_path
|
|
)
|
|
|
|
assert retried.scope_id == original.scope_id
|
|
assert retried.state == "draft"
|
|
assert conversation_files_dir(retried.scope_id, tmp_path).is_dir()
|