diff --git a/loopx/capabilities/integration_branch/core.py b/loopx/capabilities/integration_branch/core.py index 1e847c3bd2..2991c53fb6 100644 --- a/loopx/capabilities/integration_branch/core.py +++ b/loopx/capabilities/integration_branch/core.py @@ -2,6 +2,7 @@ import json import os +import re import shutil import subprocess import tempfile @@ -27,7 +28,7 @@ class IntegrationBranchError(ValueError): """A fail-closed integration-branch contract error.""" -def _retire_owned_read(process: subprocess.Popen[str]) -> None: +def _retire_owned_read(process: subprocess.Popen[Any]) -> None: """Retire the read's own process group and drop its pipes within a deadline. The shared runtime bounds termination, Windows ``taskkill`` and pipe @@ -57,15 +58,41 @@ def _git( never observed. """ + result = _git_command(repo, args, check=check, text=True) + return subprocess.CompletedProcess( + result.args, result.returncode, result.stdout, result.stderr + ) + + +def _git_bytes( + repo: Path, + *args: str, + check: bool = True, +) -> subprocess.CompletedProcess[bytes]: + """Run one Git read whose byte records are part of the caller's boundary.""" + + result = _git_command(repo, args, check=check, text=False) + return subprocess.CompletedProcess( + result.args, result.returncode, result.stdout, result.stderr + ) + + +def _git_command( + repo: Path, + args: Sequence[str], + *, + check: bool, + text: bool, +) -> subprocess.CompletedProcess[Any]: argv = ["git", "-C", str(repo), *args] process = subprocess.Popen( argv, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, - text=True, - encoding="utf-8", - errors="replace", + text=text, + encoding="utf-8" if text else None, + errors="replace" if text else None, start_new_session=os.name == "posix", ) try: @@ -84,19 +111,27 @@ def _git( raise result = subprocess.CompletedProcess(argv, process.returncode, stdout, stderr) if check and result.returncode != 0: - detail = ( - (result.stderr or "").strip() - or (result.stdout or "").strip() - or "git command failed" + stderr_detail = ( + result.stderr + if isinstance(result.stderr, str) + else result.stderr.decode("utf-8", "replace") + ) + stdout_detail = ( + result.stdout + if isinstance(result.stdout, str) + else result.stdout.decode("utf-8", "replace") + ) + raise IntegrationBranchError( + stderr_detail.strip() or stdout_detail.strip() or "git command failed" ) - raise IntegrationBranchError(detail) return result def _repository_root(repo_path: str | Path) -> Path: requested = Path(repo_path).expanduser().resolve() - result = _git(requested, "rev-parse", "--show-toplevel") - return Path(result.stdout.strip()).resolve() + return Path( + _git_path_output(requested, "rev-parse", "--show-toplevel") + ).resolve() def _plan_path(repo: Path, plan_file: str | Path | None) -> Path: @@ -576,23 +611,155 @@ def integration_branch_status( } +_LEGACY_WORKTREE_PATH_ERROR = ( + "this Git version cannot safely resolve a worktree path containing a line " + "separator because 'git worktree list -z' requires Git 2.36+; upgrade Git " + "or resolve the worktree path manually" +) + + +_GIT_VERSION_PATTERN = re.compile(r"(?:^|\s)([0-9]+)\.([0-9]+)(?:\.|$)") + + +def _worktree_list_uses_nul(repo: Path) -> bool: + version = _git(repo, "version").stdout + match = _GIT_VERSION_PATTERN.search(version) + if match is None: + raise IntegrationBranchError( + f"cannot determine whether this Git version supports NUL worktree records: {version.strip()}" + ) + return (int(match.group(1)), int(match.group(2))) >= (2, 36) + + def _worktree_for_branch(repo: Path, branch: str) -> Path | None: - result = _git(repo, "worktree", "list", "--porcelain") - current_path: Path | None = None target_ref = f"refs/heads/{branch}" - # `--porcelain` frames records on LF and prints paths verbatim, so a path - # containing U+0085/U+2028/U+2029 would be split by `str.splitlines()` and - # the prefix returned as the worktree path. Frame on the writer's separator. - for line in result.stdout.split("\n"): - if line.startswith("worktree "): - current_path = Path(line.removeprefix("worktree ")).resolve() - elif line == f"branch {target_ref}": - return current_path - elif not line: - current_path = None + if not _worktree_list_uses_nul(repo): + # NUL framing arrived in Git 2.36. Older Git rejects the switch before + # writing a record; keep ordinary paths working through the line-framed + # inventory, which stays fail-closed for values it cannot represent. + return _worktree_for_branch_line_framed(repo, target_ref) + result = _git_bytes(repo, "worktree", "list", "--porcelain", "-z") + + current_path: Path | None = None + # `-z` frames fields on NUL and records on a NUL pair. A worktree path + # may contain LF, CR or Unicode line separators, so treating its text as + # a line-oriented inventory returns a prefix and can target the wrong + # checkout for cleanup or reset. + for record in result.stdout.split(b"\0\0"): + if not record: + continue + current_path = None + for field in record.split(b"\0"): + if field.startswith(b"worktree "): + current_path = Path(os.fsdecode(field.removeprefix(b"worktree "))).resolve() + elif field == f"branch {target_ref}".encode("utf-8"): + return current_path + return None + + +def _worktree_for_branch_line_framed(repo: Path, target_ref: str) -> Path | None: + """Resolve a branch from Git versions without ``worktree list -z`` support.""" + result = _git(repo, "worktree", "list", "--porcelain") + lines = result.stdout.split("\n") + index = 0 + while index < len(lines): + if not lines[index]: + index += 1 + continue + if not lines[index].startswith("worktree "): + raise IntegrationBranchError(_LEGACY_WORKTREE_PATH_ERROR) + path = Path(lines[index].removeprefix("worktree ")).resolve() + index += 1 + saw_head = False + saw_bare = False + branch_ref: str | None = None + while index < len(lines) and lines[index]: + field = lines[index] + if field.startswith("HEAD "): + # A path containing LF or CR can shift the real HEAD field and + # leave a second one behind. Duplicate or bare/HEAD mixtures are + # not records this format can represent safely. + if saw_head or saw_bare: + raise IntegrationBranchError(_LEGACY_WORKTREE_PATH_ERROR) + saw_head = True + elif field == "bare": + if saw_head or saw_bare: + raise IntegrationBranchError(_LEGACY_WORKTREE_PATH_ERROR) + saw_bare = True + elif field.startswith("branch "): + if saw_bare or not saw_head: + raise IntegrationBranchError(_LEGACY_WORKTREE_PATH_ERROR) + branch_ref = field.removeprefix("branch ") + elif field == "detached": + if saw_bare or not saw_head: + raise IntegrationBranchError(_LEGACY_WORKTREE_PATH_ERROR) + elif field.startswith(("locked", "prunable")): + pass + else: + raise IntegrationBranchError(_LEGACY_WORKTREE_PATH_ERROR) + index += 1 + # A linked checkout has exactly one HEAD; a bare main worktree has a + # bare field instead. Anything else is a path torn across fields. + if not saw_head and not saw_bare: + raise IntegrationBranchError(_LEGACY_WORKTREE_PATH_ERROR) + if branch_ref == target_ref: + return path + index += 1 return None +def _path_key(path: Path) -> str: + return os.path.normcase(str(path.resolve())) + + +def _git_common_dir(base: Path, raw: str) -> Path: + candidate = Path(raw) + if not candidate.is_absolute(): + candidate = base / candidate + return candidate.resolve() + + +def _git_path_output(repo: Path, *args: str) -> str: + """Decode one Git path, removing only the newline Git appends to it.""" + + output = _git_bytes(repo, *args).stdout + if output.endswith(b"\n"): + output = output[:-1] + return os.fsdecode(output) + + +def _confirm_checkout_ownership(repo: Path, worktree: Path, branch: str) -> None: + """Refuse a reset unless Git confirms the checkout is this branch's root.""" + + expected_ref = f"refs/heads/{branch}" + try: + top = _git_path_output(worktree, "rev-parse", "--show-toplevel") + worktree_common = _git_path_output( + worktree, "rev-parse", "--git-common-dir" + ) + repo_common = _git_path_output(repo, "rev-parse", "--git-common-dir") + head = _git_bytes(worktree, "symbolic-ref", "--quiet", "HEAD", check=False) + except IntegrationBranchError as exc: + raise IntegrationBranchError( + f"cannot verify that {worktree} is the checkout for {branch}: {exc}" + ) from None + if not top or _path_key(Path(top)) != _path_key(worktree): + raise IntegrationBranchError( + f"worktree path {worktree} is not the top of a checkout" + ) + if _path_key(_git_common_dir(worktree, worktree_common)) != _path_key( + _git_common_dir(repo, repo_common) + ): + raise IntegrationBranchError( + f"worktree {worktree} belongs to a different repository than {repo}" + ) + head_ref = head.stdout[:-1] if head.stdout.endswith(b"\n") else head.stdout + if head.returncode != 0 or os.fsdecode(head_ref) != expected_ref: + raise IntegrationBranchError( + f"worktree {worktree} is not checked out on branch {branch}" + ) + + def _clean_worktree(path: Path) -> bool: return not _git( path, @@ -808,6 +975,7 @@ def _update_integration_branch( ) checked_out_path = _worktree_for_branch(repo, branch) if checked_out_path is not None: + _confirm_checkout_ownership(repo, checked_out_path, branch) if not _clean_worktree(checked_out_path): raise IntegrationBranchError( f"integration branch worktree is dirty: {checked_out_path}" diff --git a/tests/capabilities/test_integration_branch.py b/tests/capabilities/test_integration_branch.py index 2cc388dec6..a18a9e7f5e 100644 --- a/tests/capabilities/test_integration_branch.py +++ b/tests/capabilities/test_integration_branch.py @@ -1,6 +1,8 @@ from __future__ import annotations import json +import os +import shutil import subprocess from pathlib import Path @@ -11,6 +13,7 @@ integration_branch_status, sync_integration_branch, ) +from loopx.capabilities.integration_branch import core as integration_core from loopx.capabilities.integration_branch.core import ( IntegrationBranchError, _worktree_for_branch, @@ -56,6 +59,15 @@ def _repository(tmp_path: Path) -> Path: return repo +def _repositories_with_trailing_newline(tmp_path: Path) -> tuple[Path, Path]: + """Create two independent repositories whose root names differ by LF.""" + requested = _repository(tmp_path) + requested_with_lf = tmp_path / "repo\n" + requested.rename(requested_with_lf) + adjacent = _repository(tmp_path) + return requested_with_lf, adjacent + + def _remote_repositories(tmp_path: Path) -> tuple[Path, Path]: seed_root = tmp_path / "seed" seed_root.mkdir() @@ -527,6 +539,63 @@ def test_adopt_current_integration_head_without_touching_worktree( assert integration_branch_status(repo_path=repo)["status"] == "in_sync" +@pytest.mark.skipif(os.name != "posix", reason="newline in a path is POSIX-only") +def test_sync_root_preserves_a_trailing_newline_in_the_requested_path( + tmp_path: Path, +) -> None: + """A requested repository ending in LF must not resolve to its neighbor.""" + requested, adjacent = _repositories_with_trailing_newline(tmp_path) + _configure(requested) + _configure(adjacent) + sync_integration_branch(repo_path=requested, execute=True) + sync_integration_branch(repo_path=adjacent, execute=True) + adjacent_plan = adjacent / ".loopx/integration-branch.json" + adjacent_plan_before = adjacent_plan.read_text(encoding="utf-8") + adjacent_ref = "refs/heads/codex/local-integration" + adjacent_head_before = _git(adjacent, "rev-parse", adjacent_ref) + + _git(requested, "switch", "feature-a") + _commit(requested, "a.txt", "a2\n", "requested review update") + _git(requested, "switch", "main") + _git(adjacent, "switch", "fix-b") + _commit(adjacent, "b.txt", "b2\n", "adjacent review update") + _git(adjacent, "switch", "main") + + synced = sync_integration_branch(repo_path=requested, execute=True) + + assert synced["status"] == "synced" + assert _git(requested, "rev-parse", "refs/heads/codex/local-integration") == synced[ + "candidate_sha" + ] + assert integration_branch_status(repo_path=requested)["status"] == "in_sync" + assert _git(adjacent, "rev-parse", adjacent_ref) == adjacent_head_before + assert adjacent_plan.read_text(encoding="utf-8") == adjacent_plan_before + + +@pytest.mark.skipif(os.name != "posix", reason="newline in a path is POSIX-only") +def test_sync_root_refuses_an_unconfigured_newline_checkout_without_touching_neighbor( + tmp_path: Path, +) -> None: + """A missing local plan must fail before the adjacent repository can be read.""" + requested, adjacent = _repositories_with_trailing_newline(tmp_path) + _configure(adjacent) + sync_integration_branch(repo_path=adjacent, execute=True) + _git(adjacent, "switch", "fix-b") + _commit(adjacent, "b.txt", "b2\n", "adjacent review update") + _git(adjacent, "switch", "main") + adjacent_plan = adjacent / ".loopx/integration-branch.json" + adjacent_plan_before = adjacent_plan.read_text(encoding="utf-8") + adjacent_ref = "refs/heads/codex/local-integration" + adjacent_head_before = _git(adjacent, "rev-parse", adjacent_ref) + + with pytest.raises(IntegrationBranchError, match="plan not found"): + sync_integration_branch(repo_path=requested, execute=True) + + assert not (requested / ".loopx/integration-branch.json").exists() + assert _git(adjacent, "rev-parse", adjacent_ref) == adjacent_head_before + assert adjacent_plan.read_text(encoding="utf-8") == adjacent_plan_before + + def test_dirty_checked_out_integration_worktree_fails_closed( tmp_path: Path, ) -> None: @@ -644,6 +713,19 @@ def test_cli_rejects_plan_file_outside_loopx_state_root( assert not outside_plan.exists() +@pytest.mark.skipif(os.name != "posix", reason="line separators in paths are POSIX-only") +@pytest.mark.parametrize("line_separator", ["\n", "\r"]) +def test_worktree_for_branch_preserves_a_line_break_in_the_path( + tmp_path: Path, line_separator: str +) -> None: + """A path with LF or CR must not tear the NUL-framed worktree inventory.""" + repo = _repository(tmp_path) + odd = tmp_path / f"odd{line_separator}worktree" + _git(repo, "worktree", "add", str(odd), "feature-a") + + assert _worktree_for_branch(repo, "feature-a") == odd.resolve() + + def test_worktree_for_branch_frames_porcelain_records_on_lf(tmp_path: Path) -> None: """A worktree path containing U+0085 must resolve to the full path. @@ -657,3 +739,189 @@ def test_worktree_for_branch_frames_porcelain_records_on_lf(tmp_path: Path) -> N _git(repo, "worktree", "add", str(odd), "feature-a") assert _worktree_for_branch(repo, "feature-a") == odd.resolve() + + +def _install_legacy_git( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + *, + unknown_switch_error: str = "error: unknown switch 'z'", +) -> None: + """Put a Git wrapper first on PATH that rejects -z like Git 2.35.""" + real_git = shutil.which("git") + assert real_git + bin_dir = tmp_path / "legacy-git-bin" + bin_dir.mkdir() + wrapper = bin_dir / "git" + wrapper.write_text( + "#!/bin/sh\n" + 'if [ "$#" -eq 3 ] && [ "$1" = "-C" ] && [ "$3" = "version" ]; then\n' + " echo 'git version 2.35.8'\n" + " exit 0\n" + "fi\n" + 'for arg in "$@"; do\n' + ' if [ "$arg" = "-z" ]; then\n' + f" printf '%s\\n' \"{unknown_switch_error}\" >&2\n" + " exit 129\n" + " fi\n" + "done\n" + f'exec "{real_git}" "$@"\n', + encoding="utf-8", + ) + wrapper.chmod(0o755) + monkeypatch.setenv("PATH", f"{bin_dir}{os.pathsep}{os.environ['PATH']}") + + +@pytest.mark.skipif(os.name != "posix", reason="legacy Git stub is POSIX-only") +def test_worktree_for_branch_supports_legacy_git_without_nul_framing( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Git 2.35 rejects worktree-list -z; ordinary paths must still resolve.""" + repo = _repository(tmp_path) + ordinary = tmp_path / "ordinary worktree" + _git(repo, "worktree", "add", str(ordinary), "feature-a") + _install_legacy_git(tmp_path, monkeypatch) + + assert _worktree_for_branch(repo, "feature-a") == ordinary.resolve() + + +@pytest.mark.skipif(os.name != "posix", reason="legacy Git stub is POSIX-only") +def test_legacy_git_format_is_selected_by_version_not_error_language( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A localized usage error must not change the worktree inventory format.""" + repo = _repository(tmp_path) + ordinary = tmp_path / "ordinary worktree" + _git(repo, "worktree", "add", str(ordinary), "feature-a") + _install_legacy_git( + tmp_path, + monkeypatch, + unknown_switch_error="erreur : option inconnue 'z'", + ) + monkeypatch.setenv("LC_ALL", "fr_FR.UTF-8") + + assert _worktree_for_branch(repo, "feature-a") == ordinary.resolve() + + +@pytest.mark.skipif(os.name != "posix", reason="line separators in paths are POSIX-only") +@pytest.mark.parametrize("line_separator", ["\n", "\r"]) +def test_legacy_git_fails_closed_on_line_separator_worktree_path( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, line_separator: str +) -> None: + """A legacy line-framed inventory must refuse a path it cannot represent.""" + repo = _repository(tmp_path) + odd = tmp_path / f"odd{line_separator}worktree" + _git(repo, "worktree", "add", str(odd), "feature-a") + _install_legacy_git(tmp_path, monkeypatch) + + with pytest.raises(IntegrationBranchError, match="Git 2.36"): + _worktree_for_branch(repo, "feature-a") + + +@pytest.mark.skipif(os.name != "posix", reason="legacy Git stub is POSIX-only") +def test_sync_execute_supports_legacy_git_with_ordinary_worktree( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Git 2.35 must still sync an ordinary linked integration worktree.""" + repo = _repository(tmp_path) + _configure(repo) + sync_integration_branch(repo_path=repo, execute=True) + integration_worktree = tmp_path / "integration worktree" + _git(repo, "switch", "main") + _git(repo, "worktree", "add", str(integration_worktree), "codex/local-integration") + source_worktree = tmp_path / "source worktree" + _git(repo, "worktree", "add", str(source_worktree), "feature-a") + _commit(source_worktree, "a.txt", "a2\n", "review update") + _install_legacy_git(tmp_path, monkeypatch) + + synced = sync_integration_branch(repo_path=repo, execute=True) + + assert synced["status"] == "synced" + assert _git(integration_worktree, "rev-parse", "HEAD") == synced["candidate_sha"] + + +@pytest.mark.skipif(os.name != "posix", reason="special path names are POSIX-only") +@pytest.mark.parametrize("suffix", ["\n", "\r", " "]) +def test_sync_execute_preserves_checkout_paths_when_verifying_ownership( + tmp_path: Path, suffix: str +) -> None: + """Path output must not strip a checkout name's trailing characters.""" + repo = _repository(tmp_path) + _configure(repo) + sync_integration_branch(repo_path=repo, execute=True) + integration_worktree = tmp_path / f"integration worktree{suffix}" + _git(repo, "switch", "main") + _git(repo, "worktree", "add", str(integration_worktree), "codex/local-integration") + source_worktree = tmp_path / "source worktree" + _git(repo, "worktree", "add", str(source_worktree), "feature-a") + _commit(source_worktree, "a.txt", "a2\n", "review update") + + synced = sync_integration_branch(repo_path=repo, execute=True) + + assert synced["status"] == "synced" + assert _git(integration_worktree, "rev-parse", "HEAD") == synced["candidate_sha"] + assert integration_branch_status(repo_path=repo)["status"] == "in_sync" + + +@pytest.mark.skipif(os.name != "posix", reason="legacy Git stub is POSIX-only") +def test_legacy_git_skips_bare_record_and_syncs_linked_worktree( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A bare main record must not hide the linked integration checkout.""" + (tmp_path / "seed").mkdir() + seed = _repository(tmp_path / "seed") + bare = tmp_path / "bare.git" + _git(tmp_path, "init", "--bare", "-b", "main", str(bare)) + _git(seed, "remote", "add", "origin", str(bare)) + _git(seed, "push", "origin", "main", "feature-a", "fix-b") + linked = tmp_path / "linked" + _git(bare, "worktree", "add", str(linked), "main") + _configure(linked) + _git(bare, "branch", "codex/local-integration", "main") + integration = tmp_path / "integration worktree" + _git(bare, "worktree", "add", str(integration), "codex/local-integration") + plan = linked / ".loopx/integration-branch.json" + (integration / ".loopx").mkdir() + shutil.copy2(plan, integration / ".loopx/integration-branch.json") + source = tmp_path / "source worktree" + _git(bare, "worktree", "add", str(source), "feature-a") + _commit(source, "a.txt", "a2\n", "review update") + _install_legacy_git(tmp_path, monkeypatch) + + synced = sync_integration_branch(repo_path=integration, execute=True) + + assert synced["status"] == "synced" + assert _git(integration, "rev-parse", "HEAD") == synced["candidate_sha"] + assert integration_branch_status(repo_path=integration)["status"] == "in_sync" + + +def test_sync_refuses_a_checkout_that_is_not_the_integration_branch( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A parsed path must prove branch ownership before any reset.""" + repo = _repository(tmp_path) + _configure(repo) + sync_integration_branch(repo_path=repo, execute=True) + _git(repo, "switch", "feature-a") + _commit(repo, "a.txt", "a2\n", "review update") + _git(repo, "switch", "main") + wrong = tmp_path / "wrong branch worktree" + _git(repo, "worktree", "add", str(wrong), "fix-b") + wrong_head = _git(wrong, "rev-parse", "HEAD") + integration_head = _git(repo, "rev-parse", "codex/local-integration") + plan_path = repo / ".loopx/integration-branch.json" + plan_before = plan_path.read_text(encoding="utf-8") + + def wrong_worktree(repo_path: Path, branch: str) -> Path: + return wrong + + monkeypatch.setattr( + integration_core, "_worktree_for_branch", wrong_worktree + ) + + with pytest.raises(IntegrationBranchError, match="not checked out on branch"): + sync_integration_branch(repo_path=repo, execute=True) + + assert _git(wrong, "rev-parse", "HEAD") == wrong_head + assert _git(repo, "rev-parse", "codex/local-integration") == integration_head + assert plan_path.read_text(encoding="utf-8") == plan_before