apodex/frontier-agent-demo

1"""File editor tools — view, create, and edit files in E2B sandbox."""2 3from __future__ import annotations4 5import logging6 7from frontier_agent.core.tool import tool8from plugins.tools._deliverable_policy import output_write_error9from plugins.tools._path_auth import _authorized_local_path10from plugins.tools._sandbox import (11    aget_sandbox,12    arun_sandbox_cmd,13    asandbox_write_file,14)15 16logger = logging.getLogger(__name__)17 18_MAX_OUTPUT_CHARS = 16_00019 20 21def _truncate(text: str, max_chars: int = _MAX_OUTPUT_CHARS) -> str:22    """Truncate output if too long."""23    if len(text) > max_chars:24        return text[:max_chars] + "\n\n[... output truncated]"25    return text26 27 28@tool29async def file_editor_view(path: str, view_range: str = "") -> str:30    """View file contents or list directory in E2B sandbox.31 32    Args:33        path: Absolute path to file or directory.34        view_range: Optional line range like "1-50" to view specific lines.35 36    Returns:37        File contents with line numbers, or directory listing.38    """39    if not path or not path.strip():40        return "Error: path is required."41 42    local_path, _reason = _authorized_local_path(path)43    if local_path is not None:44        from plugins.tools.ignore_rules import discover_repo_root, should_ignore_path45        if local_path.is_dir():46            repo_root = discover_repo_root(local_path)47            entries: list[str] = []48            for entry in sorted(local_path.rglob("*")):49                if should_ignore_path(entry, repo_root):50                    continue51                try:52                    rel = entry.relative_to(local_path)53                except ValueError:54                    rel = entry55                if len(rel.parts) > 2:56                    continue57                entries.append(str(rel))58                if len(entries) >= 100:59                    break60            return "\n".join(entries) if entries else f"(empty directory: {path})"61        if not local_path.is_file():62            return f"Error: File not found: {path}"63        try:64            lines = local_path.read_text(encoding="utf-8").splitlines()65        except Exception as e:66            return f"Error viewing {path}: {e}"67        if view_range and "-" in view_range:68            start_s, end_s = view_range.split("-", 1)69            start = max(int(start_s.strip() or "1"), 1)70            end = max(int(end_s.strip() or str(start)), start)71        else:72            start, end = 1, len(lines)73        rendered = "\n".join(f"{idx}: {line}" for idx, line in enumerate(lines[start - 1:end], start=start))74        return _truncate(rendered or "(empty file)")75 76    try:77        sandbox = await aget_sandbox()78    except RuntimeError as e:79        return f"Error: {e}"80 81    try:82        check = await arun_sandbox_cmd(83            sandbox,84            f"test -d {path} && echo DIR || echo FILE", timeout=10,85        )86        is_dir = "DIR" in check.stdout87 88        if is_dir:89            result = await arun_sandbox_cmd(90                sandbox,91                f"find {path} -maxdepth 2 -not -path '*/.git/*' "92                f"-not -path '*/node_modules/*' -not -path '*/__pycache__/*' "93                f"| head -100 | sort",94                timeout=15,95            )96            return _truncate(result.stdout or f"(empty directory: {path})")97 98        # File view99        if view_range:100            parts = view_range.split("-")101            if len(parts) == 2:102                start, end = parts[0].strip(), parts[1].strip()103                cmd = f"sed -n '{start},{end}p' {path} | cat -n"104            else:105                cmd = f"cat -n {path}"106        else:107            cmd = f"cat -n {path}"108 109        result = await arun_sandbox_cmd(sandbox, cmd, timeout=15)110 111        if result.exit_code != 0:112            return f"Error: {result.stderr or 'File not found'}"113 114        return _truncate(result.stdout or "(empty file)")115 116    except Exception as e:117        return f"Error viewing {path}: {e}"118 119 120@tool121async def file_editor_create(path: str, content: str) -> str:122    """Create a new file in E2B sandbox with the given content.123 124    Use this to create new files. For modifying existing files, use file_editor_str_replace instead.125 126    Args:127        path: Absolute path for the new file.128        content: The file content to write.129 130    Returns:131        Confirmation message.132    """133    if not path or not path.strip():134        return "Error: path is required."135 136    deliverable_error = output_write_error(path)137    if deliverable_error:138        return f"Error: {deliverable_error}"139    local_path, reason = _authorized_local_path(path, write_access=True)140    if local_path is not None:141        try:142            local_path.parent.mkdir(parents=True, exist_ok=True)143            local_path.write_text(content, encoding="utf-8")144            lines = content.count("\n") + (1 if content else 0)145            return f"File created: {path} ({lines} lines, {len(content)} bytes)"146        except Exception as e:147            return f"Error creating {path}: {e}"148 149    try:150        sandbox = await aget_sandbox()151    except RuntimeError as e:152        return f"Error: {e}" if "allowed" not in reason.lower() else f"Access denied: {reason}"153 154    try:155        import os156        parent = os.path.dirname(path)157        if parent:158            await arun_sandbox_cmd(159                sandbox, f"mkdir -p {parent}", timeout=10,160            )161 162        # Write via python in sandbox (avoids files.write permission issues)163        ok, err = await asandbox_write_file(sandbox, path, content)164        if not ok:165            return f"Error creating {path}: {err}"166 167        result = await arun_sandbox_cmd(168            sandbox, f"wc -l < {path}", timeout=5,169        )170        lines = result.stdout.strip() if result.stdout else "?"171 172        return f"File created: {path} ({lines} lines, {len(content)} bytes)"173 174    except Exception as e:175        return f"Error creating {path}: {e}"176 177 178@tool179async def file_editor_str_replace(path: str, old_str: str, new_str: str) -> str:180    """Replace a string occurrence in a file in E2B sandbox.181 182    The old_str must appear exactly once in the file (for safety).183    Use file_editor_view first to see the file contents.184 185    Args:186        path: Absolute path to the file.187        old_str: The exact string to find and replace.188        new_str: The replacement string.189 190    Returns:191        Confirmation with a diff summary.192    """193    if not path or not old_str:194        return "Error: path and old_str are required."195 196    deliverable_error = output_write_error(path)197    if deliverable_error:198        return f"Error: {deliverable_error}"199    local_path, reason = _authorized_local_path(path, write_access=True)200    if local_path is not None:201        if not local_path.is_file():202            return f"Error: Cannot read {path}: file not found"203        try:204            content = local_path.read_text(encoding="utf-8")205        except Exception as e:206            return f"Error editing {path}: {e}"207 208        count = content.count(old_str)209        if count == 0:210            return (211                f"Error: old_str not found in {path}. "212                "Make sure the string matches exactly (including whitespace and indentation)."213            )214        if count > 1:215            return (216                f"Error: old_str found {count} times in {path}. "217                "Provide a more specific string that appears exactly once."218            )219        try:220            local_path.write_text(content.replace(old_str, new_str, 1), encoding="utf-8")221        except Exception as e:222            return f"Error editing {path}: {e}"223 224        old_lines = old_str.count("\n") + 1225        new_lines = new_str.count("\n") + 1226        return f"Replaced in {path}: {old_lines} line(s) → {new_lines} line(s)"227 228    try:229        sandbox = await aget_sandbox()230    except RuntimeError as e:231        return f"Error: {e}" if "allowed" not in reason.lower() else f"Access denied: {reason}"232 233    try:234        # Read current content via cat (more reliable than files.read for all paths)235        result = await arun_sandbox_cmd(sandbox, f"cat {path}", timeout=10)236        if result.exit_code != 0:237            return f"Error: Cannot read {path}: {result.stderr or 'file not found'}"238        content = result.stdout239 240        count = content.count(old_str)241        if count == 0:242            return (243                f"Error: old_str not found in {path}. "244                "Make sure the string matches exactly (including whitespace and indentation)."245            )246        if count > 1:247            return (248                f"Error: old_str found {count} times in {path}. "249                "Provide a more specific string that appears exactly once."250            )251 252        # Replace and write back via python in sandbox (avoids files.write permission issues)253        new_content = content.replace(old_str, new_str, 1)254        ok, err = await asandbox_write_file(sandbox, path, new_content)255        if not ok:256            return f"Error writing {path}: {err}"257 258        # Show diff stats259        old_lines = old_str.count("\n") + 1260        new_lines = new_str.count("\n") + 1261        return f"Replaced in {path}: {old_lines} line(s) → {new_lines} line(s)"262 263    except Exception as e:264        return f"Error editing {path}: {e}"265