ls_remote.py
python
| 1 | """muse plumbing ls-remote — list references on a remote repository. |
| 2 | |
| 3 | Plumbing command that contacts the remote and prints every branch and its |
| 4 | current commit ID without modifying any local state. Useful for scripting, |
| 5 | agent coordination, and pre-flight checks before push/pull. |
| 6 | |
| 7 | Output format (default):: |
| 8 | |
| 9 | <commit_id>\\t<branch> |
| 10 | |
| 11 | Output format (--json):: |
| 12 | |
| 13 | {"branches": {"main": "<commit_id>", ...}, "repo_id": "...", "domain": "..."} |
| 14 | |
| 15 | Plumbing contract |
| 16 | ----------------- |
| 17 | |
| 18 | - Exit 0: remote contacted, refs printed. |
| 19 | - Exit 1: remote not configured or URL looks invalid. |
| 20 | - Exit 3: transport error (network unreachable, HTTP error). |
| 21 | """ |
| 22 | |
| 23 | from __future__ import annotations |
| 24 | |
| 25 | import json |
| 26 | import logging |
| 27 | import pathlib |
| 28 | |
| 29 | import typer |
| 30 | |
| 31 | from muse.cli.config import get_auth_token, get_remote |
| 32 | from muse.core.errors import ExitCode |
| 33 | from muse.core.repo import find_repo_root |
| 34 | from muse.core.transport import HttpTransport, TransportError |
| 35 | |
| 36 | logger = logging.getLogger(__name__) |
| 37 | |
| 38 | app = typer.Typer() |
| 39 | |
| 40 | |
| 41 | @app.callback(invoke_without_command=True) |
| 42 | def ls_remote( |
| 43 | ctx: typer.Context, |
| 44 | remote_or_url: str = typer.Argument( |
| 45 | "origin", |
| 46 | help="Remote name (e.g. 'origin') or a full URL. Defaults to 'origin'.", |
| 47 | ), |
| 48 | output_json: bool = typer.Option( |
| 49 | False, "--json", help="Emit JSON for agent consumption." |
| 50 | ), |
| 51 | ) -> None: |
| 52 | """List branches and commit IDs on a remote. |
| 53 | |
| 54 | Contacts the remote and prints each branch HEAD without altering any local |
| 55 | state. Pass a remote name (configured via ``muse remote add``) or a full |
| 56 | URL. Use ``--json`` for structured output. |
| 57 | """ |
| 58 | root = find_repo_root(pathlib.Path.cwd()) |
| 59 | token: str | None = None |
| 60 | |
| 61 | url: str | None = None |
| 62 | if root is not None: |
| 63 | token = get_auth_token(root) |
| 64 | url = get_remote(remote_or_url, root) |
| 65 | |
| 66 | if url is None: |
| 67 | if remote_or_url.startswith("http://") or remote_or_url.startswith("https://"): |
| 68 | url = remote_or_url |
| 69 | else: |
| 70 | typer.echo( |
| 71 | f"❌ '{remote_or_url}' is not a configured remote and does not " |
| 72 | "look like a URL.", |
| 73 | err=True, |
| 74 | ) |
| 75 | typer.echo(" Configure it with: muse remote add <name> <url>", err=True) |
| 76 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 77 | |
| 78 | transport = HttpTransport() |
| 79 | try: |
| 80 | info = transport.fetch_remote_info(url, token) |
| 81 | except TransportError as exc: |
| 82 | typer.echo(f"❌ Cannot reach remote: {exc}", err=True) |
| 83 | raise typer.Exit(code=ExitCode.INTERNAL_ERROR) |
| 84 | |
| 85 | if output_json: |
| 86 | typer.echo( |
| 87 | json.dumps( |
| 88 | { |
| 89 | "repo_id": info["repo_id"], |
| 90 | "domain": info["domain"], |
| 91 | "default_branch": info["default_branch"], |
| 92 | "branches": info["branch_heads"], |
| 93 | }, |
| 94 | indent=2, |
| 95 | ) |
| 96 | ) |
| 97 | return |
| 98 | |
| 99 | if not info["branch_heads"]: |
| 100 | typer.echo("(no branches)") |
| 101 | return |
| 102 | |
| 103 | for branch, commit_id in sorted(info["branch_heads"].items()): |
| 104 | marker = " *" if branch == info["default_branch"] else "" |
| 105 | typer.echo(f"{commit_id}\t{branch}{marker}") |