336 lines
12 KiB
Python
336 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Git workspace isolation, scope fingerprinting, and patch transport."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import hashlib
|
|
import mimetypes
|
|
import os
|
|
import subprocess
|
|
from collections.abc import Mapping, Sequence
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from mmo_util import atomic_write_bytes
|
|
|
|
|
|
class WorkspaceTargetNotGit(RuntimeError):
|
|
"""A writable worker target has no Git repository to isolate."""
|
|
|
|
|
|
def _git(
|
|
cwd: Path, *args: str, env: Mapping[str, str] | None = None
|
|
) -> subprocess.CompletedProcess[bytes]:
|
|
process_env = os.environ.copy()
|
|
if env:
|
|
process_env.update(env)
|
|
return subprocess.run(
|
|
["git", "-C", str(cwd), *args],
|
|
stdin=subprocess.DEVNULL,
|
|
capture_output=True,
|
|
check=False,
|
|
env=process_env,
|
|
)
|
|
|
|
|
|
def _git_root(cwd: Path) -> Path | None:
|
|
result = _git(cwd, "rev-parse", "--show-toplevel")
|
|
if result.returncode != 0:
|
|
return None
|
|
root = Path(os.fsdecode(result.stdout.removesuffix(b"\n"))).resolve()
|
|
return root if root.is_dir() else None
|
|
|
|
|
|
def _content_fingerprint(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
try:
|
|
status = path.lstat()
|
|
except FileNotFoundError:
|
|
return "missing"
|
|
digest.update(f"{status.st_mode:o}\0{status.st_size}\0".encode())
|
|
if path.is_symlink():
|
|
digest.update(os.fsencode(os.readlink(path)))
|
|
elif path.is_file():
|
|
with path.open("rb") as handle:
|
|
while chunk := handle.read(1024 * 1024):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _scope_fingerprints(repo_root: Path, cwd: Path, scopes: Sequence[str]) -> dict[str, Any]:
|
|
repo_scopes: list[str] = []
|
|
for scope in scopes:
|
|
absolute = (cwd / scope).resolve(strict=False)
|
|
try:
|
|
repo_scopes.append(absolute.relative_to(repo_root).as_posix())
|
|
except ValueError as exc:
|
|
raise ValueError(f"write scope escapes Git repository: {scope}") from exc
|
|
listed = _git(
|
|
repo_root,
|
|
"ls-files",
|
|
"-co",
|
|
"--exclude-standard",
|
|
"-z",
|
|
"--",
|
|
*repo_scopes,
|
|
)
|
|
if listed.returncode != 0:
|
|
raise RuntimeError(
|
|
"unable to fingerprint write scope: "
|
|
+ listed.stderr.decode("utf-8", errors="replace")[-2000:]
|
|
)
|
|
paths = sorted(
|
|
{os.fsdecode(value) for value in listed.stdout.split(b"\0") if value} | set(repo_scopes)
|
|
)
|
|
return {
|
|
"repo_root": str(repo_root),
|
|
"scope_roots": repo_scopes,
|
|
"paths": {relative: _content_fingerprint(repo_root / relative) for relative in paths},
|
|
}
|
|
|
|
|
|
def create_isolated_worktree(
|
|
canonical_cwd: Path,
|
|
directory: Path,
|
|
scopes: Sequence[str],
|
|
attachments: Sequence[str],
|
|
) -> dict[str, Any]:
|
|
repo_root = _git_root(canonical_cwd)
|
|
if repo_root is None:
|
|
raise WorkspaceTargetNotGit("writable MCP workers require a Git worktree target")
|
|
try:
|
|
relative_cwd = canonical_cwd.relative_to(repo_root)
|
|
except ValueError as exc:
|
|
raise RuntimeError("job cwd is outside its reported Git root") from exc
|
|
index_path = directory / "synthetic.index"
|
|
index_env = {"GIT_INDEX_FILE": str(index_path)}
|
|
head = _git(repo_root, "rev-parse", "--verify", "HEAD")
|
|
if head.returncode == 0:
|
|
read_tree = _git(repo_root, "read-tree", "HEAD", env=index_env)
|
|
else:
|
|
read_tree = _git(repo_root, "read-tree", "--empty", env=index_env)
|
|
if read_tree.returncode != 0:
|
|
raise RuntimeError("unable to initialize synthetic Git index")
|
|
add = _git(repo_root, "add", "-A", "--", ".", env=index_env)
|
|
if add.returncode != 0:
|
|
raise RuntimeError(
|
|
"unable to snapshot Git workspace: "
|
|
+ add.stderr.decode("utf-8", errors="replace")[-2000:]
|
|
)
|
|
tree = _git(repo_root, "write-tree", env=index_env)
|
|
if tree.returncode != 0:
|
|
raise RuntimeError("unable to write synthetic Git tree")
|
|
# ``git commit-tree`` reads its log message from stdin unless one is
|
|
# supplied. MMO normally runs beneath an interactive Codex TUI, so
|
|
# inheriting stdin here can block a writable worker forever waiting for
|
|
# operator input. Keep the synthetic snapshot non-interactive and
|
|
# deterministic.
|
|
commit_args = [
|
|
"commit-tree",
|
|
tree.stdout.decode("ascii").strip(),
|
|
"-m",
|
|
"Codex MMO isolated workspace snapshot",
|
|
]
|
|
if head.returncode == 0:
|
|
commit_args.extend(["-p", head.stdout.decode("ascii").strip()])
|
|
commit = _git(
|
|
repo_root,
|
|
*commit_args,
|
|
env={
|
|
"GIT_AUTHOR_NAME": "Codex MMO",
|
|
"GIT_AUTHOR_EMAIL": "codex-mmo@localhost",
|
|
"GIT_COMMITTER_NAME": "Codex MMO",
|
|
"GIT_COMMITTER_EMAIL": "codex-mmo@localhost",
|
|
},
|
|
)
|
|
with contextlib.suppress(OSError):
|
|
index_path.unlink()
|
|
if commit.returncode != 0:
|
|
raise RuntimeError(
|
|
"unable to create synthetic Git commit: "
|
|
+ commit.stderr.decode("utf-8", errors="replace")[-2000:]
|
|
)
|
|
base_commit = commit.stdout.decode("ascii").strip()
|
|
worktree_root = directory / "worktree"
|
|
added = _git(repo_root, "worktree", "add", "--detach", str(worktree_root), base_commit)
|
|
if added.returncode != 0:
|
|
raise RuntimeError(
|
|
"unable to create isolated Git worktree: "
|
|
+ added.stderr.decode("utf-8", errors="replace")[-2000:]
|
|
)
|
|
execution_cwd = (worktree_root / relative_cwd).resolve()
|
|
mapped_attachments: list[str] = []
|
|
for raw in attachments:
|
|
canonical = Path(raw).resolve()
|
|
mapped_attachments.append(str(worktree_root / canonical.relative_to(repo_root)))
|
|
return {
|
|
"canonical_cwd": str(canonical_cwd),
|
|
"canonical_repo_root": str(repo_root),
|
|
"worktree_root": str(worktree_root),
|
|
"cwd": execution_cwd,
|
|
"attachments": mapped_attachments,
|
|
"base_commit": base_commit,
|
|
"base_fingerprints": _scope_fingerprints(repo_root, canonical_cwd, scopes),
|
|
}
|
|
|
|
|
|
def remove_isolated_worktree(metadata: Mapping[str, Any]) -> None:
|
|
root_value = metadata.get("canonical_repo_root")
|
|
worktree_value = metadata.get("worktree_root")
|
|
if not isinstance(root_value, str) or not isinstance(worktree_value, str):
|
|
return
|
|
_git(Path(root_value), "worktree", "remove", "--force", worktree_value)
|
|
|
|
|
|
def _nul_paths(result: subprocess.CompletedProcess[bytes]) -> set[str]:
|
|
if result.returncode != 0:
|
|
return set()
|
|
return {os.fsdecode(value) for value in result.stdout.split(b"\0") if value}
|
|
|
|
|
|
def _relative_to_job_cwd(repo_root: Path, job_cwd: Path, repo_relative: str) -> str | None:
|
|
absolute = (repo_root / repo_relative).resolve(strict=False)
|
|
try:
|
|
return absolute.relative_to(job_cwd).as_posix()
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def path_within_scope(relative: str, scopes: list[str]) -> bool:
|
|
path = Path(relative)
|
|
for raw in scopes:
|
|
scope = Path(raw)
|
|
if raw == "." or path == scope:
|
|
return True
|
|
with contextlib.suppress(ValueError):
|
|
path.relative_to(scope)
|
|
return True
|
|
return False
|
|
|
|
|
|
def capture_isolated_patch(
|
|
metadata: dict[str, Any], directory: Path, cwd: Path
|
|
) -> tuple[list[str], list[dict[str, Any]], dict[str, Any] | None]:
|
|
"""Capture one scope-checked binary patch from a disposable worker worktree."""
|
|
|
|
worktree_value = metadata.get("worktree_root")
|
|
base_commit = metadata.get("base_commit")
|
|
if not isinstance(worktree_value, str) or not isinstance(base_commit, str):
|
|
return ["writable worker lacks isolated worktree metadata"], [], None
|
|
worktree_root = Path(worktree_value)
|
|
staged = _git(worktree_root, "add", "-A", "--", ".")
|
|
if staged.returncode != 0:
|
|
return ["unable to stage isolated worker changes"], [], None
|
|
names = _git(
|
|
worktree_root,
|
|
"diff",
|
|
"--cached",
|
|
"--no-renames",
|
|
"--name-only",
|
|
"-z",
|
|
base_commit,
|
|
"--",
|
|
)
|
|
if names.returncode != 0:
|
|
return ["unable to enumerate isolated worker changes"], [], None
|
|
changed = sorted(_nul_paths(names))
|
|
violations: list[str] = []
|
|
relative_changes: list[tuple[str, str]] = []
|
|
scopes = list(metadata.get("write_scope", []))
|
|
for repo_relative in changed:
|
|
relative = _relative_to_job_cwd(worktree_root, cwd, repo_relative)
|
|
if relative is None or not path_within_scope(relative, scopes):
|
|
violations.append(repo_relative)
|
|
else:
|
|
relative_changes.append((repo_relative, relative))
|
|
if violations:
|
|
return ["out-of-scope mutation: " + ", ".join(violations[:40])], [], None
|
|
patch_result = _git(
|
|
worktree_root,
|
|
"diff",
|
|
"--cached",
|
|
"--binary",
|
|
"--full-index",
|
|
"--no-renames",
|
|
base_commit,
|
|
"--",
|
|
)
|
|
if patch_result.returncode != 0:
|
|
return ["unable to produce isolated binary patch"], [], None
|
|
patch_path = directory / "changes.patch"
|
|
atomic_write_bytes(patch_path, patch_result.stdout, 0o600)
|
|
artifacts: list[dict[str, Any]] = []
|
|
for repo_relative, relative in relative_changes:
|
|
path = worktree_root / repo_relative
|
|
if not path.is_file():
|
|
artifacts.append(
|
|
{
|
|
"relative_path": relative,
|
|
"sha256": hashlib.sha256(b"").hexdigest(),
|
|
"size": 0,
|
|
"media_type": "application/x-deleted",
|
|
"state": "deleted",
|
|
}
|
|
)
|
|
continue
|
|
content = path.read_bytes()
|
|
media_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
|
|
artifacts.append(
|
|
{
|
|
"relative_path": relative,
|
|
"sha256": hashlib.sha256(content).hexdigest(),
|
|
"size": len(content),
|
|
"media_type": media_type,
|
|
"state": "present",
|
|
}
|
|
)
|
|
patch = {
|
|
"path": str(patch_path),
|
|
"sha256": hashlib.sha256(patch_result.stdout).hexdigest(),
|
|
"size": len(patch_result.stdout),
|
|
"base_commit": base_commit,
|
|
"base_fingerprints": metadata.get("base_fingerprints"),
|
|
"changed_paths": [relative for _repo, relative in relative_changes],
|
|
}
|
|
return [], artifacts, patch
|
|
|
|
|
|
def apply_validated_patch(
|
|
*,
|
|
canonical_cwd: Path,
|
|
repo_root: Path,
|
|
scopes: Sequence[str],
|
|
base_fingerprints: Any,
|
|
patch_path: Path,
|
|
) -> None:
|
|
"""Verify a worker snapshot boundary and apply its already-authenticated patch."""
|
|
|
|
if _git_root(canonical_cwd) != repo_root:
|
|
raise RuntimeError("canonical Git repository identity changed before integration")
|
|
current = _scope_fingerprints(repo_root, canonical_cwd, scopes)
|
|
if current != base_fingerprints:
|
|
raise RuntimeError(
|
|
"canonical write scope changed after worker snapshot; integration refused"
|
|
)
|
|
check = _git(repo_root, "apply", "--check", "--binary", str(patch_path))
|
|
if check.returncode != 0:
|
|
raise RuntimeError(
|
|
"git apply --check rejected worker patch: "
|
|
+ check.stderr.decode("utf-8", errors="replace")[-2000:]
|
|
)
|
|
applied = _git(repo_root, "apply", "--binary", str(patch_path))
|
|
if applied.returncode != 0:
|
|
raise RuntimeError(
|
|
"worker patch integration failed: "
|
|
+ applied.stderr.decode("utf-8", errors="replace")[-2000:]
|
|
)
|
|
|
|
|
|
def reverse_applied_patch(repo_root: Path, patch_path: Path) -> None:
|
|
"""Reverse one patch after its lifecycle publication fails."""
|
|
|
|
rollback = _git(repo_root, "apply", "--reverse", "--binary", str(patch_path))
|
|
if rollback.returncode != 0:
|
|
raise RuntimeError(rollback.stderr.decode("utf-8", errors="replace")[-2000:])
|