"""File I/O operations with a configurable working directory.""" import os from pathlib import Path from typing import Optional from agentscope.message import TextBlock from agentscope.tool import ToolResponse from ..utils import read_file_safe, truncate_text_output, TRUNCATION_NOTICE_MARKER class FileIO: """File I/O operations with a configurable working directory.""" def __init__(self, working_dir: str | Path): """Initialize FileIO with a working directory. Args: working_dir (`str`): The working directory for resolving relative paths. """ self.working_dir = Path(working_dir) def _resolve_file_path(self, file_path: str) -> str: """Resolve file path: use absolute path as-is, resolve relative path from working_dir. Args: file_path: The input file path (absolute or relative). Returns: The resolved absolute file path as string. """ path = Path(file_path).expanduser() if path.is_absolute(): return str(path) else: return str(self.working_dir / file_path) async def read_file( # pylint: disable=too-many-return-statements self, file_path: str, start_line: Optional[int] = None, end_line: Optional[int] = None, ) -> ToolResponse: """Read a file. Relative paths resolve from WORKING_DIR. Use start_line/end_line to read a specific line range (output includes line numbers). Omit both to read the full file. Args: file_path (`str`): Path to the file. start_line (`int`, optional): First line to read (1-based, inclusive). end_line (`int`, optional): Last line to read (1-based, inclusive). """ # Convert start_line/end_line to int if they are strings if start_line is not None: try: start_line = int(start_line) except (ValueError, TypeError): return ToolResponse( content=[ TextBlock( type="text", text=f"Error: start_line must be an integer, got {start_line!r}.", ), ], ) if end_line is not None: try: end_line = int(end_line) except (ValueError, TypeError): return ToolResponse( content=[ TextBlock( type="text", text=f"Error: end_line must be an integer, got {end_line!r}.", ), ], ) file_path = self._resolve_file_path(file_path) if not os.path.exists(file_path): return ToolResponse( content=[ TextBlock( type="text", text=f"Error: The file {file_path} does not exist.", ), ], ) if not os.path.isfile(file_path): return ToolResponse( content=[ TextBlock( type="text", text=f"Error: The path {file_path} is not a file.", ), ], ) try: content = read_file_safe(file_path) all_lines = content.split("\n") total = len(all_lines) # Determine read range s = max(1, start_line if start_line is not None else 1) e = min(total, end_line if end_line is not None else total) if s > total: return ToolResponse( content=[ TextBlock( type="text", text=f"Error: start_line {s} exceeds file length ({total} lines).", ), ], ) if s > e: return ToolResponse( content=[ TextBlock( type="text", text=f"Error: start_line ({s}) > end_line ({e}).", ), ], ) # Extract selected lines selected_content = "\n".join(all_lines[s - 1 : e]) # Apply smart truncation (consistent with shell output format) text = truncate_text_output( selected_content, start_line=s, total_lines=total, file_path=file_path, ) # Add continuation hint if partial read without truncation. # Use TRUNCATION_NOTICE_MARKER format so ToolResultCompactor can # re-truncate with the correct start_line when compacting old messages. if text == selected_content and e < total: content_bytes = len(text.encode("utf-8")) notice = ( TRUNCATION_NOTICE_MARKER + f"\nThe output above was truncated." f"\nThe full content is saved to the file " f"and contains {total} lines in total." f"\nThis excerpt starts at line {s} and " f"covers the next {content_bytes} bytes." "\nIf the current content is not enough, " f"call `read_file` with file_path={file_path} start_line={e + 1} to read more." ) text = text + notice return ToolResponse( content=[TextBlock(type="text", text=text)], ) except Exception as e: return ToolResponse( content=[ TextBlock( type="text", text=f"Error: Read file failed due to \n{e}", ), ], ) async def write_file( self, file_path: str, content: str, ) -> ToolResponse: """Create or overwrite a file. Relative paths resolve from working_dir. Args: file_path (`str`): Path to the file. content (`str`): Content to write. """ if not file_path: return ToolResponse( content=[ TextBlock( type="text", text="Error: No `file_path` provided.", ), ], ) file_path = self._resolve_file_path(file_path) try: with open(file_path, "w", encoding="utf-8") as file: file.write(content) return ToolResponse( content=[ TextBlock( type="text", text=f"Wrote {len(content)} bytes to {file_path}.", ), ], ) except Exception as e: return ToolResponse( content=[ TextBlock( type="text", text=f"Error: Write file failed due to \n{e}", ), ], ) # pylint: disable=too-many-return-statements async def edit_file( self, file_path: str, old_text: str, new_text: str, ) -> ToolResponse: """Find-and-replace text in a file. All occurrences of old_text are replaced with new_text. Relative paths resolve from working_dir. Args: file_path (`str`): Path to the file. old_text (`str`): Exact text to find. new_text (`str`): Replacement text. """ if not file_path: return ToolResponse( content=[ TextBlock( type="text", text="Error: No `file_path` provided.", ), ], ) resolved_path = self._resolve_file_path(file_path) if not os.path.exists(resolved_path): return ToolResponse( content=[ TextBlock( type="text", text=f"Error: The file {resolved_path} does not exist.", ), ], ) if not os.path.isfile(resolved_path): return ToolResponse( content=[ TextBlock( type="text", text=f"Error: The path {resolved_path} is not a file.", ), ], ) try: content = read_file_safe(resolved_path) except Exception as e: return ToolResponse( content=[ TextBlock( type="text", text=f"Error: Read file failed due to \n{e}", ), ], ) if old_text not in content: return ToolResponse( content=[ TextBlock( type="text", text=f"Error: The text to replace was not found in {file_path}.", ), ], ) new_content = content.replace(old_text, new_text) write_response = await self.write_file(file_path=resolved_path, content=new_content) if write_response.content and len(write_response.content) > 0: write_text = write_response.content[0].get("text", "") if write_text.startswith("Error:"): return write_response return ToolResponse( content=[ TextBlock( type="text", text=f"Successfully replaced text in {file_path}.", ), ], ) async def append_file( self, file_path: str, content: str, ) -> ToolResponse: """Append content to the end of a file. Relative paths resolve from working_dir. Args: file_path (`str`): Path to the file. content (`str`): Content to append. """ if not file_path: return ToolResponse( content=[ TextBlock( type="text", text="Error: No `file_path` provided.", ), ], ) file_path = self._resolve_file_path(file_path) try: with open(file_path, "a", encoding="utf-8") as file: file.write(content) return ToolResponse( content=[ TextBlock( type="text", text=f"Appended {len(content)} bytes to {file_path}.", ), ], ) except Exception as e: return ToolResponse( content=[ TextBlock( type="text", text=f"Error: Append file failed due to \n{e}", ), ], )