feat: add output truncation and offset paging to all large-output tools

- coding_tools: list_files (max_entries=200, sorted), run_command
  (max_chars=8000, output_offset), grep_search (max_lines=100, offset);
  added _truncate_output helper for run_command paging
- pr_tools: get_pull_request_diff / get_pull_request_patch now truncate
  at 15k chars with hunk-boundary awareness and char_offset paging;
  get_pull_request_comments gains limit/offset paging
- issue_tools: get_issue_comments gains limit/offset paging
- file_tools: get_file_content / get_file_content_with_ref now paginate
  by line (offset=1, limit=250), matching existing read_file convention;
  added _paginate_lines helper
- research_tools: fetch_url gains char_offset parameter; _smart_truncate
  now slices from an offset and embeds next char_offset in the footer
- gitea_tools facade: all new params threaded through
This commit is contained in:
Michael
2026-06-28 19:57:14 +02:00
parent c0c9278d75
commit bf13a979c2
6 changed files with 383 additions and 69 deletions
+117 -19
View File
@@ -21,12 +21,26 @@ class CodingTools:
return path
return os.path.abspath(os.path.join(self.repo_path, path))
def list_files(self, path: str = ".") -> str:
"""List all files and directories at the given path (relative to repo root or absolute)."""
def list_files(self, path: str = ".", max_entries: int = 200) -> str:
"""List files and directories at the given path (relative to repo root or absolute).
Args:
path: Directory to list (relative to repo root or absolute).
max_entries: Maximum number of entries to return (default 200).
Pass a subdirectory path to narrow results when a directory is very large.
"""
resolved: str = self._resolve_path(path)
try:
items: list[str] = os.listdir(resolved)
return "\n".join(items)
items: list[str] = sorted(os.listdir(resolved))
total: int = len(items)
shown: list[str] = items[:max_entries]
result: str = "\n".join(shown)
if total > max_entries:
result += (
f"\n\n[{total} entries total — showing first {max_entries}. "
"Pass a subdirectory path to narrow results.]"
)
return result
except Exception as e:
return f"Error listing files: {str(e)}"
@@ -147,16 +161,58 @@ class CodingTools:
return False, "\n".join(log_output)
return True, "\n".join(log_output)
def run_command(self, command: str, timeout: int = 120) -> str:
"""Execute a shell command in the repository workspace and return stdout and stderr output. Args: command, timeout (default 120 seconds)."""
def _truncate_output(
self,
text: str,
max_chars: int,
offset: int,
label: str = "Output",
) -> str:
"""Slice [offset : offset+max_chars] from text and append a paging footer if truncated."""
total: int = len(text)
chunk: str = text[offset : offset + max_chars]
if offset > 0 or (offset + max_chars) < total:
next_offset: int = offset + len(chunk)
chunk += (
f"\n\n[{label} truncated — {total} chars total. "
f"Showing chars {offset}{next_offset}. "
f"Re-run with output_offset={next_offset} to read more.]"
)
return chunk
def run_command(
self,
command: str,
timeout: int = 120,
max_chars: int = 8000,
output_offset: int = 0,
) -> str:
"""Execute a shell command in the repository workspace and return stdout/stderr.
Args:
command: Shell command to run.
timeout: Seconds before the command is killed (default 120).
max_chars: Maximum characters of combined output to return (default 8000).
Tail (most recent lines) is preferred because errors appear there.
output_offset: Character offset into the full output to start reading from
(default 0). Increment by max_chars to page through large output.
"""
if "tea pr create" in command:
success, log_msg = self.run_verification()
if not success:
return f"Verification failed! You cannot create a pull request because the tests/checks are failing:\n\n{log_msg}\n\nPlease fix the issues and try again."
return (
f"Verification failed! You cannot create a pull request because the "
f"tests/checks are failing:\n\n{log_msg}\n\nPlease fix the issues and try again."
)
try:
process: subprocess.Popen[str] = subprocess.Popen(
command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=self.repo_path
command,
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
cwd=self.repo_path,
)
stdout: str
stderr: str
@@ -165,7 +221,11 @@ class CodingTools:
except subprocess.TimeoutExpired:
process.kill()
stdout, stderr = process.communicate()
return f"Command timed out after {timeout} seconds.\n--- STDOUT ---\n{stdout}\n--- STDERR ---\n{stderr}"
raw: str = (
f"Command timed out after {timeout} seconds.\n"
f"--- STDOUT ---\n{stdout}\n--- STDERR ---\n{stderr}"
)
return self._truncate_output(raw, max_chars, output_offset, "Output")
output: str = ""
if stdout:
@@ -173,20 +233,44 @@ class CodingTools:
if stderr:
output += f"\n--- STDERR ---\n{stderr}"
if process.returncode != 0:
return f"Command failed with exit code {process.returncode}:\n{output}"
if not output:
return "Command executed successfully (no output)."
return output if output else "Command executed successfully (no output)."
prefix: str = (
f"Command failed with exit code {process.returncode}:\n"
if process.returncode != 0
else ""
)
full: str = prefix + output
return self._truncate_output(full, max_chars, output_offset, "Output")
except Exception as e:
return f"Error running command: {str(e)}"
def grep_search(self, pattern: str, path: str = ".") -> str:
"""Search for pattern in files under path using grep (case-insensitive)."""
def grep_search(
self,
pattern: str,
path: str = ".",
max_lines: int = 100,
offset: int = 0,
) -> str:
"""Search for pattern in files under path using grep (case-insensitive).
Args:
pattern: Regular-expression / literal pattern to search for.
path: Directory or file to search (relative to repo root or absolute).
max_lines: Maximum number of matching lines to return (default 100).
offset: Line offset into the full result set for paging (default 0).
"""
resolved: str = self._resolve_path(path)
try:
command: str = f"grep -ri '{pattern}' {resolved}"
process: subprocess.Popen[str] = subprocess.Popen(
command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=self.repo_path
command,
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
cwd=self.repo_path,
)
stdout: str
stderr: str
@@ -195,14 +279,28 @@ class CodingTools:
except subprocess.TimeoutExpired:
process.kill()
stdout, stderr = process.communicate()
return f"Grep search timed out after 30 seconds.\nStdout: {stdout}\nStderr: {stderr}"
return (
f"Grep search timed out after 30 seconds.\n"
f"Stdout: {stdout}\nStderr: {stderr}"
)
if process.returncode != 0 and not stdout:
return f"No matches found for '{pattern}'."
output: str = stdout
all_lines: list[str] = stdout.splitlines()
total: int = len(all_lines)
page: list[str] = all_lines[offset : offset + max_lines]
result: str = "\n".join(page)
if total > offset + max_lines:
next_offset: int = offset + max_lines
result += (
f"\n\n[{total} matches total — showing lines {offset}{offset + len(page)}. "
f"Use offset={next_offset} to see more.]"
)
if stderr:
output += f"\nError: {stderr}"
return output
result += f"\nError: {stderr}"
return result
except Exception as e:
return f"Error during grep search: {str(e)}"