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 json |
| 27 | import logging |
| 28 | import pathlib |
| 29 | from collections import defaultdict |
| 30 | from typing import Annotated |
| 31 | |
| 32 | import typer |
| 33 | |
| 34 | from muse.core.errors import ExitCode |
| 35 | from muse.core.repo import require_repo |
| 36 | from muse.core.store import ( |
| 37 | CommitRecord, |
| 38 | get_commits_for_branch, |
| 39 | get_head_commit_id, |
| 40 | read_current_branch, |
| 41 | ) |
| 42 | from muse.core.validation import sanitize_display |
| 43 | |
| 44 | logger = logging.getLogger(__name__) |
| 45 | |
| 46 | app = typer.Typer(help="Commit summary grouped by author or agent.") |
| 47 | |
| 48 | |
| 49 | def _read_repo_id(root: pathlib.Path) -> str: |
| 50 | return str(json.loads((root / ".muse" / "repo.json").read_text(encoding="utf-8"))["repo_id"]) |
| 51 | |
| 52 | |
| 53 | def _branch_names(root: pathlib.Path) -> list[str]: |
| 54 | heads_dir = root / ".muse" / "refs" / "heads" |
| 55 | if not heads_dir.exists(): |
| 56 | return [] |
| 57 | branches: list[str] = [] |
| 58 | for ref_file in sorted(heads_dir.rglob("*")): |
| 59 | if ref_file.is_file(): |
| 60 | branches.append(str(ref_file.relative_to(heads_dir).as_posix())) |
| 61 | return branches |
| 62 | |
| 63 | |
| 64 | def _author_key(commit: CommitRecord) -> str: |
| 65 | """Return the display key for grouping: prefer author, fall back to agent_id.""" |
| 66 | if commit.author: |
| 67 | return commit.author |
| 68 | if commit.agent_id: |
| 69 | return f"{commit.agent_id} (agent)" |
| 70 | return "(unknown)" |
| 71 | |
| 72 | |
| 73 | def _build_groups( |
| 74 | commits: list[CommitRecord], |
| 75 | *, |
| 76 | by_email: bool, |
| 77 | ) -> dict[str, list[CommitRecord]]: |
| 78 | groups: dict[str, list[CommitRecord]] = defaultdict(list) |
| 79 | for c in commits: |
| 80 | key = _author_key(c) |
| 81 | if by_email and c.agent_id and c.agent_id != c.author: |
| 82 | key = f"{key} <{c.agent_id}>" |
| 83 | groups[key].append(c) |
| 84 | return dict(groups) |
| 85 | |
| 86 | |
| 87 | @app.callback(invoke_without_command=True) |
| 88 | def shortlog( |
| 89 | branch_opt: Annotated[ |
| 90 | str | None, |
| 91 | typer.Option("--branch", "-b", help="Branch to summarise (default: current branch)."), |
| 92 | ] = None, |
| 93 | all_branches: Annotated[ |
| 94 | bool, |
| 95 | typer.Option("--all", "-a", help="Summarise commits across all branches."), |
| 96 | ] = False, |
| 97 | numbered: Annotated[ |
| 98 | bool, |
| 99 | typer.Option("--numbered", "-n", help="Sort by commit count (most active first)."), |
| 100 | ] = False, |
| 101 | by_email: Annotated[ |
| 102 | bool, |
| 103 | typer.Option("--email", "-e", help="Include agent_id alongside author name."), |
| 104 | ] = False, |
| 105 | limit: Annotated[ |
| 106 | int, |
| 107 | typer.Option("--limit", "-l", help="Maximum commits to walk per branch (0 = unlimited).", min=0), |
| 108 | ] = 0, |
| 109 | fmt: Annotated[ |
| 110 | str, |
| 111 | typer.Option("--format", "-f", help="Output format: text or json."), |
| 112 | ] = "text", |
| 113 | ) -> None: |
| 114 | """Summarise commit history grouped by author or agent. |
| 115 | |
| 116 | Each group lists the author, commit count, and (in text mode) each commit |
| 117 | message indented beneath. Use ``--numbered`` to rank by activity. |
| 118 | |
| 119 | In agent pipelines, ``--format json`` returns structured data that can be |
| 120 | piped to any downstream processor. |
| 121 | |
| 122 | Examples:: |
| 123 | |
| 124 | muse shortlog # current branch |
| 125 | muse shortlog --all --numbered # all branches, ranked by count |
| 126 | muse shortlog --email # include agent_id |
| 127 | muse shortlog --format json # JSON for agent consumption |
| 128 | """ |
| 129 | if fmt not in {"text", "json"}: |
| 130 | typer.echo(f"❌ Unknown --format '{sanitize_display(fmt)}'. Choose text or json.", err=True) |
| 131 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 132 | |
| 133 | root = require_repo() |
| 134 | repo_id = _read_repo_id(root) |
| 135 | |
| 136 | branches: list[str] |
| 137 | if all_branches: |
| 138 | branches = _branch_names(root) |
| 139 | if not branches: |
| 140 | if fmt == "json": |
| 141 | typer.echo("[]") |
| 142 | else: |
| 143 | typer.echo("No commits found.") |
| 144 | return |
| 145 | else: |
| 146 | branches = [branch_opt or read_current_branch(root)] |
| 147 | |
| 148 | # Collect all commits across selected branches (deduplicated by commit_id). |
| 149 | seen_ids: set[str] = set() |
| 150 | all_commits: list[CommitRecord] = [] |
| 151 | for br in branches: |
| 152 | branch_commits = get_commits_for_branch(root, repo_id, br) |
| 153 | for c in branch_commits: |
| 154 | if c.commit_id not in seen_ids: |
| 155 | seen_ids.add(c.commit_id) |
| 156 | all_commits.append(c) |
| 157 | if limit and len(all_commits) >= limit: |
| 158 | all_commits = all_commits[:limit] |
| 159 | break |
| 160 | |
| 161 | if not all_commits: |
| 162 | if fmt == "json": |
| 163 | typer.echo("[]") |
| 164 | else: |
| 165 | typer.echo("No commits found.") |
| 166 | return |
| 167 | |
| 168 | groups = _build_groups(all_commits, by_email=by_email) |
| 169 | |
| 170 | # Sort: by count descending (if --numbered), then alphabetically. |
| 171 | sorted_keys: list[str] |
| 172 | if numbered: |
| 173 | sorted_keys = sorted(groups, key=lambda k: -len(groups[k])) |
| 174 | else: |
| 175 | sorted_keys = sorted(groups) |
| 176 | |
| 177 | if fmt == "json": |
| 178 | output = [ |
| 179 | { |
| 180 | "author": key, |
| 181 | "count": len(groups[key]), |
| 182 | "commits": [ |
| 183 | { |
| 184 | "commit_id": c.commit_id, |
| 185 | "message": c.message, |
| 186 | "committed_at": c.committed_at.isoformat(), |
| 187 | } |
| 188 | for c in groups[key] |
| 189 | ], |
| 190 | } |
| 191 | for key in sorted_keys |
| 192 | ] |
| 193 | typer.echo(json.dumps(output, indent=2)) |
| 194 | else: |
| 195 | for key in sorted_keys: |
| 196 | commits_in_group = groups[key] |
| 197 | typer.echo(f"{sanitize_display(key)} ({len(commits_in_group)}):") |
| 198 | for c in commits_in_group: |
| 199 | typer.echo(f" {sanitize_display(c.message)}") |
| 200 | typer.echo("") |