File size: 4,186 Bytes
4554903 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | """Git-aware file access for the campus repo.
Reads are jailed to the configured repo root. Writes additionally
require the target to be clean in git (no clobbering uncommitted work)
and are disabled entirely unless the operator turns them on in config —
Rivet is a colleague that suggests; writing into the repo is a
consensus-gated action.
"""
import re
from dataclasses import dataclass, field
from pathlib import Path
from tools.guard import ToolResult, check_path, run_checked
MAX_READ_BYTES = 200_000
@dataclass
class FileReadResult:
ok: bool
path: str = ""
content: str = ""
truncated: bool = False
git_status: str = "" # '' = clean/untracked-unknown, else porcelain code
error: str = ""
@dataclass
class RepoFiles:
repo_root: str
write_enabled: bool = False
files_read: list = field(default_factory=list)
def _jail(self, rel_path: str) -> tuple:
target = Path(self.repo_root) / rel_path
reason = check_path(target, [self.repo_root])
return target, reason
def read(self, rel_path: str) -> FileReadResult:
target, reason = self._jail(rel_path)
if reason:
return FileReadResult(ok=False, path=rel_path, error=reason)
if not target.exists():
return FileReadResult(ok=False, path=rel_path,
error=f"not found: {rel_path}")
if not target.is_file():
return FileReadResult(ok=False, path=rel_path,
error=f"not a file: {rel_path}")
raw = target.read_bytes()
truncated = len(raw) > MAX_READ_BYTES
content = raw[:MAX_READ_BYTES].decode("utf-8", errors="replace")
self.files_read.append(rel_path)
return FileReadResult(
ok=True, path=rel_path, content=content, truncated=truncated,
git_status=self._git_status(rel_path),
)
def write(self, rel_path: str, content: str) -> ToolResult:
if not self.write_enabled:
return ToolResult(
ok=False,
blocked_reason=("repo writes are disabled by config "
"(tools.repo_write_enabled) — Rivet suggests, "
"humans apply"),
)
target, reason = self._jail(rel_path)
if reason:
return ToolResult(ok=False, blocked_reason=reason)
status = self._git_status(rel_path)
if status and status != "??":
return ToolResult(
ok=False,
blocked_reason=(f"{rel_path} has uncommitted changes "
f"(git status '{status}') — refusing to clobber"),
)
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(content)
return ToolResult(ok=True, stdout=f"wrote {rel_path}")
def search(self, pattern: str, glob: str = "", max_results: int = 50) -> list:
"""Regex search across the repo via git grep (respects .gitignore)."""
argv = ["git", "grep", "-n", "-E", "--", pattern]
if glob:
argv += [glob]
result = run_checked(argv, cwd=self.repo_root, timeout=30)
if not result.ok and not result.stdout:
return []
hits = []
for line in result.stdout.splitlines()[:max_results]:
m = re.match(r"^([^:]+):(\d+):(.*)$", line)
if m:
hits.append({"path": m.group(1), "line": int(m.group(2)),
"text": m.group(3).strip()[:200]})
return hits
def list_dir(self, rel_path: str = ".") -> list:
target, reason = self._jail(rel_path)
if reason or not target.is_dir():
return []
return sorted(
p.name + ("/" if p.is_dir() else "")
for p in target.iterdir() if p.name != ".git"
)
def _git_status(self, rel_path: str) -> str:
result = run_checked(
["git", "status", "--porcelain", "--", rel_path],
cwd=self.repo_root, timeout=10,
)
if result.stdout.strip():
return result.stdout.strip()[:2]
return ""
|