Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions server/unified_diff.py
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,15 @@ def append_file(old_path: str = "", new_path: str = "") -> dict[str, Any] | None
current_file["newPath"] = raw_line[len("rename to ") :].strip()
current_file["path"] = current_file["newPath"]
continue
if raw_line.startswith("copy from "):
current_file["status"] = "copied"
current_file["oldPath"] = raw_line[len("copy from ") :].strip()
continue
if raw_line.startswith("copy to "):
current_file["status"] = "copied"
current_file["newPath"] = raw_line[len("copy to ") :].strip()
current_file["path"] = current_file["newPath"]
continue

hunk_match = _HUNK_HEADER.match(raw_line)
if hunk_match:
Expand Down
43 changes: 43 additions & 0 deletions tests/test_unified_diff_canvas.py
Original file line number Diff line number Diff line change
Expand Up @@ -241,6 +241,48 @@ def test_parse_unified_diff_splits_hunk_lines_only_at_newlines() -> None:
]


def test_parse_unified_diff_reads_copy_headers() -> None:
# Output of `git -c diff.renames=copies diff HEAD` after copying a
# modified file: an exact copy has no marker lines or hunks.
parsed = parse_unified_diff(
"\n".join(
[
"diff --git a/my src.txt b/my dup.txt",
"similarity index 100%",
"copy from my src.txt",
"copy to my dup.txt",
"diff --git a/lib.py b/lib copy.py",
"similarity index 80%",
"copy from lib.py",
"copy to lib copy.py",
"index 01e79c3..94ebaf9 100644",
"--- a/lib.py",
"+++ b/lib copy.py",
"@@ -1 +1,2 @@",
" base",
"+copy",
"diff --git a/my src.txt b/my src.txt",
"index b566061..b13f895 100644",
"--- a/my src.txt",
"+++ b/my src.txt",
"@@ -1 +1,2 @@",
" base",
"+source",
]
),
changed_files=["my dup.txt", "lib copy.py", "my src.txt"],
)

assert [
(item["path"], item["oldPath"], item["newPath"], item["status"], item["additions"])
for item in parsed["files"]
] == [
("my dup.txt", "my src.txt", "my dup.txt", "copied", 0),
("lib copy.py", "lib.py", "lib copy.py", "copied", 1),
("my src.txt", "my src.txt", "my src.txt", "modified", 1),
]


if __name__ == "__main__":
test_parse_unified_diff_preserves_dual_line_numbers()
test_parse_unified_diff_keeps_untracked_files_without_hunks()
Expand All @@ -250,4 +292,5 @@ def test_parse_unified_diff_splits_hunk_lines_only_at_newlines() -> None:
test_diff_canvas_payload_keeps_report_for_local_mode_switch()
test_parse_unified_diff_caps_files_without_merging_hunks()
test_parse_unified_diff_splits_hunk_lines_only_at_newlines()
test_parse_unified_diff_reads_copy_headers()
print("unified diff canvas tests passed")
48 changes: 48 additions & 0 deletions tests/test_work_artifact_registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -380,6 +380,53 @@ async def run() -> None:
asyncio.run(run())


def test_attempt_diff_lists_committed_copies_once_with_copy_detection() -> None:
async def run() -> None:
with tempfile.TemporaryDirectory(prefix="work_artifact_copy_") as temp:
root = Path(temp) / "project"
root.mkdir()
_git(root, "init")
_git(root, "config", "user.email", "amadeus-test@example.invalid")
_git(root, "config", "user.name", "Amadeus Test")
source = root / "my src.txt"
source.write_text("one\ntwo\nthree\n", encoding="utf-8")
_git(root, "add", ".")
_git(root, "commit", "-m", "baseline")
_git(root, "config", "diff.renames", "copies")

store = WorkLedgerStore(Path(temp) / "ledger.sqlite3")
project = store.create_or_get_project(root)
item = store.create_work_item(project.project_id, title="Copy notes")
attempt = store.create_attempt(item.work_item_id, provider="codex", task="Copy notes")
registry = WorkArtifactRegistry(store)
await registry.capture_baseline(attempt, item)
(root / "my dup.txt").write_text("one\ntwo\nthree\n", encoding="utf-8")
source.write_text("one\ntwo\nthree\nfour\n", encoding="utf-8")
_git(root, "add", ".")
_git(root, "commit", "-m", "provider commit")

current_attempt = store.get_attempt(attempt.attempt_id)
assert current_attempt is not None
delta = await registry.finalize_attempt(current_attempt, item)
store.close()

assert "copy to my dup.txt" in delta["committed_patch"]
parsed = parse_unified_diff(
delta["patch"],
changed_files=delta["changed_files"],
untracked=delta["untracked"],
)
assert [
(entry["path"], entry["oldPath"], entry["status"], entry["additions"])
for entry in parsed["files"]
] == [
("my dup.txt", "my src.txt", "copied", 0),
("my src.txt", "my src.txt", "modified", 1),
]

asyncio.run(run())


def test_non_git_workspace_registers_new_files_without_claiming_preexisting_changes() -> None:
async def run() -> None:
with tempfile.TemporaryDirectory(prefix="work_artifact_filesystem_") as temp:
Expand Down Expand Up @@ -610,6 +657,7 @@ def _main() -> None:
test_non_ascii_entry_revision_is_registered_as_the_real_path()
test_non_git_workspace_registers_new_files_without_claiming_preexisting_changes()
test_explicit_amend_may_change_predecessor_owned_dirty_artifact()
test_attempt_diff_lists_committed_copies_once_with_copy_detection()
print("ok: work artifact registry preserves attempt Git boundaries")


Expand Down
Loading