284 lines
10 KiB
Python
284 lines
10 KiB
Python
"""An in-memory implementation of the repository execution boundary."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Iterable, Sequence
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
import subprocess
|
|
|
|
from band_aider.judgement.repo_exec import RepoExec
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MemoryCommandInvocation:
|
|
"""One explicit command request observed by an in-memory executor."""
|
|
|
|
command: tuple[str, ...]
|
|
cwd: Path | None
|
|
env: dict[str, str] | None
|
|
timeout: int | None
|
|
unset_env: tuple[str, ...]
|
|
|
|
|
|
class MemoryRepoExec(RepoExec):
|
|
"""Store project paths and command results without host IO or process execution."""
|
|
|
|
def __init__(self, root: Path = Path("/testbed")) -> None:
|
|
self.root = root
|
|
self._files: dict[Path, bytes] = {}
|
|
self._directories: set[Path] = {root}
|
|
self.commands: list[list[str]] = []
|
|
self._queued_results: list[subprocess.CompletedProcess[str] | BaseException] = []
|
|
|
|
def queue_result(self, result: subprocess.CompletedProcess[str] | BaseException) -> None:
|
|
"""Queue one stationary CLI result for the next in-memory command."""
|
|
|
|
self._queued_results.append(result)
|
|
|
|
@property
|
|
def preserves_host_environment(self) -> bool:
|
|
"""Report that this executor never inherits the host environment."""
|
|
|
|
return False
|
|
|
|
def path_in_exec(self, path: Path) -> str:
|
|
"""Return the lexical project path used by the in-memory executor."""
|
|
|
|
return str(self._path(path))
|
|
|
|
def decorate_cli(self, command: Sequence[str], cwd: Path | None = None) -> list[str]:
|
|
"""Return an unmodified command because no executor prefix is needed."""
|
|
|
|
del cwd
|
|
return list(command)
|
|
|
|
def python_bin(self) -> str:
|
|
"""Return a stationary executable token without inspecting the host."""
|
|
|
|
return "python"
|
|
|
|
def run_cli(
|
|
self,
|
|
command: Sequence[str],
|
|
*,
|
|
cwd: Path | None = None,
|
|
env: dict[str, str] | None = None,
|
|
capture_output: bool = True,
|
|
text: bool = True,
|
|
check: bool = False,
|
|
timeout: int | None = None,
|
|
input: bytes | str | None = None,
|
|
unset_env: Iterable[str] = (),
|
|
) -> subprocess.CompletedProcess:
|
|
"""Return a successful stationary command result without launching a process."""
|
|
|
|
del cwd, env, capture_output, text, check, timeout, input, unset_env
|
|
recorded = list(command)
|
|
self.commands.append(recorded)
|
|
if self._queued_results:
|
|
queued = self._queued_results.pop(0)
|
|
if isinstance(queued, BaseException):
|
|
raise queued
|
|
return queued
|
|
if len(recorded) >= 3 and recorded[1:3] == ["-m", "pyright"]:
|
|
return subprocess.CompletedProcess(
|
|
recorded,
|
|
0,
|
|
'{"generalDiagnostics": []}',
|
|
"",
|
|
)
|
|
if len(recorded) >= 4 and recorded[1] == "-c":
|
|
candidate_path = Path(recorded[-1])
|
|
source = self._files.get(self._path(candidate_path))
|
|
if source is not None:
|
|
try:
|
|
compile(source.decode("utf-8"), str(candidate_path), "exec")
|
|
except SyntaxError as error:
|
|
return subprocess.CompletedProcess(
|
|
recorded,
|
|
1,
|
|
"",
|
|
f"SyntaxError: {error.msg}",
|
|
)
|
|
if len(recorded) >= 3 and recorded[1] == "-c":
|
|
script = recorded[2]
|
|
if "packages_distributions" in script:
|
|
return subprocess.CompletedProcess(recorded, 0, "{}", "")
|
|
if "stdlib_module_names" in script:
|
|
return subprocess.CompletedProcess(recorded, 0, "[]", "")
|
|
return subprocess.CompletedProcess(recorded, 0, "", "")
|
|
|
|
def read_text(self, path: Path, *, encoding: str = "utf-8") -> str:
|
|
"""Return stored text for one project path."""
|
|
|
|
return self._files[self._path(path)].decode(encoding)
|
|
|
|
def read_text_many(self, paths: Sequence[Path], *, encoding: str = "utf-8") -> dict[Path, str]:
|
|
"""Return stored text for each supplied project path."""
|
|
|
|
return {path: self.read_text(path, encoding=encoding) for path in paths}
|
|
|
|
def write_text(self, path: Path, content: str, *, encoding: str = "utf-8") -> None:
|
|
"""Store text and create its lexical parent directories."""
|
|
|
|
self.write_bytes(path, content.encode(encoding))
|
|
|
|
def read_bytes(self, path: Path) -> bytes:
|
|
"""Return stored bytes for one project path."""
|
|
|
|
return self._files[self._path(path)]
|
|
|
|
def write_bytes(self, path: Path, content: bytes) -> None:
|
|
"""Store bytes and create their lexical parent directories."""
|
|
|
|
target = self._path(path)
|
|
self.mkdir(target.parent)
|
|
self._files[target] = content
|
|
|
|
def exists(self, path: Path) -> bool:
|
|
"""Return whether a lexical file or directory exists in memory."""
|
|
|
|
target = self._path(path)
|
|
return target in self._files or target in self._directories
|
|
|
|
def is_file(self, path: Path) -> bool:
|
|
"""Return whether a lexical path stores bytes."""
|
|
|
|
return self._path(path) in self._files
|
|
|
|
def is_dir(self, path: Path) -> bool:
|
|
"""Return whether a lexical path is a directory."""
|
|
|
|
return self._path(path) in self._directories
|
|
|
|
def append_text(self, path: Path, content: str, *, encoding: str = "utf-8") -> None:
|
|
"""Append text to one in-memory file."""
|
|
|
|
target = self._path(path)
|
|
existing = self._files.get(target, b"")
|
|
self.write_bytes(target, existing + content.encode(encoding))
|
|
|
|
def delete(self, path: Path, *, missing_ok: bool = False) -> None:
|
|
"""Remove a lexical file or directory tree from memory."""
|
|
|
|
target = self._path(path)
|
|
matching_files = [item for item in self._files if item == target or target in item.parents]
|
|
matching_directories = [item for item in self._directories if item == target or target in item.parents]
|
|
if not matching_files and not matching_directories and not missing_ok:
|
|
raise FileNotFoundError(target)
|
|
for item in matching_files:
|
|
del self._files[item]
|
|
for item in matching_directories:
|
|
self._directories.discard(item)
|
|
|
|
def mkdir(self, path: Path, *, parents: bool = True, exist_ok: bool = True) -> None:
|
|
"""Create lexical directories without touching the host filesystem."""
|
|
|
|
del exist_ok
|
|
target = self._path(path)
|
|
current = target
|
|
while True:
|
|
self._directories.add(current)
|
|
if not parents or current.parent == current:
|
|
return
|
|
current = current.parent
|
|
|
|
def list_directory(self, path: Path) -> list[Path]:
|
|
"""Return immediate stored children in deterministic order."""
|
|
|
|
target = self._path(path)
|
|
children = [
|
|
item
|
|
for item in (*self._files, *self._directories)
|
|
if item.parent == target and item != target
|
|
]
|
|
return sorted(children)
|
|
|
|
def replace(self, source: Path, target: Path) -> None:
|
|
"""Move one stored file to its replacement location."""
|
|
|
|
source_path = self._path(source)
|
|
target_path = self._path(target)
|
|
self.mkdir(target_path.parent)
|
|
self._files[target_path] = self._files.pop(source_path)
|
|
|
|
def list_files(self, root: Path, patterns: Iterable[str], *, recursive: bool) -> list[Path]:
|
|
"""Return matching stored files in deterministic order."""
|
|
|
|
root_path = self._path(root)
|
|
requested_patterns = tuple(patterns)
|
|
matches: list[Path] = []
|
|
for path in self._files:
|
|
if path != root_path and root_path not in path.parents:
|
|
continue
|
|
relative = path.relative_to(root_path)
|
|
if not recursive and len(relative.parts) > 1:
|
|
continue
|
|
if not requested_patterns or any(relative.match(pattern) for pattern in requested_patterns):
|
|
matches.append(path)
|
|
return sorted(matches)
|
|
|
|
def move(self, source: Path, target: Path) -> None:
|
|
"""Move one stored file."""
|
|
|
|
self.replace(source, target)
|
|
|
|
def move_many(self, moves: Sequence[tuple[Path, Path]]) -> None:
|
|
"""Move each supplied stored file in order."""
|
|
|
|
for source, target in moves:
|
|
self.move(source, target)
|
|
|
|
def _path(self, path: Path) -> Path:
|
|
"""Normalize a relative path into the fixed in-memory root."""
|
|
|
|
value = Path(str(path))
|
|
return value if value.is_absolute() else self.root / value
|
|
|
|
|
|
class RecordingMemoryRepoExec(MemoryRepoExec):
|
|
"""Record exact in-memory command requests while retaining stationary results."""
|
|
|
|
def __init__(self, root: Path = Path("/testbed")) -> None:
|
|
"""Initialize the memory repository and its command observation ledger."""
|
|
|
|
super().__init__(root)
|
|
self.invocations: list[MemoryCommandInvocation] = []
|
|
|
|
def run_cli(
|
|
self,
|
|
command: Sequence[str],
|
|
*,
|
|
cwd: Path | None = None,
|
|
env: dict[str, str] | None = None,
|
|
capture_output: bool = True,
|
|
text: bool = True,
|
|
check: bool = False,
|
|
timeout: int | None = None,
|
|
input: bytes | str | None = None,
|
|
unset_env: Iterable[str] = (),
|
|
) -> subprocess.CompletedProcess:
|
|
"""Record one call and return the parent's queued stationary result."""
|
|
|
|
unset_names = tuple(unset_env)
|
|
self.invocations.append(
|
|
MemoryCommandInvocation(
|
|
command=tuple(command),
|
|
cwd=cwd,
|
|
env=env,
|
|
timeout=timeout,
|
|
unset_env=unset_names,
|
|
)
|
|
)
|
|
return super().run_cli(
|
|
command,
|
|
cwd=cwd,
|
|
env=env,
|
|
capture_output=capture_output,
|
|
text=text,
|
|
check=check,
|
|
timeout=timeout,
|
|
input=input,
|
|
unset_env=unset_names,
|
|
)
|