"""``muse tag`` — attach and query semantic tags on commits. Tags are arbitrary ``namespace:value`` strings (convention, not enforced) that annotate commits with machine-readable metadata. All tag operations are idempotent by default: adding the same tag twice to the same commit is a no-op (use ``--allow-duplicate`` to bypass). Usage:: muse tag add [] — tag a commit (HEAD if omitted) muse tag add [] --dry-run — preview without writing muse tag list [] — list tags (all or per-commit) muse tag list --match "emotion:*" — filter by glob pattern muse tag list --sort created — sort by creation time muse tag remove [] — remove matching tags from a commit Tag conventions (not enforced):: emotion:* — emotional character (emotion:melancholic, emotion:tense) section:* — song section (section:verse, section:chorus) stage:* — production stage (stage:rough-mix, stage:master) key:* — musical key (key:Am, key:Eb) tempo:* — tempo annotation (tempo:120bpm) ref:* — reference track (ref:beatles) JSON output (``--format json`` / ``--json``) schema for ``tag add``:: { "status": "tagged | already_tagged | dry_run", "tag_id": " | null", "commit_id": "", "tag": "", "namespace": "", "created_at": " | null", "dry_run": false } Exit codes:: 0 — success 1 — commit not found, tag not found, invalid tag name, invalid format 2 — not inside a Muse repository """ from __future__ import annotations import argparse import datetime import fnmatch import json import logging import re import sys import uuid from typing import TypedDict from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from muse.core.store import ( TagRecord, delete_tag, get_all_tags, get_tags_for_commit, read_current_branch, resolve_commit_ref, write_tag, ) from muse.core.validation import sanitize_display logger = logging.getLogger(__name__) # Maximum length for a tag string (prevents runaway storage). _MAX_TAG_LEN: int = 256 # Reject control characters (C0 + DEL) to prevent ANSI injection when tags # are embedded in terminal output or stored on disk. _TAG_FORBIDDEN_RE: re.Pattern[str] = re.compile(r"[\x00-\x1f\x7f]") class _TagJson(TypedDict): """Stable JSON schema for a single tag entry (list output).""" tag_id: str commit_id: str tag: str namespace: str | None created_at: str def _tag_namespace(tag: str) -> str | None: """Return the namespace prefix (part before ':'), or None if absent.""" return tag.split(":", 1)[0] if ":" in tag else None def _validate_tag_name(name: str) -> str: """Return *name* unchanged if safe; raise ``ValueError`` otherwise. Guards: - Not empty. - Max length ``_MAX_TAG_LEN`` (256 chars). - No C0 control characters or DEL (prevents ANSI injection in terminal output and on-disk storage). """ if not name: raise ValueError("Tag name must not be empty.") if len(name) > _MAX_TAG_LEN: raise ValueError( f"Tag name too long ({len(name)} chars); maximum is {_MAX_TAG_LEN}." ) if _TAG_FORBIDDEN_RE.search(name): raise ValueError( "Tag name contains forbidden control characters " "(use printable ASCII/Unicode only)." ) return name def _tag_to_json(t: TagRecord) -> _TagJson: """Convert a TagRecord to the stable JSON wire format.""" return _TagJson( tag_id=t.tag_id, commit_id=t.commit_id, tag=t.tag, namespace=_tag_namespace(t.tag), created_at=t.created_at.isoformat(), ) def _sort_tags(tags: list[TagRecord], sort_key: str) -> list[TagRecord]: """Sort *tags* by *sort_key*: ``tag`` (default), ``created``, ``commit``.""" if sort_key == "created": return sorted(tags, key=lambda t: t.created_at) if sort_key == "commit": return sorted(tags, key=lambda t: t.commit_id) return sorted(tags, key=lambda t: (t.tag, t.commit_id)) def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``muse tag`` subcommand tree.""" parser = subparsers.add_parser( "tag", help="Attach and query semantic tags on commits.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) subs = parser.add_subparsers(dest="subcommand", metavar="SUBCOMMAND") subs.required = True # -- add ------------------------------------------------------------------ add_p = subs.add_parser( "add", help="Attach a tag to a commit.", description=( "Attach a tag string to a commit (HEAD by default).\n\n" "Tags follow a 'namespace:value' convention (not enforced):\n" " emotion:joyful section:chorus stage:master\n" " key:Am tempo:120bpm ref:beatles\n\n" "Tag name rules:\n" f" - Maximum {_MAX_TAG_LEN} characters\n" " - No control characters (C0/DEL) — printable ASCII/Unicode only\n\n" "Duplicate guard (default): adding the same tag twice to the same\n" "commit is a no-op — the existing tag_id is returned in JSON output\n" "so agent workflows remain idempotent. Use --allow-duplicate to bypass.\n\n" "Agent quickstart:\n" " muse tag add emotion:joyful\n" " muse tag add emotion:joyful --json\n" " muse tag add emotion:joyful --json -j # same, short flag\n" " muse tag add emotion:joyful \n" " muse tag add emotion:joyful --dry-run --json\n\n" "JSON schema:\n" " {\"status\": \"tagged|already_tagged|dry_run\",\n" " \"tag_id\": \"|null\", \"commit_id\": \"\",\n" " \"tag\": \"\", \"namespace\": \"|null\",\n" " \"created_at\": \"|null\", \"dry_run\": false}\n\n" "Exit codes:\n" " 0 Tagged successfully (or already_tagged, or dry_run)\n" " 1 Invalid tag name, commit not found, or invalid --format\n" " 2 Not inside a Muse repository" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) add_p.add_argument("tag_name", help="Tag string (e.g. emotion:joyful).") add_p.add_argument( "ref", nargs="?", default=None, help="Commit ID or branch name (default: HEAD).", ) add_p.add_argument( "--allow-duplicate", action="store_true", help="Allow adding the same tag twice to the same commit.", ) add_p.add_argument( "-n", "--dry-run", action="store_true", help="Validate and preview without writing any data.", ) add_p.add_argument("--format", "-f", default="text", dest="fmt", help="Output format: text or json.") add_p.add_argument("--json", "-j", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.") add_p.set_defaults(func=run_add) # -- list ----------------------------------------------------------------- list_p = subs.add_parser( "list", help="List tags.", description=( "List tags in the repository, optionally filtered by commit or glob pattern.\n\n" "With no arguments, lists all tags across all commits.\n" "Pass a commit ID or branch name to list only tags on that commit.\n\n" "Glob pattern examples (--match / -m):\n" " emotion:* — all emotion-namespace tags\n" " *:joyful — any namespace, value 'joyful'\n" " section:* — all section tags\n" " * — all tags (same as no --match)\n\n" "Sort options (--sort):\n" " tag (default) — alphabetical by tag string\n" " created — chronological by creation time\n" " commit — lexicographic by commit SHA\n\n" "Agent quickstart:\n" " muse tag list\n" " muse tag list --json\n" " muse tag list -j # same, short flag\n" " muse tag list --match 'emotion:*' --json\n" " muse tag list --sort created --json\n" " muse tag list HEAD --json\n" " muse tag list --json | jq '.tags[] | .tag'\n\n" "JSON schema:\n" " {\"total\": , \"tags\": [\n" " {\"tag_id\": \"\", \"commit_id\": \"\",\n" " \"tag\": \"\", \"namespace\": \"|null\",\n" " \"created_at\": \"\"}, ...]}\n\n" "Exit codes:\n" " 0 Always (empty list is a valid result)\n" " 1 Invalid --format, or commit ref not found\n" " 2 Not inside a Muse repository" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) list_p.add_argument( "ref", nargs="?", default=None, help="Commit ID or branch name to list tags for (default: all commits).", ) list_p.add_argument( "--match", "-m", default=None, dest="match", help="Filter by glob pattern (e.g. 'emotion:*', '*:joyful').", ) list_p.add_argument( "--sort", choices=["tag", "created", "commit"], default="tag", help="Sort order: tag (default), created, commit.", ) list_p.add_argument("--format", "-f", default="text", dest="fmt", help="Output format: text or json.") list_p.add_argument("--json", "-j", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.") list_p.set_defaults(func=run_list) # -- remove --------------------------------------------------------------- remove_p = subs.add_parser( "remove", help="Remove a tag from a commit.", description=( "Remove all matching tags with the given name from a commit.\n\n" "Removal is idempotent — if the tag does not exist on the commit,\n" "exit code is still 0. Check 'removed_count' in JSON output to\n" "determine whether anything actually changed.\n\n" "When --allow-duplicate was used to add the same tag multiple times,\n" "this command removes ALL copies in a single call.\n\n" "Tag name rules (same as tag add):\n" f" - Maximum {_MAX_TAG_LEN} characters\n" " - No control characters (C0/DEL) — printable ASCII/Unicode only\n\n" "Agent quickstart:\n" " muse tag remove emotion:joyful\n" " muse tag remove emotion:joyful --json\n" " muse tag remove emotion:joyful -j # same, short flag\n" " muse tag remove emotion:joyful \n" " muse tag remove emotion:joyful --json | jq '.removed_count'\n\n" "JSON schema:\n" " {\"status\": \"removed|not_found\", \"removed_count\": ,\n" " \"tag_ids\": [\"\", ...], \"commit_id\": \"\",\n" " \"tag\": \"\"}\n\n" "Exit codes:\n" " 0 Removed (or not found — idempotent)\n" " 1 Invalid tag name, commit not found, or invalid --format\n" " 2 Not inside a Muse repository" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) remove_p.add_argument("tag_name", help="Tag string to remove (e.g. emotion:joyful).") remove_p.add_argument( "ref", nargs="?", default=None, help="Commit ID or branch name (default: HEAD).", ) remove_p.add_argument("--format", "-f", default="text", dest="fmt", help="Output format: text or json.") remove_p.add_argument("--json", "-j", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.") remove_p.set_defaults(func=run_remove) def run_add(args: argparse.Namespace) -> None: """Attach a tag to a commit. By default, adding the same tag to the same commit twice is a no-op (idempotent). Use ``--allow-duplicate`` to bypass the deduplication guard. Use ``--dry-run`` to validate the commit reference and preview the result without writing any tag to disk. Exits 0 on success, 1 on validation error. All error messages go to **stderr**; stdout is reserved for structured output. JSON schema:: { "status": "tagged | already_tagged | dry_run", "tag_id": " | null", "commit_id": "", "tag": "", "namespace": " | null", "created_at": " | null", "dry_run": false } Exit codes:: 0 — success (tagged, already_tagged, or dry_run) 1 — invalid tag name, commit not found, invalid format 2 — not inside a Muse repository """ tag_name: str = args.tag_name ref: str | None = args.ref fmt: str = args.fmt dry_run: bool = args.dry_run allow_duplicate: bool = args.allow_duplicate if fmt not in ("text", "json"): print(f"❌ Unknown --format '{sanitize_display(fmt)}'. Choose text or json.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) try: _validate_tag_name(tag_name) except ValueError as exc: print(f"❌ Invalid tag name: {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) commit = resolve_commit_ref(root, repo_id, branch, ref) if commit is None: print( f"❌ Commit '{sanitize_display(str(ref))}' not found.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) namespace = _tag_namespace(tag_name) # Dry-run: validate, then exit without writing. if dry_run: if fmt == "json": print(json.dumps({ "status": "dry_run", "tag_id": None, "commit_id": commit.commit_id, "tag": tag_name, "namespace": namespace, "created_at": None, "dry_run": True, })) else: print( f"Would tag {commit.commit_id[:8]} with '{sanitize_display(tag_name)}'" " (dry run — nothing written)" ) return # Duplicate guard: check if this exact tag already exists on the commit. if not allow_duplicate: existing = get_tags_for_commit(root, repo_id, commit.commit_id) if any(t.tag == tag_name for t in existing): if fmt == "json": # Surface the existing tag_id for idempotent agent workflows. existing_tag = next(t for t in existing if t.tag == tag_name) print(json.dumps({ "status": "already_tagged", "tag_id": existing_tag.tag_id, "commit_id": commit.commit_id, "tag": tag_name, "namespace": namespace, "created_at": existing_tag.created_at.isoformat(), "dry_run": False, })) else: print( f"Tag '{sanitize_display(tag_name)}' already on " f"{commit.commit_id[:8]} — skipped." ) return tag_id = str(uuid.uuid4()) created_at = datetime.datetime.now(datetime.timezone.utc) write_tag(root, TagRecord( tag_id=tag_id, repo_id=repo_id, commit_id=commit.commit_id, tag=tag_name, created_at=created_at, )) if fmt == "json": print(json.dumps({ "status": "tagged", "tag_id": tag_id, "commit_id": commit.commit_id, "tag": tag_name, "namespace": namespace, "created_at": created_at.isoformat(), "dry_run": False, })) else: print(f"Tagged {commit.commit_id[:8]} with '{sanitize_display(tag_name)}'") def run_list(args: argparse.Namespace) -> None: """List tags, optionally filtered by commit or glob pattern. When ``ref`` is omitted, lists all tags in the repository. Use ``--match`` to filter by glob pattern (e.g. ``'emotion:*'``). Use ``--sort`` to control ordering (tag, created, commit). All error messages go to **stderr**; stdout is reserved for structured output. JSON schema:: { "total": , "tags": [ { "tag_id": "", "commit_id": "", "tag": "", "namespace": " | null", "created_at": "" }, ... ] } Exit codes:: 0 — always (empty list is a valid result) 1 — invalid format, commit not found 2 — not inside a Muse repository """ ref: str | None = args.ref fmt: str = args.fmt match_pattern: str | None = args.match sort_key: str = args.sort if fmt not in ("text", "json"): print(f"❌ Unknown --format '{sanitize_display(fmt)}'. Choose text or json.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) if ref: commit = resolve_commit_ref(root, repo_id, branch, ref) if commit is None: print( f"❌ Commit '{sanitize_display(str(ref))}' not found.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) tags = get_tags_for_commit(root, repo_id, commit.commit_id) else: tags = get_all_tags(root, repo_id) if match_pattern: tags = [t for t in tags if fnmatch.fnmatch(t.tag, match_pattern)] tags = _sort_tags(tags, sort_key) if fmt == "json": print(json.dumps({ "total": len(tags), "tags": [_tag_to_json(t) for t in tags], })) return if not tags: print("No tags.") return for t in tags: print(f"{t.commit_id[:8]} {sanitize_display(t.tag)}") def run_remove(args: argparse.Namespace) -> None: """Remove all matching tags from a commit. Validates the tag name format before any I/O — same rules as :func:`run_add` — so invalid names produce a clear error rather than silently returning ``not_found``. Finds all tags with the exact name on the given commit and deletes them. Exits 0 even if no tags were found (idempotent removal). Use the ``removed_count`` field in JSON output to detect whether anything changed. When ``--allow-duplicate`` was used during add, all copies of the tag on that commit are removed in a single call. All error messages go to **stderr**; stdout is reserved for structured output. JSON schema:: { "status": "removed | not_found", "removed_count": , "tag_ids": ["", ...], "commit_id": "", "tag": "" } Exit codes:: 0 — success (removed or not_found — removal is idempotent) 1 — invalid tag name, commit not found, invalid format 2 — not inside a Muse repository """ tag_name: str = args.tag_name ref: str | None = args.ref fmt: str = args.fmt if fmt not in ("text", "json"): print(f"❌ Unknown --format '{sanitize_display(fmt)}'. Choose text or json.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) try: _validate_tag_name(tag_name) except ValueError as exc: print(f"❌ Invalid tag name: {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) commit = resolve_commit_ref(root, repo_id, branch, ref) if commit is None: print( f"❌ Commit '{sanitize_display(str(ref))}' not found.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) tags = get_tags_for_commit(root, repo_id, commit.commit_id) matching = [t for t in tags if t.tag == tag_name] if not matching: if fmt == "json": print(json.dumps({ "status": "not_found", "removed_count": 0, "tag_ids": [], "commit_id": commit.commit_id, "tag": tag_name, })) else: print( f"Tag '{sanitize_display(tag_name)}' not found on " f"{commit.commit_id[:8]} — nothing removed." ) return removed_ids: list[str] = [] for t in matching: delete_tag(root, repo_id, t.tag_id) removed_ids.append(t.tag_id) if fmt == "json": print(json.dumps({ "status": "removed", "removed_count": len(matching), "tag_ids": removed_ids, "commit_id": commit.commit_id, "tag": tag_name, })) else: print( f"Removed {len(matching)} tag(s) '{sanitize_display(tag_name)}' " f"from {commit.commit_id[:8]}." )