shortlog.py
python
| 1 | """``muse shortlog`` — commit summary grouped by author or agent. |
| 2 | |
| 3 | Groups the commit history by author (humans) or agent_id (agents), counts |
| 4 | commits per group, and optionally lists commit messages under each. Useful |
| 5 | for changelogs, release notes, and auditing agent contribution. |
| 6 | |
| 7 | Muse's rich commit metadata — ``author``, ``agent_id``, ``model_id`` — makes |
| 8 | shortlog especially expressive: you can see exactly which human or which agent |
| 9 | class contributed each set of commits. |
| 10 | |
| 11 | Usage:: |
| 12 | |
| 13 | muse shortlog # current branch, group by author |
| 14 | muse shortlog --all # all branches |
| 15 | muse shortlog --numbered # sort by commit count (most active first) |
| 16 | muse shortlog --format json # machine-readable |
| 17 | |
| 18 | Exit codes:: |
| 19 | |
| 20 | 0 — output produced (even if empty) |
| 21 | 1 — branch not found or ref invalid |
| 22 | """ |
| 23 | |
| 24 | from __future__ import annotations |
| 25 | |
| 26 | import argparse |
| 27 | import sys |
| 28 | |
| 29 | import json |
| 30 | import logging |
| 31 | import pathlib |
| 32 | from collections import defaultdict |
| 33 | |
| 34 | |
| 35 | from muse.core.errors import ExitCode |
| 36 | from muse.core.repo import require_repo |
| 37 | from muse.core.store import ( |
| 38 | CommitRecord, |
| 39 | get_commits_for_branch, |
| 40 | get_head_commit_id, |
| 41 | read_current_branch, |
| 42 | ) |
| 43 | from muse.core.validation import sanitize_display |
| 44 | |
| 45 | logger = logging.getLogger(__name__) |
| 46 | |
| 47 | |
| 48 | def _read_repo_id(root: pathlib.Path) -> str: |
| 49 | return str(json.loads((root / ".muse" / "repo.json").read_text(encoding="utf-8"))["repo_id"]) |
| 50 | |
| 51 | |
| 52 | def _branch_names(root: pathlib.Path) -> list[str]: |
| 53 | heads_dir = root / ".muse" / "refs" / "heads" |
| 54 | if not heads_dir.exists(): |
| 55 | return [] |
| 56 | branches: list[str] = [] |
| 57 | for ref_file in sorted(heads_dir.rglob("*")): |
| 58 | if ref_file.is_file(): |
| 59 | branches.append(str(ref_file.relative_to(heads_dir).as_posix())) |
| 60 | return branches |
| 61 | |
| 62 | |
| 63 | def _author_key(commit: CommitRecord) -> str: |
| 64 | """Return the display key for grouping: prefer author, fall back to agent_id.""" |
| 65 | if commit.author: |
| 66 | return commit.author |
| 67 | if commit.agent_id: |
| 68 | return f"{commit.agent_id} (agent)" |
| 69 | return "(unknown)" |
| 70 | |
| 71 | |
| 72 | def _build_groups( |
| 73 | commits: list[CommitRecord], |
| 74 | *, |
| 75 | by_email: bool, |
| 76 | ) -> dict[str, list[CommitRecord]]: |
| 77 | groups: dict[str, list[CommitRecord]] = defaultdict(list) |
| 78 | for c in commits: |
| 79 | key = _author_key(c) |
| 80 | if by_email and c.agent_id and c.agent_id != c.author: |
| 81 | key = f"{key} <{c.agent_id}>" |
| 82 | groups[key].append(c) |
| 83 | return dict(groups) |
| 84 | |
| 85 | |
| 86 | def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: |
| 87 | """Register the shortlog subcommand.""" |
| 88 | parser = subparsers.add_parser( |
| 89 | "shortlog", |
| 90 | help='Summarise commit history grouped by author or agent.', |
| 91 | description=__doc__, |
| 92 | ) |
| 93 | parser.add_argument( |
| 94 | "branch_opt", |
| 95 | nargs="?", |
| 96 | default=None, |
| 97 | metavar="BRANCH", |
| 98 | help="Branch to summarise (default: current branch).", |
| 99 | ) |
| 100 | parser.add_argument( |
| 101 | "--all", |
| 102 | dest="all_branches", |
| 103 | action="store_true", |
| 104 | help="Include all branches.", |
| 105 | ) |
| 106 | parser.add_argument( |
| 107 | "--numbered", |
| 108 | action="store_true", |
| 109 | help="Sort by commit count (most active first).", |
| 110 | ) |
| 111 | parser.add_argument( |
| 112 | "--email", |
| 113 | dest="by_email", |
| 114 | action="store_true", |
| 115 | help="Include agent_id alongside author.", |
| 116 | ) |
| 117 | parser.add_argument( |
| 118 | "--limit", |
| 119 | type=int, |
| 120 | default=0, |
| 121 | metavar="N", |
| 122 | help="Cap the number of commits loaded (0 = no limit).", |
| 123 | ) |
| 124 | parser.add_argument( |
| 125 | "--format", |
| 126 | dest="fmt", |
| 127 | default="text", |
| 128 | choices=["text", "json"], |
| 129 | help="Output format: text (default) or json.", |
| 130 | ) |
| 131 | parser.set_defaults(func=run) |
| 132 | |
| 133 | |
| 134 | def run(args: argparse.Namespace) -> None: |
| 135 | """Summarise commit history grouped by author or agent. |
| 136 | |
| 137 | Each group lists the author, commit count, and (in text mode) each commit |
| 138 | message indented beneath. Use ``--numbered`` to rank by activity. |
| 139 | |
| 140 | In agent pipelines, ``--format json`` returns structured data that can be |
| 141 | piped to any downstream processor. |
| 142 | |
| 143 | Examples:: |
| 144 | |
| 145 | muse shortlog # current branch |
| 146 | muse shortlog --all --numbered # all branches, ranked by count |
| 147 | muse shortlog --email # include agent_id |
| 148 | muse shortlog --format json # JSON for agent consumption |
| 149 | """ |
| 150 | branch_opt: str | None = args.branch_opt |
| 151 | all_branches: bool = args.all_branches |
| 152 | numbered: bool = args.numbered |
| 153 | by_email: bool = args.by_email |
| 154 | limit: int = args.limit |
| 155 | fmt: str = args.fmt |
| 156 | |
| 157 | if fmt not in {"text", "json"}: |
| 158 | print(f"❌ Unknown --format '{sanitize_display(fmt)}'. Choose text or json.", file=sys.stderr) |
| 159 | raise SystemExit(ExitCode.USER_ERROR) |
| 160 | |
| 161 | root = require_repo() |
| 162 | repo_id = _read_repo_id(root) |
| 163 | |
| 164 | branches: list[str] |
| 165 | if all_branches: |
| 166 | branches = _branch_names(root) |
| 167 | if not branches: |
| 168 | if fmt == "json": |
| 169 | print("[]") |
| 170 | else: |
| 171 | print("No commits found.") |
| 172 | return |
| 173 | else: |
| 174 | branches = [branch_opt or read_current_branch(root)] |
| 175 | |
| 176 | # Collect all commits across selected branches (deduplicated by commit_id). |
| 177 | seen_ids: set[str] = set() |
| 178 | all_commits: list[CommitRecord] = [] |
| 179 | for br in branches: |
| 180 | branch_commits = get_commits_for_branch(root, repo_id, br) |
| 181 | for c in branch_commits: |
| 182 | if c.commit_id not in seen_ids: |
| 183 | seen_ids.add(c.commit_id) |
| 184 | all_commits.append(c) |
| 185 | if limit and len(all_commits) >= limit: |
| 186 | all_commits = all_commits[:limit] |
| 187 | break |
| 188 | |
| 189 | if not all_commits: |
| 190 | if fmt == "json": |
| 191 | print("[]") |
| 192 | else: |
| 193 | print("No commits found.") |
| 194 | return |
| 195 | |
| 196 | groups = _build_groups(all_commits, by_email=by_email) |
| 197 | |
| 198 | # Sort: by count descending (if --numbered), then alphabetically. |
| 199 | sorted_keys: list[str] |
| 200 | if numbered: |
| 201 | sorted_keys = sorted(groups, key=lambda k: -len(groups[k])) |
| 202 | else: |
| 203 | sorted_keys = sorted(groups) |
| 204 | |
| 205 | if fmt == "json": |
| 206 | output = [ |
| 207 | { |
| 208 | "author": key, |
| 209 | "count": len(groups[key]), |
| 210 | "commits": [ |
| 211 | { |
| 212 | "commit_id": c.commit_id, |
| 213 | "message": c.message, |
| 214 | "committed_at": c.committed_at.isoformat(), |
| 215 | } |
| 216 | for c in groups[key] |
| 217 | ], |
| 218 | } |
| 219 | for key in sorted_keys |
| 220 | ] |
| 221 | print(json.dumps(output, indent=2)) |
| 222 | else: |
| 223 | for key in sorted_keys: |
| 224 | commits_in_group = groups[key] |
| 225 | print(f"{sanitize_display(key)} ({len(commits_in_group)}):") |
| 226 | for c in commits_in_group: |
| 227 | print(f" {sanitize_display(c.message)}") |
| 228 | print("") |