Day 4：Safe Edit

from __future__ import annotations

import difflib

import typer


def render_diff(old: str, new: str, path: str) -> str:
    """用 difflib 生成 unified diff，给增删行加 rich markup 着色。"""
    old_lines = old.splitlines(keepends=True)
    new_lines = new.splitlines(keepends=True)
    diff_lines = difflib.unified_diff(
        old_lines, new_lines,
        fromfile=f"a/{path}", tofile=f"b/{path}",
    )
    colored: list[str] = []
    for line in diff_lines:
        line = line.rstrip()
        if line.startswith("---") or line.startswith("+++"):
            colored.append(f"[bold]{line}[/bold]")
        elif line.startswith("-"):
            colored.append(f"[red]{line}[/red]")
        elif line.startswith("+"):
            colored.append(f"[green]{line}[/green]")
        elif line.startswith("@@"):
            colored.append(f"[cyan]{line}[/cyan]")
        else:
            colored.append(line)
    return "\n".join(colored)


def confirm_edit(path: str) -> bool:
    """让用户确认是否应用这次编辑，默认不应用。"""
    return typer.confirm(f"Apply this edit to {path}?", default=False)

def file_write(args: dict[str, Any], ctx: ToolContext) -> str:
    """整文件覆盖写入。前置校验由 agent.py 拦截块完成。"""
    path_str = args.get("file_path", "")
    content = args.get("content", "")
    if not path_str:
        return "error: missing required argument 'file_path'"
    try:
        path = resolve_in_cwd(ctx.cwd, path_str)
    except ValueError as exc:
        return f"error: {exc}"

    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8")
    # 写盘后刷新 read_state，让下一次编辑基于最新内容
    ctx.read_state.record(path, content)
    return f"Wrote {len(content)} chars to {path_str}"

    registry.register(
        Tool(
            name="file_write",
            description="Write or overwrite a file. Path is relative to cwd.",
            run=file_write,
            parameters={
                "type": "object",
                "properties": {
                    "file_path": {"type": "string", "description": "Relative path inside cwd."},
                    "content": {"type": "string", "description": "Full file content to write."},
                },
                "required": ["file_path", "content"],
            },
        )
    )

from rich.console import Console

from .diff_ui import confirm_edit, render_diff

console = Console()

    def emit(line: str) -> None:
        # 流式输出 trace：append 给测试用，print 给读者看
        trace.append(line)
        console.print(line)

            # file_write / file_edit 的 harness 拦截：先做前置校验，再渲染 diff，最后让用户确认
            if call.name in ("file_write", "file_edit"):
                path_str = call.arguments.get("file_path", "")

                # 1) 路径解析：越界 cwd 直接当 error
                try:
                    path = resolve_in_cwd(ctx.cwd, path_str)
                except (ValueError, OSError) as exc:
                    result = ToolResult(call.id, f"error: {exc}", is_error=True)
                    emit(f"observation: {result.content}")
                    tool_result_blocks.append(
                        {
                            "type": "tool_result",
                            "tool_use_id": result.tool_call_id,
                            "content": result.content,
                            "is_error": True,
                        }
                    )
                    continue

                old_content = path.read_text(encoding="utf-8") if path.exists() else ""

                # 2) 前置校验（v1 只对 file_write 做"文件存在则要求读过"）
                validation_error: str | None = None
                if call.name == "file_write" and path.exists():
                    if path not in ctx.read_state.entries:
                        validation_error = (
                            f"error: file has not been read yet. "
                            f"Read {path_str} first before editing."
                        )
                # file_edit 的校验在 v2 接上

                # 3) 算 new_content（v1 file_edit 还没接，先跳过 diff）
                new_content: str | None = None
                if call.name == "file_write":
                    new_content = call.arguments.get("content", "")

                # 4) 校验失败：不渲染 diff、不问用户，直接 error observation
                if validation_error is not None:
                    result = ToolResult(call.id, validation_error, is_error=True)
                    emit(f"observation: {result.content}")
                    tool_result_blocks.append(
                        {
                            "type": "tool_result",
                            "tool_use_id": result.tool_call_id,
                            "content": result.content,
                            "is_error": True,
                        }
                    )
                    continue

                # 5) 校验通过：渲染 diff + 用户确认
                if new_content is not None:
                    diff_text = render_diff(old_content, new_content, path_str)
                    console.print(f"\n[bold]Diff for {path_str}:[/bold]")
                    console.print(diff_text)
                    if not confirm_edit(path_str):
                        result = ToolResult(call.id, "error: edit rejected by user", is_error=True)
                        emit(f"observation: {result.content}")
                        tool_result_blocks.append(
                            {
                                "type": "tool_result",
                                "tool_use_id": result.tool_call_id,
                                "content": result.content,
                                "is_error": True,
                            }
                        )
                        continue

$ uv run agent-code "用 file_write 创建 hello.txt，内容一行 'hello from agent'"
Agent Code
cwd: /your/project
provider: anthropic  model: deepseek-v4-flash

tool_call: file_write {'file_path': 'hello.txt', 'content': 'hello from agent'}

Diff for hello.txt:
--- a/hello.txt
+++ b/hello.txt
@@ -0,0 +1 @@
+hello from agent
Apply this edit to hello.txt? [y/N]: y
observation: Wrote 16 chars to hello.txt
final: 已成功创建 hello.txt，内容为 hello from agent。

$ uv run agent-code "用 read_file 读 hello.txt，然后用 file_write 把内容改成 'goodbye from agent'"
...
tool_call: read_file {'path': 'hello.txt'}
observation: hello from agent

tool_call: file_write {'file_path': 'hello.txt', 'content': 'goodbye from agent'}

Diff for hello.txt:
--- a/hello.txt
+++ b/hello.txt
@@ -1 +1 @@
-hello from agent
+goodbye from agent
Apply this edit to hello.txt? [y/N]: y
observation: Wrote 18 chars to hello.txt
final: 已完成！hello.txt 已从 'hello from agent' 改为 'goodbye from agent'。

def ensure_read_before_edit(state: ReadFileState, path: Path) -> str | None:
    """检查文件是否在本次会话中被读过。没读过返回 error 字符串。"""
    if path not in state.entries:
        return (
            f"error: file has not been read yet. "
            f"Read {path.name} first before editing."
        )
    return None


def check_mtime_conflict(state: ReadFileState, path: Path) -> str | None:
    """检查文件在 read 之后是否被外部修改过。mtime 变了返回 error。
    教学版不做 content-equals 兜底——mtime 变就判冲突，再读一次即可。"""
    entry = state.entries.get(path)
    if entry is None:
        return None  # ensure_read_before_edit 会先拦住没读的情况
    read_mtime_ns, _ = entry
    try:
        current_mtime_ns = path.stat().st_mtime_ns
    except OSError:
        return None
    if current_mtime_ns > read_mtime_ns:
        return (
            f"error: file was modified after read. "
            f"Read {path.name} again before editing."
        )
    return None


def apply_single_replace(
    content: str, old: str, new: str, replace_all: bool
) -> tuple[str | None, str | None]:
    """在 content 中查找 old 并替换为 new。
    返回 (new_content, error)：成功时 error 为 None，失败时 new_content 为 None。"""
    if old == "":
        # str.count("") 会返回 len+1，str.replace("", x) 会在每个字符之间插入 x。
        # 这两个行为对模型完全没用，直接拒绝。
        return None, "error: old_string must not be empty."
    if old == new:
        return None, "error: old_string and new_string are exactly the same."

    count = content.count(old)
    if count == 0:
        return None, "error: string to replace not found in file."
    if count > 1 and not replace_all:
        return None, (
            f"error: found {count} matches for old_string. "
            f"Use replace_all=True to replace all, or make old_string more specific."
        )

    if replace_all:
        return content.replace(old, new), None
    else:
        return content.replace(old, new, 1), None

def file_edit(args: dict[str, Any], ctx: ToolContext) -> str:
    """字符串替换编辑。前置校验在 agent.py 拦截块里完成。"""
    path_str = args.get("file_path", "")
    old_string = args.get("old_string", "")
    new_string = args.get("new_string", "")
    replace_all = bool(args.get("replace_all", False))
    if not path_str:
        return "error: missing required argument 'file_path'"
    try:
        path = resolve_in_cwd(ctx.cwd, path_str)
    except ValueError as exc:
        return f"error: {exc}"

    try:
        content = path.read_text(encoding="utf-8")
    except (FileNotFoundError, IsADirectoryError) as exc:
        return f"error: {exc}"

    # 防 race：agent.py 已经做过一次 apply_single_replace 算 diff，
    # 如果 confirm 那一刻到现在 old_content 又被外部改过，这里会再兜一次。
    new_content, err = apply_single_replace(content, old_string, new_string, replace_all)
    if err:
        return err

    path.write_text(new_content, encoding="utf-8")
    ctx.read_state.record(path, new_content)
    return f"Edited {path_str}: replaced {len(old_string)} chars with {len(new_string)} chars"

                # 2) 前置校验：read-before-edit + mtime 冲突
                validation_error: str | None = None
                if call.name == "file_write":
                    if path.exists():
                        validation_error = (
                            ensure_read_before_edit(ctx.read_state, path)
                            or check_mtime_conflict(ctx.read_state, path)
                        )
                else:  # file_edit
                    if not path.exists():
                        validation_error = f"error: file does not exist: {path_str}"
                    else:
                        validation_error = (
                            ensure_read_before_edit(ctx.read_state, path)
                            or check_mtime_conflict(ctx.read_state, path)
                        )

                # 3) 算 new_content：file_write 直接拿 content；file_edit 试跑替换
                new_content: str | None = None
                if call.name == "file_write":
                    new_content = call.arguments.get("content", "")
                elif call.name == "file_edit" and validation_error is None:
                    new_content, replace_err = apply_single_replace(
                        old_content,
                        call.arguments.get("old_string", ""),
                        call.arguments.get("new_string", ""),
                        bool(call.arguments.get("replace_all", False)),
                    )
                    if replace_err is not None:
                        validation_error = replace_err

$ uv run agent-code "先 read_file hello.txt，再用 file_edit 把里面的 goodbye 改成 hola"
tool_call: read_file {'path': 'hello.txt'}
observation: goodbye from agent
tool_call: file_edit {'file_path': 'hello.txt', 'old_string': 'goodbye', 'new_string': 'hola'}

Diff for hello.txt:
--- a/hello.txt
+++ b/hello.txt
@@ -1 +1 @@
-goodbye from agent
+hola from agent
Apply this edit to hello.txt? [y/N]: y
observation: Edited hello.txt: replaced 7 chars with 4 chars
final: 修改成功！hello.txt 已从 goodbye from agent 变更为 hola from agent。

$ uv run agent-code "不要调用 read_file。直接用 file_edit 把 hello.txt 里的 hola 改成 hello。"
tool_call: file_edit {'file_path': 'hello.txt', 'old_string': 'hola', 'new_string': 'hello'}
observation: error: file has not been read yet. Read hello.txt first before editing.
tool_call: read_file {'path': 'hello.txt'}
observation: hola from agent
...

$ echo -e "line one: hello\nline two: hello" > dup.txt
$ uv run agent-code --max-steps 2 "先 read_file dup.txt，再用 file_edit 把里面的 hello 改成 hi。old_string 必须精确等于 hello，replace_all 必须是 false。"
tool_call: read_file {'path': 'dup.txt'}
observation: line one: hello
line two: hello

tool_call: file_edit {'file_path': 'dup.txt', 'old_string': 'hello', 'new_string': 'hi', 'replace_all': False}
observation: error: found 2 matches for old_string. Use replace_all=True to replace all, or make old_string more specific.
final: reached max_steps=2

$ uv run agent-code "先 read_file dup.txt，然后把 dup.txt 里的 hello 改成 bonjour。"

observation: error: file was modified after read. Read dup.txt again before editing.

from __future__ import annotations

from datetime import datetime, timezone
from pathlib import Path


def backup(cwd: Path, path: Path, old_content: str) -> Path | None:
    """写盘前把文件旧内容备份到 .agent/history/<rel>/<ts>。
    备份不是工具，模型看不到它——它是 harness 的全局安全网。
    失败不阻塞编辑，返回 None。"""
    try:
        rel = path.resolve().relative_to(cwd.resolve())
    except ValueError:
        return None  # 路径在 cwd 外，不备份

    ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S.%f")[:-3] + "Z"
    backup_dir = cwd / ".agent" / "history" / rel
    backup_dir.mkdir(parents=True, exist_ok=True)
    backup_path = backup_dir / ts

    try:
        backup_path.write_text(old_content, encoding="utf-8")
    except OSError:
        return None
    return backup_path

    if path.exists():
        # 备份旧内容（备份失败不阻塞写盘）
        try:
            old = path.read_text(encoding="utf-8")
            backup(ctx.cwd, path, old)
        except Exception:
            pass
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8")

    backup(ctx.cwd, path, content)  # 写盘前备份旧内容

$ uv run agent-code "先 read_file hello.txt，再把 hello 改成 hola"
...
tool_call: file_edit {'file_path': 'hello.txt', 'old_string': 'hello', 'new_string': 'hola'}
Diff for hello.txt:
...
Apply this edit to hello.txt? [y/N]: y
observation: Edited hello.txt: replaced 5 chars with 4 chars

$ ls .agent/history/hello.txt/
20260526T120000.123Z

$ cat .agent/history/hello.txt/20260526T120000.123Z
hello from agent