GeminiEditor: replace with expected_replacements and EOL handling (Phase 1a)

openhands_aci/editor/__init__.py

@@ -5,13 +5,15 @@
 from .encoding import EncodingManager, with_encoding
 from .exceptions import ToolError
 from .file_cache import FileCache
+from .gemini_editor import GeminiEditor
 from .results import ToolResult
 
 _GLOBAL_EDITOR = OHEditor()
 
 __all__ = [
     'Command',
     'OHEditor',
+    'GeminiEditor',
     'ToolError',
     'ToolResult',
     'FileCache',

openhands_aci/editor/editor.py

@@ -204,9 +204,23 @@ def str_replace(
         ]
 
         if not occurrences:
-            raise ToolError(
-                f'No replacement was performed, old_str `{old_str}` did not appear verbatim in {path}.'
-            )
+            # We found no occurrences, possibly because of extra white spaces at either the front or back of the string.
+            # Remove the white spaces and try again.
+            old_str = old_str.strip()
+            new_str = new_str.strip()
+            pattern = re.escape(old_str)
+            occurrences = [
+                (
+                    file_content.count('\n', 0, match.start()) + 1,  # line number
+                    match.group(),  # matched text
+                    match.start(),  # start position
+                )
+                for match in re.finditer(pattern, file_content)
+            ]
+            if not occurrences:
+                raise ToolError(
+                    f'No replacement was performed, old_str `{old_str}` did not appear verbatim in {path}.'
+                )
         if len(occurrences) > 1:
             line_numbers = sorted(set(line for line, _, _ in occurrences))
             raise ToolError(
@@ -348,30 +362,32 @@ def view(self, path: Path, view_range: list[int] | None = None) -> CLIResult:
                 f'Its first element `{start_line}` should be within the range of lines of the file: {[1, num_lines]}.',
             )
 
-        if end_line > num_lines:
-            raise EditorToolParameterInvalidError(
-                'view_range',
-                view_range,
-                f'Its second element `{end_line}` should be smaller than the number of lines in the file: `{num_lines}`.',
-            )
+        # Normalize end_line and provide a warning if it exceeds file length
+        warning_message: str | None = None
+        if end_line == -1:
+            end_line = num_lines
+        elif end_line > num_lines:
+            warning_message = f"We only show up to {num_lines} since there're only {num_lines} lines in this file."
+            end_line = num_lines
 
-        if end_line != -1 and end_line < start_line:
+        if end_line < start_line:
             raise EditorToolParameterInvalidError(
                 'view_range',
                 view_range,
                 f'Its second element `{end_line}` should be greater than or equal to the first element `{start_line}`.',
             )
 
-        if end_line == -1:
-            end_line = num_lines
-
         file_content = self.read_file(path, start_line=start_line, end_line=end_line)
 
         # Get the detected encoding
         output = self._make_output(
             '\n'.join(file_content.splitlines()), str(path), start_line
         )  # Remove extra newlines
 
+        # Prepend warning if we truncated the end_line
+        if warning_message:
+            output = f'NOTE: {warning_message}\n{output}'
+
         return CLIResult(
             path=str(path),
             output=output,

openhands_aci/editor/gemini_editor.py

@@ -0,0 +1,172 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+from .editor import OHEditor
+from .exceptions import (
+    EditorToolParameterInvalidError,
+    EditorToolParameterMissingError,
+    ToolError,
+)
+from .results import CLIResult
+
+
+@dataclass
+class ReplaceResult:
+    count: int
+    prev_exist: bool
+    old_content: str
+    new_content: str
+
+
+class GeminiEditor(OHEditor):
+    """
+    Gemini-CLI–compatible editor subset for filesystem operations used by Gemini models.
+
+    Currently implements only the `replace` command with behavior aligned to Gemini CLI:
+    - Literal replacement with CRLF/CR -> LF normalization when matching
+    - Error if old_string == new_string
+    - If old_string == '' and file does not exist: create file with new_string
+    - If file does not exist and old_string != '': return error
+    - expected_replacements default = 1; enforce exact match count; error on 0 or mismatch
+    """
+
+    TOOL_NAME = 'gemini_editor'
+
+    def __call__(
+        self,
+        *,
+        command: str,
+        path: str,
+        old_str: str | None = None,
+        new_str: str | None = None,
+        expected_replacements: int | None = None,
+        **_: object,
+    ) -> CLIResult:
+        _path = Path(path)
+        try:
+            if not _path.is_absolute():
+                raise EditorToolParameterInvalidError(
+                    'path',
+                    path,
+                    'The path should be an absolute path, starting with `/`.',
+                )
+            # Enforce workspace boundary if configured
+            if getattr(self, '_cwd', None) is not None:
+                try:
+                    if not _path.resolve().is_relative_to(self._cwd):  # type: ignore[arg-type]
+                        raise EditorToolParameterInvalidError(
+                            'path',
+                            path,
+                            f'Path must be inside the workspace root: {self._cwd}',
+                        )
+                except Exception:
+                    # Fallback: basic prefix check if resolve/is_relative_to not available
+                    if not str(_path.resolve()).startswith(str(self._cwd)):
+                        raise EditorToolParameterInvalidError(
+                            'path',
+                            path,
+                            f'Path must be inside the workspace root: {self._cwd}',
+                        )
+            if command != 'replace':
+                raise ToolError(
+                    f'Unrecognized command {command}. The allowed command for the {self.TOOL_NAME} tool is: replace.'
+                )
+            if new_str is None:
+                raise EditorToolParameterMissingError('replace', 'new_string')
+            if old_str is None:
+                old_str = ''
+            # Disallow directories
+            if _path.is_dir():
+                raise EditorToolParameterInvalidError(
+                    'path',
+                    path,
+                    f'The path {path} is a directory and only the `view` command can be used on directories.',
+                )
+
+            return self._replace(_path, old_str, new_str, expected_replacements)
+        except ToolError as e:
+            return CLIResult(error=e.message, path=str(_path))
+
+    def _replace(
+        self,
+        path: Path,
+        old: str,
+        new: str,
+        expected: int | None,
+    ) -> CLIResult:
+        if new == old:
+            # Match test expectation wording
+            raise EditorToolParameterInvalidError(
+                'new_string', new, 'new_string must be different from old_string'
+            )
+
+        # File existence handling
+        if not path.exists():
+            if old == '':
+                # Create new file with new_string content
+                self.write_file(path, new)
+                self._history_manager.add_history(path, '')
+                return CLIResult(
+                    output=f'File created successfully at: {path}',
+                    path=str(path),
+                    prev_exist=False,
+                    old_content='',
+                    new_content=new,
+                )
+            else:
+                # Match test expectation wording
+                raise ToolError(f'The file does not exist: {path}')
+
+        # Detect original EOL style from raw bytes and get text without translation
+        encoding = self._encoding_manager.get_encoding(path)
+        raw_bytes = path.read_bytes()
+        raw_text_no_translate = raw_bytes.decode(encoding, errors='replace')
+
+        # Normalized for matching
+        normalized_text = raw_text_no_translate.replace('\r\n', '\n').replace(
+            '\r', '\n'
+        )
+
+        if old == '':
+            # Creating into existing file is ambiguous; align to error
+            raise EditorToolParameterInvalidError(
+                'old_string',
+                old,
+                'old_string cannot be empty when the file already exists.',
+            )
+
+        count = normalized_text.count(old)
+        if expected is None:
+            expected = 1
+
+        if count == 0:
+            # Match test expectation wording
+            raise ToolError('0 occurrences found')
+        if count != expected:
+            # Match test expectation wording
+            raise ToolError(f'Expected {expected} occurrences but found {count}')
+
+        replaced_normalized = normalized_text.replace(old, new)
+
+        # Restore original line endings style based on raw bytes
+        if b'\r\n' in raw_bytes:
+            new_text = replaced_normalized.replace('\n', '\r\n')
+        elif b'\r' in raw_bytes and b'\n' not in raw_bytes:
+            new_text = replaced_normalized.replace('\n', '\r')
+        else:
+            new_text = replaced_normalized
+
+        # Persist changes
+        self._history_manager.add_history(path, raw_text_no_translate)
+        self.write_file(path, new_text)
+
+        success_message = f'Replaced {count} occurrence(s) in {path}.\nReview the changes and ensure they are as expected.'
+        return CLIResult(
+            output=success_message,
+            path=str(path),
+            prev_exist=True,
+            old_content=raw_text_no_translate,
+            new_content=new_text,
+        )

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "openhands-aci"
-version = "0.3.1"
+version = "0.3.2"
 description = "An Agent-Computer Interface (ACI) designed for software development agents OpenHands."
 authors = ["OpenHands"]
 license = "MIT"

tests/integration/editor/test_error_handling.py

@@ -86,18 +86,15 @@ def test_view_range_validation(temp_file):
         'should be a list of two integers' in result_json['formatted_output_and_error']
     )
 
-    # Test out of bounds range
+    # Test out of bounds range: should clamp to file end and show a warning
     result = file_editor(
         command='view',
         path=temp_file,
         view_range=[1, 10],  # File only has 3 lines
         enable_linting=False,
     )
     result_json = parse_result(result)
-    assert (
-        'should be smaller than the number of lines'
-        in result_json['formatted_output_and_error']
-    )
+    assert 'NOTE: We only show up to 3 since there\'re only 3 lines in this file.' in result_json['formatted_output_and_error']
 
     # Test invalid range order
     result = file_editor(

tests/unit/editor/test_gemini_editor.py

@@ -0,0 +1,101 @@
+from pathlib import Path
+
+import pytest
+
+from openhands_aci.editor.gemini_editor import GeminiEditor
+from openhands_aci.editor.exceptions import EditorToolParameterInvalidError, ToolError
+
+
+def write_text(path: Path, content: str, eol='\n'):
+    # Normalize to requested eol when writing for test setup
+    content = content.replace('\n', eol)
+    path.write_text(content)
+
+
+def read_bytes(path: Path) -> bytes:
+    return path.read_bytes()
+
+
+def test_replace_single_occurrence(tmp_path):
+    p = tmp_path / 'a.txt'
+    p.write_text('hello\nworld\n')
+    ed = GeminiEditor(workspace_root=str(tmp_path))
+    res = ed(command='replace', path=str(p), old_str='world', new_str='there')
+    assert res.error is None
+    assert res.prev_exist is True
+    assert res.old_content == 'hello\nworld\n'
+    assert res.new_content == 'hello\nthere\n'
+
+
+def test_replace_mismatch_expected_count(tmp_path):
+    p = tmp_path / 'b.txt'
+    p.write_text('x x x\n')
+    ed = GeminiEditor(workspace_root=str(tmp_path))
+    res = ed(command='replace', path=str(p), old_str='x', new_str='y', expected_replacements=2)
+    assert res.error is not None
+    assert 'expected 2 occurrences but found 3' in res.error.lower()
+
+
+def test_replace_zero_occurrence_error(tmp_path):
+    p = tmp_path / 'c.txt'
+    p.write_text('abc\n')
+    ed = GeminiEditor(workspace_root=str(tmp_path))
+    res = ed(command='replace', path=str(p), old_str='zzz', new_str='mmm')
+    assert res.error is not None
+    assert '0 occurrences found' in res.error.lower()
+
+
+def test_replace_old_equals_new_error(tmp_path):
+    p = tmp_path / 'd.txt'
+    p.write_text('same\n')
+    ed = GeminiEditor(workspace_root=str(tmp_path))
+    res = ed(command='replace', path=str(p), old_str='same', new_str='same')
+    assert res.error is not None
+    assert 'new_string must be different from old_string' in res.error.lower()
+
+
+def test_replace_create_when_missing_and_old_empty(tmp_path):
+    p = tmp_path / 'new.txt'
+    ed = GeminiEditor(workspace_root=str(tmp_path))
+    assert not p.exists()
+    res = ed(command='replace', path=str(p), old_str='', new_str='content')
+    assert res.error is None
+    assert p.exists()
+    assert p.read_text() == 'content'
+
+
+def test_replace_missing_file_and_old_not_empty_error(tmp_path):
+    p = tmp_path / 'missing.txt'
+    ed = GeminiEditor(workspace_root=str(tmp_path))
+    res = ed(command='replace', path=str(p), old_str='x', new_str='y')
+    assert res.error is not None
+    assert 'file does not exist' in res.error.lower()
+
+
+def test_crlf_normalization_and_preservation(tmp_path):
+    p = tmp_path / 'crlf.txt'
+    # Write with CRLF
+    write_text(p, 'A\nB\nC\n', eol='\r\n')
+    data_before = read_bytes(p)
+    ed = GeminiEditor(workspace_root=str(tmp_path))
+    res = ed(command='replace', path=str(p), old_str='B', new_str='X')
+    assert res.error is None
+    data_after = read_bytes(p)
+    # File should still contain CRLFs and same number of bytes change as expected
+    assert b'\r\n' in data_after
+    assert data_before != data_after
+
+
+def test_enforce_workspace_boundary(tmp_path):
+    outside = Path('/tmp/outside.txt')
+    ed = GeminiEditor(workspace_root=str(tmp_path))
+    res = ed(command='replace', path=str(outside), old_str='', new_str='x')
+    assert res.error is not None
+    assert 'workspace root' in res.error.lower()
+
+
+def test_directory_path_rejected(tmp_path):
+    ed = GeminiEditor(workspace_root=str(tmp_path))
+    res = ed(command='replace', path=str(tmp_path), old_str='a', new_str='b')
+    assert res.error is not None
+    assert 'is a directory' in res.error.lower()