gabriel / muse public
symbol_log.py python
431 lines 15.4 KB
Raw
sha256:6ed6b11aceb96bef850842a7aaac54c04843fa9fba5d1200e4bc3845e12d4142 docs: add muse-tag-guide mist (complete idiomatic guide wit… Sonnet 4.6 19 days ago
1 """muse code symbol-log -- track a single semantic symbol through commit history.
2
3 This command is impossible in Git: Git's ``git log -p src/utils.py`` shows
4 every line that changed in a file; it has no concept of a *function*.
5 ``muse code symbol-log`` tracks the full lifecycle of a single named symbol --
6 when it was created, when its implementation changed, when it was renamed,
7 and when it was deleted -- across the entire commit DAG, including all
8 branches that were ever merged in.
9
10 Usage::
11
12 muse code symbol-log "src/utils.py::calculate_total"
13 muse code symbol-log "src/models.py::User.save"
14
15 Output::
16
17 Symbol: src/utils.py::calculate_total
18 ----------------------------------------------------------------------
19
20 * a3f2c9 2026-03-14 "Refactor: extract validation logic"
21 created function calculate_total
22
23 * cb4afa 2026-03-15 "Perf: optimise total calculation"
24 modified implementation changed
25
26 * 1d2e3f 2026-03-16 "Rename: calculate_total -> compute_total"
27 renamed calculate_total -> compute_total
28 (tracking continues as src/utils.py::compute_total)
29
30 * 4a5b6c 2026-03-17 "Move: refactor to helpers module"
31 moved src/utils.py::compute_total -> src/helpers.py::compute_total
32 (tracking continues at src/helpers.py::compute_total)
33
34 4 events (created: 1 modified: 1 renamed: 1 moved: 1)
35
36 Flags::
37
38 --from <ref>
39 Start walking from this commit instead of HEAD.
40 Accepts a full or abbreviated commit SHA or a branch name.
41
42 --max <n>
43 Cap the number of commits inspected (default: 500).
44 When the cap is reached a warning is shown; increase it with
45 --max to see the full history.
46
47 --json
48 Emit a structured JSON object::
49
50 {
51 "address": "<file>::<symbol>",
52 "start_ref": "HEAD",
53 "total_commits_scanned": 282,
54 "truncated": false,
55 "events": [
56 {
57 "event": "created",
58 "commit_id": "<sha256>",
59 "message": "...",
60 "committed_at": "2026-03-14T...",
61 "address": "<file>::<symbol>",
62 "detail": "...",
63 "new_address": null
64 }
65 ]
66 }
67 """
68
69 import argparse
70 import json
71 import logging
72 import pathlib
73 import sys
74 from typing import Literal, TypedDict
75
76 from muse.core.types import short_id
77 from muse.core.envelope import EnvelopeJson, make_envelope
78 from muse.core.errors import ExitCode
79 from muse.core.repo import require_repo
80 from muse.core.refs import read_current_branch
81 from muse.core.commits import (
82 CommitRecord,
83 resolve_commit_ref,
84 )
85 from muse.core.timing import start_timer
86 from muse.domain import DomainOp
87 from muse.plugins.code._query import walk_commits_bfs
88 from muse.core.validation import clamp_int, sanitize_display
89
90 class _SymbolLogJson(EnvelopeJson):
91 address: str
92 start_ref: str
93 total_commits_scanned: int
94 truncated: bool
95 events: list[_SymbolLogDict]
96
97 type _IntMap = dict[str, int]
98 type _SymbolLogDict = dict[str, str | None]
99 logger = logging.getLogger(__name__)
100
101 _EventKind = Literal["created", "modified", "renamed", "moved", "deleted", "signature"]
102
103 # ---------------------------------------------------------------------------
104 # Repository helpers
105 # ---------------------------------------------------------------------------
106
107 # ---------------------------------------------------------------------------
108 # Event extraction
109 # ---------------------------------------------------------------------------
110
111 def _flat_ops(ops: list[DomainOp]) -> list[DomainOp]:
112 """Flatten PatchOp children into a single list for address scanning."""
113 result: list[DomainOp] = []
114 for op in ops:
115 if op["op"] == "patch":
116 result.extend(op["child_ops"])
117 else:
118 result.append(op)
119 return result
120
121 class SymbolEvent:
122 """A single event in a symbol's lifecycle."""
123
124 __slots__ = ("kind", "commit", "address", "detail", "new_address")
125
126 def __init__(
127 self,
128 kind: _EventKind,
129 commit: CommitRecord,
130 address: str,
131 detail: str,
132 new_address: str | None = None,
133 ) -> None:
134 self.kind = kind
135 self.commit = commit
136 self.address = address
137 self.detail = detail
138 self.new_address = new_address
139
140 def to_dict(self) -> _SymbolLogDict:
141 return {
142 "event": self.kind,
143 "commit_id": self.commit.commit_id,
144 "message": self.commit.message,
145 "committed_at": self.commit.committed_at.isoformat(),
146 "address": self.address,
147 "detail": self.detail,
148 "new_address": self.new_address,
149 }
150
151 def _find_events_in_commit(
152 commit: CommitRecord,
153 address: str,
154 ) -> tuple[list[SymbolEvent], str]:
155 """Scan *commit*'s structured delta for events touching *address*.
156
157 Returns ``(events, next_address)`` where *next_address* is the address to
158 track in older commits (updated on rename events so history is continuous).
159 """
160 events: list[SymbolEvent] = []
161 next_address = address
162
163 if commit.structured_delta is None:
164 return events, next_address
165
166 all_ops = _flat_ops(commit.structured_delta["ops"])
167
168 for op in all_ops:
169 op_address = op["address"]
170
171 if op["op"] == "insert" and op_address == address:
172 events.append(SymbolEvent(
173 kind="created",
174 commit=commit,
175 address=address,
176 detail=op.get("content_summary", "created"),
177 ))
178
179 elif op["op"] == "delete" and op_address == address:
180 detail = op.get("content_summary", "deleted")
181 if "moved to" in detail:
182 events.append(SymbolEvent(
183 kind="moved",
184 commit=commit,
185 address=address,
186 detail=detail,
187 new_address=None,
188 ))
189 else:
190 events.append(SymbolEvent(
191 kind="deleted",
192 commit=commit,
193 address=address,
194 detail=detail,
195 ))
196
197 elif op["op"] == "replace" and op_address == address:
198 new_summary: str = op.get("new_summary", "")
199 if new_summary.startswith("renamed to "):
200 new_name = new_summary.removeprefix("renamed to ").strip()
201 file_prefix = address.rsplit("::", 1)[0]
202 new_addr = f"{file_prefix}::{new_name}"
203 events.append(SymbolEvent(
204 kind="renamed",
205 commit=commit,
206 address=address,
207 detail=f"{address.rsplit('::', 1)[-1]} → {new_name}",
208 new_address=new_addr,
209 ))
210 next_address = new_addr
211 elif new_summary.startswith("moved to "):
212 events.append(SymbolEvent(
213 kind="moved",
214 commit=commit,
215 address=address,
216 detail=new_summary,
217 new_address=None,
218 ))
219 elif "signature" in new_summary:
220 events.append(SymbolEvent(
221 kind="signature",
222 commit=commit,
223 address=address,
224 detail=new_summary,
225 ))
226 else:
227 events.append(SymbolEvent(
228 kind="modified",
229 commit=commit,
230 address=address,
231 detail=new_summary or "modified",
232 ))
233
234 return events, next_address
235
236 # ---------------------------------------------------------------------------
237 # Output
238 # ---------------------------------------------------------------------------
239
240 def _print_human(
241 address: str,
242 events: list[SymbolEvent],
243 total_commits: int,
244 truncated: bool,
245 ) -> None:
246 print(f"\nSymbol: {sanitize_display(address)}")
247 print("─" * 62)
248
249 if truncated:
250 print(
251 f"\n⚠️ History may be incomplete — scanned {total_commits:,} commits "
252 f"(use --max to increase the limit).",
253 )
254
255 if not events:
256 print(" (no events found — symbol may not exist in this range)")
257 return
258
259 # Collected newest-first; reverse for chronological (oldest-first) display.
260 chrono = list(reversed(events))
261
262 counts: _IntMap = {}
263 for ev in chrono:
264 counts[ev.kind] = counts.get(ev.kind, 0) + 1
265 date_str = ev.commit.committed_at.strftime("%Y-%m-%d")
266 cid = short_id(ev.commit.commit_id)
267 print(f'\n● {cid} {date_str} "{sanitize_display(ev.commit.message)}"')
268 print(f" {ev.kind:<12} {ev.detail}")
269 if ev.new_address:
270 print(f" (tracking continues as {ev.new_address})")
271
272 total = len(events)
273 summary_parts = [f"{k}: {v}" for k, v in sorted(counts.items())]
274 print(f"\n{total} event(s) ({', '.join(summary_parts)})")
275
276 # ---------------------------------------------------------------------------
277 # Argument parser registration
278 # ---------------------------------------------------------------------------
279
280 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
281 """Register the symbol-log subcommand.
282
283 Arguments
284 ---------
285 ADDRESS
286 Fully-qualified symbol address, e.g. ``src/utils.py::calculate_total``.
287 Must contain ``::``.
288 --from REF
289 Start walking from this commit / branch (default: HEAD).
290 --max N
291 Maximum commits to inspect (default: 500).
292 --json / -j
293 Emit the full event list as structured JSON with schema_version,
294 exit_code, and duration_ms in the envelope.
295 """
296 parser = subparsers.add_parser(
297 "symbol-log",
298 help="Track a single symbol through the entire commit history.",
299 description=__doc__,
300 formatter_class=argparse.RawDescriptionHelpFormatter,
301 )
302 parser.add_argument(
303 "address",
304 metavar="ADDRESS",
305 help=(
306 'Fully-qualified symbol address, e.g. "src/utils.py::calculate_total" '
307 'or "src/models.py::User.save". Must contain "::".'
308 ),
309 )
310 parser.add_argument(
311 "--from",
312 dest="from_ref",
313 default=None,
314 metavar="REF",
315 help="Start walking from this commit / branch (default: HEAD).",
316 )
317 parser.add_argument(
318 "--max",
319 dest="max_commits",
320 type=int,
321 default=500,
322 metavar="N",
323 help="Maximum number of commits to inspect (default: 500).",
324 )
325 parser.add_argument(
326 "--json", "-j",
327 dest="json_out",
328 action="store_true",
329 help="Emit the full event list as structured JSON.",
330 )
331 parser.set_defaults(func=run, json_out=False)
332
333 # ---------------------------------------------------------------------------
334 # Command entry point
335 # ---------------------------------------------------------------------------
336
337 def run(args: argparse.Namespace) -> None:
338 """Track a single symbol through the entire commit history.
339
340 ``muse code symbol-log`` is impossible in Git: Git tracks file lines, not
341 semantic symbols. This command follows a function, class, or method
342 across every commit — detecting creation, implementation changes, renames,
343 cross-file moves, and deletion, including events on merged feature branches.
344
345 Agent quickstart::
346
347 muse code symbol-log "src/utils.py::calculate_total" --json
348 muse code symbol-log "src/models.py::User.save" --json
349 muse code symbol-log "src/utils.py::Fn" --from v1.0.0 --json
350 muse code symbol-log "src/utils.py::Fn" --max 1000 --json
351
352 JSON fields::
353
354 address Symbol address that was tracked.
355 start_ref Ref the walk started from (``"HEAD"`` by default).
356 total_commits_scanned Number of commits actually walked.
357 truncated ``true`` when ``--max`` capped the walk.
358 events Lifecycle events (oldest first): ``event``, ``commit_id``,
359 ``message``, ``committed_at``, ``address``, ``detail``, ``new_address``.
360 muse_version Muse release that produced this output.
361 schema Envelope schema version (int).
362 exit_code ``0`` on success.
363 duration_ms Wall-clock milliseconds for the command.
364 timestamp ISO-8601 UTC timestamp of command completion.
365 warnings List of non-fatal advisory messages.
366
367 Exit codes::
368
369 0 Success.
370 1 User error (invalid address, bad ref).
371 """
372 elapsed = start_timer()
373 address: str = args.address
374 from_ref: str | None = args.from_ref
375 max_commits: int = clamp_int(args.max_commits, 1, 100000, 'max_commits')
376 json_out: bool = args.json_out
377
378 # ── Input validation ──────────────────────────────────────────────────────
379
380 if "::" not in address:
381 print(
382 f"❌ '{address}' is not a valid symbol address.\n"
383 " Expected 'file_path::SymbolName', "
384 "e.g. 'src/utils.py::calculate_total'.",
385 file=sys.stderr,
386 )
387 raise SystemExit(ExitCode.USER_ERROR)
388
389 if max_commits < 1:
390 print("❌ --max must be at least 1.", file=sys.stderr)
391 raise SystemExit(ExitCode.USER_ERROR)
392
393 # ── Repo / commit resolution ──────────────────────────────────────────────
394
395 root = require_repo()
396 branch = read_current_branch(root)
397
398 start_commit = resolve_commit_ref(root, branch, from_ref)
399 if start_commit is None:
400 label = from_ref or "HEAD"
401 print(f"❌ Commit '{label}' not found.", file=sys.stderr)
402 raise SystemExit(ExitCode.USER_ERROR)
403
404 # ── DAG walk ──────────────────────────────────────────────────────────────
405
406 commits, truncated = walk_commits_bfs(root, start_commit.commit_id, max_commits)
407
408 # ── Event scan ────────────────────────────────────────────────────────────
409 # Walk commits newest-first, tracking address changes on rename events.
410
411 current_address = address
412 all_events: list[SymbolEvent] = []
413
414 for commit in commits:
415 evs, current_address = _find_events_in_commit(commit, current_address)
416 all_events.extend(evs)
417
418 # ── Output ────────────────────────────────────────────────────────────────
419
420 if json_out:
421 print(json.dumps(_SymbolLogJson(
422 **make_envelope(elapsed),
423 address=address,
424 start_ref=from_ref or "HEAD",
425 total_commits_scanned=len(commits),
426 truncated=truncated,
427 events=[e.to_dict() for e in reversed(all_events)],
428 )))
429 return
430
431 _print_human(address, all_events, total_commits=len(commits), truncated=truncated)
File History 1 commit
sha256:6ed6b11aceb96bef850842a7aaac54c04843fa9fba5d1200e4bc3845e12d4142 docs: add muse-tag-guide mist (complete idiomatic guide wit… Sonnet 4.6 19 days ago