feat(git): resolve exact two-dot and three-dot ranges

diffgraph/git_snapshot.py

@@ -1,13 +1,14 @@
 """Resolve exact Git object identities for index and working-tree changes.
 
-This module deliberately models only the two local snapshot pairs:
+This module models the two local snapshot pairs and immutable commit ranges:
 
 * staged: ``HEAD`` -> index
 * unstaged: index -> working tree
+* two-dot: the requested base commit -> the requested head commit
+* three-dot: the merge base of the requested commits -> the requested head
 
-Commit/range resolution belongs to a separate layer.  Paths returned by Git are
-read using its NUL-delimited raw format, so tabs, newlines, and other unusual
-filename bytes are not delimiters.
+Paths returned by Git are read using its NUL-delimited raw format, so tabs,
+newlines, and other unusual filename bytes are not delimiters.
 """
 
 from __future__ import annotations
@@ -55,6 +56,25 @@ class SnapshotResolution:
     warnings: Tuple[ResolutionWarning, ...]
 
 
+@dataclass(frozen=True)
+class CommitRangeResolution:
+    """Exact endpoints and entries for a two-dot or three-dot comparison.
+
+    ``comparison_base_oid`` equals ``base_oid`` for two-dot comparisons and
+    the resolved merge-base object ID for three-dot comparisons. Recording
+    both prevents display refs from being mistaken for immutable provenance.
+    """
+
+    entries: Tuple[SnapshotEntry, ...]
+    warnings: Tuple[ResolutionWarning, ...]
+    base_ref: str
+    head_ref: str
+    base_oid: Optional[str]
+    head_oid: Optional[str]
+    comparison_base_oid: Optional[str]
+    three_dot: bool
+
+
 class GitSnapshotError(RuntimeError):
     """A Git or working-tree operation required for exact snapshots failed."""
 
@@ -97,6 +117,129 @@ def resolve_unstaged(
     return _resolve(repository, pathspecs, staged=False)
 
 
+def resolve_commit_range(
+    repository: str,
+    base_ref: str,
+    head_ref: str,
+    *,
+    three_dot: bool = False,
+    pathspecs: Optional[Sequence[str]] = None,
+) -> CommitRangeResolution:
+    """Resolve exact tree identities for a two-dot or three-dot comparison.
+
+    Refs are resolved to commits before diffing. Two-dot compares those two
+    commits directly; three-dot compares their merge base with the resolved
+    head, matching ``git diff base...head`` semantics. Failures are returned
+    as structured warnings and never as fabricated changes.
+    """
+
+    warnings: List[ResolutionWarning] = []
+    root = _repository_root(repository, warnings)
+    if root is None:
+        return _commit_range_result(base_ref, head_ref, three_dot, warnings=warnings)
+
+    base_oid = _resolve_commit(root, base_ref, "invalid_base_ref", warnings)
+    head_oid = _resolve_commit(root, head_ref, "invalid_head_ref", warnings)
+    if base_oid is None or head_oid is None:
+        return _commit_range_result(
+            base_ref, head_ref, three_dot, warnings=warnings,
+            base_oid=base_oid, head_oid=head_oid,
+        )
+
+    comparison_base_oid = base_oid
+    if three_dot:
+        output = _run(
+            ["git", "merge-base", base_oid, head_oid], root, warnings,
+            "merge_base_failed",
+        )
+        if output is None:
+            return _commit_range_result(
+                base_ref, head_ref, three_dot, warnings=warnings,
+                base_oid=base_oid, head_oid=head_oid,
+            )
+        comparison_base_oid = os.fsdecode(output).strip()
+        if not _is_hex_oid(comparison_base_oid):
+            warnings.append(ResolutionWarning(
+                "malformed_merge_base",
+                "Git returned an invalid merge-base object ID",
+            ))
+            return _commit_range_result(
+                base_ref, head_ref, three_dot, warnings=warnings,
+                base_oid=base_oid, head_oid=head_oid,
+            )
+
+    command = [
+        "git", "diff", "--raw", "-z", "--no-abbrev", "--no-ext-diff",
+        "--find-renames=50%", comparison_base_oid, head_oid,
+    ]
+    scoped_pathspecs = _root_relative_pathspecs(repository, root, pathspecs)
+    if scoped_pathspecs:
+        command.append("--")
+        command.extend(scoped_pathspecs)
+    output = _run(command, root, warnings, "git_diff_failed")
+    entries: List[SnapshotEntry] = []
+    if output is not None:
+        for raw in _parse_raw(output, warnings):
+            entry = _exact_staged_entry(raw, warnings)
+            if entry is not None:
+                entries.append(entry)
+        entries.sort(key=_entry_sort_key)
+
+    return _commit_range_result(
+        base_ref, head_ref, three_dot, entries=entries, warnings=warnings,
+        base_oid=base_oid, head_oid=head_oid,
+        comparison_base_oid=comparison_base_oid,
+    )
+
+
+def _commit_range_result(
+    base_ref: str,
+    head_ref: str,
+    three_dot: bool,
+    *,
+    entries: Sequence[SnapshotEntry] = (),
+    warnings: Sequence[ResolutionWarning] = (),
+    base_oid: Optional[str] = None,
+    head_oid: Optional[str] = None,
+    comparison_base_oid: Optional[str] = None,
+) -> CommitRangeResolution:
+    return CommitRangeResolution(
+        tuple(entries), tuple(warnings), base_ref, head_ref, base_oid, head_oid,
+        comparison_base_oid, three_dot,
+    )
+
+
+def _resolve_commit(
+    root: str, ref: str, warning_code: str,
+    warnings: List[ResolutionWarning],
+) -> Optional[str]:
+    output = _run(
+        [
+            "git",
+            "rev-parse",
+            "--verify",
+            "--end-of-options",
+            "{}^{{commit}}".format(ref),
+        ],
+        root, warnings, warning_code,
+    )
+    if output is None:
+        return None
+    oid = os.fsdecode(output).strip()
+    if not _is_hex_oid(oid):
+        warnings.append(ResolutionWarning(
+            warning_code, "Git returned an invalid commit object ID"
+        ))
+        return None
+    return oid
+
+
+def _is_hex_oid(value: str) -> bool:
+    return bool(value) and all(
+        character in "0123456789abcdef" for character in value
+    )
+
+
 def _resolve(
     repository: str, pathspecs: Optional[Sequence[str]], staged: bool
 ) -> SnapshotResolution:

tests/test_git_snapshot.py

@@ -1,7 +1,11 @@
 import os
 import subprocess
 
-from diffgraph.git_snapshot import resolve_staged, resolve_unstaged
+from diffgraph.git_snapshot import (
+    resolve_commit_range,
+    resolve_staged,
+    resolve_unstaged,
+)
 
 
 def git(repo, *args, input_bytes=None):
@@ -288,3 +292,112 @@ def test_unstaged_symlink_hashes_raw_target_without_attributes(tmp_path):
     assert entry.new_oid == git(
         repo, "hash-object", "--stdin", input_bytes=b"replacement.py"
     ).decode("ascii").strip()
+
+
+def test_two_dot_range_records_exact_endpoints_and_tree_identities(tmp_path):
+    repo = make_repo(tmp_path)
+    write(repo, "shared.txt", b"common\n")
+    commit_all(repo, "common")
+    git(repo, "branch", "base")
+
+    write(repo, "head-only.txt", b"head\n")
+    commit_all(repo, "head change")
+    git(repo, "branch", "head")
+    head_oid = oid(repo, "head")
+
+    git(repo, "switch", "base")
+    write(repo, "base-only.txt", b"base\n")
+    commit_all(repo, "base change")
+    base_oid = oid(repo, "base")
+
+    result = resolve_commit_range(str(repo), "base", "head")
+    entries = {entry.new_path or entry.old_path: entry for entry in result.entries}
+
+    assert result.warnings == ()
+    assert result.base_oid == result.comparison_base_oid == base_oid
+    assert result.head_oid == head_oid
+    assert result.three_dot is False
+    assert set(entries) == {"base-only.txt", "head-only.txt"}
+    assert entries["base-only.txt"].status == "D"
+    assert entries["base-only.txt"].old_oid == oid(repo, "base:base-only.txt")
+    assert entries["base-only.txt"].new_oid is None
+    assert entries["head-only.txt"].status == "A"
+    assert entries["head-only.txt"].old_oid is None
+    assert entries["head-only.txt"].new_oid == oid(repo, "head:head-only.txt")
+
+
+def test_three_dot_range_uses_merge_base_and_excludes_base_only_changes(tmp_path):
+    repo = make_repo(tmp_path)
+    write(repo, "shared.txt", b"common\n")
+    commit_all(repo, "common")
+    common_oid = oid(repo, "HEAD")
+    git(repo, "branch", "base")
+
+    write(repo, "head-only.txt", b"head\n")
+    commit_all(repo, "head change")
+    git(repo, "branch", "head")
+
+    git(repo, "switch", "base")
+    write(repo, "base-only.txt", b"base\n")
+    commit_all(repo, "base change")
+
+    result = resolve_commit_range(str(repo), "base", "head", three_dot=True)
+
+    assert result.warnings == ()
+    assert result.comparison_base_oid == common_oid
+    assert result.base_oid == oid(repo, "base")
+    assert result.head_oid == oid(repo, "head")
+    assert result.three_dot is True
+    assert [entry.new_path or entry.old_path for entry in result.entries] == [
+        "head-only.txt"
+    ]
+
+
+def test_commit_range_pathspec_is_relative_to_calling_subdirectory(tmp_path):
+    repo = make_repo(tmp_path)
+    write(repo, "inside/a.txt", b"old a\n")
+    write(repo, "outside/b.txt", b"old b\n")
+    commit_all(repo)
+    git(repo, "branch", "before")
+    write(repo, "inside/a.txt", b"new a\n")
+    write(repo, "outside/b.txt", b"new b\n")
+    commit_all(repo, "both changed")
+
+    result = resolve_commit_range(
+        str(repo / "inside"), "before", "HEAD", pathspecs=["a.txt"]
+    )
+
+    assert result.warnings == ()
+    assert [entry.new_path for entry in result.entries] == ["inside/a.txt"]
+
+
+def test_invalid_commit_range_ref_is_a_warning_not_a_change(tmp_path):
+    repo = make_repo(tmp_path)
+    write(repo, "tracked.txt", b"content\n")
+    commit_all(repo)
+
+    result = resolve_commit_range(str(repo), "missing-ref", "HEAD")
+
+    assert result.entries == ()
+    assert result.base_oid is None
+    assert result.head_oid == oid(repo, "HEAD")
+    assert [warning.code for warning in result.warnings] == ["invalid_base_ref"]
+
+
+def test_three_dot_without_merge_base_is_a_warning_not_a_change(tmp_path):
+    repo = make_repo(tmp_path)
+    write(repo, "first.txt", b"first\n")
+    commit_all(repo)
+    git(repo, "branch", "first")
+    git(repo, "switch", "--orphan", "second")
+    first_path = repo / "first.txt"
+    if first_path.exists():
+        first_path.unlink()
+    write(repo, "second.txt", b"second\n")
+    commit_all(repo, "unrelated root")
+
+    result = resolve_commit_range(str(repo), "first", "second", three_dot=True)
+
+    assert result.entries == ()
+    assert result.comparison_base_oid is None
+    assert [warning.code for warning in result.warnings] == ["merge_base_failed"]