"""``muse annotate`` — attach CRDT-backed metadata to an existing commit. Annotations use real CRDT semantics so that multiple agents can annotate the same commit concurrently without conflicts: - ``--reviewed-by`` merges into the commit's ``reviewed_by`` field using **ORSet** semantics (set union — once added, a reviewer is never lost). - ``--remove-reviewer`` removes a reviewer from the stored list. - ``--test-run`` increments the commit's ``test_runs`` field using **GCounter** semantics (monotonically increasing). These annotations are persisted directly in the commit JSON on disk. Commit reference ---------------- The *commit* argument accepts any reference that ``resolve_commit_ref`` understands: - ``HEAD`` or omitted — the most recent commit on the current branch. - ``HEAD~N`` — *N* first-parent steps back from HEAD. - A full 64-character hex commit ID. - A short hex prefix (e.g. ``abc1234``) — resolved by prefix scan. Security model -------------- - All reviewer names are validated before storage: control characters, ANSI escapes, and names longer than 200 characters are rejected. - All user-controlled values are sanitized via ``sanitize_display()`` before appearing in human-readable terminal output. - All error messages go to **stderr**; **stdout** carries only data. - Commit references are resolved through the safe-prefix scan in ``resolve_commit_ref`` — glob metacharacters cannot escape the scan. Agent UX -------- Pass ``--json`` to receive a machine-readable object on stdout. Usage:: muse annotate # show HEAD annotations muse annotate abc1234 # show commit annotations muse annotate abc1234 --reviewed-by agent-x muse annotate abc1234 --reviewed-by alice --reviewed-by bob muse annotate abc1234 --reviewed-by 'alice,claude-v4' muse annotate abc1234 --remove-reviewer agent-old muse annotate abc1234 --test-run muse annotate abc1234 --dry-run --reviewed-by agent-x muse annotate abc1234 --json JSON schema (``--json``):: { "commit_id": "", "message": "", "branch": "", "author": "", "committed_at": "", "reviewed_by": ["alice", "bob"], "test_runs": 3, "changed": true, "dry_run": false } Exit codes ---------- - 0 — success (show or mutation applied) - 1 — user error (bad reviewer name, commit not found, invalid args) - 2 — not inside a Muse repository """ from __future__ import annotations import argparse import json import logging import sys from typing import TYPE_CHECKING, TypedDict from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from muse.core.store import ( CommitRecord, overwrite_commit, read_current_branch, resolve_commit_ref, ) from muse.core.validation import sanitize_display, sanitize_provenance if TYPE_CHECKING: import pathlib logger = logging.getLogger(__name__) # Maximum length for a reviewer name — prevents outsized stored values. _MAX_REVIEWER_LEN: int = 200 # --------------------------------------------------------------------------- # JSON TypedDict — stable machine-readable output schema # --------------------------------------------------------------------------- class _AnnotateJson(TypedDict): """JSON output of ``muse annotate [--json]``.""" commit_id: str message: str branch: str author: str committed_at: str reviewed_by: list[str] test_runs: int changed: bool dry_run: bool # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _validate_reviewer(name: str) -> str: """Return *name* unchanged if it is an acceptable reviewer identifier. Raises: SystemExit(USER_ERROR): if *name* contains control characters or exceeds ``_MAX_REVIEWER_LEN`` characters. """ if not name: print("❌ reviewer name must not be empty.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR.value) if len(name) > _MAX_REVIEWER_LEN: print( f"❌ reviewer name too long ({len(name)} chars, max {_MAX_REVIEWER_LEN}).", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR.value) # Use sanitize_provenance to detect — if sanitised form differs, it # contained control characters. sanitised = sanitize_provenance(name) if sanitised != name: print( f"❌ reviewer name contains control characters: {name!r}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR.value) return name def _parse_reviewer_list(raw_values: list[str]) -> list[str]: """Expand comma-separated reviewer lists and validate each name. Accepts both ``--reviewed-by alice --reviewed-by bob`` (multiple flags, each value in *raw_values*) and ``--reviewed-by 'alice,bob'`` (a single comma-separated value). Whitespace is stripped from each name. Returns: Deduplicated, validated list of reviewer names (insertion order). """ seen: set[str] = set() result: list[str] = [] for raw in raw_values: for part in raw.split(","): name = part.strip() if not name: continue _validate_reviewer(name) if name not in seen: seen.add(name) result.append(name) return result def _commit_to_json( record: CommitRecord, *, changed: bool, dry_run: bool, ) -> _AnnotateJson: """Build the JSON payload from a ``CommitRecord``.""" return _AnnotateJson( commit_id=record.commit_id, message=record.message, branch=record.branch, author=record.author, committed_at=record.committed_at.isoformat(), reviewed_by=list(record.reviewed_by), test_runs=record.test_runs, changed=changed, dry_run=dry_run, ) # --------------------------------------------------------------------------- # Command registration # --------------------------------------------------------------------------- def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the ``annotate`` subcommand. Every output path accepts ``--json`` for machine-readable stdout. All diagnostic messages go to stderr. """ parser = subparsers.add_parser( "annotate", help="Attach CRDT-backed annotations to an existing commit.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "commit_arg", nargs="?", default=None, metavar="COMMIT", help=( "Commit to annotate: full SHA, short prefix, HEAD~N, or omit for HEAD." ), ) parser.add_argument( "--reviewed-by", action="append", dest="reviewed_by", default=None, metavar="NAME[,NAME…]", help=( "Add a reviewer (ORSet semantics — once added, never lost). " "Accepts comma-separated names or multiple flags." ), ) parser.add_argument( "--remove-reviewer", action="append", dest="remove_reviewer", default=None, metavar="NAME[,NAME…]", help=( "Remove a reviewer from the stored list. " "Comma-separated or multiple flags." ), ) parser.add_argument( "--test-run", action="store_true", dest="test_run", help="Increment the GCounter test-run count for this commit.", ) parser.add_argument( "--dry-run", action="store_true", dest="dry_run", help="Show what would change without writing to disk.", ) parser.add_argument( "--json", action="store_true", dest="output_json", default=False, help="Emit a JSON object to stdout.", ) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Command handler # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Attach CRDT-backed annotations to an existing commit. When called with no mutation flags, displays the current annotations for the specified commit (or HEAD). Mutation flags modify the stored record: ``--reviewed-by NAME`` ORSet union — merges new names into the existing reviewer set. A name once added is never removed by concurrent additions. ``--remove-reviewer NAME`` Removes a name from the stored reviewer list. Not CRDT-safe for concurrent removals, but useful for correcting mistakes. ``--test-run`` GCounter increment — monotonically increases ``test_runs`` by 1. ``--dry-run`` Resolves the commit and computes the new annotation state without writing anything to disk. Useful for previewing changes. """ commit_arg: str | None = args.commit_arg reviewed_by_raw: list[str] | None = args.reviewed_by remove_reviewer_raw: list[str] | None = args.remove_reviewer test_run: bool = args.test_run dry_run: bool = args.dry_run output_json: bool = args.output_json root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) # Resolve the commit reference (supports short IDs, HEAD~N, HEAD). record = resolve_commit_ref(root, repo_id, branch, commit_arg) if record is None: ref_display = sanitize_display(commit_arg or "HEAD") print(f"❌ commit {ref_display!r} not found.", file=sys.stderr) raise SystemExit(ExitCode.NOT_FOUND.value) # Parse and validate reviewer arguments before any mutation. add_reviewers: list[str] = _parse_reviewer_list(reviewed_by_raw or []) remove_reviewers: list[str] = _parse_reviewer_list(remove_reviewer_raw or []) is_show_mode = not add_reviewers and not remove_reviewers and not test_run if is_show_mode: _show(record, output_json=output_json) return # Apply mutations to a copy of the record fields. current_set: set[str] = set(record.reviewed_by) added: list[str] = [] removed: list[str] = [] if add_reviewers: for name in add_reviewers: if name not in current_set: current_set.add(name) added.append(name) if remove_reviewers: for name in remove_reviewers: if name in current_set: current_set.discard(name) removed.append(name) new_reviewed_by: list[str] = sorted(current_set) new_test_runs: int = record.test_runs + (1 if test_run else 0) changed = ( new_reviewed_by != sorted(record.reviewed_by) or new_test_runs != record.test_runs ) if not dry_run and changed: record.reviewed_by = new_reviewed_by record.test_runs = new_test_runs overwrite_commit(root, record) if output_json: # If dry_run: report the *prospective* state (not yet written). payload = _AnnotateJson( commit_id=record.commit_id, message=record.message, branch=record.branch, author=record.author, committed_at=record.committed_at.isoformat(), reviewed_by=new_reviewed_by, test_runs=new_test_runs, changed=changed, dry_run=dry_run, ) print(json.dumps(payload, indent=2)) return # Human-readable output. cid = sanitize_display(record.commit_id[:8]) prefix = "[dry-run] " if dry_run else "" for name in added: print(f"✅ {prefix}Added reviewer: {sanitize_display(name)}") for name in removed: print(f"✅ {prefix}Removed reviewer: {sanitize_display(name)}") # Warn about names that were requested for removal but weren't present. removed_set: set[str] = set(removed) for name in remove_reviewers: if name not in removed_set: print( f"⚠️ reviewer {sanitize_display(name)!r} was not present.", file=sys.stderr, ) if test_run: print(f"✅ {prefix}Test run recorded (total: {new_test_runs})") if changed: print(f"[{cid}] annotation {'would be ' if dry_run else ''}updated") else: print(f"[{cid}] no changes (annotations already up to date)") def _show(record: CommitRecord, *, output_json: bool) -> None: """Display the current annotations for *record*.""" if output_json: payload = _commit_to_json(record, changed=False, dry_run=False) print(json.dumps(payload, indent=2)) return cid = sanitize_display(record.commit_id[:8]) print(f"ℹ️ commit {cid}") if record.reviewed_by: reviewers = ", ".join(sanitize_display(r) for r in sorted(record.reviewed_by)) print(f" reviewed-by: {reviewers}") else: print(" reviewed-by: (none)") print(f" test-runs: {record.test_runs}")