gabriel / muse public
reflog.py python
381 lines 13.8 KB
Raw
sha256:e6465e8a9b7fa8e6223ed4a3576e96c568c913ae2caeb9c31f15e7a81b250b40 docs: add | jq convention to --json section of agent-guide Sonnet 4.6 1 day ago
1 """``muse reflog`` — inspect the history of HEAD and branch movements.
2
3 The reflog is a chronological journal of every time a ref moved: commits,
4 checkouts, merges, resets, cherry-picks, shelf pops. It is your safety net
5 when you need to undo an operation that moved HEAD.
6
7 Usage::
8
9 muse reflog # HEAD reflog, last 20 entries
10 muse reflog --branch dev # dev branch reflog
11 muse reflog --limit 100 # show more entries
12 muse reflog --all # list all refs that have a reflog
13 muse reflog --operation commit # only commit events
14 muse reflog --author alice # only events by alice
15 muse reflog --since 2026-01-01 # entries on or after date
16 muse reflog --until 2026-03-01 # entries on or before date
17 muse reflog --json # machine-readable JSON
18
19 Each text row shows::
20
21 @{N} <new_sha12> (<old_sha12>) <when> <author> <operation>
22
23 The ``@{N}`` syntax mirrors Git so scripts that already understand Git
24 reflogs need no translation.
25
26 Security model
27 --------------
28 - Branch names are validated via ``validate_branch_name`` before being used
29 to construct a filesystem path — prevents path traversal.
30 - All user-controlled values (operation, author, commit IDs, branch names)
31 are passed through ``sanitize_display()`` before terminal output —
32 prevents ANSI injection from stored reflog data.
33 - Date filter values are validated as ``YYYY-MM-DD`` before use.
34 - Error messages go to **stderr**; **stdout** carries only data.
35
36 Agent UX
37 --------
38 Pass ``--json`` for a stable ``_ReflogResultJson`` object on stdout.
39 All fields are always present. Apply ``--operation``, ``--author``,
40 ``--since``, ``--until`` to narrow without changing the JSON schema.
41
42 JSON schema (``--json``)::
43
44 {
45 "ref": "refs/heads/main",
46 "total": 3,
47 "limit": 20,
48 "duration_ms": 4.123,
49 "exit_code": 0,
50 "entries": [
51 {
52 "index": 0,
53 "new_id": "sha256:<64-hex>",
54 "old_id": "sha256:<64-hex or 000…>",
55 "timestamp": "2026-03-16T12:00:00+00:00",
56 "operation": "commit: add verse",
57 "author": "alice"
58 }
59 ]
60 }
61
62 ``new_id`` and ``old_id`` are always ``sha256:<64-hex>`` canonical form —
63 consistent with ``muse read-commit`` and all other ID-bearing commands.
64 The on-disk reflog stores bare hex; the JSON layer normalises them.
65
66 Exit codes
67 ----------
68 - 0 — success
69 - 1 — invalid arguments (bad branch, bad format, bad date)
70 - 2 — not inside a Muse repository
71 """
72
73 import argparse
74 import datetime
75 import json
76 import logging
77 import sys
78 import time
79 from typing import TypedDict
80
81 from muse.core.envelope import EnvelopeJson, make_envelope
82 from muse.core.errors import ExitCode
83 from muse.core.reflog import ReflogEntry, list_reflog_refs, read_reflog
84 from muse.core.repo import require_repo
85 from muse.core.validation import clamp_int, sanitize_display, validate_branch_name
86 from muse.core.types import NULL_COMMIT_ID, long_id, short_id
87 from muse.core.timing import start_timer
88
89
90 logger = logging.getLogger(__name__)
91
92 # ---------------------------------------------------------------------------
93 # JSON TypedDicts — stable machine-readable output schemas
94 # ---------------------------------------------------------------------------
95
96 class _ReflogEntryJson(TypedDict):
97 """One entry in the JSON reflog output."""
98
99 index: int
100 new_id: str
101 old_id: str
102 timestamp: str
103 operation: str
104 author: str
105
106 class _ReflogResultJson(EnvelopeJson):
107 """Top-level JSON object returned by ``muse reflog --json``.
108
109 All fields are always present. Envelope fields (``duration_ms``,
110 ``exit_code``, etc.) are command metadata — not reflog data — and are
111 always emitted regardless of filters applied.
112
113 ``new_id`` and ``old_id`` in each entry are ``sha256:<64-hex>``
114 canonical form, matching the Muse ID convention everywhere else.
115 """
116
117 ref: str
118 total: int
119 limit: int
120 entries: list[_ReflogEntryJson]
121
122 class _ReflogAllJson(EnvelopeJson):
123 """JSON object returned by ``muse reflog --all --json``."""
124
125 refs: list[str]
126 count: int
127
128 # ---------------------------------------------------------------------------
129 # Formatting helpers
130 # ---------------------------------------------------------------------------
131
132 def _fmt_entry(idx: int, entry: ReflogEntry, short: int = 12) -> str:
133 """Format one reflog entry for terminal display.
134
135 All user-controlled fields (new_id, old_id, author, operation) are
136 passed through ``sanitize_display()`` to prevent ANSI injection.
137 Short IDs are rendered as ``sha256:<12-hex>`` (19 chars) for
138 consistency with all other Muse commands.
139 """
140 new_short = sanitize_display(short_id(entry.new_id))
141 old_short = (
142 "initial"
143 if entry.old_id == NULL_COMMIT_ID
144 else sanitize_display(short_id(entry.old_id))
145 )
146 when = entry.timestamp.strftime("%Y-%m-%d %H:%M:%S UTC")
147 safe_op = sanitize_display(entry.operation)
148 safe_author = sanitize_display(entry.author or "")
149 author_col = f" {safe_author}" if safe_author else ""
150 return f"@{{{idx}}} {new_short} ({old_short}) {when}{author_col} {safe_op}"
151
152 def _parse_date(value: str, flag: str) -> datetime.datetime:
153 """Parse *value* as ``YYYY-MM-DD`` into a UTC-aware datetime.
154
155 Raises SystemExit(USER_ERROR) for invalid format.
156 """
157 try:
158 d = datetime.date.fromisoformat(value)
159 except ValueError:
160 print(
161 f"❌ Invalid date for {flag}: {sanitize_display(value)!r} — "
162 "expected YYYY-MM-DD.",
163 file=sys.stderr,
164 )
165 raise SystemExit(ExitCode.USER_ERROR)
166 return datetime.datetime(d.year, d.month, d.day, tzinfo=datetime.timezone.utc)
167
168 # ---------------------------------------------------------------------------
169 # Argument parser registration
170 # ---------------------------------------------------------------------------
171
172 def register(
173 subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]",
174 ) -> None:
175 """Register the ``reflog`` subcommand."""
176 parser = subparsers.add_parser(
177 "reflog",
178 help="Show the history of HEAD and branch-ref movements.",
179 description=__doc__,
180 formatter_class=argparse.RawDescriptionHelpFormatter,
181 )
182 parser.add_argument(
183 "--branch", "-b", default=None,
184 help="Branch to show reflog for (default: HEAD).",
185 )
186 parser.add_argument(
187 "--limit", type=int, default=20,
188 help="Maximum number of entries to show (after filters, default: 20).",
189 )
190 parser.add_argument(
191 "--all", action="store_true", dest="all_refs",
192 help="List all refs that have a reflog.",
193 )
194 parser.add_argument(
195 "--operation", default=None, metavar="PATTERN", dest="operation_filter",
196 help="Filter to entries whose operation contains PATTERN (case-insensitive).",
197 )
198 parser.add_argument(
199 "--author", default=None, metavar="PATTERN", dest="author_filter",
200 help="Filter to entries whose author contains PATTERN (case-insensitive).",
201 )
202 parser.add_argument(
203 "--since", default=None, metavar="YYYY-MM-DD", dest="since",
204 help="Show only entries on or after this date.",
205 )
206 parser.add_argument(
207 "--until", default=None, metavar="YYYY-MM-DD", dest="until",
208 help="Show only entries on or before this date.",
209 )
210 parser.add_argument(
211 "--json", "-j", action="store_true", dest="json_out",
212 help="Emit machine-readable JSON instead of human text.",
213 )
214 parser.set_defaults(func=run)
215
216 # ---------------------------------------------------------------------------
217 # Command entry point
218 # ---------------------------------------------------------------------------
219
220 def run(args: argparse.Namespace) -> None:
221 """Show the history of HEAD and branch-ref movements.
222
223 Every time HEAD or a branch ref moves — commit, checkout, merge, reset,
224 cherry-pick, shelf pop — Muse appends an entry to the reflog. Use this
225 command to find lost commits and undo accidental resets.
226
227 The ``--limit`` cap applies *after* all filters, so ``--limit 10
228 --operation commit`` always returns exactly 10 commit events (or fewer
229 if the history is shorter).
230
231 Agent quickstart::
232
233 muse reflog --json
234 muse reflog --branch dev --json
235 muse reflog --operation commit --limit 5 --json
236 muse reflog --all --json
237
238 JSON fields::
239
240 ref str full ref name (e.g. "refs/heads/main" or "HEAD")
241 total int number of entries after all filters applied
242 limit int display cap that was requested
243 entries list [{index, new_id, old_id, timestamp, operation, author}]
244
245 JSON fields (--all mode)::
246
247 refs list full ref names that have a reflog
248 count int number of refs
249
250 Exit codes::
251
252 0 Success.
253 1 Invalid arguments (bad branch, bad format, bad date).
254 2 Not inside a Muse repository.
255 """
256 elapsed = start_timer()
257 branch: str | None = args.branch
258 limit: int = clamp_int(args.limit, 1, 100_000, "limit")
259 all_refs: bool = args.all_refs
260 json_out: bool = args.json_out
261 operation_filter: str | None = args.operation_filter
262 author_filter: str | None = args.author_filter
263 since_str: str | None = args.since
264 until_str: str | None = args.until
265
266 # Parse date filters before repo access so bad dates fail fast.
267 since_dt: datetime.datetime | None = (
268 _parse_date(since_str, "--since") if since_str else None
269 )
270 until_dt: datetime.datetime | None = (
271 _parse_date(until_str, "--until") if until_str else None
272 )
273 if since_dt and until_dt and since_dt > until_dt:
274 print("❌ --since must not be after --until.", file=sys.stderr)
275 raise SystemExit(ExitCode.USER_ERROR)
276
277 repo_root = require_repo()
278
279 # ── --all mode ────────────────────────────────────────────────────────────
280
281 if all_refs:
282 refs = list_reflog_refs(repo_root)
283 full_refs = [f"refs/heads/{r}" for r in refs]
284 if json_out:
285 payload = _ReflogAllJson(
286 **make_envelope(elapsed),
287 refs=full_refs,
288 count=len(full_refs),
289 )
290 print(json.dumps(payload))
291 else:
292 if not refs:
293 print("No reflog entries found.")
294 return
295 print("Refs with reflog entries:")
296 for ref in refs:
297 print(f" refs/heads/{sanitize_display(ref)}")
298 return
299
300 # ── Branch validation ─────────────────────────────────────────────────────
301
302 if branch is not None:
303 try:
304 validate_branch_name(branch)
305 except ValueError as exc:
306 print(
307 f"❌ Invalid branch name: {sanitize_display(str(exc))}",
308 file=sys.stderr,
309 )
310 raise SystemExit(ExitCode.USER_ERROR)
311
312 # ── Read all entries (limit is applied after filtering) ───────────────────
313
314 # Read a generous cap so post-filter limit can be satisfied.
315 raw_limit = min(limit * 50, 100_000)
316 entries = read_reflog(repo_root, branch=branch, limit=raw_limit)
317
318 # ── Apply filters ─────────────────────────────────────────────────────────
319
320 filtered: list[ReflogEntry] = entries
321
322 if operation_filter is not None:
323 needle = operation_filter.lower()
324 filtered = [e for e in filtered if needle in e.operation.lower()]
325
326 if author_filter is not None:
327 needle_a = author_filter.lower()
328 filtered = [e for e in filtered if needle_a in (e.author or "").lower()]
329
330 if since_dt is not None:
331 filtered = [e for e in filtered if e.timestamp >= since_dt]
332
333 if until_dt is not None:
334 # until is inclusive: entries on or before the end of until_dt day
335 until_end = until_dt + datetime.timedelta(days=1)
336 filtered = [e for e in filtered if e.timestamp < until_end]
337
338 # Apply display limit after all filters.
339 displayed = filtered[:limit]
340
341 # ── Output ────────────────────────────────────────────────────────────────
342
343 ref_name = f"refs/heads/{branch}" if branch else "HEAD"
344
345 if json_out:
346 json_entries: list[_ReflogEntryJson] = [
347 _ReflogEntryJson(
348 index=idx,
349 new_id=long_id(e.new_id),
350 old_id=long_id(e.old_id),
351 timestamp=e.timestamp.isoformat(),
352 operation=e.operation,
353 author=e.author,
354 )
355 for idx, e in enumerate(displayed)
356 ]
357 payload_result = _ReflogResultJson(
358 **make_envelope(elapsed),
359 ref=ref_name,
360 total=len(filtered),
361 limit=limit,
362 entries=json_entries,
363 )
364 print(json.dumps(payload_result))
365 return
366
367 safe_label = sanitize_display(ref_name)
368 if not displayed:
369 if filtered:
370 print(f"No reflog entries for {safe_label} match the active filters.")
371 else:
372 print(f"No reflog entries for {safe_label}.")
373 return
374
375 print(f"Reflog for {safe_label} (newest first)\n")
376 for idx, entry in enumerate(displayed):
377 print(_fmt_entry(idx, entry))
378
379 if len(filtered) > limit:
380 remaining = len(filtered) - limit
381 print(f"\n … {remaining} older entry/entries — increase --limit to see more.")
File History 1 commit
sha256:e6465e8a9b7fa8e6223ed4a3576e96c568c913ae2caeb9c31f15e7a81b250b40 docs: add | jq convention to --json section of agent-guide Sonnet 4.6 1 day ago