"""muse hub — MuseHub fabric connection management. The hub is not just a remote. It is the shared fabric where versioned multidimensional state flows across agents, humans, and repositories. Connecting a repo to a hub anchors it to the synchronisation layer that enables push/pull, plugin discovery, and multi-agent coordination. Separation of concerns ----------------------- - ``muse remote`` manages generic push/pull endpoints (any Muse server). - ``muse hub`` manages the *primary identity fabric* — the one hub this repo belongs to for authentication, discovery, and coordination. A repo has at most **one** hub. It may have many remotes. Security model -------------- - ``_normalise_url`` enforces HTTPS for non-loopback hosts at connect time. - ``_hub_api`` validates the URL scheme before every network request, preventing SSRF from a tampered ``config.toml``. - ``_hub_api`` caps the response body at ``_MAX_API_RESPONSE_BYTES`` to prevent OOM from a hostile hub. - All user-controlled values (proposal titles, branch names, hub URLs) are passed through ``sanitize_display()`` before being printed. - All diagnostic messages go to **stderr**; **stdout** is reserved for JSON and machine-readable data. Subcommands ----------- :: muse hub connect Attach this repo to a MuseHub instance. muse hub status [--json] Show connection and identity information. muse hub disconnect [--json] Remove the hub association from this repo. muse hub ping [--json] Test HTTP connectivity to the hub. muse hub proposal list [--json] List proposals on MuseHub. muse hub proposal read [--json] Show a single proposal. muse hub proposal create [--json] Open a new proposal. muse hub proposal merge [--json] Merge a proposal. JSON schemas ------------ ``muse hub connect --json``:: {"status": "ok", "hub_url": "", "hostname": "", "authenticated": true|false, "identity_name": "", "identity_type": ""} ``muse hub status --json``:: {"hub_url": "", "hostname": "", "authenticated": true|false, "identity_type": "", "identity_name": "", "identity_id": ""} ``muse hub disconnect --json``:: {"status": "ok"|"nothing_to_do", "hostname": ""} ``muse hub ping --json``:: {"status": "ok"|"error", "hub_url": "", "hostname": "", "reachable": true|false, "message": ""} Agent workflow examples ----------------------- :: # Verify hub auth before starting work muse hub status --json | python3 -c "import json,sys; d=json.load(sys.stdin); sys.exit(0 if d['authenticated'] else 1)" # Create a proposal and capture the ID muse hub proposal create --title "feat: x" --json | python3 -c "import json,sys; print(json.load(sys.stdin)['proposalId'])" # List open proposals (machine-readable) muse hub proposal list --json # Merge with squash strategy muse hub proposal merge af54753d --strategy squash --json """ from __future__ import annotations import argparse import http.client import json import logging import sys import textwrap import urllib.error import urllib.parse import urllib.request from collections.abc import Mapping from typing import IO, TypedDict from muse.cli.config import ( clear_hub_url, get_hub_url, get_remote, list_remotes, set_hub_url, ) from muse.core.errors import ExitCode from muse.core.identity import IdentityEntry, load_identity from muse.core.repo import find_repo_root from muse.core.store import read_current_branch from muse.core._types import Metadata from muse.core.validation import clamp_int, sanitize_display logger = logging.getLogger(__name__) type _HubPayload = dict[str, str | bool | list[str]] type _ProposalPayload = dict[str, str | list[str]] _CONNECT_TIMEOUT = 8 # seconds for ping/status health check _MAX_API_RESPONSE_BYTES = 4 * 1024 * 1024 # 4 MiB cap on API responses # Only allow http and https — no file://, ftp://, data://, etc. _ALLOWED_API_SCHEMES = frozenset({"http", "https"}) # Maximum proposals fetched when resolving an 8-char prefix — repos with more proposals # than this will not find older ones by prefix. Intentionally conservative; # raise if real-world repos hit the limit. _PROPOSAL_PREFIX_RESOLVE_LIMIT = 200 _MAX_PROPOSAL_BODY_LINES = 20 # body lines shown in text mode before truncation hint _MAX_PROPOSAL_TITLE_LEN = 512 # client-side guard: titles longer than this are rejected _MAX_ISSUE_TITLE_LEN = 512 # client-side guard: issue titles longer than this are rejected _MAX_REPO_NAME_LEN = 255 # mirrors CreateRepoRequest.name max_length on the server _MAX_REPO_DESC_LEN = 1024 # reasonable cap on description before sending to server _MAX_ISSUE_LABEL_LEN = 255 # mirrors label name max_length on the server _MAX_ISSUE_COMMENT_LEN = 50_000 # mirrors IssueCommentCreate.body max_length on the server _MAX_LABEL_NAME_LEN = 50 # mirrors LabelCreate.name max_length on the server _MAX_LABEL_DESC_LEN = 200 # mirrors LabelCreate.description max_length on the server # ── TypedDicts ──────────────────────────────────────────────────────────────── class _ConnectJson(TypedDict): """JSON schema for ``muse hub connect --json``.""" status: str # "ok" hub_url: str hostname: str authenticated: bool identity_name: str # display name or "" identity_type: str # "human" | "agent" | "" class _StatusJson(TypedDict): """JSON schema for ``muse hub status --json``. All keys are always present so agents can access them without guard checks. """ hub_url: str hostname: str authenticated: bool identity_type: str # "" when not authenticated identity_name: str # "" when not authenticated identity_id: str # "" when not authenticated capabilities: list[str] # [] when not authenticated or not an agent class _DisconnectJson(TypedDict): """JSON schema for ``muse hub disconnect --json``.""" status: str # "ok" | "nothing_to_do" hub_url: str # full normalised URL removed, or "" when nothing was connected hostname: str # host[:port] display form, or "" when nothing was connected class _PingJson(TypedDict): """JSON schema for ``muse hub ping --json``.""" status: str # "ok" | "error" hub_url: str hostname: str reachable: bool message: str class _RepoEntry(TypedDict, total=False): """A single repo entry returned by the MuseHub search/list API.""" owner: str slug: str repoId: str class _ProposalEntry(TypedDict, total=False): """A single proposal entry returned by the MuseHub proposals API.""" proposalId: str title: str state: str # "open" | "merged" | "closed" fromBranch: str toBranch: str author: str createdAt: str class _RepoJson(TypedDict): """JSON schema for ``muse hub repo create --json``.""" repo_id: str name: str owner: str slug: str visibility: str # "public" | "private" description: str clone_url: str tags: list[str] created_at: str # ISO-8601 UTC class _HubApiResponse(TypedDict, total=False): """Generic MuseHub API response envelope (all keys optional). Different endpoints populate different subsets of these keys. Callers must validate presence before accessing. """ repo_id: str repos: list[_RepoEntry] proposals: list[_ProposalEntry] class _LabelEntry(TypedDict, total=False): """One label row returned by the MuseHub labels API.""" label_id: str repo_id: str name: str color: str description: str # ── Security — redirect refusal for ping ────────────────────────────────────── class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): """Refuse all HTTP redirects for the ping health check. No credentials travel on the ping request, but silently following redirects (potentially cross-scheme or cross-host) is misleading about what was actually reached and normalises insecure redirect behaviour. """ def redirect_request( self, req: urllib.request.Request, fp: IO[bytes], code: int, msg: str, headers: http.client.HTTPMessage, newurl: str, ) -> urllib.request.Request | None: raise urllib.error.HTTPError( req.full_url, code, f"Redirect refused ({code}): hub redirected to {newurl!r}. Update the hub URL.", headers, fp, ) _PING_OPENER = urllib.request.build_opener(_NoRedirectHandler()) # ── URL / hostname helpers ──────────────────────────────────────────────────── def _normalise_url(url: str) -> str: """Normalise *url* to an https:// URL (or http:// for loopback addresses). Adds ``https://`` when no scheme is present. Raises ``ValueError`` when an explicit ``http://`` scheme is given for a non-loopback host — sending sending credentials over cleartext HTTP to a remote server is never acceptable. Exception: ``http://localhost``, ``http://127.0.0.1``, and ``http://[::1]`` (with or without a port) are allowed because loopback traffic never leaves the machine and cannot be intercepted in transit. Args: url: Raw user-supplied URL. Returns: Normalised URL without a trailing slash. Raises: ValueError: If the URL uses ``http://`` for a non-loopback host, or uses a disallowed scheme (``file://``, ``ftp://``, etc.). """ stripped = url.strip().rstrip("/") # Check for an explicit URL scheme BEFORE adding https://, catching # file:///etc/passwd, ftp://, data:text/plain, javascript:, etc. # # We can't unconditionally rely on urlparse's scheme detection because # bare host:port inputs like "musehub.ai:8443" are misidentified: # urlparse("musehub.ai:8443").scheme == "musehub.ai". # # Heuristic: if the URL contains "://" OR if the part after the first ":" # is NOT a pure port number (digits), treat the prefix as a scheme. # Examples: # "musehub.ai:8443" → after ":" is "8443" (digits) → host:port, skip # "localhost:8080" → after ":" is "8080" (digits) → host:port, skip # "data:text/plain,x" → after ":" is "text/plain,x" → scheme, check # "javascript:alert()" → after ":" is "alert()" → scheme, check # "https://musehub.ai" → "://" present → scheme, check colon_idx = stripped.find(":") if colon_idx > 0: after_colon = stripped[colon_idx + 1:].lstrip("/") is_port = after_colon.isdigit() # bare port: "8443", "10003", etc. if not is_port: pre_scheme = urllib.parse.urlparse(stripped).scheme.lower() if pre_scheme and pre_scheme not in _ALLOWED_API_SCHEMES: raise ValueError( f"URL scheme '{pre_scheme}' is not allowed. " "Use https:// (or http:// for localhost)." ) if not stripped.startswith(("http://", "https://")): stripped = f"https://{stripped}" if stripped.startswith("http://"): # Use urlparse to correctly extract the hostname — handles IPv6 # bracket notation (e.g. http://[::1]:8080) where manual split(":")[0] # would yield "[" instead of "::1". _loopback = {"localhost", "127.0.0.1", "::1"} parsed_host = urllib.parse.urlparse(stripped).hostname or "" if parsed_host not in _loopback: host = stripped[len("http://"):] raise ValueError( f"Insecure URL rejected: {stripped!r}\n" f"MuseHub requires HTTPS for non-loopback hosts. Did you mean: https://{host}" ) return stripped def _hub_hostname(url: str) -> str: """Extract the display hostname (host:port) from a hub URL.""" stripped = url.strip().rstrip("/") if "://" in stripped: stripped = stripped.split("://", 1)[1] return stripped.split("/")[0] def _ping_hub(url: str) -> tuple[bool, str]: """Attempt an HTTP GET to ``/health``. Returns a ``(reachable, message)`` tuple. Never raises — all errors are captured and surfaced as human-readable strings. Security note ------------- Only ``http`` and ``https`` schemes are attempted. Any other scheme (``file://``, ``ftp://``, etc.) returns ``(False, "scheme not allowed")`` without opening a socket — prevents callers from accidentally probing the local filesystem via a ``--hub`` override. Exception coverage ------------------ - ``urllib.error.HTTPError`` — non-2xx from a reachable server - ``urllib.error.URLError`` — DNS failure, connection refused - ``TimeoutError`` — connect/read timeout - ``OSError`` — network-layer errors (SSL, connection reset, etc.) - ``http.client.HTTPException`` — malformed HTTP response (``BadStatusLine``, ``InvalidURL``) from a misbehaving server; NOT a subclass of OSError so must be caught separately """ # Scheme guard — never probe non-HTTP(S) targets. scheme = urllib.parse.urlparse(url).scheme.lower() if scheme not in _ALLOWED_API_SCHEMES: return False, f"URL scheme '{scheme}' is not allowed (use http or https)" health_url = f"{url.rstrip('/')}/health" try: req = urllib.request.Request(health_url, method="GET") with _PING_OPENER.open(req, timeout=_CONNECT_TIMEOUT) as resp: status = resp.status if 200 <= status < 300: return True, f"HTTP {status} OK" return False, f"HTTP {status}" except urllib.error.HTTPError as exc: return False, f"HTTP {exc.code} {exc.reason}" except urllib.error.URLError as exc: return False, str(exc.reason) except TimeoutError: return False, "timed out" except OSError as exc: return False, str(exc) except http.client.HTTPException as exc: # Malformed HTTP response (BadStatusLine, InvalidURL, etc.). # Not an OSError subclass — must be caught explicitly. return False, f"malformed response: {type(exc).__name__}" # ── MuseHub HTTP helper ─────────────────────────────────────────────────────── def _hub_api( hub_url: str, identity: IdentityEntry, method: str, path: str, body: Mapping[str, str | bool | list[str] | None] | None = None, timeout: float = 10.0, ) -> _HubApiResponse: """Make an authenticated JSON request to the MuseHub API. Security: - Validates the hub URL scheme (``http``/``https`` only) before opening any socket — prevents SSRF if ``config.toml`` is tampered with. - Caps the response body at ``_MAX_API_RESPONSE_BYTES`` to prevent OOM from a hostile or misbehaving hub. - Sanitizes error detail from HTTP responses before printing. Args: hub_url: Repository-level hub URL (e.g. ``http://localhost:10003/gabriel/muse``). identity: Identity entry from ``~/.muse/identity.toml``. method: HTTP method (``GET``, ``POST``, ``DELETE``). path: API path appended to the server root (e.g. ``/api/v1/repos/{id}/proposals``). body: Optional JSON body for POST/PUT requests. timeout: Connect+read timeout in seconds. Returns: Parsed JSON response body as a dict. Raises: SystemExit: On invalid URL scheme, missing token, network errors, or non-2xx responses. """ parsed = urllib.parse.urlparse(hub_url) scheme = parsed.scheme.lower() if scheme not in _ALLOWED_API_SCHEMES: print( f"❌ Hub URL scheme '{sanitize_display(scheme)}' is not allowed. " "Use http or https.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) server_root = f"{parsed.scheme}://{parsed.netloc}" url = f"{server_root}{path}" # Resolve signing identity — try the passed identity first, then fall back to config. # This fixes the bug where the identity parameter was silently ignored. from muse.core.transport import SigningIdentity signing: SigningIdentity | None = None _handle = identity.get("handle", "") _key_path_str = identity.get("key_path", "") if _handle and _key_path_str: import pathlib as _pl _key_path = _pl.Path(_key_path_str) if _key_path.is_file(): try: from cryptography.hazmat.primitives.serialization import load_pem_private_key _pem = _key_path.read_bytes() _private_key = load_pem_private_key(_pem, password=None) signing = SigningIdentity(handle=_handle, private_key=_private_key) except Exception: pass if signing is None: from muse.cli.config import get_signing_identity signing = get_signing_identity(remote_url=hub_url) if not signing: print( "❌ No signing identity for this hub. Run:\n" " muse auth keygen --hub \n" " muse auth register --hub --handle ", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) from muse.core.transport import build_msign_header data: bytes | None = None body_bytes: bytes | None = None if body is not None: body_bytes = json.dumps(body).encode() data = body_bytes headers: Metadata = { "Authorization": build_msign_header(signing, method, url, body_bytes), "Accept": "application/json", } if body_bytes is not None: headers["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=timeout) as resp: # noqa: S310 raw = resp.read(_MAX_API_RESPONSE_BYTES + 1) if len(raw) > _MAX_API_RESPONSE_BYTES: print( f"❌ Response from {sanitize_display(url)} exceeds size limit.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) text = raw.decode("utf-8", errors="replace") parsed = json.loads(text) if text.strip() else {} return parsed if isinstance(parsed, dict) else {} except urllib.error.HTTPError as exc: raw_body = exc.read().decode(errors="replace") try: detail = json.loads(raw_body).get("detail", raw_body) except (json.JSONDecodeError, AttributeError): detail = raw_body[:200] print( f"❌ MuseHub API error {exc.code}: {sanitize_display(str(detail))}", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) from exc except urllib.error.URLError as exc: print( f"❌ Cannot reach MuseHub: {sanitize_display(str(exc.reason))}", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) from exc # ── Repo / identity resolution helpers ─────────────────────────────────────── def _enrich_hub_url_from_remote(hub_base_url: str) -> str: """Append owner/slug to a bare hub URL by cross-referencing configured remotes. When the hub URL stored by ``muse hub connect`` is a bare base URL (e.g. ``http://localhost:10003``) with no owner/slug path component, this function searches the configured remotes for one whose netloc matches the hub netloc and extracts the owner/slug from its path. Returns the enriched URL on a match; returns *hub_base_url* unchanged when no matching remote is found or when the caller is not inside a repo. """ base_parsed = urllib.parse.urlparse(hub_base_url) if len(base_parsed.path.strip("/").split("/")) >= 2 and base_parsed.path.strip("/"): # Already has owner/slug — nothing to do. return hub_base_url root = find_repo_root() if root is None: return hub_base_url for remote in list_remotes(root): r = urllib.parse.urlparse(remote["url"]) if r.netloc == base_parsed.netloc: parts = r.path.strip("/").split("/") if len(parts) >= 2 and parts[0]: return f"{base_parsed.scheme}://{base_parsed.netloc}/{parts[0]}/{parts[1]}" return hub_base_url def _resolve_repo_id(hub_url: str, identity: IdentityEntry) -> str: """Look up the MuseHub repo UUID from the hub URL's owner/slug path. The hub URL is ``http://host/owner/slug``. This calls ``GET /{owner}/{slug}/refs`` to obtain the UUID, falling back to the search endpoint if the refs response does not include ``repo_id``. When the hub URL is a bare base URL (no owner/slug), the function first attempts to enrich it from the configured remotes via :func:`_enrich_hub_url_from_remote`. Raises: SystemExit: If the URL has no owner/slug or the repo ID cannot be resolved. """ hub_url = _enrich_hub_url_from_remote(hub_url) parsed = urllib.parse.urlparse(hub_url) parts = parsed.path.strip("/").split("/") if len(parts) < 2: print( f"❌ Hub URL '{sanitize_display(hub_url)}' does not contain " "owner/slug — cannot resolve repo ID.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) owner, slug = parts[0], parts[1] data = _hub_api(hub_url, identity, "GET", f"/{owner}/{slug}/refs") repo_id: str = str(data.get("repo_id", "")) if not repo_id: # Fallback: search endpoint search = _hub_api( hub_url, identity, "GET", f"/api/search?q={urllib.parse.quote(slug)}&limit=5", ) repos_val = search.get("repos", []) repos_list: list[_RepoEntry] = ( [r for r in repos_val if isinstance(r, dict)] if isinstance(repos_val, list) else [] ) for repo in repos_list: if repo.get("owner") == owner and repo.get("slug") == slug: repo_id = str(repo.get("repoId", "")) break if not repo_id: print( f"❌ Could not resolve repo ID for {sanitize_display(owner)}/{sanitize_display(slug)}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) return repo_id def _get_hub_and_identity( remote: str | None = None, hub_url_override: str | None = None, ) -> tuple[str, IdentityEntry]: """Load hub URL and identity for the current repo, or exit with a helpful error. Falls back to a named remote's URL when no hub is configured — useful for local development where the remote IS the hub. When *hub_url_override* is provided it takes precedence over the hub URL stored in ``.muse/config.toml``. This is the value supplied via the ``--hub`` CLI flag and lets callers (e.g. containerised agents) target a hub at a different address than the one recorded in the repo config without having to modify ``config.toml``. Raises: SystemExit: If not inside a repo, no hub/remote is configured, or not authenticated. """ root = find_repo_root() if root is None: print("❌ Not inside a Muse repository.", file=sys.stderr) raise SystemExit(ExitCode.REPO_NOT_FOUND) hub_url = hub_url_override or get_hub_url(root) if hub_url is None: # No hub configured — try the specified remote, then 'local', then any remote. candidate_name = remote or "local" candidate_url = get_remote(candidate_name, root) if candidate_url is None and remote is None: remotes = list_remotes(root) candidate_url = remotes[0]["url"] if remotes else None if candidate_url: candidate_name = remotes[0]["name"] if candidate_url: hub_url = candidate_url logger.debug( "No hub configured — using remote '%s' as hub URL: %s", candidate_name, hub_url, ) else: print("❌ No hub connected and no remotes configured.", file=sys.stderr) print( " Run `muse hub connect ` or `muse remote add local `.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) identity = load_identity(hub_url) if identity is None: print( f"❌ Not authenticated for {sanitize_display(hub_url)}.", file=sys.stderr, ) print(f" Run: muse auth register --hub {hub_url}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) return hub_url, identity def _resolve_proposal_id( hub_url: str, identity: IdentityEntry, repo_id: str, proposal_id_or_prefix: str, ) -> str: """Resolve a proposal ID prefix to a full UUID. If *proposal_id_or_prefix* looks like a full UUID (≥32 chars with hyphens), it is returned as-is. Otherwise the most recent ``_PROPOSAL_PREFIX_RESOLVE_LIMIT`` proposals are fetched and the first match whose ``proposalId`` starts with the prefix is returned. All user-supplied and API-sourced strings are sanitized before printing. Note ---- Prefix resolution is capped at :data:`_PROPOSAL_PREFIX_RESOLVE_LIMIT` proposals. On repositories with many more open proposals, very old proposals may not be findable by prefix — pass the full UUID in that case. Raises: SystemExit: When no match is found or multiple proposals match the prefix. """ if "-" in proposal_id_or_prefix and len(proposal_id_or_prefix) >= 32: return proposal_id_or_prefix # looks like a full UUID data = _hub_api( hub_url, identity, "GET", f"/api/repos/{repo_id}/proposals?limit={_PROPOSAL_PREFIX_RESOLVE_LIMIT}", ) proposals_val = data.get("proposals", []) proposals: list[_ProposalEntry] = ( [r for r in proposals_val if isinstance(r, dict)] if isinstance(proposals_val, list) else [] ) matches = [p for p in proposals if str(p.get("proposalId", "")).startswith(proposal_id_or_prefix)] if not matches: print( f"❌ No proposal found with ID prefix '{sanitize_display(proposal_id_or_prefix)}'.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if len(matches) > 1: print( f"❌ Ambiguous prefix '{sanitize_display(proposal_id_or_prefix)}' " f"— matches {len(matches)} proposals:", file=sys.stderr, ) for m in matches: print( f" {sanitize_display(str(m.get('proposalId', '')))} " f"{sanitize_display(str(m.get('title', '')))}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) return str(matches[0].get("proposalId", proposal_id_or_prefix)) def _format_proposal(entry: _ProposalEntry, *, verbose: bool = False) -> str: """Format a single proposal for human-readable display. All fields from the API response are sanitized via :func:`sanitize_display` before inclusion in the returned string — including ``author`` and ``createdAt`` in verbose mode. Args: entry: Raw proposal dict from the MuseHub API. verbose: When ``True``, append author name and creation date (YYYY-MM-DD). """ proposal_id = sanitize_display(str(entry.get("proposalId", ""))[:8]) title = sanitize_display(str(entry.get("title", "(no title)"))) state = str(entry.get("state", "?")) from_b = sanitize_display(str(entry.get("fromBranch", "?"))) to_b = sanitize_display(str(entry.get("toBranch", "?"))) state_icon = {"open": "🟢", "merged": "🟣", "closed": "⛔"}.get(state, "❓") line = f" {state_icon} {proposal_id} {from_b} → {to_b} {title}" if verbose: author = sanitize_display(str(entry.get("author", ""))) created = sanitize_display(str(entry.get("createdAt", ""))[:10]) line += f"\n by {author or '?'} {created}" return line # ── register ────────────────────────────────────────────────────────────────── def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``muse hub`` subcommand tree and all its flags. Every subcommand that produces output accepts ``--json`` for machine-readable output on stdout. All progress and diagnostic messages go to stderr. """ parser = subparsers.add_parser( "hub", help="MuseHub fabric connection management.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) subs = parser.add_subparsers(dest="subcommand", metavar="SUBCOMMAND") subs.required = True # ── connect ─────────────────────────────────────────────────────────────── connect_p = subs.add_parser( "connect", help="Attach this repository to a MuseHub instance.", description=( "Write [hub] url to .muse/config.toml and confirm auth status.\n" "Does not touch credentials — authenticate with 'muse auth register'.\n\n" "URL is normalised: bare hostnames gain https://, trailing slashes\n" "are stripped, http:// is rejected for non-loopback hosts.\n\n" "Agent quickstart:\n" " muse hub connect https://musehub.ai --json && muse auth register --agent --json\n\n" "JSON output keys: status, hub_url, hostname, authenticated,\n" " identity_name, identity_type" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) connect_p.add_argument( "url", metavar="URL", help="MuseHub URL (e.g. https://musehub.ai or just musehub.ai).", ) connect_p.add_argument( "--json", "-j", action="store_true", dest="json_output", default=False, help="Emit a JSON object to stdout on success.", ) connect_p.set_defaults(func=run_connect) # ── disconnect ──────────────────────────────────────────────────────────── disconnect_p = subs.add_parser( "disconnect", help="Remove the hub association from this repository.", description=( "Remove [hub] url from .muse/config.toml. Credentials in\n" "~/.muse/identity.toml are preserved — use 'muse auth logout'\n" "to remove them too. Makes no network calls.\n\n" "Operation is idempotent: exits 0 with status 'nothing_to_do'\n" "when no hub is configured.\n\n" "Agent quickstart:\n" " muse hub disconnect --json # get hub_url for cleanup\n\n" "JSON keys: status, hub_url, hostname" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) disconnect_p.add_argument( "--json", "-j", action="store_true", dest="json_output", default=False, help="Emit a JSON object to stdout on completion.", ) disconnect_p.set_defaults(func=run_disconnect) # ── repo ────────────────────────────────────────────────────────────────── repo_p = subs.add_parser( "repo", help="Manage repositories on MuseHub.", formatter_class=argparse.RawDescriptionHelpFormatter, ) repo_subs = repo_p.add_subparsers(dest="repo_subcommand", metavar="REPO_COMMAND") repo_subs.required = True repo_create_p = repo_subs.add_parser( "create", help="Create a new repository on MuseHub.", description=( "Create a new remote Muse repository on MuseHub.\n\n" "Name must be non-empty and ≤ 255 characters.\n" "Owner defaults to the authenticated identity's handle.\n" "The repo is initialized with an empty commit by default so it\n" "is immediately pushable. Pass --no-init to skip initialization\n" "(useful when you are about to push existing history).\n\n" "Agent quickstart:\n" " muse hub repo create --name my-repo --json\n" " muse hub repo create --name my-repo --private --no-init --json\n\n" "JSON output keys: repo_id, name, owner, slug, visibility,\n" " description, clone_url, tags, created_at\n\n" "Exit codes: 0 created, 1 validation/conflict/auth error,\n" " 2 not in repo, 3 API/network error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) repo_create_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config (e.g. http://localhost:10003/owner/repo).", ) repo_create_p.add_argument( "--name", "-n", required=True, help="Repository name (used to generate the URL slug).", ) repo_create_p.add_argument( "--owner", dest="owner", default="", metavar="OWNER", help="Owner username. Defaults to the authenticated identity's handle.", ) repo_create_p.add_argument( "--description", "-d", default="", help="Short description shown on the explore page.", ) repo_create_p.add_argument( "--private", action="store_true", default=False, help="Create as a private repository (default: public).", ) repo_create_p.add_argument( "--tag", dest="tags", action="append", default=[], metavar="TAG", help="Tag to apply (repeatable, e.g. --tag jazz --tag piano).", ) repo_create_p.add_argument( "--no-init", dest="no_init", action="store_true", default=False, help=( "Skip the initial empty commit. Use this when you are about to " "push existing history." ), ) repo_create_p.add_argument( "--default-branch", dest="default_branch", default="main", metavar="BRANCH", help="Name of the default branch created on initialization (default: main).", ) repo_create_p.add_argument( "--json", "-j", action="store_true", dest="json_output", default=False, help="Emit a JSON object to stdout on success.", ) repo_create_p.set_defaults(func=run_repo_create) # ── repo delete ─────────────────────────────────────────────────────────── repo_delete_p = repo_subs.add_parser( "delete", help="Soft-delete a repository (owner only).", formatter_class=argparse.RawDescriptionHelpFormatter, description=textwrap.dedent( """\ Soft-delete a MuseHub repository. Only the repository owner may delete. All data is retained for audit purposes; subsequent reads return 404. TARGET may be OWNER/SLUG or a repo UUID. When omitted the repo is resolved from the current directory's hub remote config. Examples: muse hub repo delete --yes muse hub repo delete gabriel/my-repo --yes muse hub repo delete a3f2c9d1-... --yes --json """ ), ) repo_delete_p.add_argument("target", nargs="?", default=None, metavar="OWNER/SLUG|REPO_ID", help="Repo to delete: OWNER/SLUG or UUID (default: current dir).") repo_delete_p.add_argument("--yes", "-y", action="store_true", dest="yes", default=False, help="Confirm deletion (required).") repo_delete_p.add_argument("--hub", dest="hub", default=None, metavar="URL", help="MuseHub base URL (overrides config).") repo_delete_p.add_argument("--json", "-j", action="store_true", dest="json_output", default=False, help="Emit JSON on success.") repo_delete_p.set_defaults(func=run_repo_delete) # ── repo settings ───────────────────────────────────────────────────────── repo_update_p = repo_subs.add_parser( "update", help="View or update repository settings (owner/admin).", formatter_class=argparse.RawDescriptionHelpFormatter, description=textwrap.dedent( """\ View or patch mutable settings for a MuseHub repository. Omit all patch flags to read current settings. The repository is resolved from the current directory's hub remote config. Examples: muse hub repo update muse hub repo update --description "New description" --visibility private muse hub repo update --json """ ), ) repo_update_p.add_argument("--name", dest="name", default=None, metavar="NAME", help="New repository name.") repo_update_p.add_argument("--description", dest="description", default=None, metavar="TEXT", help="New markdown description.") repo_update_p.add_argument("--visibility", dest="visibility", default=None, choices=["public", "private"], help="New visibility: public or private.") repo_update_p.add_argument("--default-branch", dest="default_branch", default=None, metavar="BRANCH", help="New default branch name.") repo_update_p.add_argument("--homepage-url", dest="homepage_url", default=None, metavar="URL", help="Project homepage URL.") repo_update_p.add_argument("--hub", dest="hub", default=None, metavar="URL", help="MuseHub base URL (overrides config).") repo_update_p.add_argument("--json", "-j", action="store_true", dest="json_output", default=False, help="Emit JSON on success.") repo_update_p.set_defaults(func=run_repo_update) # ── repo transfer ───────────────────────────────────────────────────────── repo_transfer_ownership_p = repo_subs.add_parser( "transfer-ownership", help="Transfer repository ownership to another user (owner only).", formatter_class=argparse.RawDescriptionHelpFormatter, description=textwrap.dedent( """\ Transfer ownership of a MuseHub repository to another user. Only the current owner may initiate a transfer. The repository is resolved from the current directory's hub remote config. Examples: muse hub repo transfer-ownership --new-owner bob muse hub repo transfer-ownership --new-owner bob --json """ ), ) repo_transfer_ownership_p.add_argument("--new-owner", dest="new_owner", required=True, metavar="HANDLE", help="MSign handle of the new owner.") repo_transfer_ownership_p.add_argument("--hub", dest="hub", default=None, metavar="URL", help="MuseHub base URL (overrides config).") repo_transfer_ownership_p.add_argument("--json", "-j", action="store_true", dest="json_output", default=False, help="Emit JSON on success.") repo_transfer_ownership_p.set_defaults(func=run_repo_transfer_ownership) # ── repo list ───────────────────────────────────────────────────────────── repo_list_p = repo_subs.add_parser( "list", help="List repositories owned by or collaborated on by the authenticated user.", formatter_class=argparse.RawDescriptionHelpFormatter, description=textwrap.dedent( """\ List MuseHub repositories owned by or collaborated on by the authenticated user. Results are ordered newest-first. Examples: muse hub repo list --json muse hub repo list --limit 50 --json muse hub repo list --cursor "" --json """ ), ) repo_list_p.add_argument( "--limit", dest="limit", type=int, default=20, metavar="N", help="Maximum repos per page (default 20, max 100).", ) repo_list_p.add_argument( "--cursor", dest="cursor", default=None, metavar="CURSOR", help="Pagination cursor from a previous next_cursor field.", ) repo_list_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="MuseHub base URL (overrides config).", ) repo_list_p.add_argument( "--json", "-j", action="store_true", dest="json_output", default=False, help="Emit JSON to stdout.", ) repo_list_p.set_defaults(func=run_repo_list) # ── repo show ───────────────────────────────────────────────────────────── repo_show_p = repo_subs.add_parser( "show", help="Show metadata for a single MuseHub repository.", formatter_class=argparse.RawDescriptionHelpFormatter, description=textwrap.dedent( """\ Show metadata for a MuseHub repository. Pass OWNER/SLUG to target a specific repo, or omit to resolve from the current directory's hub remote config. Examples: muse hub repo show gabriel/jazz-standards --json muse hub repo show --json """ ), ) repo_show_p.add_argument( "target", nargs="?", default=None, metavar="OWNER/SLUG", help="Repository to show (e.g. gabriel/my-repo). Omit to use current repo.", ) repo_show_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="MuseHub base URL (overrides config).", ) repo_show_p.add_argument( "--json", "-j", action="store_true", dest="json_output", default=False, help="Emit JSON to stdout.", ) repo_show_p.set_defaults(func=run_repo_show) repo_p.set_defaults(func=lambda a: repo_p.print_help()) # ── issue ───────────────────────────────────────────────────────────────── issue_p = subs.add_parser( "issue", help="Manage issues on MuseHub.", formatter_class=argparse.RawDescriptionHelpFormatter, ) issue_subs = issue_p.add_subparsers(dest="issue_subcommand", metavar="ISSUE_COMMAND") issue_subs.required = True issue_create_p = issue_subs.add_parser( "create", help="Open a new issue.", description=( "Open a new issue on MuseHub.\n\n" f"Title must be non-empty and ≤ {_MAX_ISSUE_TITLE_LEN} characters.\n" "Use --label (repeatable) to apply labels at creation time.\n" "Use --anchor (repeatable) to link to Muse symbols (path/to/file.py::Symbol).\n" "Use --commit-anchor (repeatable) to link to specific Muse commits.\n" "Use --agent-id / --model-id when filing on behalf of an AI agent.\n" "In text mode the issue URL is printed to stdout — capture it with $().\n\n" "Agent quickstart:\n" " muse hub issue create --title 'bug: crash in create_issue' \\\n" " --anchor musehub/services/musehub_issues.py::create_issue \\\n" " --agent-id agentception-worker-42 --model-id claude-sonnet-4-6 --json\n" " URL=$(muse hub issue create --title 'feat: X' --label enhancement)\n\n" "Exit codes: 0 created, 1 validation/auth error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) issue_create_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config (e.g. http://host.docker.internal:10003/owner/repo).", ) issue_create_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Alias for --hub: specify repo as owner/repo (hub base URL taken from config).", ) issue_create_p.add_argument("--title", "-t", required=True, help="Issue title.") issue_create_p.add_argument("--body", "-b", default="", help="Issue body.") issue_create_p.add_argument( "--label", "-l", dest="labels", action="append", default=[], help="Label name to apply (repeatable).", ) issue_create_p.add_argument( "--anchor", "-a", dest="symbol_anchors", action="append", default=[], metavar="FILE::SYMBOL", help=( "Muse symbol address to anchor this issue to (repeatable). " "Format: path/to/file.py::SymbolName. " "Example: --anchor musehub/services/musehub_issues.py::create_issue" ), ) issue_create_p.add_argument( "--commit-anchor", dest="commit_anchors", action="append", default=[], metavar="COMMIT_ID", help="Muse commit ID to anchor this issue to (repeatable).", ) issue_create_p.add_argument( "--agent-id", dest="agent_id", default="", help="Agent identifier when filing on behalf of an AI agent (e.g. agentception-worker-42).", ) issue_create_p.add_argument( "--model-id", dest="model_id", default="", help="Model identifier when filing on behalf of an AI agent (e.g. claude-sonnet-4-6).", ) issue_create_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON with the created issue.", ) issue_create_p.set_defaults(func=run_issue_create) issue_update_p = issue_subs.add_parser( "update", help="Update an existing issue.", description=( "Update an existing issue on MuseHub.\n\n" "Updates title, body, and/or anchors; omitted fields are left unchanged.\n" "At least one of --title, --body, --anchor, or --commit-anchor must be provided.\n" f"If --title is given it must be non-empty and ≤ {_MAX_ISSUE_TITLE_LEN} characters.\n" "--anchor / --commit-anchor replace the full anchor list (send all desired anchors).\n\n" "Agent quickstart:\n" " muse hub issue update 42 --body 'updated description' --json\n" " muse hub issue update 42 \\\n" " --anchor musehub/services/musehub_issues.py::create_issue \\\n" " --anchor musehub/db/musehub_models.py::MusehubIssue --json\n\n" "Exit codes: 0 updated, 1 validation/auth error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) issue_update_p.add_argument("number", type=int, help="Issue number.") issue_update_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) issue_update_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Alias for --hub: specify repo as owner/repo (hub base URL taken from config).", ) issue_update_p.add_argument("--title", "-t", default=None, help="New title.") issue_update_p.add_argument("--body", "-b", default=None, help="New body.") issue_update_p.add_argument( "--anchor", "-a", dest="symbol_anchors", action="append", default=None, metavar="FILE::SYMBOL", help=( "Replacement symbol anchor (repeatable; replaces all existing anchors). " "Format: path/to/file.py::SymbolName." ), ) issue_update_p.add_argument( "--commit-anchor", dest="commit_anchors", action="append", default=None, metavar="COMMIT_ID", help="Replacement commit anchor (repeatable; replaces all existing anchors).", ) issue_update_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON with the updated issue.", ) issue_update_p.set_defaults(func=run_issue_update) issue_read_p = issue_subs.add_parser( "read", help="Show a single issue.", description=( "Fetch a single issue by its per-repo number.\n\n" "Agent quickstart:\n" " muse hub issue read 42 --json\n" " muse hub issue read 42 --json | jq '{number,title,state}'\n\n" "Exit codes: 0 success, 1 auth error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) issue_read_p.add_argument("number", type=int, help="Issue number.") issue_read_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) issue_read_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) issue_read_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON with the issue.", ) issue_read_p.set_defaults(func=run_issue_read) issue_list_p = issue_subs.add_parser( "list", help="List issues for this repo.", description=( "List issues on MuseHub for the current repo.\n\n" "Agent quickstart:\n" " muse hub issue list --json\n" " muse hub issue list --state closed --json\n" " muse hub issue list --label bug --json\n\n" "Exit codes: 0 success (including empty list), 1 auth error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) issue_list_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) issue_list_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) issue_list_p.add_argument( "--state", dest="state", default="open", choices=["open", "closed", "all"], help="Filter by state (default: open).", ) issue_list_p.add_argument( "--label", dest="label", default=None, metavar="LABEL", help="Filter by label string.", ) issue_list_p.add_argument( "--limit", dest="limit", type=int, default=100, metavar="N", help="Maximum number of issues to return (default: 100).", ) issue_list_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON array of issues.", ) issue_list_p.set_defaults(func=run_issue_list) issue_close_p = issue_subs.add_parser( "close", help="Close an open issue.", description=( "Close an issue on MuseHub.\n\n" "Agent quickstart:\n" " muse hub issue close 42\n" " muse hub issue close 42 --json\n\n" "Exit codes: 0 success, 1 auth error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) issue_close_p.add_argument("number", type=int, help="Issue number.") issue_close_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) issue_close_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) issue_close_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON with the updated issue.", ) issue_close_p.set_defaults(func=run_issue_close) issue_reopen_p = issue_subs.add_parser( "reopen", help="Reopen a closed issue.", description=( "Reopen a closed issue on MuseHub.\n\n" "Agent quickstart:\n" " muse hub issue reopen 42\n" " muse hub issue reopen 42 --json\n\n" "Exit codes: 0 success, 1 auth error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) issue_reopen_p.add_argument("number", type=int, help="Issue number.") issue_reopen_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) issue_reopen_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) issue_reopen_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON with the updated issue.", ) issue_reopen_p.set_defaults(func=run_issue_reopen) issue_comment_p = issue_subs.add_parser( "comment", help="Post a comment on an issue.", description=( "Post a Markdown comment on an issue on MuseHub.\n\n" "Agent quickstart:\n" " muse hub issue comment 42 --body 'Fixed in commit abc123'\n" " muse hub issue comment 42 --body 'see also #43' --json\n\n" "Exit codes: 0 success, 1 validation/auth error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) issue_comment_p.add_argument("number", type=int, help="Issue number.") issue_comment_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) issue_comment_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) issue_comment_p.add_argument( "--body", "-b", required=True, dest="body", help="Comment body (Markdown).", ) issue_comment_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON with the updated comment list.", ) issue_comment_p.set_defaults(func=run_issue_comment) issue_comment_delete_p = issue_subs.add_parser( "comment-delete", help="Soft-delete a comment from an issue.", description=( "Soft-delete a comment on an issue on MuseHub.\n\n" "The comment is hidden from list results but preserved in the audit log.\n" "Requires write/admin access or repo ownership.\n\n" "Agent quickstart:\n" " muse hub issue comment-delete 42 --comment-id \n" " muse hub issue comment-delete 42 --comment-id --json\n\n" "Exit codes: 0 success, 1 validation/auth error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) issue_comment_delete_p.add_argument("number", type=int, help="Issue number.") issue_comment_delete_p.add_argument( "--comment-id", dest="comment_id", required=True, metavar="UUID", help="UUID of the comment to delete.", ) issue_comment_delete_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) issue_comment_delete_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) issue_comment_delete_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON confirmation on success.", ) issue_comment_delete_p.set_defaults(func=run_issue_comment_delete) issue_assign_p = issue_subs.add_parser( "assign", help="Assign or unassign a collaborator on an issue.", description=( "Assign or unassign a collaborator on an issue on MuseHub.\n\n" "Agent quickstart:\n" " muse hub issue assign 42 --assignee gabriel\n" " muse hub issue assign 42 --assignee '' # unassign\n" " muse hub issue assign 42 --assignee gabriel --json\n\n" "Exit codes: 0 success, 1 auth error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) issue_assign_p.add_argument("number", type=int, help="Issue number.") issue_assign_p.add_argument( "--assignee", dest="assignee", required=True, metavar="USER", help="Username to assign, or empty string to unassign.", ) issue_assign_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) issue_assign_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) issue_assign_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON with the updated issue.", ) issue_assign_p.set_defaults(func=run_issue_assign) issue_label_p = issue_subs.add_parser( "label", help="Manage labels on an issue.", description=( "Add or remove labels on an issue on MuseHub.\n\n" "Agent quickstart:\n" " muse hub issue label 42 --set bug enhancement\n" " muse hub issue label 42 --remove bug\n" " muse hub issue label 42 --set bug --json\n\n" "--set replaces the entire label list. --remove removes a single label.\n" "Exit codes: 0 success, 1 auth error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) issue_label_p.add_argument("number", type=int, help="Issue number.") _issue_label_mutex = issue_label_p.add_mutually_exclusive_group(required=True) _issue_label_mutex.add_argument( "--set", dest="set_labels", nargs="+", metavar="LABEL", help="Replace the entire label list with these labels.", ) _issue_label_mutex.add_argument( "--remove", dest="remove_label", metavar="LABEL", help="Remove a single label from the issue.", ) issue_label_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) issue_label_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) issue_label_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON with the updated issue.", ) issue_label_p.set_defaults(func=run_issue_label) issue_p.set_defaults(func=lambda a: issue_p.print_help()) # ── label ───────────────────────────────────────────────────────────────── label_p = subs.add_parser( "label", help="Manage labels on MuseHub.", formatter_class=argparse.RawDescriptionHelpFormatter, ) label_subs = label_p.add_subparsers(dest="label_subcommand", metavar="LABEL_COMMAND") label_subs.required = True label_create_p = label_subs.add_parser( "create", help="Create a new label.", description=( "Create a repo-scoped label with a name, hex colour, and optional description.\n\n" f"Name must be non-empty and ≤ {_MAX_LABEL_NAME_LEN} characters.\n" "Colour must be a 7-character hex string starting with '#' (e.g. '#d73a4a').\n" "Names must be unique within the repository.\n\n" "Agent quickstart:\n" " muse hub label create --name bug --color '#d73a4a' --json\n" " muse hub label create --name enhancement --color '#a2eeef' " "--description 'New feature or request' --json\n\n" "JSON output keys: label_id, repo_id, name, color, description\n\n" "Exit codes: 0 created, 1 validation/conflict/auth error,\n" " 2 not in repo, 3 API/network error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) label_create_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) label_create_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo (hub base URL taken from config).", ) label_create_p.add_argument( "--name", "-n", required=True, help="Label name (must be unique within the repo).", ) label_create_p.add_argument( "--color", "-c", required=True, help="Hex colour string, e.g. '#d73a4a'.", ) label_create_p.add_argument( "--description", "-d", default=None, help="Optional label description (≤ 200 characters).", ) label_create_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON with the created label.", ) label_create_p.set_defaults(func=run_label_create) label_list_p = label_subs.add_parser( "list", help="List all labels for the current repo.", description=( "List every label defined in the repository.\n\n" "The list endpoint is public — no authentication required for public repos.\n\n" "Agent quickstart:\n" " muse hub label list --json\n\n" "JSON output: array of label objects (label_id, name, color, description)\n\n" "Exit codes: 0 success, 1 auth error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) label_list_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) label_list_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) label_list_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON array of labels.", ) label_list_p.set_defaults(func=run_label_list) label_update_p = label_subs.add_parser( "update", help="Update an existing label's name, colour, or description.", description=( "Partially update a label identified by its current name.\n\n" "Only provided flags are changed; omitted fields are left unchanged.\n" f"New name must be ≤ {_MAX_LABEL_NAME_LEN} characters.\n" "Colour must be a 7-character hex string starting with '#'.\n\n" "Agent quickstart:\n" " muse hub label update --name bug --new-color '#b60205' --json\n" " muse hub label update --name bug --new-name bug-report --json\n\n" "JSON output keys: label_id, repo_id, name, color, description\n\n" "Exit codes: 0 updated, 1 validation/auth error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) label_update_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) label_update_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) label_update_p.add_argument( "--name", "-n", required=True, help="Current name of the label to update.", ) label_update_p.add_argument( "--new-name", dest="new_name", default=None, help="New name for the label.", ) label_update_p.add_argument( "--new-color", dest="new_color", default=None, help="New hex colour, e.g. '#b60205'.", ) label_update_p.add_argument( "--new-description", dest="new_description", default=None, help="New description (pass empty string to clear).", ) label_update_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON with the updated label.", ) label_update_p.set_defaults(func=run_label_update) label_delete_p = label_subs.add_parser( "delete", help="Delete a label and remove it from all issues.", description=( "Permanently delete a label by name and remove it from every\n" "issue and proposal it is currently attached to.\n\n" "This operation is irreversible.\n\n" "Agent quickstart:\n" " muse hub label delete --name bug --json\n\n" "Exit codes: 0 deleted, 1 auth/not-found error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) label_delete_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) label_delete_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) label_delete_p.add_argument( "--name", "-n", required=True, help="Name of the label to delete.", ) label_delete_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON confirmation on success.", ) label_delete_p.set_defaults(func=run_label_delete) label_p.set_defaults(func=lambda a: label_p.print_help()) # ── collaborator ────────────────────────────────────────────────────────── collab_p = subs.add_parser( "collaborator", help="Manage repository collaborators on MuseHub.", formatter_class=argparse.RawDescriptionHelpFormatter, ) collab_subs = collab_p.add_subparsers(dest="collaborator_subcommand", metavar="COLLAB_COMMAND") collab_subs.required = True collab_list_p = collab_subs.add_parser( "list", help="List collaborators on a repo.", description=( "List all collaborators and their permission levels for a repository.\n\n" "Agent quickstart:\n" " muse hub collaborator list --json\n\n" "Exit codes: 0 success, 1 auth error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) collab_list_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) collab_list_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) collab_list_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON list of collaborators.", ) collab_list_p.set_defaults(func=run_collaborator_list) collab_invite_p = collab_subs.add_parser( "invite", help="Invite a user as a collaborator.", description=( "Invite a user to collaborate on a repository.\n" "Requires admin or owner access.\n\n" "Agent quickstart:\n" " muse hub collaborator invite carol --permission write --json\n\n" "Exit codes: 0 success, 1 auth/conflict error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) collab_invite_p.add_argument("handle", help="MSign handle of the user to invite.") collab_invite_p.add_argument( "--permission", "-p", default="write", choices=["read", "write", "admin"], help="Permission level: read, write (default), or admin.", ) collab_invite_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) collab_invite_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) collab_invite_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON collaborator record on success.", ) collab_invite_p.set_defaults(func=run_collaborator_invite) collab_update_permission_p = collab_subs.add_parser( "update-permission", help="Update a collaborator's permission level.", description=( "Update an existing collaborator's permission level.\n" "Requires admin or owner access.\n\n" "Agent quickstart:\n" " muse hub collaborator update-permission carol --permission admin --json\n\n" "Exit codes: 0 success, 1 auth/not-found error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) collab_update_permission_p.add_argument("handle", help="MSign handle of the collaborator to update.") collab_update_permission_p.add_argument( "--permission", "-p", required=True, choices=["read", "write", "admin"], help="New permission level.", ) collab_update_permission_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) collab_update_permission_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) collab_update_permission_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON with the updated collaborator record.", ) collab_update_permission_p.set_defaults(func=run_collaborator_update_permission) collab_remove_p = collab_subs.add_parser( "remove", help="Remove a collaborator from a repo.", description=( "Remove a collaborator from a repository.\n" "Requires admin or owner access. The owner cannot be removed.\n\n" "Agent quickstart:\n" " muse hub collaborator remove carol --json\n\n" "Exit codes: 0 success, 1 auth/not-found error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) collab_remove_p.add_argument("handle", help="MSign handle of the collaborator to remove.") collab_remove_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) collab_remove_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) collab_remove_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON confirmation on success.", ) collab_remove_p.set_defaults(func=run_collaborator_remove) collab_p.set_defaults(func=lambda a: collab_p.print_help()) # ── webhook ─────────────────────────────────────────────────────────────── webhook_p = subs.add_parser( "webhook", help="Manage webhook subscriptions on MuseHub.", formatter_class=argparse.RawDescriptionHelpFormatter, ) webhook_subs = webhook_p.add_subparsers(dest="webhook_subcommand", metavar="WEBHOOK_COMMAND") webhook_subs.required = True webhook_create_p = webhook_subs.add_parser( "create", help="Register a new webhook subscription.", description=( "Register a webhook that receives HTTP POST payloads for repository events.\n" "Valid event types: push, proposal, issue, release, branch, tag, session, analysis.\n" "Requires write/admin access or repo ownership.\n\n" "Agent quickstart:\n" " muse hub webhook create --url https://example.com/hook --events push release --json\n\n" "Exit codes: 0 success, 1 validation/auth error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) webhook_create_p.add_argument( "--url", required=True, help="HTTPS URL to receive event payloads.", ) webhook_create_p.add_argument( "--events", nargs="+", required=True, metavar="EVENT", help="Event types to subscribe to (e.g. push release issue).", ) webhook_create_p.add_argument( "--secret", default="", help="HMAC-SHA256 signing secret (optional, plaintext).", ) webhook_create_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) webhook_create_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) webhook_create_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON webhook record on success.", ) webhook_create_p.set_defaults(func=run_webhook_create) webhook_list_p = webhook_subs.add_parser( "list", help="List webhook subscriptions for a repo.", description=( "List all registered webhook subscriptions for a repository.\n\n" "Agent quickstart:\n" " muse hub webhook list --json\n\n" "Exit codes: 0 success, 1 auth error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) webhook_list_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) webhook_list_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) webhook_list_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON list of webhooks.", ) webhook_list_p.set_defaults(func=run_webhook_list) webhook_delete_p = webhook_subs.add_parser( "delete", help="Delete a webhook subscription.", description=( "Delete a webhook subscription and all its delivery history.\n" "Requires write/admin access or repo ownership.\n\n" "Agent quickstart:\n" " muse hub webhook delete --json\n\n" "Exit codes: 0 success, 1 auth/not-found error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) webhook_delete_p.add_argument("webhook_id", help="UUID of the webhook to delete.") webhook_delete_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) webhook_delete_p.add_argument( "--repo", dest="repo", default=None, metavar="OWNER/REPO", help="Specify repo as owner/repo.", ) webhook_delete_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON confirmation on success.", ) webhook_delete_p.set_defaults(func=run_webhook_delete) webhook_p.set_defaults(func=lambda a: webhook_p.print_help()) # ── ping ────────────────────────────────────────────────────────────────── ping_p = subs.add_parser( "ping", help="Test HTTP connectivity to the configured hub.", description=( "Send GET /health and report reachability. No auth token\n" "is sent — the health endpoint is intentionally public.\n" "HTTP redirects are refused.\n\n" "Agent readiness loop:\n" " until muse hub ping --json 2>/dev/null; do sleep 2; done\n\n" "JSON keys: status, hub_url, hostname, reachable, message\n" "Exit 0 = reachable, 5 = unreachable, 1 = no hub, 2 = no repo" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) ping_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) ping_p.add_argument( "--json", "-j", action="store_true", dest="json_output", default=False, help="Emit a JSON object to stdout with the ping result.", ) ping_p.set_defaults(func=run_ping) # ── proposal ────────────────────────────────────────────────────────────── proposal_p = subs.add_parser( "proposal", help="Manage proposals on MuseHub.", formatter_class=argparse.RawDescriptionHelpFormatter, ) proposal_subs = proposal_p.add_subparsers(dest="proposal_subcommand", metavar="PROPOSAL_COMMAND") proposal_subs.required = True proposal_create_p = proposal_subs.add_parser( "create", help="Open a new proposal.", description=( "Open a new proposal on MuseHub.\n\n" "Uses the current branch as the source if --from-branch is omitted.\n" "Both Muse-native flags (--from-branch / --to-branch) and familiar\n" "aliases (--head / --base) are accepted.\n\n" f"Title must be non-empty and ≤ {_MAX_PROPOSAL_TITLE_LEN} characters.\n" "Detached HEAD is rejected — pass --from-branch explicitly.\n\n" "Agent quickstart:\n" " muse hub proposal create --title 'feat: x' --from-branch feat/x --json\n" " PROPOSAL=$(muse hub proposal create --title 'feat: x' --json)\n" " echo \"$PROPOSAL\" | jq '.proposalId'\n\n" "Exit codes: 0 created, 1 validation/branch error, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) proposal_create_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) proposal_create_p.add_argument("--title", "-t", required=True, help="Proposal title.") proposal_create_p.add_argument("--body", "-b", default="", help="Proposal body / description.") proposal_create_p.add_argument( "--from-branch", "--head", "-f", dest="from_branch", default=None, help=( "Source branch (default: current branch). " "--head is accepted as a git-familiar alias." ), ) proposal_create_p.add_argument( "--to-branch", "--base", dest="to_branch", default="dev", help="Target branch (default: dev). --base is accepted as a git-familiar alias.", ) proposal_create_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON with the created proposal.", ) proposal_create_p.set_defaults(func=run_proposal_create) proposal_list_p = proposal_subs.add_parser( "list", help="List proposals.", description=( "List proposals on the configured MuseHub.\n\n" "Filters by state (default: open). --limit caps the number of results.\n" "Use --state all to see every proposal regardless of state.\n" "Use --verbose to also show author and creation date per proposal.\n\n" "Agent quickstart:\n" " muse hub proposal list --state open --json\n" " muse hub proposal list --state all -n 100 --json | jq '.[] | .proposalId'\n\n" "Exit codes: 0 success, 1 not authenticated, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) proposal_list_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config (e.g. http://host.docker.internal:10003/owner/repo).", ) proposal_list_p.add_argument( "--state", default="open", choices=["open", "merged", "closed", "all"], help="Filter by state: open (default), merged, closed, all.", ) proposal_list_p.add_argument( "--limit", "-n", type=int, default=20, help="Maximum number of proposals to return (default: 20).", ) proposal_list_p.add_argument( "--verbose", "-v", action="store_true", dest="verbose", help="Show author and creation date for each proposal.", ) proposal_list_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON to stdout instead of human-readable output.", ) proposal_list_p.set_defaults(func=run_proposal_list) proposal_merge_p = proposal_subs.add_parser( "merge", help="Merge a proposal.", description=( "Merge a proposal immediately — no confirmation dialog.\n\n" "The source branch is deleted by default; pass --no-delete-branch\n" "to keep it. Merge strategies: merge_commit (default), squash, rebase.\n\n" "Exit code 3 is returned on merge rejection in BOTH text and JSON mode\n" "so agent pipelines can always rely on the exit code:\n" " muse hub proposal merge af54753d --json && echo ok || echo conflict\n\n" "Agent quickstart:\n" " muse hub proposal merge af54753d --strategy squash --json\n" " # → {\"merged\": true, \"mergeCommitId\": \"...\", ...}\n\n" "Exit codes: 0 merged, 1 not found/not authenticated,\n" " 2 not in repo, 3 merge rejected or API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) proposal_merge_p.add_argument("proposal_id", help="Proposal ID (full UUID or 8-char prefix).") proposal_merge_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) proposal_merge_p.add_argument( "--strategy", default="merge_commit", choices=["merge_commit", "squash", "rebase"], help="Merge strategy (default: merge_commit).", ) proposal_merge_p.add_argument( "--no-delete-branch", dest="delete_branch", action="store_false", help="Do not delete the source branch after merging.", ) proposal_merge_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON with the proposal merge result.", ) proposal_merge_p.set_defaults(func=run_proposal_merge, delete_branch=True) proposal_read_p = proposal_subs.add_parser( "read", help="Show a single proposal.", description=( "Show a single proposal by ID.\n\n" "Accepts the full UUID or an 8-character prefix. Prefix resolution\n" f"fetches the most recent {_PROPOSAL_PREFIX_RESOLVE_LIMIT} proposals — use the full UUID\n" "on large repositories to avoid ambiguity.\n\n" "Text output shows state, title, ID, branches, author, date, and\n" f"up to {_MAX_PROPOSAL_BODY_LINES} body lines (truncation hint printed when exceeded).\n" "JSON output is the raw API response (camelCase field names).\n\n" "Agent quickstart:\n" " muse hub proposal read af54753d --json | jq '.state'\n" " muse hub proposal read af54753d --json | jq '{state,title,author}'\n\n" "Exit codes: 0 found, 1 not found/ambiguous, 2 not in repo, 3 API error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) proposal_read_p.add_argument("proposal_id", help="Proposal ID (full UUID or 8-char prefix).") proposal_read_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config.", ) proposal_read_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON.", ) proposal_read_p.set_defaults(func=run_proposal_read) # ── proposal comment ────────────────────────────────────────────────────── proposal_comment_p = proposal_subs.add_parser( "comment", help="Manage proposal comments.", description="Manage review comments on a merge proposal.", formatter_class=argparse.RawDescriptionHelpFormatter, ) proposal_comment_subs = proposal_comment_p.add_subparsers( dest="proposal_comment_subcommand", metavar="COMMENT_COMMAND", ) proposal_comment_subs.required = True proposal_comment_list_p = proposal_comment_subs.add_parser( "list", help="List all comments on a proposal (threaded).", ) proposal_comment_list_p.add_argument("proposal_id", help="Proposal ID (full UUID or 8-char prefix).") proposal_comment_list_p.add_argument("--hub", dest="hub", default=None, metavar="URL") proposal_comment_list_p.add_argument("--json", "-j", action="store_true", dest="json_output") proposal_comment_list_p.set_defaults(func=run_proposal_comment_list) proposal_comment_p.set_defaults(func=lambda a: proposal_comment_p.print_help()) # ── proposal reviewer ───────────────────────────────────────────────────── proposal_reviewer_p = proposal_subs.add_parser( "reviewer", help="Manage proposal reviewers.", description="Request or remove reviewers on a merge proposal.", formatter_class=argparse.RawDescriptionHelpFormatter, ) proposal_reviewer_subs = proposal_reviewer_p.add_subparsers( dest="proposal_reviewer_subcommand", metavar="REVIEWER_COMMAND", ) proposal_reviewer_subs.required = True proposal_reviewer_request_p = proposal_reviewer_subs.add_parser( "request", help="Request reviewers on a proposal.", ) proposal_reviewer_request_p.add_argument("proposal_id", help="Proposal ID (full UUID or 8-char prefix).") proposal_reviewer_request_p.add_argument("reviewers", nargs="+", help="MSign handles to request.") proposal_reviewer_request_p.add_argument("--hub", dest="hub", default=None, metavar="URL") proposal_reviewer_request_p.add_argument("--json", "-j", action="store_true", dest="json_output") proposal_reviewer_request_p.set_defaults(func=run_proposal_reviewer_request) proposal_reviewer_remove_p = proposal_reviewer_subs.add_parser( "remove", help="Remove a pending reviewer from a proposal.", ) proposal_reviewer_remove_p.add_argument("proposal_id", help="Proposal ID (full UUID or 8-char prefix).") proposal_reviewer_remove_p.add_argument("reviewer", help="MSign handle of the reviewer to remove.") proposal_reviewer_remove_p.add_argument("--hub", dest="hub", default=None, metavar="URL") proposal_reviewer_remove_p.add_argument("--json", "-j", action="store_true", dest="json_output") proposal_reviewer_remove_p.set_defaults(func=run_proposal_reviewer_remove) proposal_reviewer_p.set_defaults(func=lambda a: proposal_reviewer_p.print_help()) # ── proposal review ─────────────────────────────────────────────────────── proposal_review_p = proposal_subs.add_parser( "review", help="Manage proposal reviews.", description="List or filter formal reviews on a merge proposal.", formatter_class=argparse.RawDescriptionHelpFormatter, ) proposal_review_subs = proposal_review_p.add_subparsers( dest="proposal_review_subcommand", metavar="REVIEW_COMMAND", ) proposal_review_subs.required = True proposal_review_list_p = proposal_review_subs.add_parser( "list", help="List all reviews on a proposal.", ) proposal_review_list_p.add_argument("proposal_id", help="Proposal ID (full UUID or 8-char prefix).") proposal_review_list_p.add_argument( "--state", choices=["pending", "approved", "changes_requested", "dismissed"], default=None, help="Filter by review state.", ) proposal_review_list_p.add_argument("--hub", dest="hub", default=None, metavar="URL") proposal_review_list_p.add_argument("--json", "-j", action="store_true", dest="json_output") proposal_review_list_p.set_defaults(func=run_proposal_review_list) proposal_review_p.set_defaults(func=lambda a: proposal_review_p.print_help()) proposal_p.set_defaults(func=lambda a: proposal_p.print_help()) # ── status ──────────────────────────────────────────────────────────────── status_p = subs.add_parser( "status", help="Show the hub connection and identity for this repository.", description=( "Display the hub URL stored in .muse/config.toml and the identity\n" "stored in ~/.muse/identity.toml. Makes no network calls.\n\n" "Agent quickstart:\n" " muse hub status --json || muse hub connect https://musehub.ai --json\n\n" "JSON keys (always present): hub_url, hostname, authenticated,\n" " identity_type, identity_name, identity_id, capabilities" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) status_p.add_argument( "--hub", dest="hub", default=None, metavar="URL", help="Override the hub URL from config (e.g. http://host.docker.internal:10003/owner/repo).", ) status_p.add_argument( "--json", "-j", action="store_true", dest="json_output", help="Emit JSON to stdout instead of human-readable output.", ) status_p.set_defaults(func=run_status) # ── Proposal subcommand handlers ────────────────────────────────────────────── def run_proposal_list(args: argparse.Namespace) -> None: """List proposals on MuseHub. Filters by state (default: ``open``) and respects ``--limit`` (default: 20). Use ``--state all`` to list every proposal regardless of state. Use ``--verbose`` / ``-v`` to show author and creation date per proposal. JSON output goes to stdout; human-readable text goes to stderr. With ``--json`` the response is always a JSON **array** (``[]`` when empty). ``--verbose`` has no effect when ``--json`` is set — callers receive the full API response and can select any field with ``jq``. Agent quickstart ---------------- :: muse hub proposal list --state open --json # open proposals as JSON array muse hub proposal list --state all -n 100 --json | jq '.[] | .proposalId' muse hub proposal list --verbose # show author + date in text mode Exit codes ---------- 0 Success (including empty list). 1 Not authenticated or no hub configured. 2 Not inside a Muse repository. 3 API / network error. """ state: str = args.state limit: int = clamp_int(args.limit, 1, 10000, "limit") json_output: bool = args.json_output verbose: bool = getattr(args, "verbose", False) hub_url, identity = _get_hub_and_identity(hub_url_override=args.hub) repo_id = _resolve_repo_id(hub_url, identity) params = f"?limit={limit}" if state != "all": params += f"&state={state}" data = _hub_api(hub_url, identity, "GET", f"/api/repos/{repo_id}/proposals{params}") proposals_val = data.get("proposals", []) proposals: list[_ProposalEntry] = ( [r for r in proposals_val if isinstance(r, dict)] if isinstance(proposals_val, list) else [] ) if json_output: print(json.dumps(proposals)) return if not proposals: print(f" No proposals found (state={state}).", file=sys.stderr) return print( f"\n Proposals — {sanitize_display(hub_url)} ({len(proposals)} shown)", file=sys.stderr, ) print(" " + "─" * 60, file=sys.stderr) for proposal in proposals: print(_format_proposal(proposal, verbose=verbose), file=sys.stderr) print("", file=sys.stderr) def run_proposal_read(args: argparse.Namespace) -> None: """Show a single proposal by ID. Accepts the full UUID or an 8-character prefix. Prefix resolution fetches the most recent :data:`_PROPOSAL_PREFIX_RESOLVE_LIMIT` proposals — use the full UUID on large repositories to avoid ambiguity. Text output (stderr) shows: state icon, title, ID, branches, author, creation date, and the first :data:`_MAX_PROPOSAL_BODY_LINES` lines of the body. When the body is truncated a hint is printed; pass ``--json`` to retrieve the full body. JSON output is the raw API response object (passthrough) — field names follow MuseHub's camelCase convention (``proposalId``, ``fromBranch``, etc.). Agent quickstart ---------------- :: muse hub proposal view af54753d --json | jq '.state' muse hub proposal view af54753d --json | jq '{state,title,author}' Exit codes ---------- 0 Proposal found and printed. 1 Prefix not found, ambiguous, or not authenticated. 2 Not inside a Muse repository. 3 API / network error. """ proposal_id: str = args.proposal_id json_output: bool = args.json_output hub_url, identity = _get_hub_and_identity(hub_url_override=args.hub) repo_id = _resolve_repo_id(hub_url, identity) proposal_id = _resolve_proposal_id(hub_url, identity, repo_id, proposal_id) data = _hub_api(hub_url, identity, "GET", f"/api/repos/{repo_id}/proposals/{proposal_id}") if json_output: print(json.dumps(data)) return title = sanitize_display(str(data.get("title", "(no title)"))) state = str(data.get("state", "?")) from_b = sanitize_display(str(data.get("fromBranch", "?"))) to_b = sanitize_display(str(data.get("toBranch", "?"))) full_id = sanitize_display(str(data.get("proposalId", proposal_id))) author = sanitize_display(str(data.get("author", ""))) created = sanitize_display(str(data.get("createdAt", ""))[:10]) body = str(data.get("body", "")) # state_icon lookup uses the raw state string as a dict key — safe; the # fallback "❓" handles any unexpected value. The display string is # sanitized separately so ANSI-injected state values don't reach the terminal. state_icon = {"open": "🟢", "merged": "🟣", "closed": "⛔"}.get(state, "❓") state_display = sanitize_display(state) print(f"\n {state_icon} [{state_display.upper()}] {title}", file=sys.stderr) print(f" ID: {full_id}", file=sys.stderr) print(f" Branch: {from_b} → {to_b}", file=sys.stderr) if author: print(f" By: {author} {created}", file=sys.stderr) if body: body_lines = body.strip().splitlines() print(" Body:", file=sys.stderr) for line in body_lines[:_MAX_PROPOSAL_BODY_LINES]: print(f" {sanitize_display(line)}", file=sys.stderr) if len(body_lines) > _MAX_PROPOSAL_BODY_LINES: remaining = len(body_lines) - _MAX_PROPOSAL_BODY_LINES print( f" ... ({remaining} more line{'s' if remaining != 1 else ''}" " — use --json to see full body)", file=sys.stderr, ) print("", file=sys.stderr) def run_proposal_create(args: argparse.Namespace) -> None: """Open a new proposal on MuseHub. Uses the current branch as the source if ``--from-branch`` is not given. Both Muse-native (``--from-branch`` / ``--to-branch``) and git-familiar aliases (``--head`` / ``--base``) are accepted. All local validation (title length, branch detection) runs before any network I/O, so errors are reported immediately without contacting the hub. Branch detection falls back to the current branch in ``.muse/HEAD``. When the repository is in detached HEAD state and ``--from-branch`` is not given, a clear error is printed and the command exits with code 1. Title is validated client-side: must be non-empty and no longer than :data:`_MAX_PROPOSAL_TITLE_LEN` characters. JSON output is the raw API response (passthrough). Human-readable text goes to stderr. Agent quickstart ---------------- :: muse hub proposal create --title "feat: x" --from-branch feat/x --json # → {"proposalId": "...", "state": "open", ...} # Full idiomatic flow: PROPOSAL=$(muse hub proposal create --title "feat: x" --json) echo "$PROPOSAL" | jq '.proposalId' Exit codes ---------- 0 Proposal created. 1 Title empty/too long, branch cannot be determined, or not authenticated. 2 Not inside a Muse repository. 3 API / network error. """ title: str = args.title body: str = args.body to_branch: str = args.to_branch json_output: bool = args.json_output # ── Local validation first — fail fast before any network I/O ──────────── # Client-side title validation — better UX than a raw API 400. if not title.strip(): print("❌ Proposal title must not be empty.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if len(title) > _MAX_PROPOSAL_TITLE_LEN: print( f"❌ Proposal title is too long ({len(title)} chars); " f"maximum is {_MAX_PROPOSAL_TITLE_LEN}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # Resolve source branch before making any network calls so that a # detached-HEAD error or missing branch is caught locally and quickly. from_branch: str = args.from_branch or "" if not from_branch: ctx = find_repo_root() if ctx is not None: try: from_branch = read_current_branch(ctx) except ValueError as exc: # Detached HEAD — read_current_branch raises ValueError. print(f"❌ {exc}", file=sys.stderr) print( " Use --from-branch to specify the source branch explicitly.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) from exc if not from_branch: print( "❌ Could not determine current branch. Use --from-branch.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # ── Network calls ───────────────────────────────────────────────────────── hub_url, identity = _get_hub_and_identity(hub_url_override=args.hub) repo_id = _resolve_repo_id(hub_url, identity) payload: _HubPayload = { "title": title, "body": body, "fromBranch": from_branch, "toBranch": to_branch, } data = _hub_api( hub_url, identity, "POST", f"/api/repos/{repo_id}/proposals", body=payload, ) if json_output: print(json.dumps(data)) return proposal_id = str(data.get("proposalId", "")) print( f"✅ Proposal created: {sanitize_display(proposal_id[:8])}", file=sys.stderr, ) print( f" {sanitize_display(from_branch)} → {sanitize_display(to_branch)}: " f"{sanitize_display(title)}", file=sys.stderr, ) parsed = urllib.parse.urlparse(hub_url) parts = parsed.path.strip("/").split("/") if len(parts) >= 2: owner, slug = parts[0], parts[1] server_root = f"{parsed.scheme}://{parsed.netloc}" print( f" URL: {sanitize_display(server_root)}/{sanitize_display(owner)}" f"/{sanitize_display(slug)}/proposals/{sanitize_display(proposal_id)}", file=sys.stderr, ) def run_proposal_merge(args: argparse.Namespace) -> None: """Merge a proposal. Merges immediately — no confirmation dialog. The source branch is deleted by default; pass ``--no-delete-branch`` to keep it. Merge strategies: ``merge_commit`` (default), ``squash``, ``rebase``. The hub may return HTTP 200 with ``{"merged": false}`` when it refuses the merge (e.g. conflict, branch-protection rule). This command treats that as a failure in **both** text and JSON modes — exit code 3 is returned so that agent pipelines can reliably use the exit code to detect success:: muse hub proposal merge af54753d --json && echo "merged" || echo "conflict" JSON output is the raw API response (passthrough) to stdout; human-readable text goes to stderr. Agent quickstart ---------------- :: muse hub proposal merge af54753d --strategy squash --json # → {"merged": true, "mergeCommitId": "...", ...} # Safe pipeline — exit code 3 on merge failure, even with --json: PROPOSAL_RESULT=$(muse hub proposal merge af54753d --json) || exit 1 echo "$PROPOSAL_RESULT" | jq '.mergeCommitId' Exit codes ---------- 0 Merged successfully. 1 Prefix not found, ambiguous, or not authenticated. 2 Not inside a Muse repository. 3 Merge rejected by hub (conflict, branch protection, etc.) or API error. """ proposal_id: str = args.proposal_id strategy: str = args.strategy delete_branch: bool = args.delete_branch json_output: bool = args.json_output hub_url, identity = _get_hub_and_identity(hub_url_override=args.hub) repo_id = _resolve_repo_id(hub_url, identity) proposal_id = _resolve_proposal_id(hub_url, identity, repo_id, proposal_id) payload: _HubPayload = { "mergeStrategy": strategy, "deleteBranch": delete_branch, } data = _hub_api( hub_url, identity, "POST", f"/api/repos/{repo_id}/proposals/{proposal_id}/merge", body=payload, ) merged: bool = bool(data.get("merged", False)) if json_output: print(json.dumps(data)) # Exit nonzero even in JSON mode so agent pipelines can rely on the # exit code — a hub-side merge rejection must never silently exit 0. if not merged: raise SystemExit(ExitCode.INTERNAL_ERROR) return commit_id: str = str(data.get("mergeCommitId", "")) or "" if merged: sha = sanitize_display(commit_id[:8]) if commit_id else "(no SHA)" print( f"✅ Proposal {sanitize_display(proposal_id[:8])} merged → {sha}", file=sys.stderr, ) if delete_branch: print(" Source branch deleted.", file=sys.stderr) else: msg = str(data.get("message", "unknown error")) print(f"❌ Merge failed: {sanitize_display(msg)}", file=sys.stderr) raise SystemExit(ExitCode.INTERNAL_ERROR) # ── Proposal comment / reviewer / review handlers ───────────────────────────── def run_proposal_comment_list(args: argparse.Namespace) -> None: """List all comments on a proposal, threaded by parent. Each top-level comment includes its direct replies nested under it.:: muse hub proposal comment list af54753d muse hub proposal comment list af54753d --json Exit codes: 0 Success. 1 Auth error or proposal not found. 2 Not inside a Muse repository. 3 API error. """ proposal_id_or_prefix: str = args.proposal_id json_output: bool = args.json_output hub_url, identity = _get_hub_and_identity(hub_url_override=args.hub) repo_id = _resolve_repo_id(hub_url, identity) proposal_id = _resolve_proposal_id(hub_url, identity, repo_id, proposal_id_or_prefix) data = _hub_api( hub_url, identity, "GET", f"/api/repos/{repo_id}/proposals/{proposal_id}/comments", ) if json_output: print(json.dumps(data)) return comments = data.get("comments", []) total = data.get("total", len(comments)) if not comments: print(f"No comments on proposal {sanitize_display(proposal_id[:8])}.", file=sys.stderr) return print(f"Comments ({total}):", file=sys.stderr) for c in comments: author = sanitize_display(str(c.get("author", ""))) body = sanitize_display(str(c.get("body", "")))[:80] print(f" [{author}] {body}") for r in c.get("replies", []): r_author = sanitize_display(str(r.get("author", ""))) r_body = sanitize_display(str(r.get("body", "")))[:70] print(f" ↳ [{r_author}] {r_body}") def run_proposal_reviewer_request(args: argparse.Namespace) -> None: """Request one or more reviewers on a proposal. Reviewer handles are passed as positional arguments:: muse hub proposal reviewer request af54753d alice bob muse hub proposal reviewer request af54753d alice --json Exit codes: 0 Success. 1 Auth error or not authorized. 2 Not inside a Muse repository. 3 API error. """ proposal_id_or_prefix: str = args.proposal_id reviewers: list[str] = args.reviewers json_output: bool = args.json_output hub_url, identity = _get_hub_and_identity(hub_url_override=args.hub) repo_id = _resolve_repo_id(hub_url, identity) proposal_id = _resolve_proposal_id(hub_url, identity, repo_id, proposal_id_or_prefix) data = _hub_api( hub_url, identity, "POST", f"/api/repos/{repo_id}/proposals/{proposal_id}/reviewers", body={"reviewers": reviewers}, ) if json_output: print(json.dumps(data)) return reviews = data.get("reviews", []) print(f"✅ Reviewers requested ({len(reviews)} total):", file=sys.stderr) for rv in reviews: handle = sanitize_display(str(rv.get("reviewerUsername", rv.get("reviewer", "")))) state = sanitize_display(str(rv.get("state", ""))) print(f" {handle:<32} {state}") def run_proposal_reviewer_remove(args: argparse.Namespace) -> None: """Remove a pending reviewer from a proposal. Only pending review requests can be removed — submitted reviews are immutable:: muse hub proposal reviewer remove af54753d alice muse hub proposal reviewer remove af54753d alice --json Exit codes: 0 Success. 1 Auth error, reviewer not found, or review already submitted. 2 Not inside a Muse repository. 3 API error. """ proposal_id_or_prefix: str = args.proposal_id reviewer: str = args.reviewer json_output: bool = args.json_output hub_url, identity = _get_hub_and_identity(hub_url_override=args.hub) repo_id = _resolve_repo_id(hub_url, identity) proposal_id = _resolve_proposal_id(hub_url, identity, repo_id, proposal_id_or_prefix) data = _hub_api( hub_url, identity, "DELETE", f"/api/repos/{repo_id}/proposals/{proposal_id}/reviewers/{reviewer}", ) if json_output: print(json.dumps(data)) return reviews = data.get("reviews", []) print( f"✅ Reviewer {sanitize_display(reviewer)} removed " f"({len(reviews)} reviewers remaining).", file=sys.stderr, ) def run_proposal_review_list(args: argparse.Namespace) -> None: """List all reviews for a proposal. Includes pending assignments and submitted reviews (approved, changes_requested, dismissed). Pass ``--state`` to filter:: muse hub proposal review list af54753d muse hub proposal review list af54753d --state approved --json Exit codes: 0 Success. 1 Auth error or proposal not found. 2 Not inside a Muse repository. 3 API error. """ proposal_id_or_prefix: str = args.proposal_id state: str | None = getattr(args, "state", None) or None json_output: bool = args.json_output hub_url, identity = _get_hub_and_identity(hub_url_override=args.hub) repo_id = _resolve_repo_id(hub_url, identity) proposal_id = _resolve_proposal_id(hub_url, identity, repo_id, proposal_id_or_prefix) path = f"/api/repos/{repo_id}/proposals/{proposal_id}/reviews" if state: path += f"?state={state}" data = _hub_api(hub_url, identity, "GET", path) if json_output: print(json.dumps(data)) return reviews = data.get("reviews", []) total = data.get("total", len(reviews)) if not reviews: print(f"No reviews on proposal {sanitize_display(proposal_id[:8])}.", file=sys.stderr) return print(f"Reviews ({total}):", file=sys.stderr) for rv in reviews: handle = sanitize_display(str(rv.get("reviewerUsername", rv.get("reviewer", "")))) s = sanitize_display(str(rv.get("state", ""))) print(f" {handle:<32} {s}") # ── Repo subcommand handlers ────────────────────────────────────────────────── def run_repo_create(args: argparse.Namespace) -> None: # noqa: C901 """Create a new repository on MuseHub. All local validation (name format, length) runs before any network I/O so errors are reported immediately without contacting the hub. The owner defaults to the authenticated identity's handle. Pass ``--owner`` to create under a different handle (requires the caller to have permission on the server). The repo is created with ``initialize=True`` by default so the default branch exists and the repo is immediately browsable and pushable. Pass ``--no-init`` to skip the initial commit (useful when you are about to push an existing history). JSON output (``--json``, stdout) -------------------------------- :: { "repo_id": "", "name": "", "owner": "", "slug": "", "visibility": "public" | "private", "description": "", "clone_url": "", "tags": ["", ...], "created_at": "" } Agent quickstart ---------------- :: muse hub repo create --name my-repo --json # → {"repo_id": "...", "slug": "my-repo", "clone_url": "...", ...} # Create private repo, push immediately: muse hub repo create --name my-repo --private --no-init --json muse push main Exit codes ---------- 0 Repo created. 1 Validation error, name conflict (409), or not authenticated. 2 Not inside a Muse repository (when hub URL is inferred from config). 3 API / network error. """ name: str = args.name.strip() owner_override: str = getattr(args, "owner", "") or "" description: str = getattr(args, "description", "") or "" visibility: str = "private" if getattr(args, "private", False) else "public" tags: list[str] = getattr(args, "tags", []) or [] initialize: bool = not getattr(args, "no_init", False) default_branch: str = getattr(args, "default_branch", "main") or "main" json_output: bool = args.json_output # ── Local validation — fail fast before any network I/O ────────────────── if not name: print("❌ Repo name must not be empty.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if len(name) > _MAX_REPO_NAME_LEN: print( f"❌ Repo name is too long ({len(name)} chars); " f"maximum is {_MAX_REPO_NAME_LEN}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if len(description) > _MAX_REPO_DESC_LEN: print( f"❌ Description is too long ({len(description)} chars); " f"maximum is {_MAX_REPO_DESC_LEN}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if visibility not in ("public", "private"): print( f"❌ Invalid visibility '{sanitize_display(visibility)}'. " "Use 'public' or 'private'.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if not default_branch.strip(): print("❌ Default branch name must not be empty.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # ── Network calls ───────────────────────────────────────────────────────── hub_url, identity = _get_hub_and_identity(hub_url_override=getattr(args, "hub", None)) # Resolve owner: use --owner if given, else fall back to authenticated handle. owner = owner_override.strip() or str(identity.get("handle", "")) if not owner: print( "❌ Could not determine owner. Pass --owner or authenticate with " "`muse auth register`.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) payload: _HubPayload = { "name": name, "owner": owner, "visibility": visibility, "description": description, "tags": tags, "initialize": initialize, "defaultBranch": default_branch, } # _hub_api exits on 4xx/5xx — but 409 Conflict deserves a friendlier message. # We intercept it by wrapping the call so we can re-raise with a clear hint. parsed_hub = urllib.parse.urlparse(hub_url) server_root = f"{parsed_hub.scheme}://{parsed_hub.netloc}" # Determine the correct API path. _hub_api prepends server_root internally. api_path = "/api/repos" try: data = _hub_api(hub_url, identity, "POST", api_path, body=payload) except SystemExit as exc: # _hub_api already printed the error; re-raise unless we can improve it. # We can't distinguish 409 at this point (it was already printed), so just propagate. raise exc slug = str(data.get("slug", name)) repo_id = str(data.get("repoId", data.get("repo_id", ""))) clone_url = str(data.get("cloneUrl", data.get("clone_url", ""))) created_at = str(data.get("createdAt", data.get("created_at", ""))) resp_tags: list[str] = [str(t) for t in data.get("tags", [])] if isinstance(data.get("tags"), list) else [] repo_url = f"{server_root}/{sanitize_display(owner)}/{sanitize_display(slug)}" if json_output: out: _RepoJson = { "repo_id": repo_id, "name": name, "owner": owner, "slug": slug, "visibility": visibility, "description": description, "clone_url": clone_url, "tags": resp_tags, "created_at": created_at, } print(json.dumps(out)) return print( f"✅ Repository created: {sanitize_display(owner)}/{sanitize_display(slug)}", file=sys.stderr, ) print(f" URL: {sanitize_display(repo_url)}", file=sys.stderr) if clone_url: print(f" Clone: {sanitize_display(clone_url)}", file=sys.stderr) print( f" Visibility: {sanitize_display(visibility)} " f"Branch: {sanitize_display(default_branch)} " f"Init: {'yes' if initialize else 'no'}", file=sys.stderr, ) print( f"\n To push an existing repo:\n" f" muse remote add origin {sanitize_display(repo_url)}\n" f" muse push origin {sanitize_display(default_branch)}", file=sys.stderr, ) # ── Repo management handlers ───────────────────────────────────────────────── def run_repo_delete(args: argparse.Namespace) -> None: """Soft-delete a repository on MuseHub. Only the repository owner may delete. All data is retained for audit; subsequent reads return 404. Requires explicit confirmation via ``--yes``. When ``target`` is provided it may be ``OWNER/SLUG`` or a repo UUID, allowing bulk deletion without being inside the target repo's directory. When omitted the repo is resolved from the current directory's remote config (original behaviour):: muse hub repo delete --yes muse hub repo delete gabriel/my-repo --yes muse hub repo delete a3f2c9d1-... --yes --json Exit codes: 0 Deleted successfully. 1 Auth error or not authorized. 2 Not inside a Muse repository (and no target given). 3 API error. """ yes: bool = args.yes json_output: bool = args.json_output target: str | None = getattr(args, "target", None) if not yes: print( "❌ Pass --yes to confirm deletion. This action cannot be undone.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) hub_url, identity = _get_hub_and_identity(hub_url_override=args.hub) if target is not None: if "/" in target: # OWNER/SLUG — resolve to repo_id via the show endpoint owner, slug = target.split("/", 1) resp = _hub_api(hub_url, identity, "GET", f"/api/{owner}/{slug}") repo_id = resp.get("repoId") or resp.get("repo_id", "") else: # Treat as a bare repo_id UUID repo_id = target else: repo_id = _resolve_repo_id(hub_url, identity) _hub_api(hub_url, identity, "DELETE", f"/api/repos/{repo_id}") if json_output: print(json.dumps({"deleted": True, "repo_id": repo_id})) return print(f"✅ Repository {sanitize_display(repo_id[:8])} deleted.", file=sys.stderr) def run_repo_update(args: argparse.Namespace) -> None: """Show or update repository settings on MuseHub. Without flags, prints the current settings. Pass update flags to patch specific fields — only provided flags are written:: muse hub repo settings muse hub repo settings --visibility private muse hub repo settings --description "My project" --json Exit codes: 0 Success. 1 Auth error or not authorized. 2 Not inside a Muse repository. 3 API error. """ json_output: bool = args.json_output # Build patch body from flags patch: dict[str, object] = {} for field in ("name", "description", "visibility", "default_branch", "homepage_url"): val = getattr(args, field, None) if val is not None: patch[field] = val for bool_field in ("has_issues", "has_wiki", "allow_merge_commit", "allow_squash_merge", "allow_rebase_merge", "delete_branch_on_merge"): val = getattr(args, bool_field, None) if val is not None: patch[bool_field] = val hub_url, identity = _get_hub_and_identity(hub_url_override=args.hub) repo_id = _resolve_repo_id(hub_url, identity) if patch: data = _hub_api(hub_url, identity, "PATCH", f"/api/repos/{repo_id}/settings", body=patch) else: data = _hub_api(hub_url, identity, "GET", f"/api/repos/{repo_id}/settings") if json_output: print(json.dumps(data)) return print(f"Name: {sanitize_display(str(data.get('name', '')))}", file=sys.stderr) print(f"Description: {sanitize_display(str(data.get('description', '')))}", file=sys.stderr) print(f"Visibility: {sanitize_display(str(data.get('visibility', '')))}", file=sys.stderr) print(f"Branch: {sanitize_display(str(data.get('defaultBranch', '')))}", file=sys.stderr) print(f"Issues: {data.get('hasIssues', True)}", file=sys.stderr) print(f"Wiki: {data.get('hasWiki', False)}", file=sys.stderr) def run_repo_list(args: argparse.Namespace) -> None: """List repositories owned by or collaborated on by the authenticated user. Results are ordered newest-first. Pass ``--limit`` to control page size and ``--cursor`` to page through results using the ``next_cursor`` value from a previous call. JSON output (``--json``, stdout) -------------------------------- :: { "total": 42, "next_cursor": " | null", "repos": [ { "repo_id": "", "name": "", "owner": "", "slug": "", "visibility": "public" | "private", "description": "", "tags": ["", ...], "default_branch": "", "created_at": "", "pushed_at": "" }, ... ] } Agent quickstart ---------------- :: muse hub repo list --json muse hub repo list --limit 50 --json muse hub repo list --cursor "" --json Exit codes ---------- 0 Success. 1 Not authenticated. 2 Not inside a Muse repository (when hub URL is inferred from config). 3 API / network error. """ limit: int = getattr(args, "limit", 20) or 20 cursor: str | None = getattr(args, "cursor", None) or None json_output: bool = args.json_output hub_url, identity = _get_hub_and_identity(hub_url_override=getattr(args, "hub", None)) params: list[str] = [f"limit={min(max(1, limit), 100)}"] if cursor: params.append(f"cursor={cursor}") api_path = "/api/repos?" + "&".join(params) data = _hub_api(hub_url, identity, "GET", api_path) repos: list[dict[str, object]] = data.get("repos", []) # type: ignore[assignment] total: int = int(data.get("total", len(repos))) next_cursor: str | None = data.get("nextCursor") or data.get("next_cursor") # type: ignore[assignment] if json_output: out: dict[str, object] = { "total": total, "next_cursor": next_cursor, "repos": [ { "repo_id": str(r.get("repoId", r.get("repo_id", ""))), "name": str(r.get("name", "")), "owner": str(r.get("owner", "")), "slug": str(r.get("slug", "")), "visibility": str(r.get("visibility", "public")), "description": str(r.get("description", "")), "tags": r.get("tags", []), "default_branch": str(r.get("defaultBranch", r.get("default_branch", "main"))), "created_at": str(r.get("createdAt", r.get("created_at", ""))), "pushed_at": str(r.get("pushedAt", r.get("pushed_at", ""))), } for r in repos if isinstance(r, dict) ], } print(json.dumps(out)) return if not repos: print("No repositories found.", file=sys.stderr) return print(f"Repositories ({total} total):", file=sys.stderr) for r in repos: if not isinstance(r, dict): continue owner = r.get("owner", "") slug = r.get("slug", r.get("name", "")) visibility = r.get("visibility", "public") desc = r.get("description", "") marker = "🔒 " if visibility == "private" else " " print(f" {marker}{sanitize_display(str(owner))}/{sanitize_display(str(slug))}", file=sys.stderr) if desc: print(f" {sanitize_display(str(desc))[:72]}", file=sys.stderr) if next_cursor: print(f"\n (more results — pass --cursor {sanitize_display(str(next_cursor))} to continue)", file=sys.stderr) def run_repo_show(args: argparse.Namespace) -> None: """Show metadata for a single MuseHub repository. Resolves by ``owner/slug`` (positional ``OWNER/SLUG`` argument) or by the hub remote config of the current directory when no argument is given. JSON output (``--json``, stdout) -------------------------------- :: { "repo_id": "", "name": "", "owner": "", "slug": "", "visibility": "public" | "private", "description": "", "tags": ["", ...], "default_branch": "", "clone_url": "", "created_at": "", "updated_at": "", "pushed_at": "" } Agent quickstart ---------------- :: muse hub repo show gabriel/my-repo --json muse hub repo show --json # resolves from current repo's hub config Exit codes ---------- 0 Success. 1 Not authenticated or not found (404). 2 Not inside a Muse repository (when hub URL is inferred from config). 3 API / network error. """ target: str | None = getattr(args, "target", None) json_output: bool = args.json_output hub_url, identity = _get_hub_and_identity(hub_url_override=getattr(args, "hub", None)) if target and "/" in target: # owner/slug form — use the human-readable URL parts = target.split("/", 1) owner_part = parts[0].strip() slug_part = parts[1].strip() api_path = f"/api/{owner_part}/{slug_part}" else: # Resolve from the current repo's hub remote config repo_id = _resolve_repo_id(hub_url, identity) api_path = f"/api/repos/{repo_id}" data = _hub_api(hub_url, identity, "GET", api_path) repo_id_val = str(data.get("repoId", data.get("repo_id", ""))) name = str(data.get("name", "")) owner = str(data.get("owner", "")) slug = str(data.get("slug", "")) visibility = str(data.get("visibility", "public")) description = str(data.get("description", "")) tags: list[str] = [str(t) for t in data.get("tags", [])] if isinstance(data.get("tags"), list) else [] default_branch = str(data.get("defaultBranch", data.get("default_branch", "main"))) clone_url = str(data.get("cloneUrl", data.get("clone_url", ""))) created_at = str(data.get("createdAt", data.get("created_at", ""))) updated_at = str(data.get("updatedAt", data.get("updated_at", ""))) pushed_at = str(data.get("pushedAt", data.get("pushed_at", ""))) if json_output: out: dict[str, object] = { "repo_id": repo_id_val, "name": name, "owner": owner, "slug": slug, "visibility": visibility, "description": description, "tags": tags, "default_branch": default_branch, "clone_url": clone_url, "created_at": created_at, "updated_at": updated_at, "pushed_at": pushed_at, } print(json.dumps(out)) return visibility_icon = "🔒 private" if visibility == "private" else "public" print(f" {sanitize_display(owner)}/{sanitize_display(slug)} [{visibility_icon}]", file=sys.stderr) if description: print(f" {sanitize_display(description)}", file=sys.stderr) if tags: print(f" Tags: {', '.join(sanitize_display(t) for t in tags)}", file=sys.stderr) print(f" Branch: {sanitize_display(default_branch)}", file=sys.stderr) if clone_url: print(f" Clone: {sanitize_display(clone_url)}", file=sys.stderr) if pushed_at: print(f" Last push: {sanitize_display(pushed_at)}", file=sys.stderr) elif created_at: print(f" Created: {sanitize_display(created_at)}", file=sys.stderr) def run_repo_transfer_ownership(args: argparse.Namespace) -> None: """Transfer ownership of a repository to another user. Only the current owner may initiate. After transfer the calling user loses owner privileges immediately. Requires ``--new-owner``:: muse hub repo transfer --new-owner alice muse hub repo transfer --new-owner alice --json Exit codes: 0 Transfer succeeded. 1 Auth error or not authorized. 2 Not inside a Muse repository. 3 API error. """ new_owner: str = args.new_owner json_output: bool = args.json_output hub_url, identity = _get_hub_and_identity(hub_url_override=args.hub) repo_id = _resolve_repo_id(hub_url, identity) data = _hub_api( hub_url, identity, "POST", f"/api/repos/{repo_id}/transfer", body={"newOwner": new_owner}, ) if json_output: print(json.dumps(data)) return new_owner_out = sanitize_display(str(data.get("ownerUserId", new_owner))) print(f"✅ Repository transferred to {new_owner_out}.", file=sys.stderr) # ── Issue subcommand handlers ───────────────────────────────────────────────── def _resolve_hub_override(args: argparse.Namespace) -> str | None: """Return the effective hub URL override, resolving --repo as an alias for --hub. ``--repo owner/repo`` is the idiomatic git/gh CLI style. When provided, the hub base URL is read from the repo config and the owner/repo path is appended, producing a full ``--hub``-compatible URL. """ if args.hub: return args.hub repo_arg: str | None = getattr(args, "repo", None) if not repo_arg: return None # Read the hub base URL from config (without owner/repo path). root = find_repo_root() if root is not None: configured = get_hub_url(root) if configured: # Strip any trailing owner/repo path — keep only scheme://host[:port]. parsed = urllib.parse.urlparse(configured) base = f"{parsed.scheme}://{parsed.netloc}" return f"{base}/{repo_arg.strip('/')}" # Fallback: no config — caller will get a helpful error from _get_hub_and_identity. return None def run_issue_create(args: argparse.Namespace) -> None: """Open a new issue on MuseHub. Validates title length and emptiness before making any network calls. Prints the issue URL to stdout in text mode (scriptable); use ``--json`` for the full API response:: muse hub issue create --title "feat: add thing" muse hub issue create --title "fix: bug" --body "details" --label bug --json muse hub issue create --hub http://host.docker.internal:10003/owner/repo \\ --title "feat: X" --label "phase/1" Exit codes: 0 Issue created successfully. 1 Validation error or not authenticated. 2 Not inside a Muse repository. 3 API error. """ title: str = args.title body: str = args.body labels: list[str] = args.labels symbol_anchors: list[str] = args.symbol_anchors commit_anchors: list[str] = args.commit_anchors agent_id: str = args.agent_id model_id: str = args.model_id json_output: bool = args.json_output # ── Local validation first — fail fast before any network I/O ──────────── if not title.strip(): print("❌ Issue title must not be empty.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if len(title) > _MAX_ISSUE_TITLE_LEN: print( f"❌ Issue title is too long ({len(title)} chars); " f"maximum is {_MAX_ISSUE_TITLE_LEN}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # ── Network calls ───────────────────────────────────────────────────────── hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) payload: _ProposalPayload = { "title": title, "body": body, "labels": labels, "symbol_anchors": symbol_anchors, "commit_anchors": commit_anchors, "agent_id": agent_id, "model_id": model_id, } data = _hub_api(hub_url, identity, "POST", f"/api/repos/{repo_id}/issues", body=payload) _num = data.get("number") try: number = int(_num) if _num is not None else 0 except (ValueError, TypeError): number = 0 parsed = urllib.parse.urlparse(hub_url) parts = parsed.path.strip("/").split("/") if len(parts) >= 2: owner, slug = parts[0], parts[1] server_root = f"{parsed.scheme}://{parsed.netloc}" issue_url = f"{server_root}/{owner}/{slug}/issues/{number}" else: issue_url = str(data.get("issueId", "")) if json_output: print(json.dumps(data)) return print(f"✅ Issue #{number} created.", file=sys.stderr) print(sanitize_display(issue_url)) def run_issue_update(args: argparse.Namespace) -> None: """Update an existing issue on MuseHub. Updates title and/or body; omitted fields are left unchanged. All validation runs before any network I/O: - Issue number must be a positive integer. - If ``--title`` is given it must be non-empty and ≤ ``_MAX_ISSUE_TITLE_LEN`` chars. - At least one of ``--title`` or ``--body`` must be provided. :: muse hub issue edit 42 --body "updated description" muse hub issue edit 42 --title "new title" --body "new body" --json Exit codes: 0 Issue updated successfully. 1 Validation error or not authenticated. 2 Not inside a Muse repository. 3 API error. """ number: int = args.number json_output: bool = args.json_output # ── Local validation first — fail fast before any network I/O ──────────── if number <= 0: print(f"❌ Issue number must be a positive integer, got {number}.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) patch: _ProposalPayload = {} if args.title is not None: if not args.title.strip(): print("❌ Issue title must not be empty.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if len(args.title) > _MAX_ISSUE_TITLE_LEN: print( f"❌ Issue title is too long ({len(args.title)} chars); " f"maximum is {_MAX_ISSUE_TITLE_LEN}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) patch["title"] = args.title if args.body is not None: patch["body"] = args.body if args.symbol_anchors is not None: patch["symbol_anchors"] = args.symbol_anchors if args.commit_anchors is not None: patch["commit_anchors"] = args.commit_anchors if not patch: print( "❌ Nothing to update — provide --title, --body, --anchor, and/or --commit-anchor.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # ── Network calls ───────────────────────────────────────────────────────── hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) data = _hub_api( hub_url, identity, "PATCH", f"/api/repos/{repo_id}/issues/{number}", body=patch, ) if json_output: print(json.dumps(data)) return print(f"✅ Issue #{number} updated.", file=sys.stderr) def run_issue_read(args: argparse.Namespace) -> None: """Fetch a single issue by its per-repo number. Prints a one-line summary to stderr in text mode; use ``--json`` for the full API response:: muse hub issue read 42 --json | jq '{number,title,state}' Exit codes: 0 Issue found and printed. 1 Not authenticated. 2 Not inside a Muse repository. 3 API error (includes 404 if the issue does not exist). """ number: int = args.number json_output: bool = args.json_output if number <= 0: print(f"❌ Issue number must be a positive integer, got {number}.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) data = _hub_api(hub_url, identity, "GET", f"/api/repos/{repo_id}/issues/{number}") if json_output: print(json.dumps(data)) return title = sanitize_display(str(data.get("title", "(no title)"))) state = sanitize_display(str(data.get("state", "?"))) state_icon = {"open": "🟢", "closed": "⛔"}.get(str(data.get("state", "")), "❓") author = sanitize_display(str(data.get("author", ""))) created = sanitize_display(str(data.get("createdAt", ""))[:10]) print(f"\n {state_icon} [{state.upper()}] #{number} — {title}", file=sys.stderr) print(f" Author: {author} Created: {created}", file=sys.stderr) def run_issue_list(args: argparse.Namespace) -> None: """List issues for the current repo. Returns open issues by default; filter with ``--state`` and ``--label``. User inputs are sanitized before use: - ``--state`` is constrained by argparse ``choices`` and URL-encoded. - ``--label`` is length-capped locally then percent-encoded in the query string — no raw label value is ever interpolated into the URL directly. - ``--limit`` is clamped to a safe integer range by :func:`clamp_int`. :: muse hub issue list --json muse hub issue list --state closed --label bug --json Exit codes: 0 Success (including empty list). 1 Validation error or not authenticated. 2 Not inside a Muse repository. 3 API error. """ state: str = args.state label: str | None = args.label limit: int = clamp_int(args.limit, 1, 10000, "limit") json_output: bool = args.json_output # ── Local validation first — fail fast before any network I/O ──────────── if label is not None and len(label) > _MAX_ISSUE_LABEL_LEN: print( f"❌ Label is too long ({len(label)} chars); " f"maximum is {_MAX_ISSUE_LABEL_LEN}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # ── Network calls ───────────────────────────────────────────────────────── hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) # URL-encode state even though argparse already constrains it to a known # set — defense-in-depth against any future code path that bypasses argparse. params = f"?state={urllib.parse.quote(state, safe='')}&per_page={limit}" if label: params += f"&label={urllib.parse.quote(label, safe='')}" data = _hub_api(hub_url, identity, "GET", f"/api/repos/{repo_id}/issues{params}") issues_val = data.get("issues", []) issues: list[_HubApiResponse] = ( [r for r in issues_val if isinstance(r, dict)] if isinstance(issues_val, list) else [] ) if json_output: print(json.dumps(issues)) return if not issues: print(f" No issues found (state={sanitize_display(state)}).", file=sys.stderr) return print(f"\n Issues — {sanitize_display(hub_url)} ({len(issues)} shown)", file=sys.stderr) print(" " + "─" * 60, file=sys.stderr) for issue in issues: _state = str(issue.get("state", "?")) state_icon = {"open": "🟢", "closed": "⛔"}.get(_state, "❓") _num = sanitize_display(str(issue.get("number", "?"))) _title = sanitize_display(str(issue.get("title", "(no title)"))) _author = sanitize_display(str(issue.get("author", ""))) print(f" {state_icon} #{_num} {_title} [{_author}]", file=sys.stderr) print("", file=sys.stderr) def run_issue_close(args: argparse.Namespace) -> None: """Close an open issue on MuseHub. Idempotent — closing an already-closed issue returns the issue unchanged:: muse hub issue close 42 muse hub issue close 42 --json Exit codes: 0 Issue closed successfully. 1 Not authenticated. 2 Not inside a Muse repository. 3 API error (includes 404 if the issue does not exist). """ number: int = args.number json_output: bool = args.json_output if number <= 0: print(f"❌ Issue number must be a positive integer, got {number}.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) data = _hub_api( hub_url, identity, "POST", f"/api/repos/{repo_id}/issues/{number}/close", ) if json_output: print(json.dumps(data)) return print(f"✅ Issue #{number} closed.", file=sys.stderr) def run_issue_reopen(args: argparse.Namespace) -> None: """Reopen a closed issue on MuseHub. Idempotent — reopening an already-open issue returns the issue unchanged:: muse hub issue reopen 42 muse hub issue reopen 42 --json Exit codes: 0 Issue reopened successfully. 1 Not authenticated. 2 Not inside a Muse repository. 3 API error (includes 404 if the issue does not exist). """ number: int = args.number json_output: bool = args.json_output if number <= 0: print(f"❌ Issue number must be a positive integer, got {number}.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) data = _hub_api( hub_url, identity, "POST", f"/api/repos/{repo_id}/issues/{number}/reopen", ) if json_output: print(json.dumps(data)) return print(f"✅ Issue #{number} reopened.", file=sys.stderr) def run_issue_comment(args: argparse.Namespace) -> None: """Post a Markdown comment on an issue on MuseHub. Returns the updated comment list (all comments on the issue):: muse hub issue comment 42 --body 'Fixed in abc123' muse hub issue comment 42 --body 'see also #43' --json Exit codes: 0 Comment posted successfully. 1 Validation or auth error. 2 Not inside a Muse repository. 3 API error (includes 404 if the issue does not exist). """ number: int = args.number body: str = args.body json_output: bool = args.json_output if number <= 0: print(f"❌ Issue number must be a positive integer, got {number}.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if not body.strip(): print("❌ Comment body must not be empty.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if len(body) > _MAX_ISSUE_COMMENT_LEN: print( f"❌ Comment body is too long ({len(body)} chars); " f"maximum is {_MAX_ISSUE_COMMENT_LEN}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) payload: _ProposalPayload = {"body": body} data = _hub_api( hub_url, identity, "POST", f"/api/repos/{repo_id}/issues/{number}/comments", body=payload, ) if json_output: print(json.dumps(data)) return comments_val = data.get("comments", []) count = len(comments_val) if isinstance(comments_val, list) else 0 print(f"✅ Comment posted on issue #{number} ({count} comment(s) total).", file=sys.stderr) def run_issue_comment_delete(args: argparse.Namespace) -> None: """Soft-delete a comment from an issue on MuseHub. Deleted comments are hidden from list results but preserved in the audit log. Requires write/admin access or repo ownership:: muse hub issue comment-delete 42 --comment-id muse hub issue comment-delete 42 --comment-id --json Exit codes: 0 Comment deleted successfully. 1 Validation or auth error. 2 Not inside a Muse repository. 3 API error (includes 404 if the comment does not exist). """ number: int = args.number comment_id: str = args.comment_id json_output: bool = args.json_output if number <= 0: print(f"❌ Issue number must be a positive integer, got {number}.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if not comment_id.strip(): print("❌ --comment-id must not be empty.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) _hub_api( hub_url, identity, "DELETE", f"/api/repos/{repo_id}/issues/{number}/comments/{comment_id}", ) if json_output: print(json.dumps({"deleted": True, "comment_id": comment_id})) return print(f"✅ Comment {comment_id} deleted from issue #{number}.", file=sys.stderr) def run_issue_assign(args: argparse.Namespace) -> None: """Assign or unassign a collaborator on an issue on MuseHub. Pass an empty string for ``--assignee`` to clear the current assignee:: muse hub issue assign 42 --assignee gabriel muse hub issue assign 42 --assignee '' # unassign muse hub issue assign 42 --assignee gabriel --json Exit codes: 0 Assignee updated successfully. 1 Not authenticated. 2 Not inside a Muse repository. 3 API error (includes 404 if the issue does not exist). """ number: int = args.number assignee: str = args.assignee json_output: bool = args.json_output if number <= 0: print(f"❌ Issue number must be a positive integer, got {number}.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) payload: dict[str, str | None] = {"assignee": assignee if assignee else None} data = _hub_api( hub_url, identity, "POST", f"/api/repos/{repo_id}/issues/{number}/assign", body=payload, ) if json_output: print(json.dumps(data)) return if assignee: print(f"✅ Issue #{number} assigned to {assignee}.", file=sys.stderr) else: print(f"✅ Issue #{number} unassigned.", file=sys.stderr) def run_issue_label(args: argparse.Namespace) -> None: """Manage labels on an issue on MuseHub. Use ``--set`` to replace the entire label list, or ``--remove`` to strip a single label without affecting others:: muse hub issue label 42 --set bug enhancement muse hub issue label 42 --remove bug muse hub issue label 42 --set bug --json Exit codes: 0 Labels updated successfully. 1 Not authenticated. 2 Not inside a Muse repository. 3 API error (includes 404 if the issue does not exist). """ number: int = args.number set_labels: list[str] | None = args.set_labels remove_label: str | None = args.remove_label json_output: bool = args.json_output if number <= 0: print(f"❌ Issue number must be a positive integer, got {number}.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) if set_labels is not None: label_body: _ProposalPayload = {"labels": set_labels} data = _hub_api( hub_url, identity, "POST", f"/api/repos/{repo_id}/issues/{number}/labels", body=label_body, ) if json_output: print(json.dumps(data)) return print(f"✅ Issue #{number} labels set to {set_labels}.", file=sys.stderr) else: # remove_label is guaranteed non-None (mutually exclusive group) label_name: str = remove_label or "" data = _hub_api( hub_url, identity, "DELETE", f"/api/repos/{repo_id}/issues/{number}/labels/{urllib.parse.quote(label_name, safe='')}", ) if json_output: print(json.dumps(data)) return print(f"✅ Label '{label_name}' removed from issue #{number}.", file=sys.stderr) # ── Label subcommand handlers ───────────────────────────────────────────────── def _validate_hex_color(color: str) -> bool: """Return True if *color* is a valid 7-character hex color string (e.g. '#d73a4a'). Validates without importing ``re`` — checks length, ``#`` prefix, and hex digit range explicitly so there is no import overhead on every CLI invocation. """ if len(color) != 7 or color[0] != "#": return False try: int(color[1:], 16) except ValueError: return False return True def _lookup_label_by_name( hub_url: str, identity: IdentityEntry, repo_id: str, name: str, ) -> _LabelEntry | None: """Fetch the label list and return the entry whose name matches *name*. Returns ``None`` if no label with that name exists. Case-sensitive match. """ data = _hub_api(hub_url, identity, "GET", f"/api/repos/{repo_id}/labels") items_val = data.get("items", []) items: list[_LabelEntry] = ( [r for r in items_val if isinstance(r, dict)] if isinstance(items_val, list) else [] ) for item in items: if item.get("name") == name: return item return None def run_label_create(args: argparse.Namespace) -> None: """Create a new label on MuseHub. Validates name length and colour format locally before any network call. Prints the label_id to stdout in text mode; use ``--json`` for the full API response:: muse hub label create --name bug --color '#d73a4a' muse hub label create --name enhancement --color '#a2eeef' --json Exit codes: 0 Label created. 1 Validation error, conflict (name already exists), or not authenticated. 2 Not inside a Muse repository. 3 API error. """ name: str = args.name.strip() color: str = args.color.strip() description: str | None = args.description json_output: bool = args.json_output # ── Local validation — fail fast before any network I/O ────────────────── if not name: print("❌ Label name must not be empty.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if len(name) > _MAX_LABEL_NAME_LEN: print( f"❌ Label name is too long ({len(name)} chars); " f"maximum is {_MAX_LABEL_NAME_LEN}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if not _validate_hex_color(color): print( f"❌ Invalid colour '{sanitize_display(color)}'. " "Must be a 7-character hex string starting with '#' (e.g. '#d73a4a').", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if description is not None and len(description) > _MAX_LABEL_DESC_LEN: print( f"❌ Description is too long ({len(description)} chars); " f"maximum is {_MAX_LABEL_DESC_LEN}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # ── Network calls ───────────────────────────────────────────────────────── hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) payload: _HubPayload = {"name": name, "color": color} if description is not None: payload["description"] = description data = _hub_api(hub_url, identity, "POST", f"/api/repos/{repo_id}/labels", body=payload) if json_output: print(json.dumps(data)) return label_id = sanitize_display(str(data.get("label_id", ""))) print(f"✅ Label '{sanitize_display(name)}' created ({label_id}).", file=sys.stderr) print(label_id) def run_label_list(args: argparse.Namespace) -> None: """List all labels for the current repo. The list endpoint is public — no authentication required for public repos:: muse hub label list muse hub label list --json Exit codes: 0 Success (including empty list). 1 Not authenticated (private repo). 2 Not inside a Muse repository. 3 API error. """ json_output: bool = args.json_output hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) data = _hub_api(hub_url, identity, "GET", f"/api/repos/{repo_id}/labels") items_val = data.get("items", []) items: list[_LabelEntry] = ( [r for r in items_val if isinstance(r, dict)] if isinstance(items_val, list) else [] ) if json_output: print(json.dumps(items)) return if not items: print(" No labels defined in this repo.", file=sys.stderr) return print(f"\n Labels — {sanitize_display(hub_url)}", file=sys.stderr) print(" " + "─" * 50, file=sys.stderr) for lbl in items: _name = sanitize_display(str(lbl.get("name", "?"))) _color = sanitize_display(str(lbl.get("color", ""))) _desc = sanitize_display(str(lbl.get("description") or "")) suffix = f" {_desc}" if _desc else "" print(f" {_color} {_name}{suffix}", file=sys.stderr) print("", file=sys.stderr) def run_label_update(args: argparse.Namespace) -> None: """Update an existing label's name, colour, or description on MuseHub. Looks up the label by its current name, then sends a PATCH request with only the provided fields. At least one of --new-name, --new-color, or --new-description must be supplied:: muse hub label update --name bug --new-color '#b60205' muse hub label update --name bug --new-name bug-report --json Exit codes: 0 Label updated. 1 Validation error, label not found, or not authenticated. 2 Not inside a Muse repository. 3 API error. """ name: str = args.name.strip() new_name: str | None = args.new_name new_color: str | None = args.new_color new_description: str | None = args.new_description json_output: bool = args.json_output # ── Local validation ────────────────────────────────────────────────────── if not name: print("❌ Label name must not be empty.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if new_name is None and new_color is None and new_description is None: print( "❌ Provide at least one of --new-name, --new-color, --new-description.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if new_name is not None: new_name = new_name.strip() if not new_name: print("❌ New label name must not be empty.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if len(new_name) > _MAX_LABEL_NAME_LEN: print( f"❌ New name is too long ({len(new_name)} chars); " f"maximum is {_MAX_LABEL_NAME_LEN}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if new_color is not None: new_color = new_color.strip() if not _validate_hex_color(new_color): print( f"❌ Invalid colour '{sanitize_display(new_color)}'. " "Must be a 7-character hex string starting with '#' (e.g. '#d73a4a').", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if new_description is not None and len(new_description) > _MAX_LABEL_DESC_LEN: print( f"❌ Description is too long ({len(new_description)} chars); " f"maximum is {_MAX_LABEL_DESC_LEN}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # ── Network calls ───────────────────────────────────────────────────────── hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) # Look up label by name to get the UUID. label = _lookup_label_by_name(hub_url, identity, repo_id, name) if label is None: print( f"❌ Label '{sanitize_display(name)}' not found in this repo.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) label_id = str(label.get("label_id", "")) patch: _HubPayload = {} if new_name is not None: patch["name"] = new_name if new_color is not None: patch["color"] = new_color if new_description is not None: patch["description"] = new_description data = _hub_api( hub_url, identity, "PATCH", f"/api/repos/{repo_id}/labels/{urllib.parse.quote(label_id, safe='')}", body=patch, ) if json_output: print(json.dumps(data)) return display_name = sanitize_display(str(data.get("name", name))) print(f"✅ Label '{sanitize_display(name)}' updated → '{display_name}'.", file=sys.stderr) def run_label_delete(args: argparse.Namespace) -> None: """Delete a label from MuseHub and remove it from all issues and proposals. Looks up the label by name, then sends a DELETE request. This operation is irreversible:: muse hub label delete --name bug muse hub label delete --name obsolete-label --json Exit codes: 0 Label deleted. 1 Label not found, or not authenticated. 2 Not inside a Muse repository. 3 API error. """ name: str = args.name.strip() json_output: bool = args.json_output if not name: print("❌ Label name must not be empty.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) # Look up label by name to get the UUID. label = _lookup_label_by_name(hub_url, identity, repo_id, name) if label is None: print( f"❌ Label '{sanitize_display(name)}' not found in this repo.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) label_id = str(label.get("label_id", "")) # DELETE returns 204 with no body; _hub_api returns {} for empty responses. _hub_api( hub_url, identity, "DELETE", f"/api/repos/{repo_id}/labels/{urllib.parse.quote(label_id, safe='')}", ) if json_output: print(json.dumps({"deleted": True, "name": name, "label_id": label_id})) return print(f"✅ Label '{sanitize_display(name)}' deleted.", file=sys.stderr) # ── Collaborator commands ────────────────────────────────────────────────────── def run_collaborator_list(args: argparse.Namespace) -> None: """List all collaborators and their permission levels for a repository. Returns the list ordered by permission level:: muse hub collaborator list muse hub collaborator list --json Exit codes: 0 Success. 1 Auth error. 2 Not inside a Muse repository. 3 API error. """ json_output: bool = args.json_output hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) data = _hub_api(hub_url, identity, "GET", f"/api/repos/{repo_id}/collaborators") if json_output: print(json.dumps(data)) return collaborators = data.get("collaborators", []) if not collaborators: print("No collaborators.", file=sys.stderr) return for c in collaborators: handle = c.get("handle", "") perm = c.get("permission", "") print(f" {handle:<32} {perm}") def run_collaborator_invite(args: argparse.Namespace) -> None: """Invite a user as a collaborator on a repository. Requires admin or owner access:: muse hub collaborator invite carol --permission write muse hub collaborator invite carol --permission admin --json Exit codes: 0 Collaborator invited. 1 Auth or conflict error. 2 Not inside a Muse repository. 3 API error. """ handle: str = args.handle permission: str = args.permission json_output: bool = args.json_output if not handle.strip(): print("❌ Handle must not be empty.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) data = _hub_api( hub_url, identity, "POST", f"/api/repos/{repo_id}/collaborators", body={"handle": handle, "permission": permission}, ) if json_output: print(json.dumps(data)) return print(f"✅ Invited '{sanitize_display(handle)}' as {permission} collaborator.", file=sys.stderr) def run_collaborator_update_permission(args: argparse.Namespace) -> None: """Update a collaborator's permission level. Requires admin or owner access:: muse hub collaborator update carol --permission admin muse hub collaborator update carol --permission read --json Exit codes: 0 Permission updated. 1 Auth or not-found error. 2 Not inside a Muse repository. 3 API error. """ handle: str = args.handle permission: str = args.permission json_output: bool = args.json_output hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) data = _hub_api( hub_url, identity, "PUT", f"/api/repos/{repo_id}/collaborators/{handle}/permission", body={"permission": permission}, ) if json_output: print(json.dumps(data)) return print(f"✅ Updated '{sanitize_display(handle)}' permission to {permission}.", file=sys.stderr) def run_collaborator_remove(args: argparse.Namespace) -> None: """Remove a collaborator from a repository. Requires admin or owner access. The owner cannot be removed:: muse hub collaborator remove carol muse hub collaborator remove carol --json Exit codes: 0 Collaborator removed. 1 Auth or not-found error. 2 Not inside a Muse repository. 3 API error. """ handle: str = args.handle json_output: bool = args.json_output hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) _hub_api(hub_url, identity, "DELETE", f"/api/repos/{repo_id}/collaborators/{handle}") if json_output: print(json.dumps({"removed": True, "handle": handle})) return print(f"✅ Removed '{sanitize_display(handle)}' from collaborators.", file=sys.stderr) # ── Webhook commands ─────────────────────────────────────────────────────────── def run_webhook_create(args: argparse.Namespace) -> None: """Register a new webhook subscription for a repository. Requires write/admin access or repo ownership:: muse hub webhook create --url https://example.com/hook --events push release muse hub webhook create --url https://ci.example.com/hook --events push --json Exit codes: 0 Webhook registered. 1 Validation or auth error. 2 Not inside a Muse repository. 3 API error. """ url: str = args.url events: list[str] = args.events secret: str = args.secret json_output: bool = args.json_output if not url.strip(): print("❌ --url must not be empty.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if not events: print("❌ --events must include at least one event type.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) data = _hub_api( hub_url, identity, "POST", f"/api/repos/{repo_id}/webhooks", body={"url": url, "events": events, "secret": secret}, ) if json_output: print(json.dumps(data)) return webhook_id = data.get("webhookId", data.get("webhook_id", "")) print(f"✅ Webhook registered: {webhook_id}", file=sys.stderr) print(f" URL: {url}", file=sys.stderr) print(f" Events: {', '.join(events)}", file=sys.stderr) def run_webhook_list(args: argparse.Namespace) -> None: """List all webhook subscriptions for a repository. Authentication required:: muse hub webhook list muse hub webhook list --json Exit codes: 0 Success. 1 Auth error. 2 Not inside a Muse repository. 3 API error. """ json_output: bool = args.json_output hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) data = _hub_api(hub_url, identity, "GET", f"/api/repos/{repo_id}/webhooks") if json_output: print(json.dumps(data)) return webhooks = data.get("webhooks", []) if not webhooks: print("No webhooks registered.", file=sys.stderr) return for wh in webhooks: wid = wh.get("webhookId", wh.get("webhook_id", "")) wurl = wh.get("url", "") evts = ", ".join(wh.get("events", [])) print(f" {wid} {wurl} [{evts}]") def run_webhook_delete(args: argparse.Namespace) -> None: """Delete a webhook subscription and all its delivery history. Requires write/admin access or repo ownership:: muse hub webhook delete muse hub webhook delete --json Exit codes: 0 Webhook deleted. 1 Auth or not-found error. 2 Not inside a Muse repository. 3 API error. """ webhook_id: str = args.webhook_id json_output: bool = args.json_output hub_url, identity = _get_hub_and_identity(hub_url_override=_resolve_hub_override(args)) repo_id = _resolve_repo_id(hub_url, identity) _hub_api(hub_url, identity, "DELETE", f"/api/repos/{repo_id}/webhooks/{webhook_id}") if json_output: print(json.dumps({"deleted": True, "webhook_id": webhook_id})) return print(f"✅ Webhook {webhook_id} deleted.", file=sys.stderr) # ── Connection management commands ──────────────────────────────────────────── def run_connect(args: argparse.Namespace) -> None: """Attach this repository to a MuseHub instance. Writes ``[hub] url`` to ``.muse/config.toml``. Does **not** modify credentials — authenticate separately with ``muse auth register``. URL normalisation ----------------- - Bare hostnames (``musehub.ai``) are promoted to ``https://musehub.ai``. - Trailing slashes are stripped. - ``http://`` is rejected for non-loopback hosts; loopback addresses (``localhost``, ``127.0.0.1``, ``[::1]``) are accepted for local dev. - Disallowed schemes (``file://``, ``ftp://``, etc.) are rejected. Idempotent ---------- Re-connecting to the same hub URL is a no-op (no warning, no write). Connecting to a *different* hub prints a warning on stderr and overwrites the stored URL. Agent quickstart ---------------- :: muse hub connect https://musehub.ai --json && muse auth register --agent --json JSON output (``--json``, stdout) -------------------------------- :: { "status": "ok", "hub_url": "https://musehub.ai", ← normalised URL, no trailing slash "hostname": "musehub.ai", ← host[:port] display string "authenticated": true | false, ← true if identity stored "identity_name": "" | "", ← display name or empty string "identity_type": "human" | "agent" | "" ← identity type or empty string } All diagnostic messages (warnings, errors) always go to stderr. Exit codes ---------- 0 Connected successfully (or no-op re-connect to same hub). 1 Bad URL: disallowed scheme, http:// for non-loopback host. 2 Not inside a Muse repository. """ url: str = args.url json_output: bool = args.json_output root = find_repo_root() if root is None: print("❌ Not inside a Muse repository. Run `muse init` first.", file=sys.stderr) raise SystemExit(ExitCode.REPO_NOT_FOUND) try: normalised = _normalise_url(url) except ValueError as exc: print(f"❌ {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) from exc hostname = _hub_hostname(normalised) # Warn before overwriting an existing connection. existing = get_hub_url(root) if existing and existing != normalised: existing_host = _hub_hostname(existing) print( f"⚠️ This repo was connected to {sanitize_display(existing_host)}.\n" f" Switching to {sanitize_display(hostname)}.\n" f" Your credentials for {sanitize_display(existing_host)} remain " "in ~/.muse/identity.toml.\n" f" To remove them: muse auth logout --hub {sanitize_display(existing_host)}", file=sys.stderr, ) set_hub_url(normalised, root) identity = load_identity(normalised) authenticated = identity is not None identity_name = "" identity_type = "" if identity is not None: identity_name = str(identity.get("handle") or "") identity_type = str(identity.get("type") or "") if json_output: out: _ConnectJson = { "status": "ok", "hub_url": normalised, "hostname": hostname, "authenticated": authenticated, "identity_name": identity_name, "identity_type": identity_type, } print(json.dumps(out)) else: print(f"✅ Connected to {sanitize_display(hostname)}", file=sys.stderr) if authenticated: print( f" Authenticated as {sanitize_display(identity_type)} " f"'{sanitize_display(identity_name)}'", file=sys.stderr, ) else: print(" No identity stored yet — run: muse auth register", file=sys.stderr) def run_status(args: argparse.Namespace) -> None: """Show the hub connection and identity for this repository. Reads ``.muse/config.toml`` for the hub URL and ``~/.muse/identity.toml`` for the stored identity. Makes **no network calls**. ``--hub`` override ------------------ Pass ``--hub `` to inspect a hub URL that differs from the one stored in ``.muse/config.toml``. Useful for containerised agents that reach the hub at a different address (e.g. ``http://host.docker.internal:10003``). The override is not persisted. Agent quickstart ---------------- :: muse hub status --json || muse hub connect https://musehub.ai --json JSON output (``--json``, stdout) -------------------------------- All keys are always present — agents never receive a ``KeyError``:: { "hub_url": "https://musehub.ai", ← URL as stored in config "hostname": "musehub.ai", ← host[:port] display form "authenticated": true | false, "identity_type": "human" | "agent" | "", ← "" when not authenticated "identity_name": "" | "", "identity_id": "" | "", "capabilities": ["read:*", ...] | [] ← [] for humans / unauthenticated } All text output (labels, warnings, errors) goes to stderr. Exit codes ---------- 0 Status printed successfully. 1 No hub connected (no ``[hub] url`` in config and no ``--hub`` override). 2 Not inside a Muse repository. """ json_output: bool = args.json_output root = find_repo_root() if root is None: print("❌ Not inside a Muse repository.", file=sys.stderr) raise SystemExit(ExitCode.REPO_NOT_FOUND) hub_url = args.hub or get_hub_url(root) if hub_url is None: print("No hub connected.\nRun: muse hub connect ", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) hostname = _hub_hostname(hub_url) identity = load_identity(hub_url) authenticated = identity is not None identity_type = str(identity.get("type") or "") if identity else "" identity_name = str(identity.get("handle") or "") if identity else "" identity_id = str(identity.get("fingerprint") or "") if identity else "" capabilities: list[str] = list(identity.get("capabilities") or []) if identity else [] if json_output: out: _StatusJson = { "hub_url": hub_url, "hostname": hostname, "authenticated": authenticated, "identity_type": identity_type, "identity_name": identity_name, "identity_id": identity_id, "capabilities": capabilities, } print(json.dumps(out)) return print("", file=sys.stderr) print(" Hub", file=sys.stderr) print(f" URL: {sanitize_display(hub_url)}", file=sys.stderr) if not authenticated: print(" Auth: not authenticated — run `muse auth register`", file=sys.stderr) else: handle = identity.get("handle", "") if identity else "" fingerprint = identity.get("fingerprint", "") if identity else "" print(f" Type: {sanitize_display(identity_type) or 'unknown'}", file=sys.stderr) print(f" Name: {sanitize_display(identity_name) or '—'}", file=sys.stderr) print(f" ID: {sanitize_display(identity_id) or '—'}", file=sys.stderr) print( f" Auth: {'Ed25519 key set (handle: ' + handle + ')' if handle else 'not set — run muse auth keygen'}", file=sys.stderr, ) if capabilities: caps_display = " ".join(sanitize_display(str(c)) for c in capabilities) print(f" Caps: {caps_display}", file=sys.stderr) print("", file=sys.stderr) def run_disconnect(args: argparse.Namespace) -> None: """Remove the hub association from this repository. Removes ``[hub] url`` from ``.muse/config.toml``. Credentials in ``~/.muse/identity.toml`` are **preserved** — use ``muse auth logout`` to remove them as well. Makes no network calls. Idempotent ---------- Disconnecting when no hub is configured exits 0 with ``status: "nothing_to_do"`` — safe to call unconditionally in scripts. Agent quickstart ---------------- Full teardown (disconnect + revoke credentials):: muse hub disconnect --json | python3 -c " import json, subprocess, sys d = json.load(sys.stdin) if d['hub_url']: subprocess.run(['muse', 'auth', 'logout', '--hub', d['hub_url']], check=True) " JSON output (``--json``, stdout) -------------------------------- :: { "status": "ok" | "nothing_to_do", "hub_url": "" | "", ← full normalised URL; "" on nothing_to_do "hostname": "" | "" ← host[:port]; "" on nothing_to_do } All text (success messages, hints) goes to stderr. Exit codes ---------- 0 Disconnected successfully, or nothing was connected. 2 Not inside a Muse repository. """ json_output: bool = args.json_output root = find_repo_root() if root is None: print("❌ Not inside a Muse repository.", file=sys.stderr) raise SystemExit(ExitCode.REPO_NOT_FOUND) hub_url = get_hub_url(root) if hub_url is None: if json_output: out: _DisconnectJson = { "status": "nothing_to_do", "hub_url": "", "hostname": "", } print(json.dumps(out)) else: print("No hub connected — nothing to do.", file=sys.stderr) return hostname = _hub_hostname(hub_url) clear_hub_url(root) if json_output: result: _DisconnectJson = { "status": "ok", "hub_url": hub_url, "hostname": hostname, } print(json.dumps(result)) else: print(f"✅ Disconnected from {sanitize_display(hostname)}.", file=sys.stderr) print( " Credentials in ~/.muse/identity.toml are preserved.\n" f" To remove them too: muse auth logout --hub {sanitize_display(hub_url)}", file=sys.stderr, ) def run_ping(args: argparse.Namespace) -> None: """Test HTTP connectivity to the configured hub. Sends a ``GET /health`` request and reports the result. No authentication token is sent — the health endpoint is intentionally unauthenticated. HTTP redirects are refused (the hub URL in config should be the final destination). ``--hub`` override ------------------ Pass ``--hub `` to test a URL that differs from the one in config (e.g. for containerised agents: ``--hub http://host.docker.internal:10003``). The URL is not persisted. Agent quickstart ---------------- Health-check before any operation:: muse hub ping --json || { echo "hub unreachable"; exit 1; } Startup readiness loop:: until muse hub ping --json 2>/dev/null; do sleep 2; done JSON output (``--json``, stdout) -------------------------------- :: { "status": "ok" | "error", "hub_url": "", ← URL that was pinged "hostname": "", "reachable": true | false, "message": "HTTP 200 OK" | "" } All text output (progress, errors) goes to stderr. Exit codes ---------- 0 Hub reachable (HTTP 2xx). 1 No hub connected (no ``[hub] url`` in config and no ``--hub`` flag). 2 Not inside a Muse repository. 5 Hub unreachable (connection refused, timeout, non-2xx, bad response). """ json_output: bool = args.json_output root = find_repo_root() if root is None: print("❌ Not inside a Muse repository.", file=sys.stderr) raise SystemExit(ExitCode.REPO_NOT_FOUND) hub_url = args.hub or get_hub_url(root) if hub_url is None: print("No hub connected.\nRun: muse hub connect ", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) hostname = _hub_hostname(hub_url) if not json_output: print(f"Pinging {sanitize_display(hostname)}…", end="", flush=True, file=sys.stderr) reachable, message = _ping_hub(hub_url) if json_output: out: _PingJson = { "status": "ok" if reachable else "error", "hub_url": hub_url, "hostname": hostname, "reachable": reachable, "message": message, } print(json.dumps(out)) if not reachable: raise SystemExit(ExitCode.REMOTE_ERROR) else: if reachable: print(f" ✅ {sanitize_display(message)}", file=sys.stderr) else: print(f" ❌ {sanitize_display(message)}", file=sys.stderr) raise SystemExit(ExitCode.REMOTE_ERROR)