diff --git a/src/kimi_cli/tools/file/replace.py b/src/kimi_cli/tools/file/replace.py index 4f551de4f4..e2e4e9eea1 100644 --- a/src/kimi_cli/tools/file/replace.py +++ b/src/kimi_cli/tools/file/replace.py @@ -79,12 +79,63 @@ async def _validate_path(self, path: KaosPath) -> ToolError | None: return None def _apply_edit(self, content: str, edit: Edit) -> str: - """Apply a single edit to the content.""" + """Apply a single edit to the content (string form; for display/tests).""" if edit.replace_all: return content.replace(edit.old, edit.new) else: return content.replace(edit.old, edit.new, 1) + @staticmethod + def _detect_line_ending(content: bytes) -> bytes: + """Return the file's dominant newline bytes by counting occurrences. + + ReadFile exposes lines with universal newlines, so the model always + supplies ``\\n`` in multi-line ``old``/``new``. Byte matching must + re-apply the on-disk ending or CRLF files reject every multi-line edit. + + Dominance is by count (not “any CRLF wins”): a mostly-LF file with a + stray ``\\r\\n`` keeps LF so multi-line ``old`` still matches. + Ties prefer LF. + """ + crlf = content.count(b"\r\n") + # Newlines that are not the second byte of a CRLF pair. + lf_only = content.count(b"\n") - crlf + if crlf > lf_only: + return b"\r\n" + return b"\n" + + @staticmethod + def _encode_edit_text(text: str, line_ending: bytes) -> bytes: + """Encode model text as UTF-8 using the file's on-disk newlines.""" + # Model / tool JSON always uses LF; normalize any mixed endings first. + normalized = text.replace("\r\n", "\n").replace("\r", "\n") + encoded = normalized.encode("utf-8") + if line_ending == b"\r\n": + return encoded.replace(b"\n", b"\r\n") + return encoded + + def _apply_edit_bytes(self, content: bytes, edit: Edit) -> bytes: + """Apply a single edit on raw bytes so non-UTF-8 regions stay intact. + + ``old``/``new`` come from the model as Unicode and are encoded as UTF-8, + with newlines rewritten to match the file's dominant line ending. + Searching/replacing in the raw byte stream avoids the + decode(errors=replace) → edit → re-encode round-trip that permanently + rewrites invalid sequences (e.g. ``\\xff`` → U+FFFD / ``EF BF BD``) + far from the requested edit (#2591). + """ + line_ending = self._detect_line_ending(content) + old_b = self._encode_edit_text(edit.old, line_ending) + new_b = self._encode_edit_text(edit.new, line_ending) + if not old_b: + return content + if edit.replace_all: + return content.replace(old_b, new_b) + idx = content.find(old_b) + if idx < 0: + return content + return content[:idx] + new_b + content[idx + len(old_b) :] + @override async def __call__(self, params: Params) -> ToolReturnValue: if not params.path: @@ -128,23 +179,33 @@ async def __call__(self, params: Params) -> ToolReturnValue: brief="Invalid path", ) - # Read the file content - content = await p.read_text(errors="replace") - - original_content = content + # Read raw bytes so non-UTF-8 sequences outside the edit are preserved + # (#2591 / same whole-file rewrite class as #2191). + raw = await p.read_bytes() + original_raw = raw edits = [params.edit] if isinstance(params.edit, Edit) else params.edit - # Apply all edits for edit in edits: - content = self._apply_edit(content, edit) + # Empty old is invalid: str.replace("", ...) is not a meaningful + # edit, and the byte path intentionally no-ops on empty needles. + if edit.old == "": + return ToolError( + message="The old string to replace cannot be empty.", + brief="Empty old string", + ) + + for edit in edits: + raw = self._apply_edit_bytes(raw, edit) - # Check if any changes were made - if content == original_content: + if raw == original_raw: return ToolError( message="No replacements were made. The old string was not found in the file.", brief="No replacements made", ) + # Diff is display-only: lossy decode is fine for the approval UI. + original_content = original_raw.decode("utf-8", errors="replace") + content = raw.decode("utf-8", errors="replace") diff_blocks: list[DisplayBlock] = await build_diff_blocks( str(p), original_content, content ) @@ -166,16 +227,19 @@ async def __call__(self, params: Params) -> ToolReturnValue: if not result: return result.rejection_error() - # Write the modified content back to the file - await p.write_text(content, errors="replace") + await p.write_bytes(raw) - # Count changes for success message + # Count changes for success message (byte-accurate for the edit strings) + line_ending = self._detect_line_ending(original_raw) total_replacements = 0 for edit in edits: + old_b = self._encode_edit_text(edit.old, line_ending) + if not old_b: + continue if edit.replace_all: - total_replacements += original_content.count(edit.old) + total_replacements += original_raw.count(old_b) else: - total_replacements += 1 if edit.old in original_content else 0 + total_replacements += 1 if old_b in original_raw else 0 return ToolReturnValue( is_error=False, diff --git a/tests/tools/test_str_replace_file.py b/tests/tools/test_str_replace_file.py index a16dad303b..aa3a20b690 100644 --- a/tests/tools/test_str_replace_file.py +++ b/tests/tools/test_str_replace_file.py @@ -246,3 +246,100 @@ async def test_replace_empty_strings( assert not result.is_error assert "successfully edited" in result.message assert await file_path.read_text() == "Hello !" + + +async def test_replace_rejects_empty_old_string( + str_replace_file_tool: StrReplaceFile, temp_work_dir: KaosPath +): + """Empty old is not a valid edit (Copilot review on byte-path semantics).""" + file_path = temp_work_dir / "test.txt" + await file_path.write_text("hello") + + result = await str_replace_file_tool(Params(path=str(file_path), edit=Edit(old="", new="x"))) + + assert result.is_error + assert "cannot be empty" in result.message + assert await file_path.read_text() == "hello" + + +async def test_replace_preserves_invalid_utf8_bytes_outside_edit( + str_replace_file_tool: StrReplaceFile, temp_work_dir: KaosPath +): + """Invalid UTF-8 far from the edit must not become U+FFFD (#2591).""" + file_path = temp_work_dir / "mixed.bin" + # 25 bytes: valid text with a lone 0xff between spaces + original = b"alpha\nbeta \xff gamma\ndelta\n" + await file_path.write_bytes(original) + + result = await str_replace_file_tool( + Params(path=str(file_path), edit=Edit(old="alpha", new="ALPHA")) + ) + + assert not result.is_error + out = await file_path.read_bytes() + assert out == b"ALPHA\nbeta \xff gamma\ndelta\n" + assert b"\xef\xbf\xbd" not in out # U+FFFD as UTF-8 + assert len(out) == len(original) + (len(b"ALPHA") - len(b"alpha")) + + +async def test_replace_multiline_crlf_file( + str_replace_file_tool: StrReplaceFile, temp_work_dir: KaosPath +): + """Multi-line edits must match CRLF files (model always supplies LF).""" + file_path = temp_work_dir / "crlf.txt" + original = b"Line 1\r\nLine 2\r\nLine 3\r\n" + await file_path.write_bytes(original) + + result = await str_replace_file_tool( + Params( + path=str(file_path), + edit=Edit(old="Line 2\nLine 3", new="Modified 2\nModified 3"), + ) + ) + + assert not result.is_error + out = await file_path.read_bytes() + assert out == b"Line 1\r\nModified 2\r\nModified 3\r\n" + # Non-edited region stays bit-identical (CRLF preserved) + assert out.startswith(b"Line 1\r\n") + + +async def test_replace_multiline_mostly_lf_with_stray_crlf( + str_replace_file_tool: StrReplaceFile, temp_work_dir: KaosPath +): + """A stray CRLF must not force whole-file CRLF rewriting of old/new (Copilot).""" + file_path = temp_work_dir / "mostly-lf.txt" + # Three LF newlines dominate a single CRLF elsewhere in the file. + original = b"Line 1\nLine 2\nLine 3\ntrailer\r\n" + await file_path.write_bytes(original) + + result = await str_replace_file_tool( + Params( + path=str(file_path), + edit=Edit(old="Line 2\nLine 3", new="Modified 2\nModified 3"), + ) + ) + + assert not result.is_error + out = await file_path.read_bytes() + assert out == b"Line 1\nModified 2\nModified 3\ntrailer\r\n" + + +async def test_replace_preserves_invalid_utf8_with_crlf_multiline( + str_replace_file_tool: StrReplaceFile, temp_work_dir: KaosPath +): + """CRLF multi-line edit must not corrupt invalid UTF-8 elsewhere.""" + file_path = temp_work_dir / "mixed-crlf.bin" + original = b"head\r\nkeep \xff me\r\ntail\r\n" + await file_path.write_bytes(original) + + result = await str_replace_file_tool( + Params( + path=str(file_path), + edit=Edit(old="head\nkeep", new="HEAD\nKEEP"), + ) + ) + assert not result.is_error + out = await file_path.read_bytes() + assert out == b"HEAD\r\nKEEP \xff me\r\ntail\r\n" + assert b"\xef\xbf\xbd" not in out