"""muse fetch — download commits, snapshots, and objects from a remote. Fetches the latest state of a remote branch without touching the local branch HEAD or working tree. After a successful fetch: - All new commits, snapshots, and objects from the remote are stored locally. - The remote tracking pointer ``.muse/remotes//`` is updated. Use ``muse pull`` to fetch *and* merge into the current branch, or run ``muse merge`` after fetching to integrate on your own schedule. Flags ----- ``--all`` Fetch every configured remote instead of just one. When combined with ``--branch``, that branch is fetched from every remote. ``--prune / -p`` After fetching, delete local remote-tracking refs (pointers under ``.muse/remotes//``) for branches that no longer exist on the remote. Mirrors ``git fetch --prune``. ``--dry-run / -n`` Show what would be fetched without writing anything. ``--tags`` Also fetch tags from the remote (default behaviour when tags exist). ``--no-tags`` Do not fetch tags from the remote. ``--format {text,json}`` / ``--json`` Emit a machine-readable JSON object on stdout instead of human text. Human-readable diagnostics always go to stderr regardless of format. JSON output schema ------------------ Always emits a single JSON object on stdout:: { "results": [ { "remote": "", "branch": "", "status": "fetched | up_to_date | dry_run | branch_missing", "commits_received": , "objects_written": , "head": " | null", "pruned": ["/", ...], "dry_run": false } ], "dry_run": false } Exit codes:: 0 — success (fetched, up_to_date, dry_run, or branch_missing + prune) 1 — remote not configured, network error, or no remotes when using --all """ from __future__ import annotations import argparse import json import logging import pathlib import sys from typing import TYPE_CHECKING, TypedDict from muse.cli.config import ( get_signing_identity, get_remote, get_remote_head, list_remotes, set_remote_head, ) from muse.core.errors import ExitCode from muse.core.pack import apply_pack from muse.core.repo import require_repo from muse.core.store import get_all_commits, read_current_branch from muse.core.transport import TransportError, make_transport, negotiate_have from muse.core.validation import sanitize_display from muse.core._types import BranchHeads if TYPE_CHECKING: from muse.core.transport import HttpTransport, LocalFileTransport logger = logging.getLogger(__name__) class _RemoteResultJson(TypedDict): """Per-remote/branch result inside a ``muse fetch --json`` response.""" remote: str branch: str status: str # "fetched" | "up_to_date" | "dry_run" | "branch_missing" commits_received: int objects_written: int head: str | None # remote commit-id after the fetch, null when not fetched pruned: list[str] # "/" strings for each pruned ref dry_run: bool class _FetchJson(TypedDict): """Top-level JSON object emitted by ``muse fetch --json``.""" results: list[_RemoteResultJson] dry_run: bool def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``muse fetch`` subcommand and all its flags.""" parser = subparsers.add_parser( "fetch", help="Download commits, snapshots, and objects from a remote.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "remote", nargs="?", default="origin", help="Remote name to fetch from (default: origin). Ignored when --all is set.", ) parser.add_argument( "--branch", "-b", default=None, help=( "Remote branch to fetch (default: current branch). " "When combined with --all, this branch is fetched from every remote." ), ) parser.add_argument( "--all", action="store_true", default=False, help="Fetch all configured remotes.", ) parser.add_argument( "--prune", "-p", action="store_true", default=False, help=( "Remove local remote-tracking refs for branches that no longer exist " "on the remote (mirrors git fetch --prune)." ), ) parser.add_argument( "--dry-run", "-n", action="store_true", default=False, dest="dry_run", help="Show what would be fetched without writing any objects or tracking refs.", ) # Tag handling flags — reserved for future use when tag storage is added. tag_group = parser.add_mutually_exclusive_group() tag_group.add_argument( "--tags", action="store_true", default=None, dest="tags", help="Fetch tags from the remote (default).", ) tag_group.add_argument( "--no-tags", action="store_false", dest="tags", help="Do not fetch tags from the remote.", ) fmt_group = parser.add_mutually_exclusive_group() fmt_group.add_argument( "--format", choices=["text", "json"], default="text", dest="format", help="Output format: 'text' (default) or 'json'.", ) fmt_group.add_argument( "--json", action="store_const", const="json", dest="format", help="Shorthand for --format json.", ) parser.set_defaults(func=run) def _stale_ref_names( root: pathlib.Path, remote: str, live_branch_heads: BranchHeads, ) -> list[str]: """Return branch names whose local tracking refs are absent from *live_branch_heads*. Branch names may contain slashes (e.g. ``feat/my-thing``), so the refs are stored as nested files under ``.muse/remotes//``. We walk the tree recursively and compute the relative path from ``refs_dir`` to get the full branch name to compare against *live_branch_heads*. Symlinks inside the refs directory are skipped to prevent path-traversal attacks via a malicious remote name or branch name. """ refs_dir = root / ".muse" / "remotes" / remote if not refs_dir.is_dir(): return [] stale: list[str] = [] for ref_file in refs_dir.rglob("*"): if ref_file.is_symlink() or not ref_file.is_file(): continue branch_name = str(ref_file.relative_to(refs_dir)) if branch_name not in live_branch_heads: stale.append(branch_name) return stale def _prune_stale_refs( root: pathlib.Path, remote: str, live_branch_heads: BranchHeads, *, dry_run: bool, ) -> list[str]: """Prune stale remote-tracking refs, returning the list of pruned branch names. Walks ``.muse/remotes//`` recursively (branch names may contain slashes and are stored as nested paths) and, unless *dry_run* is True, deletes any file whose relative path is not a key in *live_branch_heads*. Prints a ``- [deleted]`` or ``Would prune`` line for each, mirroring ``git fetch --prune`` output. All output goes to stderr so stdout stays clean for structured JSON. """ refs_dir = root / ".muse" / "remotes" / remote pruned: list[str] = [] for branch_name in sorted(_stale_ref_names(root, remote, live_branch_heads)): safe_ref = f"{sanitize_display(remote)}/{sanitize_display(branch_name)}" if dry_run: print(f" Would prune {safe_ref}", file=sys.stderr) else: ref_file = refs_dir / branch_name ref_file.unlink() # Remove empty parent directories left behind (e.g. feat/ after feat/my-thing). for parent in ref_file.parents: if parent == refs_dir: break try: parent.rmdir() except OSError: break logger.debug("🗑 Pruned stale tracking ref %s/%s", remote, branch_name) print(f" - [deleted] {safe_ref}", file=sys.stderr) pruned.append(f"{remote}/{branch_name}") return pruned def _fetch_one( root: pathlib.Path, remote: str, branch: str, *, prune: bool, dry_run: bool, ) -> _RemoteResultJson: """Fetch a single remote/branch pair. Returns a :class:`_RemoteResultJson` describing the outcome. Raises ``SystemExit`` on unrecoverable errors (unknown remote, network failure). Writes nothing when *dry_run* is True. MWP negotiation (depth-limited ``have`` list) is used to minimise the amount of history sent to the server, matching the behaviour of ``muse pull``. """ result: _RemoteResultJson = { "remote": remote, "branch": branch, "status": "fetched", "commits_received": 0, "objects_written": 0, "head": None, "pruned": [], "dry_run": dry_run, } url = get_remote(remote, root) if url is None: print( f"❌ Remote '{sanitize_display(remote)}' is not configured.", file=sys.stderr, ) print( f" Add it with: muse remote add {sanitize_display(remote)} ", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) token = get_signing_identity(root, remote_url=url) transport = make_transport(url) try: info = transport.fetch_remote_info(url, token) except TransportError as exc: print( f"❌ Cannot reach remote '{sanitize_display(remote)}': " f"{sanitize_display(str(exc))}", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) remote_commit_id = info["branch_heads"].get(branch) if remote_commit_id is None: if prune: # The branch we were tracking is gone from the remote. # Prune it (and any other stale refs) then return cleanly — # this mirrors `git fetch --prune` behaviour where a deleted # upstream branch produces " - [deleted] remote/branch" rather # than an error. pruned = _prune_stale_refs(root, remote, info["branch_heads"], dry_run=dry_run) result["status"] = "branch_missing" result["pruned"] = pruned return result available = ", ".join( sanitize_display(b) for b in sorted(info["branch_heads"]) ) print( f"❌ Branch '{sanitize_display(branch)}' does not exist on " f"remote '{sanitize_display(remote)}'.", file=sys.stderr, ) print(f" Available branches: {available}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) already_known = get_remote_head(remote, branch, root) if already_known == remote_commit_id: print( f"✅ {sanitize_display(remote)}/{sanitize_display(branch)} " f"is already up to date ({remote_commit_id[:8]})", file=sys.stderr, ) if prune and not dry_run: pruned = _prune_stale_refs(root, remote, info["branch_heads"], dry_run=False) result["pruned"] = pruned result["status"] = "up_to_date" result["head"] = remote_commit_id return result if dry_run: print( f" Would fetch {sanitize_display(remote)}/{sanitize_display(branch)} " f"→ {remote_commit_id[:8]}", file=sys.stderr, ) if prune: pruned = _prune_stale_refs(root, remote, info["branch_heads"], dry_run=True) result["pruned"] = pruned result["status"] = "dry_run" result["head"] = remote_commit_id return result # ── MWP negotiation: find minimal have-list before fetching ────────────── all_local = [c.commit_id for c in get_all_commits(root)] try: have_for_fetch = negotiate_have( transport, url, token, [remote_commit_id], all_local ) except TransportError: logger.debug("negotiate not supported — falling back to full have list") have_for_fetch = all_local print(f"Fetching {sanitize_display(remote)}/{sanitize_display(branch)} …", file=sys.stderr) try: bundle = transport.fetch_pack( url, token, want=[remote_commit_id], have=have_for_fetch ) except TransportError as exc: print(f"❌ Fetch failed: {sanitize_display(str(exc))}", file=sys.stderr) raise SystemExit(ExitCode.INTERNAL_ERROR) apply_result = apply_pack(root, bundle) set_remote_head(remote, branch, remote_commit_id, root) commits_received: int = apply_result["commits_written"] objects_written: int = apply_result["objects_written"] print( f"✅ Fetched {commits_received} commit(s), " f"{objects_written} new object(s) " f"from {sanitize_display(remote)}/{sanitize_display(branch)} " f"({remote_commit_id[:8]})", file=sys.stderr, ) if prune: pruned = _prune_stale_refs(root, remote, info["branch_heads"], dry_run=False) result["pruned"] = pruned result["commits_received"] = commits_received result["objects_written"] = objects_written result["head"] = remote_commit_id return result def run(args: argparse.Namespace) -> None: """Download commits, snapshots, and objects from a remote. Updates the remote tracking pointer but does NOT change the local branch HEAD or working tree. Run ``muse pull`` to fetch and merge in one step. When ``--all`` is passed every configured remote is fetched; the positional ``remote`` argument is ignored. When ``--branch`` is also given, that specific branch is fetched from every remote. When ``--prune`` is passed, stale remote-tracking refs (branches that no longer exist on the remote) are deleted after a successful fetch. JSON output (``--format json`` / ``--json``) always goes to stdout; human-readable progress and errors always go to stderr. """ root = require_repo() current_branch = read_current_branch(root) dry_run: bool = args.dry_run prune: bool = args.prune fmt: str = args.format if dry_run: print("(dry run — no objects or refs will be written)", file=sys.stderr) results: list[_RemoteResultJson] = [] if args.all: remotes = list_remotes(root) if not remotes: print("❌ No remotes configured.", file=sys.stderr) print(" Add one with: muse remote add ", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # --branch with --all: fetch the specified branch from every remote. branch: str = args.branch or current_branch for remote_cfg in remotes: result = _fetch_one( root, remote_cfg["name"], branch, prune=prune, dry_run=dry_run, ) results.append(result) else: remote: str = args.remote # Use the explicitly passed branch, or fall back to the current local branch. # Do NOT use get_upstream() here — that returns the remote *name*, not branch. branch_single: str = args.branch or current_branch result = _fetch_one(root, remote, branch_single, prune=prune, dry_run=dry_run) results.append(result) if fmt == "json": output: _FetchJson = {"results": results, "dry_run": dry_run} print(json.dumps(output))