mirror of
https://github.com/abhigyanpatwari/GitNexus.git
synced 2026-10-16 04:28:11 +00:00
c723d420d5
trusted_gitnexus_runtime_mounts and sanitized_graph both hand the ~290 MiB graph index through TaskAssetSnapshot.materialize(), which reflinks into each arm clone or falls back to a buffered copy under a 16 MiB budget. Neither ext4 (CI runners) nor 9p (this container) support FICLONE, so every real materialization fell to the buffered path and blew the budget instantly (CI run 29750271566). Raises MAX_BUFFERED_FALLBACK_BYTES to 512 MiB - well above the real index size, still a full 4x below MAX_TASK_ASSET_BYTES so a genuinely oversized declaration still fails closed.
412 lines
17 KiB
Python
412 lines
17 KiB
Python
"""Copy-on-write task-asset snapshot contracts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import stat
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from workflow_bench.proposer_sandbox import SandboxError
|
|
from workflow_bench.oracle_assets import TaskOracleSnapshot
|
|
from workflow_bench.runner_tasks import resolve_task_bindings
|
|
from workflow_bench.task_assets import TaskAssetCache, stage_task_assets
|
|
from workflow_bench import task_assets
|
|
|
|
|
|
SHA = "a" * 40
|
|
|
|
|
|
def _repo_and_task(tmp_path: Path, files: dict[str, bytes]) -> tuple[Path, dict[str, object]]:
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
for relative, payload in files.items():
|
|
target = repo / relative
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(payload)
|
|
return repo, {"sandbox_copy": sorted(files), "sandbox_dependencies": []}
|
|
|
|
|
|
def _git(repo: Path, *args: str) -> str:
|
|
return subprocess.run(
|
|
["git", "-C", str(repo), *args],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
).stdout.strip()
|
|
|
|
|
|
def test_snapshot_is_reused_frozen_and_isolates_arm_writes(monkeypatch, tmp_path: Path) -> None:
|
|
repo, task = _repo_and_task(tmp_path, {"assets/index": b"original"})
|
|
clone_a = tmp_path / "clone-a"
|
|
clone_b = tmp_path / "clone-b"
|
|
clone_a.mkdir()
|
|
clone_b.mkdir()
|
|
reflink_calls: list[tuple[int, int]] = []
|
|
|
|
def fake_reflink(source: int, destination: int) -> bool:
|
|
reflink_calls.append((source, destination))
|
|
while chunk := os.read(source, 1024):
|
|
os.write(destination, chunk)
|
|
return True
|
|
|
|
monkeypatch.setattr(task_assets, "_try_reflink", fake_reflink)
|
|
monkeypatch.setattr(task_assets, "MAX_BUFFERED_FALLBACK_BYTES", 0)
|
|
|
|
with TaskAssetCache(tmp_path / "cache") as cache:
|
|
snapshot = cache.prepare(task, repo=repo, resolved_sha=SHA)
|
|
assert cache.prepare(task, repo=repo, resolved_sha=SHA) is snapshot
|
|
snapshot_file = snapshot.root / "sandbox-copy" / "assets" / "index"
|
|
assert stat.S_IMODE(snapshot.root.stat().st_mode) == 0o500
|
|
assert stat.S_IMODE(snapshot_file.stat().st_mode) == 0o400
|
|
|
|
stage_task_assets(task, repo=repo, clone=clone_a, snapshot=snapshot)
|
|
stage_task_assets(task, repo=repo, clone=clone_b, snapshot=snapshot)
|
|
(clone_a / "assets" / "index").write_bytes(b"arm-a")
|
|
|
|
assert snapshot_file.read_bytes() == b"original"
|
|
assert (clone_b / "assets" / "index").read_bytes() == b"original"
|
|
assert (clone_a / "assets" / "index").stat().st_ino != snapshot_file.stat().st_ino
|
|
assert (clone_b / "assets" / "index").stat().st_ino != snapshot_file.stat().st_ino
|
|
assert len(reflink_calls) == 2
|
|
|
|
|
|
def test_snapshot_digest_binds_content_declaration_repo_and_sha(tmp_path: Path) -> None:
|
|
repo, task = _repo_and_task(tmp_path, {"one": b"1", "two": b"2"})
|
|
with TaskAssetCache(tmp_path / "cache-a") as cache:
|
|
original = cache.prepare(task, repo=repo, resolved_sha=SHA)
|
|
other_sha = cache.prepare(task, repo=repo, resolved_sha="b" * 40)
|
|
reordered = cache.prepare(
|
|
{"sandbox_copy": ["two", "one"]},
|
|
repo=repo,
|
|
resolved_sha=SHA,
|
|
)
|
|
with TaskAssetCache(tmp_path / "cache-b") as cache:
|
|
identical = cache.prepare(task, repo=repo, resolved_sha=SHA)
|
|
(repo / "one").write_bytes(b"changed")
|
|
with TaskAssetCache(tmp_path / "cache-c") as cache:
|
|
changed = cache.prepare(task, repo=repo, resolved_sha=SHA)
|
|
|
|
assert identical.digest == original.digest
|
|
assert identical.manifest_digest == original.manifest_digest
|
|
assert other_sha.digest != original.digest
|
|
assert other_sha.manifest_digest == original.manifest_digest
|
|
assert reordered.digest != original.digest
|
|
assert changed.digest != original.digest
|
|
assert changed.manifest_digest != original.manifest_digest
|
|
|
|
|
|
def test_small_assets_use_a_bounded_buffered_fallback(monkeypatch, tmp_path: Path) -> None:
|
|
repo, task = _repo_and_task(tmp_path, {"first": b"abc", "second": b"def"})
|
|
clone = tmp_path / "clone"
|
|
clone.mkdir()
|
|
monkeypatch.setattr(task_assets, "_try_reflink", lambda *_args: False)
|
|
monkeypatch.setattr(task_assets, "MAX_BUFFERED_FALLBACK_BYTES", 6)
|
|
|
|
with TaskAssetCache(tmp_path / "cache") as cache:
|
|
snapshot = cache.prepare(task, repo=repo, resolved_sha=SHA)
|
|
snapshot.materialize(clone)
|
|
|
|
assert (clone / "first").read_bytes() == b"abc"
|
|
assert (clone / "second").read_bytes() == b"def"
|
|
|
|
|
|
def test_default_buffered_fallback_budget_covers_a_realistic_large_asset(
|
|
monkeypatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
# 20 MiB exceeds the old 16 MiB default but must fit comfortably under
|
|
# the current default, proving the real (non-monkeypatched) budget
|
|
# constant is sized for a realistic large sandbox_copy asset such as the
|
|
# harness's own pre-built graph index, not just tiny fixtures.
|
|
payload = os.urandom(20 * 1024 * 1024)
|
|
repo, task = _repo_and_task(tmp_path, {"large": payload})
|
|
clone = tmp_path / "clone"
|
|
clone.mkdir()
|
|
monkeypatch.setattr(task_assets, "_try_reflink", lambda *_args: False)
|
|
|
|
with TaskAssetCache(tmp_path / "cache") as cache:
|
|
snapshot = cache.prepare(task, repo=repo, resolved_sha=SHA)
|
|
snapshot.materialize(clone)
|
|
|
|
assert (clone / "large").read_bytes() == payload
|
|
|
|
|
|
def test_large_asset_without_reflink_fails_before_publish_and_cleans_staging(
|
|
monkeypatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
repo, task = _repo_and_task(tmp_path, {"large": b"12345"})
|
|
clone = tmp_path / "clone"
|
|
clone.mkdir()
|
|
monkeypatch.setattr(task_assets, "_try_reflink", lambda *_args: False)
|
|
monkeypatch.setattr(task_assets, "MAX_BUFFERED_FALLBACK_BYTES", 4)
|
|
|
|
with TaskAssetCache(tmp_path / "cache") as cache:
|
|
snapshot = cache.prepare(task, repo=repo, resolved_sha=SHA)
|
|
with pytest.raises(SandboxError, match="cannot reflink"):
|
|
snapshot.materialize(clone)
|
|
|
|
assert not (clone / "large").exists()
|
|
assert not list(tmp_path.glob(".wfbench-assets-*"))
|
|
|
|
|
|
@pytest.mark.skipif(os.name == "nt", reason="symlink and FIFO contracts are POSIX-only")
|
|
@pytest.mark.parametrize("kind", ["symlink", "fifo"])
|
|
def test_snapshot_rejects_links_and_special_files(tmp_path: Path, kind: str) -> None:
|
|
repo = tmp_path / "repo"
|
|
assets = repo / "assets"
|
|
assets.mkdir(parents=True)
|
|
if kind == "symlink":
|
|
(repo / "outside").write_text("secret")
|
|
(assets / "bad").symlink_to(repo / "outside")
|
|
else:
|
|
os.mkfifo(assets / "bad")
|
|
|
|
with TaskAssetCache(tmp_path / "cache") as cache:
|
|
with pytest.raises(SandboxError, match="regular files and directories|symlink"):
|
|
cache.prepare({"sandbox_copy": ["assets"]}, repo=repo, resolved_sha=SHA)
|
|
|
|
|
|
def test_snapshot_rejects_a_file_mutated_during_capture(monkeypatch, tmp_path: Path) -> None:
|
|
repo, task = _repo_and_task(tmp_path, {"asset": b"original"})
|
|
original_read = task_assets._read_source_chunk
|
|
changed = False
|
|
|
|
def mutate_after_first_read(descriptor: int, size: int) -> bytes:
|
|
nonlocal changed
|
|
chunk = original_read(descriptor, size)
|
|
if chunk and not changed:
|
|
changed = True
|
|
with (repo / "asset").open("ab") as source:
|
|
source.write(b"!")
|
|
return chunk
|
|
|
|
monkeypatch.setattr(task_assets, "_read_source_chunk", mutate_after_first_read)
|
|
with TaskAssetCache(tmp_path / "cache") as cache:
|
|
with pytest.raises(SandboxError, match="changed while snapshotting"):
|
|
cache.prepare(task, repo=repo, resolved_sha=SHA)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("limit", "value", "task_files", "message"),
|
|
[
|
|
("MAX_TASK_ASSET_ENTRIES", 1, {"nested/file": b"x"}, "entry limit"),
|
|
("MAX_TASK_ASSET_PATH_BYTES", 4, {"long-name": b"x"}, "path byte limit"),
|
|
("MAX_TASK_ASSET_BYTES", 4, {"asset": b"12345"}, "total byte limit"),
|
|
],
|
|
)
|
|
def test_snapshot_enforces_hard_walk_limits(
|
|
monkeypatch,
|
|
tmp_path: Path,
|
|
limit: str,
|
|
value: int,
|
|
task_files: dict[str, bytes],
|
|
message: str,
|
|
) -> None:
|
|
repo, task = _repo_and_task(tmp_path, task_files)
|
|
monkeypatch.setattr(task_assets, limit, value)
|
|
with TaskAssetCache(tmp_path / "cache") as cache:
|
|
with pytest.raises(SandboxError, match=message):
|
|
cache.prepare(task, repo=repo, resolved_sha=SHA)
|
|
|
|
|
|
def test_snapshot_rejects_overlapping_declarations(tmp_path: Path) -> None:
|
|
repo, _task = _repo_and_task(tmp_path, {"assets/index": b"index"})
|
|
with TaskAssetCache(tmp_path / "cache") as cache:
|
|
with pytest.raises(SandboxError, match="overlap"):
|
|
cache.prepare(
|
|
{"sandbox_copy": ["assets", "assets/index"]},
|
|
repo=repo,
|
|
resolved_sha=SHA,
|
|
)
|
|
|
|
|
|
def test_directory_materialization_removes_stale_children_exactly(tmp_path: Path) -> None:
|
|
repo, task = _repo_and_task(tmp_path, {"assets/current": b"captured"})
|
|
task["sandbox_copy"] = ["assets"]
|
|
clone = tmp_path / "clone"
|
|
(clone / "assets" / "nested").mkdir(parents=True)
|
|
(clone / "assets" / "current").write_bytes(b"old")
|
|
(clone / "assets" / "stale").write_bytes(b"stale")
|
|
(clone / "assets" / "nested" / "stale").write_bytes(b"stale")
|
|
|
|
with TaskAssetCache(tmp_path / "cache") as cache:
|
|
snapshot = cache.prepare(task, repo=repo, resolved_sha=SHA)
|
|
snapshot.materialize(clone)
|
|
|
|
assert sorted(path.relative_to(clone).as_posix() for path in clone.rglob("*")) == [
|
|
"assets",
|
|
"assets/current",
|
|
]
|
|
assert (clone / "assets" / "current").read_bytes() == b"captured"
|
|
|
|
|
|
@pytest.mark.parametrize("source_kind", ["file", "directory"])
|
|
def test_materialization_replaces_file_directory_type_conflicts(
|
|
tmp_path: Path,
|
|
source_kind: str,
|
|
) -> None:
|
|
if source_kind == "file":
|
|
repo, task = _repo_and_task(tmp_path, {"asset": b"file"})
|
|
else:
|
|
repo, task = _repo_and_task(tmp_path, {"asset/child": b"directory"})
|
|
task["sandbox_copy"] = ["asset"]
|
|
clone = tmp_path / "clone"
|
|
clone.mkdir()
|
|
if source_kind == "file":
|
|
(clone / "asset").mkdir()
|
|
(clone / "asset" / "stale").write_bytes(b"stale")
|
|
else:
|
|
(clone / "asset").write_bytes(b"stale file")
|
|
|
|
with TaskAssetCache(tmp_path / "cache") as cache:
|
|
cache.prepare(task, repo=repo, resolved_sha=SHA).materialize(clone)
|
|
|
|
if source_kind == "file":
|
|
assert (clone / "asset").is_file()
|
|
assert (clone / "asset").read_bytes() == b"file"
|
|
else:
|
|
assert (clone / "asset").is_dir()
|
|
assert (clone / "asset" / "child").read_bytes() == b"directory"
|
|
|
|
|
|
@pytest.mark.skipif(os.name == "nt", reason="symlink containment is POSIX-only")
|
|
def test_exact_tree_removal_does_not_follow_stale_child_symlinks(tmp_path: Path) -> None:
|
|
repo, task = _repo_and_task(tmp_path, {"assets/current": b"captured"})
|
|
task["sandbox_copy"] = ["assets"]
|
|
clone = tmp_path / "clone"
|
|
outside = tmp_path / "outside"
|
|
(clone / "assets").mkdir(parents=True)
|
|
outside.mkdir()
|
|
(outside / "canary").write_bytes(b"outside")
|
|
(clone / "assets" / "stale-link").symlink_to(outside, target_is_directory=True)
|
|
|
|
with TaskAssetCache(tmp_path / "cache") as cache:
|
|
cache.prepare(task, repo=repo, resolved_sha=SHA).materialize(clone)
|
|
|
|
assert (outside / "canary").read_bytes() == b"outside"
|
|
assert not (clone / "assets" / "stale-link").exists()
|
|
|
|
|
|
def test_dependency_snapshot_mounts_bound_bytes_and_rejects_later_live_drift(tmp_path: Path) -> None:
|
|
repo, _ = _repo_and_task(tmp_path, {"dependency/package.json": b'{"version":1}'})
|
|
task = {
|
|
"sandbox_copy": [],
|
|
"sandbox_dependencies": [{"source": "dependency", "target": "node_modules/dependency"}],
|
|
}
|
|
clone = tmp_path / "clone"
|
|
clone.mkdir()
|
|
|
|
with TaskAssetCache(tmp_path / "cache-a") as cache:
|
|
snapshot = cache.prepare(task, repo=repo, resolved_sha=SHA)
|
|
binding = snapshot.dependency_binding
|
|
mounts = stage_task_assets(task, repo=repo, clone=clone, snapshot=snapshot)
|
|
assert mounts[0].source != (repo / "dependency").resolve()
|
|
assert (mounts[0].source / "package.json").read_bytes() == b'{"version":1}'
|
|
|
|
(repo / "dependency" / "package.json").write_bytes(b'{"version":2}')
|
|
with TaskAssetCache(tmp_path / "cache-b") as cache:
|
|
with pytest.raises(SandboxError, match="changed after task binding"):
|
|
cache.prepare(
|
|
task,
|
|
repo=repo,
|
|
resolved_sha=SHA,
|
|
expected_dependency_binding=binding,
|
|
)
|
|
|
|
|
|
def test_dependency_content_and_manifest_digests_bind_distinct_contracts(tmp_path: Path) -> None:
|
|
repo, _ = _repo_and_task(tmp_path, {"dependency/file": b"one"})
|
|
task = {
|
|
"sandbox_dependencies": [{"source": "dependency", "target": "dependency"}],
|
|
}
|
|
retargeted = {
|
|
"sandbox_dependencies": [{"source": "dependency", "target": "vendor/dependency"}],
|
|
}
|
|
with TaskAssetCache(tmp_path / "cache-a") as cache:
|
|
original = cache.prepare(task, repo=repo, resolved_sha=SHA)
|
|
changed_target = cache.prepare(retargeted, repo=repo, resolved_sha=SHA)
|
|
|
|
assert changed_target.dependency_content_digest == original.dependency_content_digest
|
|
assert changed_target.dependency_manifest_digest != original.dependency_manifest_digest
|
|
(repo / "dependency" / "file").write_bytes(b"two")
|
|
with TaskAssetCache(tmp_path / "cache-b") as cache:
|
|
changed_content = cache.prepare(task, repo=repo, resolved_sha=SHA)
|
|
assert changed_content.dependency_content_digest != original.dependency_content_digest
|
|
assert changed_content.dependency_manifest_digest != original.dependency_manifest_digest
|
|
|
|
|
|
@pytest.mark.skipif(os.name == "nt", reason="dependency symlink fixtures are POSIX-only")
|
|
def test_dependency_snapshot_preserves_internal_symlinks_and_executable_files(tmp_path: Path) -> None:
|
|
repo, _ = _repo_and_task(tmp_path, {"dependency/package/bin": b"#!/bin/sh\nexit 0\n"})
|
|
executable = repo / "dependency" / "package" / "bin"
|
|
executable.chmod(0o755)
|
|
(repo / "dependency" / ".bin").mkdir()
|
|
(repo / "dependency" / ".bin" / "tool").symlink_to("../package/bin")
|
|
task = {
|
|
"sandbox_dependencies": [{"source": "dependency", "target": "dependency"}],
|
|
}
|
|
clone = tmp_path / "clone"
|
|
clone.mkdir()
|
|
|
|
with TaskAssetCache(tmp_path / "cache") as cache:
|
|
snapshot = cache.prepare(task, repo=repo, resolved_sha=SHA)
|
|
mount = stage_task_assets(task, repo=repo, clone=clone, snapshot=snapshot)[0]
|
|
link = mount.source / ".bin" / "tool"
|
|
captured_executable = mount.source / "package" / "bin"
|
|
assert link.is_symlink()
|
|
assert os.readlink(link) == "../package/bin"
|
|
assert stat.S_IMODE(captured_executable.stat().st_mode) == 0o500
|
|
|
|
|
|
@pytest.mark.skipif(os.name == "nt", reason="dependency symlink fixtures are POSIX-only")
|
|
def test_dependency_snapshot_rejects_links_that_escape_the_sandbox_workspace(tmp_path: Path) -> None:
|
|
repo, _ = _repo_and_task(tmp_path, {"dependency/kept": b"payload"})
|
|
(repo / "dependency" / "escape").symlink_to("../../../../outside")
|
|
task = {
|
|
"sandbox_dependencies": [{"source": "dependency", "target": "dependency"}],
|
|
}
|
|
|
|
with TaskAssetCache(tmp_path / "cache") as cache:
|
|
with pytest.raises(SandboxError, match="escapes the sandbox workspace"):
|
|
cache.prepare(task, repo=repo, resolved_sha=SHA)
|
|
|
|
|
|
def test_resolved_task_binding_carries_dependency_digests_and_rejects_live_drift(tmp_path: Path) -> None:
|
|
repo, _ = _repo_and_task(tmp_path, {"dependency/package.json": b'{"version":1}'})
|
|
_git(repo, "init", "--quiet")
|
|
_git(repo, "config", "user.name", "Workflow Bench Test")
|
|
_git(repo, "config", "user.email", "workflow-bench@example.invalid")
|
|
_git(repo, "add", ".")
|
|
_git(repo, "commit", "--quiet", "-m", "fixture")
|
|
task = {
|
|
"id": "dependency-binding",
|
|
"class": "test",
|
|
"repo": str(repo),
|
|
"prompt": "inspect dependency",
|
|
"verify": "true",
|
|
"sandbox_copy": [],
|
|
"sandbox_dependencies": [{"source": "dependency", "target": "dependency"}],
|
|
}
|
|
oracle = TaskOracleSnapshot(
|
|
command="true",
|
|
command_digest="1" * 64,
|
|
manifest_digest="2" * 64,
|
|
digest="3" * 64,
|
|
files=(),
|
|
)
|
|
with TaskAssetCache(tmp_path / "binding-cache") as cache:
|
|
binding = resolve_task_bindings(
|
|
[task],
|
|
oracle_snapshots=[oracle],
|
|
task_asset_cache=cache,
|
|
)[0]
|
|
|
|
assert len(binding["sandbox_dependency_content_digest"]) == 64
|
|
assert len(binding["sandbox_dependency_manifest_digest"]) == 64
|
|
(repo / "dependency" / "package.json").write_bytes(b'{"version":2}')
|
|
with pytest.raises(ValueError, match="definition drifted"):
|
|
resolve_task_bindings([task], [binding], oracle_snapshots=[oracle])
|