diff --git a/docs/dev/architecture.md b/docs/dev/architecture.md index 856b58a..abad9d9 100644 --- a/docs/dev/architecture.md +++ b/docs/dev/architecture.md @@ -1,7 +1,7 @@ # Yada architecture Yada is a deliberately small DeepSeek-native coding harness: one model loop, -five tools, an append-only conversation, checked patch application, and a +six tools, an append-only conversation, checked patch application, and a verification gate. This document describes the internal boundaries contributors must preserve. @@ -34,7 +34,7 @@ run/cli.py │ │ └── tools/runner.py │ │ ├── environments/workspace.py │ │ ├── environments/approval.py - │ │ └── tools/{search,read,patch,command,finish}.py + │ │ └── tools/{search,read,replace,patch,command,finish}.py │ ├── models/base.py ← models/deepseek.py │ └── traces/jsonl.py ← traces/report.py └── evals/cli.py @@ -94,12 +94,13 @@ without weakening the workspace and tool contracts. ## Tool system -Yada exposes five tools: +Yada exposes six tools: | Tool | Responsibility | | --- | --- | | `search_code` | Search repository text with ripgrep and a Python fallback. | | `read_file` | Return bounded, numbered text plus a SHA-256 content hash. | +| `replace_text` | Apply exact unique replacements to existing UTF-8 files. | | `apply_patch` | Validate and apply a Git-style unified diff. | | `run_command` | Run an approved argv array and return bounded structured output. | | `finish` | End only after verification of the latest revision. | @@ -126,6 +127,12 @@ This is an optimistic transaction: a file changed after the model read it is a conflict, not permission to apply a stale edit. Every successful patch increments the workspace revision and invalidates earlier verification. +`replace_text` uses the same transaction rather than introducing another write +path. It validates every SHA and exact unique match in memory, applies same-file +edits in declaration order, generates a standard-library unified diff, and sends +the complete result through `apply_patch`. Zero or ambiguous matches fail closed; +no file changes until the generated multi-file patch passes validation. + ## Commands and verification `run_command` accepts argv, not a shell string. It checks the executable diff --git a/src/yada/agents/prompts.py b/src/yada/agents/prompts.py index 48ed717..55802a8 100644 --- a/src/yada/agents/prompts.py +++ b/src/yada/agents/prompts.py @@ -7,8 +7,8 @@ Rules: 1. Search before reading, and read a file before editing it. -2. read_file returns a SHA-256. apply_patch requires the current SHA-256 for every - existing file it touches, or the literal NEW for a new file. +2. read_file returns a SHA-256. replace_text and apply_patch require the current + SHA-256 for every existing file they touch; apply_patch uses NEW for a new file. 3. Prefer small unified diffs. Do not rewrite unrelated code. 4. Run the most relevant available tests after the last patch. A successful inspection command is not a test. @@ -22,6 +22,7 @@ Tool strategy: - search_code: locate symbols and references. - read_file: inspect bounded line ranges and obtain a file hash. +- replace_text: make exact, unique, version-checked replacements in existing text. - apply_patch: make a version-checked unified-diff edit. - run_command: inspect or verify with an argv array; no shell syntax. - finish: submit only after the verification gate is satisfied. diff --git a/src/yada/tools/replace.py b/src/yada/tools/replace.py new file mode 100644 index 0000000..4d9d996 --- /dev/null +++ b/src/yada/tools/replace.py @@ -0,0 +1,427 @@ +"""Exact SHA-bound transactional text replacement.""" + +from __future__ import annotations + +import difflib +import hashlib +import json +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from yada.exceptions import ToolError +from yada.tools.base import ToolContext +from yada.tools.patch import _resolve_patch_target, apply_patch + +_MAX_EDITS = 100 +_MAX_EDIT_TEXT_CHARS = 100_000 +_MAX_TOTAL_TEXT_CHARS = 200_000 +_MAX_FILE_BYTES = 1_000_000 +_MAX_GENERATED_PATCH_CHARS = 250_000 +_MAX_MATCH_LINES = 20 +_MAX_ERROR_CHARS = 2_000 +_MAX_PATH_CHARS = 300 +_MAX_PATH_INPUT_CHARS = 4_096 +_EDIT_FIELDS = frozenset({"path", "sha256", "old_text", "new_text"}) + + +@dataclass +class _FileReplacement: + path: str + sha256: str + before: str + after: str + + +def replace_text( + context: ToolContext, + edits: list[dict[str, str]], +) -> dict[str, Any]: + """Apply ordered exact replacements as one checked patch transaction. + + Every path must identify an existing regular UTF-8 text file, and every + replacement is evaluated against an in-memory version in declaration order. + No workspace file is mutated until all edits validate. + """ + + if not isinstance(edits, list) or not edits: + raise _replace_error( + "edits must be a non-empty array", + "invalid_edit", + recovery="Provide at least one exact text replacement.", + ) + if len(edits) > _MAX_EDITS: + raise _replace_error( + f"edits exceeds the {_MAX_EDITS}-item limit", + "invalid_edit", + recovery="Split the replacements into smaller transactions.", + ) + + files: dict[str, _FileReplacement] = {} + total_text_chars = 0 + for index, item in enumerate(edits): + path, sha256, old_text, new_text = _validate_edit(item, index) + total_text_chars += len(old_text) + len(new_text) + if total_text_chars > _MAX_TOTAL_TEXT_CHARS: + raise _replace_error( + f"combined old_text and new_text exceed the " + f"{_MAX_TOTAL_TEXT_CHARS}-character limit", + "invalid_edit", + path=path, + details={"edit_index": index}, + recovery="Split the replacements into smaller transactions.", + ) + + try: + file_path = _resolve_patch_target(context, path) + except ToolError: + raise + except (OSError, RuntimeError, ValueError) as exc: + raise _replace_error( + f"could not resolve replace_text target {_clip_path(path)}: {exc}", + "unsupported_target", + path=path, + ) from exc + relative_path = context.workspace.display(file_path) + state = files.get(relative_path) + if state is None: + state = _load_file(relative_path, file_path, sha256) + files[relative_path] = state + elif sha256 != state.sha256: + raise _replace_error( + f"all edits for {_clip_path(relative_path)} must use the same " + "starting sha256", + "invalid_edit", + path=relative_path, + details={"edit_index": index}, + recovery="Use the SHA-256 from the same read_file result.", + ) + + offsets = _find_match_offsets(state.after, old_text) + if not offsets: + raise _replace_error( + f"old_text was not found in {_clip_path(relative_path)}", + "no_match", + path=relative_path, + details={ + "edit_index": index, + "current_sha256": state.sha256, + }, + recovery="Read the current file and retry with exact source text.", + ) + if len(offsets) > 1: + match_lines = list( + dict.fromkeys( + state.after.count("\n", 0, offset) + 1 for offset in offsets + ) + )[:_MAX_MATCH_LINES] + raise _replace_error( + f"old_text is ambiguous in {_clip_path(relative_path)}", + "ambiguous_match", + path=relative_path, + details={"edit_index": index, "match_lines": match_lines}, + recovery="Read a narrower range and provide a larger unique anchor.", + ) + + offset = offsets[0] + state.after = ( + state.after[:offset] + new_text + state.after[offset + len(old_text) :] + ) + + for state in files.values(): + if state.after == state.before: + raise _replace_error( + f"edits produce no net change for {_clip_path(state.path)}", + "invalid_edit", + path=state.path, + recovery="Remove cancelling edits or provide the intended replacement.", + ) + + patch = _build_patch(files) + if len(patch) > _MAX_GENERATED_PATCH_CHARS: + raise _replace_error( + "generated patch exceeds the 250 KB transaction limit", + "invalid_edit", + details={"paths": _bounded_paths(files)}, + recovery="Split the replacements into smaller transactions.", + ) + + expected_files = [ + {"path": state.path, "sha256": state.sha256} + for state in sorted(files.values(), key=lambda item: item.path) + ] + try: + result = apply_patch(context, patch, expected_files) + except ToolError as exc: + if exc.error_code not in {"invalid_patch", "patch_context_mismatch"}: + raise + details = dict(exc.details or {}) + details["cause_error_code"] = exc.error_code + raise _replace_error( + "validated replacement transaction could not be applied", + "apply_failed", + details=details, + recovery="Read the current files and retry from the latest workspace state.", + ) from exc + + result["message"] = "text replaced; run a relevant test before finish" + return result + + +def _validate_edit( + item: Any, + index: int, +) -> tuple[str, str, str, str]: + if not isinstance(item, dict): + raise _replace_error( + "each edits entry must be an object", + "invalid_edit", + details={"edit_index": index}, + ) + if set(item) != _EDIT_FIELDS: + raise _replace_error( + "each edits entry must contain exactly path, sha256, old_text, and " + "new_text", + "invalid_edit", + details={ + "edit_index": index, + "missing_fields": sorted(_EDIT_FIELDS - set(item)), + "unexpected_fields": sorted( + str(field) for field in set(item) - _EDIT_FIELDS + )[:20], + }, + ) + path = item.get("path") + sha256 = item.get("sha256") + old_text = item.get("old_text") + new_text = item.get("new_text") + if not all(isinstance(value, str) for value in (path, sha256, old_text, new_text)): + raise _replace_error( + "edits entries require string path, sha256, old_text, and new_text", + "invalid_edit", + details={"edit_index": index}, + ) + if not path.strip(): + raise _replace_error( + "edit path must be a non-empty string", + "invalid_edit", + details={"edit_index": index}, + ) + if len(path) > _MAX_PATH_INPUT_CHARS: + raise _replace_error( + f"edit path exceeds the {_MAX_PATH_INPUT_CHARS}-character limit", + "unsupported_target", + path=path, + details={"edit_index": index}, + ) + if any(character in path for character in ("\0", "\n", "\r", "\t")): + raise _replace_error( + "edit path contains an unsupported control character", + "unsupported_target", + path=path, + details={"edit_index": index}, + ) + try: + path.encode("utf-8") + except UnicodeEncodeError as exc: + raise _replace_error( + "edit path is not valid UTF-8", + "unsupported_target", + details={"edit_index": index}, + ) from exc + if not re.fullmatch(r"[0-9a-f]{64}", sha256): + raise _replace_error( + f"invalid sha256 for {_clip_path(path)}", + "invalid_edit", + path=path, + details={"edit_index": index}, + recovery="Use the SHA-256 returned by read_file.", + ) + if not old_text: + raise _replace_error( + "old_text must be non-empty", + "invalid_edit", + path=path, + details={"edit_index": index}, + ) + if old_text == new_text: + raise _replace_error( + "old_text and new_text must differ", + "invalid_edit", + path=path, + details={"edit_index": index}, + ) + if "\0" in old_text or "\0" in new_text: + raise _replace_error( + "old_text and new_text must not contain NUL bytes", + "invalid_edit", + path=path, + details={"edit_index": index}, + ) + try: + old_text.encode("utf-8") + new_text.encode("utf-8") + except UnicodeEncodeError as exc: + raise _replace_error( + "old_text and new_text must be valid UTF-8", + "invalid_edit", + path=path, + details={"edit_index": index}, + ) from exc + if max(len(old_text), len(new_text)) > _MAX_EDIT_TEXT_CHARS: + raise _replace_error( + f"old_text and new_text are limited to {_MAX_EDIT_TEXT_CHARS} " + "characters per edit", + "invalid_edit", + path=path, + details={"edit_index": index}, + recovery="Split the replacement into smaller edits.", + ) + return path, sha256, old_text, new_text + + +def _load_file( + path: str, + file_path: Path, + declared_sha256: str, +) -> _FileReplacement: + if not file_path.exists(): + raise _replace_error( + f"replace_text only supports existing files: {_clip_path(path)}", + "unsupported_target", + path=path, + recovery="Use apply_patch to create a new file.", + ) + if not file_path.is_file(): + raise _replace_error( + f"replace_text target is not a regular file: {_clip_path(path)}", + "unsupported_target", + path=path, + ) + try: + with file_path.open("rb") as handle: + content = handle.read(_MAX_FILE_BYTES + 1) + except OSError as exc: + raise _replace_error( + f"could not read replace_text target {_clip_path(path)}: {exc}", + "unsupported_target", + path=path, + ) from exc + if len(content) > _MAX_FILE_BYTES: + raise _replace_error( + f"replace_text target exceeds the {_MAX_FILE_BYTES}-byte limit", + "unsupported_target", + path=path, + recovery="Use apply_patch for an unsuitable large structural edit.", + ) + if b"\0" in content: + raise _replace_error( + f"replace_text target appears to be binary: {_clip_path(path)}", + "unsupported_target", + path=path, + ) + try: + text = content.decode("utf-8") + except UnicodeDecodeError as exc: + raise _replace_error( + f"replace_text target is not valid UTF-8: {_clip_path(path)}", + "unsupported_target", + path=path, + ) from exc + + actual_sha256 = hashlib.sha256(content).hexdigest() + if declared_sha256 != actual_sha256: + raise _replace_error( + f"stale file hash for {_clip_path(path)}: expected {declared_sha256}, " + f"current {actual_sha256}", + "stale_hash", + path=path, + details={ + "expected_sha256": declared_sha256, + "current_sha256": actual_sha256, + }, + recovery="Read the current file and retry with its latest SHA-256.", + ) + return _FileReplacement(path, declared_sha256, text, text) + + +def _find_match_offsets(text: str, old_text: str) -> list[int]: + offsets: list[int] = [] + start = 0 + while len(offsets) <= _MAX_MATCH_LINES: + offset = text.find(old_text, start) + if offset < 0: + break + offsets.append(offset) + start = offset + 1 + return offsets + + +def _build_patch(files: dict[str, _FileReplacement]) -> str: + chunks: list[str] = [] + for state in sorted(files.values(), key=lambda item: item.path): + old_path = _quote_diff_path(f"a/{state.path}") + new_path = _quote_diff_path(f"b/{state.path}") + chunks.append(f"diff --git {old_path} {new_path}\n") + diff = difflib.unified_diff( + _split_lf_lines(state.before), + _split_lf_lines(state.after), + fromfile=old_path, + tofile=new_path, + lineterm="\n", + ) + for line in diff: + chunks.append(line) + if line[:1] in {" ", "+", "-"} and not line.endswith("\n"): + chunks.append("\n\\ No newline at end of file\n") + return "".join(chunks) + + +def _split_lf_lines(text: str) -> list[str]: + parts = text.split("\n") + lines = [f"{part}\n" for part in parts[:-1]] + if parts[-1]: + lines.append(parts[-1]) + return lines + + +def _quote_diff_path(path: str) -> str: + if all(ord(character) >= 33 and character not in {'"', "\\"} for character in path): + return path + return json.dumps(path, ensure_ascii=False) + + +def _replace_error( + message: str, + error_code: str, + *, + path: str | None = None, + details: dict[str, Any] | None = None, + recovery: str | None = None, +) -> ToolError: + payload = dict(details or {}) + if path is not None: + payload["paths"] = [_clip_path(path)] + if recovery is not None: + payload["recovery"] = _clip(recovery) + return ToolError( + _clip(message), + error_code=error_code, + details=payload or None, + ) + + +def _bounded_paths(paths: Any) -> list[str]: + values = paths.keys() if isinstance(paths, dict) else paths + return [_clip_path(str(path)) for path in sorted(values)[:20]] + + +def _clip_path(path: str) -> str: + return _clip(path, _MAX_PATH_CHARS) + + +def _clip(value: str, limit: int = _MAX_ERROR_CHARS) -> str: + if len(value) <= limit: + return value + return f"{value[: limit - 3]}..." diff --git a/src/yada/tools/runner.py b/src/yada/tools/runner.py index 12f2fb0..d864828 100644 --- a/src/yada/tools/runner.py +++ b/src/yada/tools/runner.py @@ -12,6 +12,7 @@ from yada.tools.finish import final_state, finish from yada.tools.patch import apply_patch from yada.tools.read import read_file +from yada.tools.replace import replace_text from yada.tools.schemas import TOOL_SCHEMAS from yada.tools.search import search_code @@ -42,6 +43,7 @@ def __init__( "search_code": search_code, "read_file": read_file, "apply_patch": apply_patch, + "replace_text": replace_text, "run_command": run_command, } diff --git a/src/yada/tools/schemas.py b/src/yada/tools/schemas.py index e761a0b..4d522c1 100644 --- a/src/yada/tools/schemas.py +++ b/src/yada/tools/schemas.py @@ -1,4 +1,4 @@ -"""Stable DeepSeek function schemas for Yada's five tools.""" +"""Stable DeepSeek function schemas for Yada's six tools.""" from __future__ import annotations @@ -79,6 +79,41 @@ }, }, }, + { + "type": "function", + "function": { + "name": "replace_text", + "description": "Replace exact unique text in existing UTF-8 files as one SHA-bound transaction. Edits to the same file run in declared order.", + "parameters": { + "type": "object", + "properties": { + "edits": { + "type": "array", + "minItems": 1, + "maxItems": 100, + "items": { + "type": "object", + "properties": { + "path": {"type": "string"}, + "sha256": {"type": "string"}, + "old_text": {"type": "string"}, + "new_text": {"type": "string"}, + }, + "required": [ + "path", + "sha256", + "old_text", + "new_text", + ], + "additionalProperties": False, + }, + } + }, + "required": ["edits"], + "additionalProperties": False, + }, + }, + }, { "type": "function", "function": { diff --git a/tests/tools/test_replace.py b/tests/tools/test_replace.py new file mode 100644 index 0000000..64ea8b3 --- /dev/null +++ b/tests/tools/test_replace.py @@ -0,0 +1,380 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from yada.agents.executor import Executor +from yada.exceptions import ToolError +from yada.tools import ToolRunner +from yada.traces import TraceWriter, read_trace + + +def _edit( + tool_runner: ToolRunner, + path: str, + old_text: str, + new_text: str, + *, + sha256: str | None = None, +) -> dict[str, str]: + file_path = tool_runner.workspace.resolve(path) + return { + "path": path, + "sha256": sha256 or tool_runner.workspace.sha256(file_path), + "old_text": old_text, + "new_text": new_text, + } + + +def test_replace_text_is_public_and_updates_edit_state( + git_workspace: Path, tool_runner: ToolRunner +) -> None: + names = [schema["function"]["name"] for schema in tool_runner.schemas] + assert "replace_text" in names + tool_runner.context.state.verified_revision = 0 + + result = tool_runner.execute( + "replace_text", + {"edits": [_edit(tool_runner, "app.py", "return 41", "return 42")]}, + ) + + assert result.data["ok"], result.data + assert (git_workspace / "app.py").read_text() == "def answer():\n return 42\n" + assert result.data["changed_files"] == [ + { + "path": "app.py", + "sha256": tool_runner.workspace.sha256(git_workspace / "app.py"), + } + ] + assert tool_runner.context.state.revision == 1 + assert tool_runner.context.state.patch_count == 1 + assert tool_runner.context.state.touched_files == {"app.py"} + assert tool_runner.context.state.verified_revision == -1 + assert not tool_runner.execute("finish", {"summary": "done"}).data["ok"] + + +def test_no_match_is_structured_and_does_not_change_state( + git_workspace: Path, tool_runner: ToolRunner +) -> None: + tool_runner.context.state.verified_revision = 0 + + result = tool_runner.execute( + "replace_text", + {"edits": [_edit(tool_runner, "app.py", "return 99", "return 42")]}, + ) + + assert result.data["error_code"] == "no_match" + assert result.data["details"]["paths"] == ["app.py"] + assert "recovery" in result.data["details"] + assert (git_workspace / "app.py").read_text() == "def answer():\n return 41\n" + assert tool_runner.context.state.revision == 0 + assert tool_runner.context.state.patch_count == 0 + assert tool_runner.context.state.verified_revision == 0 + + +def test_overlapping_matches_are_ambiguous( + git_workspace: Path, tool_runner: ToolRunner +) -> None: + repeated = git_workspace / "repeated.txt" + repeated.write_text("aaaa\n", encoding="utf-8") + + result = tool_runner.execute( + "replace_text", + {"edits": [_edit(tool_runner, "repeated.txt", "aa", "b")]}, + ) + + assert result.data["error_code"] == "ambiguous_match" + assert result.data["details"]["match_lines"] == [1] + assert repeated.read_text() == "aaaa\n" + + +def test_stale_hash_reports_current_hash( + git_workspace: Path, tool_runner: ToolRunner +) -> None: + result = tool_runner.execute( + "replace_text", + { + "edits": [ + _edit( + tool_runner, + "app.py", + "return 41", + "return 42", + sha256="0" * 64, + ) + ] + }, + ) + + assert result.data["error_code"] == "stale_hash" + assert result.data["details"]["current_sha256"] == tool_runner.workspace.sha256( + git_workspace / "app.py" + ) + assert "return 41" in (git_workspace / "app.py").read_text() + + +def test_same_file_edits_run_in_order_against_memory( + git_workspace: Path, tool_runner: ToolRunner +) -> None: + target = git_workspace / "ordered.txt" + target.write_text("alpha beta\n", encoding="utf-8") + digest = tool_runner.workspace.sha256(target) + + result = tool_runner.execute( + "replace_text", + { + "edits": [ + _edit( + tool_runner, + "ordered.txt", + "alpha", + "gamma", + sha256=digest, + ), + _edit( + tool_runner, + "ordered.txt", + "gamma beta", + "done", + sha256=digest, + ), + ] + }, + ) + + assert result.data["ok"], result.data + assert target.read_text() == "done\n" + + +def test_same_file_edits_require_one_starting_hash( + git_workspace: Path, tool_runner: ToolRunner +) -> None: + digest = tool_runner.workspace.sha256(git_workspace / "app.py") + result = tool_runner.execute( + "replace_text", + { + "edits": [ + _edit( + tool_runner, + "app.py", + "return 41", + "return 42", + sha256=digest, + ), + _edit( + tool_runner, + "app.py", + "return 42", + "return 43", + sha256="0" * 64, + ), + ] + }, + ) + + assert result.data["error_code"] == "invalid_edit" + assert "return 41" in (git_workspace / "app.py").read_text() + + +def test_later_file_failure_rolls_back_every_edit( + git_workspace: Path, tool_runner: ToolRunner +) -> None: + second = git_workspace / "second.py" + second.write_text("VALUE = 1\n", encoding="utf-8") + + result = tool_runner.execute( + "replace_text", + { + "edits": [ + _edit(tool_runner, "app.py", "return 41", "return 42"), + _edit(tool_runner, "second.py", "VALUE = 9", "VALUE = 2"), + ] + }, + ) + + assert result.data["error_code"] == "no_match" + assert (git_workspace / "app.py").read_text() == "def answer():\n return 41\n" + assert second.read_text() == "VALUE = 1\n" + assert tool_runner.context.state.revision == 0 + + +def test_unsupported_text_targets_are_rejected( + git_workspace: Path, tool_runner: ToolRunner +) -> None: + (git_workspace / "folder").mkdir() + (git_workspace / "link.py").symlink_to("app.py") + binary = git_workspace / "binary.dat" + binary.write_bytes(b"valid utf-8\0binary") + invalid_utf8 = git_workspace / "invalid.dat" + invalid_utf8.write_bytes(b"\xff\xfe") + oversized = git_workspace / "oversized.txt" + oversized.write_bytes(b"x" * 1_000_001) + cases = [ + ("missing.py", "0" * 64), + ("folder", "0" * 64), + ("link.py", "0" * 64), + ("binary.dat", tool_runner.workspace.sha256(binary)), + ("invalid.dat", tool_runner.workspace.sha256(invalid_utf8)), + ("oversized.txt", tool_runner.workspace.sha256(oversized)), + (".git/config", "0" * 64), + ("../outside.py", "0" * 64), + ] + + for path, digest in cases: + result = tool_runner.execute( + "replace_text", + { + "edits": [ + { + "path": path, + "sha256": digest, + "old_text": "old", + "new_text": "new", + } + ] + }, + ) + assert result.data["error_code"] == "unsupported_target", result.data + + assert tool_runner.context.state.revision == 0 + assert "return 41" in (git_workspace / "app.py").read_text() + + +def test_invalid_and_oversized_edits_are_bounded( + tool_runner: ToolRunner, +) -> None: + empty = tool_runner.execute("replace_text", {"edits": []}) + unchanged = tool_runner.execute( + "replace_text", + {"edits": [_edit(tool_runner, "app.py", "return 41", "return 41")]}, + ) + oversized = tool_runner.execute( + "replace_text", + {"edits": [_edit(tool_runner, "app.py", "return 41", "x" * 100_001)]}, + ) + + assert empty.data["error_code"] == "invalid_edit" + assert unchanged.data["error_code"] == "invalid_edit" + assert oversized.data["error_code"] == "invalid_edit" + assert len(oversized.content) < 4_000 + + +def test_whole_file_can_be_replaced_with_empty_text( + git_workspace: Path, tool_runner: ToolRunner +) -> None: + target = git_workspace / "empty-me.txt" + target.write_bytes(b"no final newline") + + result = tool_runner.execute( + "replace_text", + {"edits": [_edit(tool_runner, "empty-me.txt", "no final newline", "")]}, + ) + + assert result.data["ok"], result.data + assert target.read_bytes() == b"" + + +def test_unicode_crlf_empty_replacement_and_missing_final_newline( + git_workspace: Path, tool_runner: ToolRunner +) -> None: + crlf = git_workspace / "你好.txt" + crlf.write_bytes("first\r\nsecond\r\n".encode()) + no_newline = git_workspace / "no newline.txt" + no_newline.write_bytes(b"alpha omega") + + result = tool_runner.execute( + "replace_text", + { + "edits": [ + _edit(tool_runner, "你好.txt", "second", "第二"), + _edit(tool_runner, "no newline.txt", " omega", ""), + ] + }, + ) + + assert result.data["ok"], result.data + assert crlf.read_bytes() == "first\r\n第二\r\n".encode() + assert no_newline.read_bytes() == b"alpha" + + +def test_cancelling_edits_are_rejected_without_mutation( + git_workspace: Path, tool_runner: ToolRunner +) -> None: + digest = tool_runner.workspace.sha256(git_workspace / "app.py") + result = tool_runner.execute( + "replace_text", + { + "edits": [ + _edit( + tool_runner, + "app.py", + "return 41", + "return 42", + sha256=digest, + ), + _edit( + tool_runner, + "app.py", + "return 42", + "return 41", + sha256=digest, + ), + ] + }, + ) + + assert result.data["error_code"] == "invalid_edit" + assert "return 41" in (git_workspace / "app.py").read_text() + + +def test_generated_patch_failure_is_reported_as_apply_failed( + monkeypatch, git_workspace: Path, tool_runner: ToolRunner +) -> None: + def reject_patch(*args, **kwargs): + raise ToolError( + "generated patch did not apply", + error_code="patch_context_mismatch", + details={"phase": "check", "paths": ["app.py"]}, + ) + + monkeypatch.setattr("yada.tools.replace.apply_patch", reject_patch) + + result = tool_runner.execute( + "replace_text", + {"edits": [_edit(tool_runner, "app.py", "return 41", "return 42")]}, + ) + + assert result.data["error_code"] == "apply_failed" + assert result.data["details"]["cause_error_code"] == ("patch_context_mismatch") + assert "return 41" in (git_workspace / "app.py").read_text() + assert tool_runner.context.state.revision == 0 + + +def test_replace_text_execution_is_auditable( + tmp_path: Path, tool_runner: ToolRunner +) -> None: + trace_path = tmp_path / "replace-trace.jsonl" + executor = Executor( + tools=tool_runner, + trace=TraceWriter(trace_path, level="debug"), + emit=lambda _: None, + ) + arguments = {"edits": [_edit(tool_runner, "app.py", "return 41", "return 42")]} + call = { + "id": "replace-1", + "type": "function", + "function": { + "name": "replace_text", + "arguments": json.dumps(arguments), + }, + } + + executed = executor.execute_batch(1, (call,)) + events = read_trace(trace_path) + + assert executed[0].execution.data["ok"] + assert [event["data"]["tool"] for event in events] == [ + "replace_text", + "replace_text", + ] + assert events[-1]["data"]["result"]["ok"]