dongarandClaude Opus 5.5
1a7836167d
From its failed review of #156: read_file stopped at 20,000 characters with no way on, so the reviewer hunted each file's end a definition at a time and ran out of turns; its retry carried every read before it, and a model call timed out. Both drafts were lost. - read_file takes a start line and says where the next part starts. - Each run's view starts at its brief (Transcript.since); the log keeps the rest. - A run out of turns with points drafted posts them as a partial review. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk
265 lines
9.9 KiB
Python
265 lines
9.9 KiB
Python
"""The repository, read-only, as the reviewer's executor (R3, L13).
|
|
|
|
Reads are bound to the item's repository by construction: no tool names one.
|
|
Code comes from Gitea's archive of the PR's head and of its base branch, one
|
|
API call each, unpacked into a scratch folder on the agent host and read in
|
|
Python: no git, no shell, nothing a model chose is ever run. Issues come
|
|
from the Gitea API, for the same repository.
|
|
"""
|
|
|
|
import io
|
|
import posixpath
|
|
import shutil
|
|
import tarfile
|
|
from collections.abc import Mapping
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from nisse_core.gitea import GiteaClient, GiteaError
|
|
from nisse_core.loop.driver import ToolOutcome
|
|
from nisse_core.loop.tools import tool
|
|
|
|
MAX_FILE_CHARS = 20000
|
|
MAX_LISTED = 500
|
|
MAX_FOUND = 50 # matching lines from search_code
|
|
MAX_ISSUES = 20
|
|
|
|
REF = "head (the PR's code) or base (the branch it merges into)."
|
|
READ_FILE = tool(
|
|
"read_file",
|
|
"Read a file in the repository. Its lines are numbered, for comment. A "
|
|
"long file comes in parts: the end of each says where the next starts.",
|
|
path="The file's path from the repository root.",
|
|
ref=REF,
|
|
start="Optional: the line to start from (1, the default, is the top).",
|
|
)
|
|
LIST_FILES = tool(
|
|
"list_files",
|
|
"List the files under a folder in the repository.",
|
|
path="The folder, from the repository root; empty for the whole tree.",
|
|
ref=REF,
|
|
)
|
|
SEARCH_CODE = tool(
|
|
"search_code",
|
|
"Find lines in the repository's files containing some text (ignoring "
|
|
"case): file, line number and the line.",
|
|
text="What to look for.",
|
|
ref=REF,
|
|
)
|
|
READ_ISSUE = tool(
|
|
"read_issue",
|
|
"Read one of the repository's issues (or PRs) by number, with its text.",
|
|
number="The issue's number.",
|
|
)
|
|
LIST_ISSUES = tool(
|
|
"list_issues",
|
|
"Find the repository's issues, open or closed, whose text matches.",
|
|
text="What to look for.",
|
|
)
|
|
READER_TOOLS = [READ_FILE, LIST_FILES, SEARCH_CODE, READ_ISSUE, LIST_ISSUES]
|
|
|
|
|
|
class Repository:
|
|
"""``refs``: what ``head`` and ``base`` name (commits or branches).
|
|
``scratch``: a folder of this run's own, removed by ``close``."""
|
|
|
|
def __init__(
|
|
self,
|
|
gitea: GiteaClient,
|
|
repo: str,
|
|
refs: Mapping[str, str],
|
|
scratch: Path,
|
|
) -> None:
|
|
self._gitea = gitea
|
|
self.repo = repo
|
|
self._refs = dict(refs)
|
|
self._scratch = scratch
|
|
self._trees: dict[str, Path] = {}
|
|
|
|
def close(self) -> None:
|
|
shutil.rmtree(self._scratch, ignore_errors=True)
|
|
|
|
def tree(self, ref: str) -> Path:
|
|
"""The repository at ``ref``, unpacked on first use."""
|
|
if ref not in self._refs:
|
|
raise ValueError(f"ref is head or base, not {ref!r}")
|
|
if ref not in self._trees:
|
|
root = self._scratch / ref
|
|
shutil.rmtree(root, ignore_errors=True)
|
|
root.mkdir(parents=True)
|
|
_unpack(self._gitea.archive(self.repo, self._refs[ref]), root)
|
|
self._trees[ref] = root
|
|
return self._trees[ref]
|
|
|
|
def lines(self, path: str, ref: str = "head") -> list[str] | None:
|
|
"""A file's lines, or None if there's no such file."""
|
|
found = self._file(path, ref)
|
|
if found is None:
|
|
return None
|
|
return found.read_text(errors="replace").splitlines()
|
|
|
|
def _file(self, path: str, ref: str) -> Path | None:
|
|
clean = _clean(path)
|
|
if clean is None:
|
|
return None
|
|
found = self.tree(ref) / clean
|
|
return found if found.is_file() and not found.is_symlink() else None
|
|
|
|
def dispatch(self, tool: str, arguments: dict[str, Any]) -> ToolOutcome:
|
|
try:
|
|
match tool:
|
|
case "read_file":
|
|
return self._read(arguments)
|
|
case "list_files":
|
|
return self._list(arguments)
|
|
case "search_code":
|
|
return self._search(arguments)
|
|
case "read_issue":
|
|
return self._issue(arguments)
|
|
case "list_issues":
|
|
return self._issues(arguments)
|
|
except (ValueError, KeyError) as exc:
|
|
return ToolOutcome(str(exc), False)
|
|
except GiteaError as exc:
|
|
return ToolOutcome(f"Gitea didn't answer: {exc}", False)
|
|
return ToolOutcome(f"no such tool: {tool}", False)
|
|
|
|
def _ref(self, arguments: Mapping[str, Any]) -> str:
|
|
return str(arguments.get("ref") or "head").strip().lower()
|
|
|
|
def _read(self, arguments: Mapping[str, Any]) -> ToolOutcome:
|
|
path = str(arguments.get("path", ""))
|
|
ref = self._ref(arguments)
|
|
lines = self.lines(path, ref)
|
|
if lines is None:
|
|
return ToolOutcome(f"{path}: no such file at {ref}", False)
|
|
raw = str(arguments.get("start") or "1").strip()
|
|
if not raw.isdigit() or not 1 <= int(raw) <= max(len(lines), 1):
|
|
return ToolOutcome(f"start is a line number from 1 to {len(lines)}", False)
|
|
# A part of at most MAX_FILE_CHARS, in whole lines; the rest is a
|
|
# read away, never out of reach (#156: a long file's end was).
|
|
first = int(raw)
|
|
shown: list[str] = []
|
|
size = 0
|
|
for n in range(first, len(lines) + 1):
|
|
row = f"{n:>5} {lines[n - 1]}"
|
|
if shown and size + len(row) + 1 > MAX_FILE_CHARS:
|
|
break
|
|
shown.append(row[:MAX_FILE_CHARS])
|
|
size += len(row) + 1
|
|
last = first + len(shown) - 1
|
|
where = f"lines {first}-{last} of {len(lines)}"
|
|
more = (
|
|
f"\n[{where}: read on with start={last + 1}]" if last < len(lines) else ""
|
|
)
|
|
text = "\n".join(shown)
|
|
return ToolOutcome(f"[{self.repo}@{ref}: {path}, {where}]\n{text}{more}", True)
|
|
|
|
def _list(self, arguments: Mapping[str, Any]) -> ToolOutcome:
|
|
folder = str(arguments.get("path", "")).strip().strip("/")
|
|
ref = self._ref(arguments)
|
|
root = self.tree(ref)
|
|
start = root if not folder else root / (_clean(folder) or "\0")
|
|
if not start.is_dir():
|
|
return ToolOutcome(f"{folder}: no such folder at {ref}", False)
|
|
found = sorted(
|
|
str(p.relative_to(root))
|
|
for p in start.rglob("*")
|
|
if p.is_file() and not p.is_symlink()
|
|
)
|
|
shown = found[:MAX_LISTED]
|
|
more = f"\n[{len(found) - len(shown)} more]" if len(found) > len(shown) else ""
|
|
return ToolOutcome(
|
|
f"[{self.repo}@{ref}: {folder or 'every file'}]\n"
|
|
+ ("\n".join(shown) or "no files")
|
|
+ more,
|
|
True,
|
|
)
|
|
|
|
def _search(self, arguments: Mapping[str, Any]) -> ToolOutcome:
|
|
text = str(arguments.get("text", ""))
|
|
if not text.strip():
|
|
return ToolOutcome("search for some text", False)
|
|
ref = self._ref(arguments)
|
|
root = self.tree(ref)
|
|
needle = text.lower()
|
|
hits: list[str] = []
|
|
for path in sorted(root.rglob("*")):
|
|
if not path.is_file() or path.is_symlink():
|
|
continue
|
|
try:
|
|
content = path.read_text()
|
|
except (UnicodeDecodeError, OSError):
|
|
continue # binary
|
|
for n, line in enumerate(content.splitlines(), 1):
|
|
if needle in line.lower():
|
|
hits.append(f"{path.relative_to(root)}:{n}: {line.strip()[:200]}")
|
|
if len(hits) > MAX_FOUND:
|
|
break
|
|
if len(hits) > MAX_FOUND:
|
|
break
|
|
more = "\n[more not shown: narrow the search]" if len(hits) > MAX_FOUND else ""
|
|
body = "\n".join(hits[:MAX_FOUND]) or "nothing"
|
|
return ToolOutcome(f"[{self.repo}@{ref} search: {text}]\n{body}{more}", True)
|
|
|
|
def _issue(self, arguments: Mapping[str, Any]) -> ToolOutcome:
|
|
number = _number(arguments.get("number"))
|
|
issue = self._gitea.issue(self.repo, number)
|
|
return ToolOutcome(
|
|
f"[{self.repo}#{issue.number}, by {issue.author}] {issue.title}\n\n"
|
|
f"{issue.body}",
|
|
True,
|
|
)
|
|
|
|
def _issues(self, arguments: Mapping[str, Any]) -> ToolOutcome:
|
|
text = str(arguments.get("text", "")).strip()
|
|
found = self._gitea.search_issues(self.repo, text)[:MAX_ISSUES]
|
|
lines = [f"#{i.number}: {i.title}" for i in found]
|
|
return ToolOutcome(
|
|
f"[{self.repo} issues: {text}]\n" + ("\n".join(lines) or "nothing"), True
|
|
)
|
|
|
|
|
|
def _number(raw: Any) -> int:
|
|
try:
|
|
number = int(str(raw).strip().lstrip("#"))
|
|
except ValueError:
|
|
raise ValueError(f"{raw!r}: an issue's number, like 14") from None
|
|
if number <= 0:
|
|
raise ValueError(f"{raw!r}: an issue's number, like 14")
|
|
return number
|
|
|
|
|
|
def _clean(path: str) -> str | None:
|
|
"""A path inside the tree, or None: never absolute, never up and out."""
|
|
path = path.strip()
|
|
if not path or path.startswith("/") or "\\" in path:
|
|
return None
|
|
clean = posixpath.normpath(path)
|
|
if clean == "." or clean.startswith(".."):
|
|
return None
|
|
return clean
|
|
|
|
|
|
def _unpack(archive: bytes, root: Path) -> None:
|
|
"""Gitea's archive, without its top folder; only plain files and
|
|
folders, each inside ``root``."""
|
|
with tarfile.open(fileobj=io.BytesIO(archive), mode="r:gz") as tar:
|
|
for member in tar.getmembers():
|
|
if not (member.isfile() or member.isdir()):
|
|
continue # links and devices are never followed
|
|
parts = member.name.split("/", 1)
|
|
if len(parts) < 2 or not parts[1]:
|
|
continue
|
|
inner = _clean(parts[1])
|
|
if inner is None:
|
|
continue
|
|
target = root / inner
|
|
if member.isdir():
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
continue
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
source = tar.extractfile(member)
|
|
if source is not None:
|
|
target.write_bytes(source.read())
|