reflog.py
python
| 1 | """Reflog — record every HEAD and branch-ref movement. |
| 2 | |
| 3 | The reflog is a append-only per-ref journal that records every time a branch |
| 4 | pointer moves, regardless of cause: commit, checkout, merge, reset, cherry-pick, |
| 5 | stash pop. It is a safety net — if a ``muse reset --hard`` goes wrong the |
| 6 | reflog tells you exactly what ``HEAD`` was before. |
| 7 | |
| 8 | Storage layout:: |
| 9 | |
| 10 | .muse/logs/ |
| 11 | HEAD — log for the symbolic HEAD pointer |
| 12 | refs/ |
| 13 | heads/ |
| 14 | main — log for the main branch ref |
| 15 | dev — log for the dev branch ref |
| 16 | … |
| 17 | |
| 18 | Each log file is a plain-text append-only sequence of lines:: |
| 19 | |
| 20 | <old_id> <new_id> <author> <timestamp_unix> <tz_offset> \\t<operation> |
| 21 | |
| 22 | This format mirrors Git's reflog so tooling that understands both can be |
| 23 | built without translation. |
| 24 | |
| 25 | Fields |
| 26 | ------ |
| 27 | old_id 64-hex SHA-256 or ``0000…0000`` when there is no predecessor |
| 28 | (initial commit on a branch, new branch creation). |
| 29 | new_id 64-hex SHA-256 of the new commit. |
| 30 | author ``Name <email>`` or a short label for automated operations. |
| 31 | timestamp_unix Unix seconds as a decimal integer. |
| 32 | tz_offset UTC offset in ``+HHMM`` / ``-HHMM`` form. |
| 33 | operation Free-form description, e.g. ``commit: add verse``, |
| 34 | ``checkout: moving from main to dev``, |
| 35 | ``reset: moving to <sha12>``. |
| 36 | """ |
| 37 | |
| 38 | from __future__ import annotations |
| 39 | |
| 40 | import datetime |
| 41 | import logging |
| 42 | import pathlib |
| 43 | from dataclasses import dataclass |
| 44 | |
| 45 | logger = logging.getLogger(__name__) |
| 46 | |
| 47 | _NULL_ID = "0" * 64 |
| 48 | _LOG_DIR_NAME = "logs" |
| 49 | |
| 50 | |
| 51 | # --------------------------------------------------------------------------- |
| 52 | # Types |
| 53 | # --------------------------------------------------------------------------- |
| 54 | |
| 55 | |
| 56 | @dataclass(frozen=True) |
| 57 | class ReflogEntry: |
| 58 | """One line of a reflog.""" |
| 59 | |
| 60 | old_id: str |
| 61 | new_id: str |
| 62 | author: str |
| 63 | timestamp: datetime.datetime |
| 64 | operation: str |
| 65 | |
| 66 | |
| 67 | # --------------------------------------------------------------------------- |
| 68 | # Path helpers |
| 69 | # --------------------------------------------------------------------------- |
| 70 | |
| 71 | |
| 72 | def _logs_dir(repo_root: pathlib.Path) -> pathlib.Path: |
| 73 | return repo_root / ".muse" / _LOG_DIR_NAME |
| 74 | |
| 75 | |
| 76 | def _head_log_path(repo_root: pathlib.Path) -> pathlib.Path: |
| 77 | return _logs_dir(repo_root) / "HEAD" |
| 78 | |
| 79 | |
| 80 | def _ref_log_path(repo_root: pathlib.Path, branch: str) -> pathlib.Path: |
| 81 | return _logs_dir(repo_root) / "refs" / "heads" / branch |
| 82 | |
| 83 | |
| 84 | # --------------------------------------------------------------------------- |
| 85 | # Write |
| 86 | # --------------------------------------------------------------------------- |
| 87 | |
| 88 | |
| 89 | def append_reflog( |
| 90 | repo_root: pathlib.Path, |
| 91 | branch: str, |
| 92 | old_id: str | None, |
| 93 | new_id: str, |
| 94 | author: str, |
| 95 | operation: str, |
| 96 | ) -> None: |
| 97 | """Append one entry to both the branch log and the HEAD log. |
| 98 | |
| 99 | Args: |
| 100 | repo_root: Root of the Muse repository. |
| 101 | branch: Branch whose ref is moving (e.g. ``"main"``). |
| 102 | old_id: Previous commit SHA-256 (``None`` for initial commit). |
| 103 | new_id: New commit SHA-256. |
| 104 | author: ``"Name <email>"`` or a short label. |
| 105 | operation: Human-readable description of the operation. |
| 106 | """ |
| 107 | effective_old = old_id or _NULL_ID |
| 108 | now = datetime.datetime.now(tz=datetime.timezone.utc) |
| 109 | ts = int(now.timestamp()) |
| 110 | tz_offset = "+0000" # we always store UTC; offset is for display parity |
| 111 | |
| 112 | line = f"{effective_old} {new_id} {author} {ts} {tz_offset}\t{operation}\n" |
| 113 | |
| 114 | for log_path in (_ref_log_path(repo_root, branch), _head_log_path(repo_root)): |
| 115 | log_path.parent.mkdir(parents=True, exist_ok=True) |
| 116 | with log_path.open("a", encoding="utf-8") as fh: |
| 117 | fh.write(line) |
| 118 | |
| 119 | |
| 120 | # --------------------------------------------------------------------------- |
| 121 | # Read |
| 122 | # --------------------------------------------------------------------------- |
| 123 | |
| 124 | |
| 125 | def _parse_line(line: str) -> ReflogEntry | None: |
| 126 | """Parse one reflog line; return None on malformed input.""" |
| 127 | line = line.rstrip("\n") |
| 128 | # Split on the tab that separates the metadata from the operation string. |
| 129 | parts = line.split("\t", 1) |
| 130 | if len(parts) != 2: |
| 131 | return None |
| 132 | meta, operation = parts |
| 133 | tokens = meta.split() |
| 134 | if len(tokens) < 5: |
| 135 | return None |
| 136 | old_id, new_id = tokens[0], tokens[1] |
| 137 | # author may contain spaces — everything between tokens[2] and the last |
| 138 | # two tokens (timestamp, tz_offset) is the author. |
| 139 | ts_str = tokens[-2] |
| 140 | author = " ".join(tokens[2:-2]) |
| 141 | try: |
| 142 | ts = datetime.datetime.fromtimestamp(int(ts_str), tz=datetime.timezone.utc) |
| 143 | except (ValueError, OSError): |
| 144 | ts = datetime.datetime.now(tz=datetime.timezone.utc) |
| 145 | return ReflogEntry( |
| 146 | old_id=old_id, |
| 147 | new_id=new_id, |
| 148 | author=author, |
| 149 | timestamp=ts, |
| 150 | operation=operation, |
| 151 | ) |
| 152 | |
| 153 | |
| 154 | def read_reflog( |
| 155 | repo_root: pathlib.Path, |
| 156 | branch: str | None = None, |
| 157 | limit: int = 100, |
| 158 | ) -> list[ReflogEntry]: |
| 159 | """Return reflog entries newest-first. |
| 160 | |
| 161 | Args: |
| 162 | repo_root: Root of the Muse repository. |
| 163 | branch: Branch name, or ``None`` to read the HEAD log. |
| 164 | limit: Maximum number of entries to return. |
| 165 | """ |
| 166 | log_path = ( |
| 167 | _head_log_path(repo_root) if branch is None else _ref_log_path(repo_root, branch) |
| 168 | ) |
| 169 | if not log_path.exists(): |
| 170 | return [] |
| 171 | try: |
| 172 | lines = log_path.read_text(encoding="utf-8").splitlines() |
| 173 | except OSError as exc: |
| 174 | logger.warning("⚠️ Could not read reflog %s: %s", log_path, exc) |
| 175 | return [] |
| 176 | entries: list[ReflogEntry] = [] |
| 177 | for line in reversed(lines): |
| 178 | if not line.strip(): |
| 179 | continue |
| 180 | entry = _parse_line(line) |
| 181 | if entry is not None: |
| 182 | entries.append(entry) |
| 183 | if len(entries) >= limit: |
| 184 | break |
| 185 | return entries |
| 186 | |
| 187 | |
| 188 | def list_reflog_refs(repo_root: pathlib.Path) -> list[str]: |
| 189 | """Return branch names that have a reflog, sorted.""" |
| 190 | refs_dir = _logs_dir(repo_root) / "refs" / "heads" |
| 191 | if not refs_dir.exists(): |
| 192 | return [] |
| 193 | return sorted(p.name for p in refs_dir.iterdir() if p.is_file()) |