aaronrene / gabriel-muse public
blame.py python
217 lines 7.0 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago
1 """Core VCS blame — attribute each line of a text file to the commit that last changed it.
2
3 This is the domain-agnostic layer. The MIDI domain has ``note-blame``
4 (per-bar attribution); the code domain has ``muse code blame`` (per-symbol
5 attribution). This module provides line-level blame for *any text file*
6 tracked in ``state/``, making it useful for any domain that stores text —
7 configuration files, lyrics, scripts, prose.
8
9 Algorithm
10 ---------
11 Walk the commit graph from the requested ref backwards to the root:
12
13 1. At the starting commit, every line is "owned" by that commit.
14 2. At each parent commit, compute the *unified diff* between the parent's
15 version and the child's version of the file.
16 3. Lines that appear in both versions (context/unchanged) are attributed to
17 the *earliest* commit that produced them; we update the attribution when
18 we encounter a commit where those lines are *unchanged from the parent* —
19 i.e. they existed before this commit.
20 4. Lines that are *added* by a commit stay attributed to that commit.
21
22 This is equivalent to the ``git blame`` algorithm for single-parent chains.
23 For merge commits (two parents), we take the parent whose file content most
24 closely matches the merge result to avoid over-attributing lines to merges.
25
26 Output
27 ------
28 A list of ``BlameLine`` objects, one per line of the file at the requested
29 ref.
30 """
31
32 from __future__ import annotations
33
34 import difflib
35 import logging
36 import pathlib
37 from dataclasses import dataclass
38
39 from muse.core.object_store import object_path
40 from muse.core.store import read_commit, read_snapshot, resolve_commit_ref
41
42
43 type _CommitCache = dict[str, tuple[str, str, str]]
44 logger = logging.getLogger(__name__)
45
46
47 # ---------------------------------------------------------------------------
48 # Types
49 # ---------------------------------------------------------------------------
50
51
52 @dataclass(frozen=True)
53 class BlameLine:
54 """Attribution for one line of text."""
55
56 lineno: int
57 """1-based line number in the final version of the file."""
58
59 commit_id: str
60 """Commit that last changed this line."""
61
62 author: str
63 """Author field from the commit record."""
64
65 committed_at: str
66 """ISO timestamp of the commit."""
67
68 message: str
69 """First line of the commit message."""
70
71 content: str
72 """The line content (without trailing newline)."""
73
74
75 # ---------------------------------------------------------------------------
76 # File reading helper
77 # ---------------------------------------------------------------------------
78
79
80 def _read_file_at_commit(
81 repo_root: pathlib.Path,
82 commit_id: str,
83 rel_path: str,
84 ) -> list[str] | None:
85 """Return lines of *rel_path* as it existed at *commit_id*, or None."""
86 commit = read_commit(repo_root, commit_id)
87 if commit is None:
88 return None
89 snap = read_snapshot(repo_root, commit.snapshot_id)
90 if snap is None:
91 return None
92 obj_id = snap.manifest.get(rel_path)
93 if obj_id is None:
94 return None
95 obj = object_path(repo_root, obj_id)
96 if not obj.exists():
97 return None
98 try:
99 return obj.read_text(encoding="utf-8", errors="replace").splitlines()
100 except OSError:
101 return None
102
103
104 # ---------------------------------------------------------------------------
105 # Commit graph walker
106 # ---------------------------------------------------------------------------
107
108
109 def _walk_ancestry(
110 repo_root: pathlib.Path,
111 start_id: str,
112 ) -> list[str]:
113 """Return commit IDs from *start_id* to the root, newest-first."""
114 visited: list[str] = []
115 seen: set[str] = set()
116 queue = [start_id]
117 while queue:
118 cid = queue.pop(0)
119 if cid in seen:
120 continue
121 seen.add(cid)
122 visited.append(cid)
123 commit = read_commit(repo_root, cid)
124 if commit is None:
125 continue
126 # Prefer first parent for a clean linear walk; visit both for merges.
127 if commit.parent_commit_id:
128 queue.insert(0, commit.parent_commit_id)
129 if commit.parent2_commit_id:
130 queue.append(commit.parent2_commit_id)
131 return visited
132
133
134 # ---------------------------------------------------------------------------
135 # Public API
136 # ---------------------------------------------------------------------------
137
138
139 def blame_file(
140 repo_root: pathlib.Path,
141 rel_path: str,
142 commit_id: str,
143 ) -> list[BlameLine] | None:
144 """Attribute each line of *rel_path* to the commit that last modified it.
145
146 Args:
147 repo_root: Repository root.
148 rel_path: Path relative to ``state/``, e.g. ``"README.md"``.
149 commit_id: The commit to start the blame from (usually HEAD).
150
151 Returns:
152 A list of :class:`BlameLine` objects (1-indexed), or ``None`` if the
153 file does not exist at *commit_id*.
154 """
155 current_lines = _read_file_at_commit(repo_root, commit_id, rel_path)
156 if current_lines is None:
157 return None
158
159 n = len(current_lines)
160 # attribution[i] = commit_id that last changed line i (0-indexed)
161 attribution: list[str] = [commit_id] * n
162
163 ancestry = _walk_ancestry(repo_root, commit_id)
164
165 # Walk from the commit towards the root. When a parent has the same line
166 # content the attribution moves back to the parent (older is better).
167 child_lines = current_lines[:]
168 child_id = commit_id
169
170 for parent_id in ancestry[1:]: # skip the starting commit itself
171 parent_lines = _read_file_at_commit(repo_root, parent_id, rel_path)
172 if parent_lines is None:
173 # File didn't exist in this ancestor — stop the walk.
174 break
175
176 # Use SequenceMatcher to align lines.
177 sm = difflib.SequenceMatcher(None, parent_lines, child_lines, autojunk=False)
178 for tag, i1, i2, j1, j2 in sm.get_opcodes():
179 if tag == "equal":
180 # These lines are unchanged from parent → child.
181 # They may be attributed to an even older commit; update those
182 # that are currently attributed to child_id.
183 for k in range(j2 - j1):
184 if attribution[j1 + k] == child_id:
185 attribution[j1 + k] = parent_id
186
187 child_lines = parent_lines
188 child_id = parent_id
189
190 # Build the final BlameLine list.
191 result: list[BlameLine] = []
192 commit_cache: _CommitCache = {}
193
194 def _commit_meta(cid: str) -> tuple[str, str, str]:
195 if cid in commit_cache:
196 return commit_cache[cid]
197 c = read_commit(repo_root, cid)
198 if c is None:
199 meta = ("unknown", "", "")
200 else:
201 first_line = c.message.split("\n", 1)[0] if c.message else ""
202 meta = (c.author or "unknown", c.committed_at.isoformat(), first_line)
203 commit_cache[cid] = meta
204 return meta
205
206 for idx, (line, cid) in enumerate(zip(current_lines, attribution)):
207 author, committed_at, message = _commit_meta(cid)
208 result.append(BlameLine(
209 lineno=idx + 1,
210 commit_id=cid,
211 author=author,
212 committed_at=committed_at,
213 message=message,
214 content=line,
215 ))
216
217 return result
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago