feat(file_edit): implement streaming file I/O with async operations, …

[ykf173]

…atomic writes, and progress reporting

Implements streaming file editing capabilities based on write tool optimization proposal:

Features:

• Async non-blocking I/O using aiofiles (prevents UI freezing)
• Atomic writes via temp file + rename pattern (prevents data corruption)
• Real-time progress events via ToolCallProgressEvent
• Configurable timeout protection (asyncio.timeout)
• Chunked streaming for large files (default 64KB chunks)

APIs:

• StreamingFileEditor: Main class with async I/O and progress
• StreamingWriteTool: Sync-style wrapper for simple usage
• streaming_write_file()/streaming_edit_file(): Convenience functions

Testing:

• 24 comprehensive tests covering basic ops, progress events, atomic writes, timeouts, errors, edge cases
• All tests passing

Documentation:

• Full docstrings and type hints
• Implementation report (STREAMING_WRITE_IMPLEMENTATION_REPORT.md)
• Interactive demo script (examples/streaming_file_write_demo.py)

Backward compatible: Legacy edit_file_tool unchanged

[gemini-code-assist]

Code Review

This pull request implements a streaming file editing system featuring async I/O via aiofiles, atomic writes, and real-time progress events. The reviewer provided several actionable suggestions to improve the robustness and idiomatic quality of the code, such as using os.replace for cross-platform atomic renames, utilizing full UUIDs for temporary files to prevent collisions, and employing async with contexts for safer file handling. Additionally, feedback was given regarding the consolidation of local imports and the removal of redundant ones to enhance maintainability.

[gemini-code-assist]

os.rename can fail on Windows if the destination file already exists, raising a FileExistsError. os.replace is the cross-platform standard for atomic replacement that works even if the target exists.

Suggested change

	await asyncio.to_thread(os.rename, str(temp_path), str(path))
	await asyncio.to_thread(os.replace, str(temp_path), str(path))

[gemini-code-assist]

Using only the first 8 characters of a UUID for a temporary file suffix increases the risk of collisions in high-concurrency environments. Consider using a longer portion or the full UUID to ensure uniqueness.

Suggested change

	operation_id = str(uuid4())[:8]
	operation_id = str(uuid4())

[gemini-code-assist]

Scattered local imports can make the code harder to maintain and slightly slower if the function is called frequently. It is generally preferred to place all imports at the top of the module unless there is a specific reason for lazy loading (e.g., avoiding circular dependencies).

[gemini-code-assist]

Using async with for aiofiles.open is more idiomatic and safer than manually calling close() in a finally block. It ensures the file is closed correctly even if an exception occurs during the iteration.

            try:
                async with aiofiles.open(temp_path, mode="wb") as f:
                    for chunk_num in range(total_chunks):
                        start_idx = chunk_num * self.chunk_size
                        end_idx = min(start_idx + self.chunk_size, total_size)
                        chunk = content_bytes[start_idx:end_idx]

                        await f.write(chunk)
                        bytes_written += len(chunk)

                        # Calculate progress metrics
                        elapsed = time.monotonic() - start_time
                        bytes_per_second = bytes_written / elapsed if elapsed > 0 else 0
                        percentage = bytes_written / total_size if total_size > 0 else 1.0
                        remaining_bytes = total_size - bytes_written
                        estimated_seconds_remaining = (
                            remaining_bytes / bytes_per_second if bytes_per_second > 0 else 0
                        )

                        # Yield progress event every few chunks or at start/end
                        if chunk_num % 4 == 0 or chunk_num == total_chunks - 1:
                            progress_items: list[ToolCallContentItem] = [
                                LocationContentItem(path=str(path), line=0),
                                TextContentItem(
                                    text=(
                                        f"Writing: {bytes_written}/{total_size} bytes "
                                        f"({percentage * 100:.1f}%) "
                                        f"@ {bytes_per_second / 1024:.1f} KB/s"
                                    )
                                ),
                            ]

                            yield ToolCallProgressEvent(
                                tool_call_id=tool_call_id,
                                status="in_progress",
                                title=f"Writing {path.name}: {percentage * 100:.0f}%",
                                items=progress_items,
                                progress=chunk_num + 1,
                                total=total_chunks,
                            )

[gemini-code-assist]

This local import is redundant as compute_unified_diff is already imported at the top of the file on line 25.