"""muse clone — create a local copy of a remote Muse repository. Downloads the complete commit history, snapshots, and objects from a remote MuseHub repository into a new local directory. After cloning: - A full ``.muse/`` directory is created with the remote's repo_id and domain. - The ``origin`` remote is configured to point at the source URL. - The default branch is checked out into the working tree. Usage ----- muse clone Clone into a directory named after the last URL segment. muse clone Clone into a specific directory. muse clone --branch dev Clone and check out 'dev'. muse clone --dry-run Show what would happen without writing anything. muse clone --no-checkout Skip working-tree restore after cloning. muse clone --json Emit a machine-readable JSON result to stdout. Auth ---- Signing identities are read from ``~/.muse/identity.toml`` keyed by hostname. No signing identity is required for public repositories. JSON schema (``--json``) ------------------------ :: { "status": "cloned | dry_run | already_exists", "url": "", "directory": "", "branch": "", "commits_received": , "objects_written": , "head": " | null", "domain": "", "dry_run": false } Exit codes ---------- 0 — success (including dry-run) 1 — user error (target already exists, empty repository, unknown branch) 2 — internal / transport error """ from __future__ import annotations import argparse import datetime import json import logging import pathlib import shutil import sys import uuid from typing import TYPE_CHECKING, TypedDict from muse._version import __version__ as _SCHEMA_VERSION from muse.cli.config import get_signing_identity, set_remote, set_remote_head, set_upstream from muse.core.errors import ExitCode from muse.core.object_store import has_object, write_object from muse.core.pack import apply_pack from muse.core.store import ( read_commit, read_snapshot, write_head_branch, write_text_atomic, ) from muse.core.transport import TransportError, make_transport from muse.core.validation import sanitize_display from muse.core.workdir import apply_manifest type _RepoMeta = dict[str, str] if TYPE_CHECKING: from muse.core.pack import ApplyResult logger = logging.getLogger(__name__) # Canonical set of subdirectories — must match muse init's _INIT_SUBDIRS. _CLONE_SUBDIRS: tuple[str, ...] = ( "refs", "refs/heads", "objects", "commits", "snapshots", "tags", ) _DEFAULT_CONFIG = """\ [user] name = "" email = "" [remotes] [domain] # Domain-specific configuration keys depend on the active domain. """ class _CloneJson(TypedDict): """Stable JSON schema emitted by ``muse clone --json``.""" status: str # "cloned" | "dry_run" | "already_exists" url: str directory: str # resolved local path branch: str # branch checked out commits_received: int objects_written: int head: str | None # HEAD commit ID after clone, null on dry-run domain: str dry_run: bool def _infer_dir_name(url: str) -> str: """Derive a safe local directory name from the last non-empty segment of *url*. Strips query strings, fragments, and path-traversal components so that a crafted URL like ``http://evil.example.com/../../../../etc`` cannot escape the current working directory. """ # Drop fragment and query before splitting on path separators. stripped = url.split("#")[0].split("?")[0].rstrip("/") last = stripped.rsplit("/", 1)[-1] # pathlib.Path.name always strips leading dots and directory separators, # eliminating traversal attempts like ".." or "../../secret". safe = pathlib.PurePosixPath(last).name return safe if safe and safe not in (".", "..") else "muse-repo" def _init_muse_dir( target: pathlib.Path, repo_id: str, domain: str, default_branch: str, ) -> None: """Create the ``.muse/`` directory tree inside *target*. Uses the same subdirectory set as ``muse init`` so that every command that relies on the standard layout (tags, objects, etc.) works out of the box. """ muse_dir = target / ".muse" for subdir in _CLONE_SUBDIRS: (muse_dir / subdir).mkdir(parents=True, exist_ok=True) repo_meta: _RepoMeta = { "repo_id": repo_id, "schema_version": _SCHEMA_VERSION, "created_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), "domain": domain, } write_text_atomic(muse_dir / "repo.json", json.dumps(repo_meta, indent=2) + "\n") write_head_branch(muse_dir.parent, default_branch) write_text_atomic(muse_dir / "refs" / "heads" / default_branch, "") write_text_atomic(muse_dir / "config.toml", _DEFAULT_CONFIG) def _restore_working_tree(root: pathlib.Path, commit_id: str) -> None: """Restore the working tree to the snapshot referenced by *commit_id*. Logs a warning to stderr (rather than silently returning) if the commit or snapshot cannot be read — this surfaces bugs where apply_pack did not write the expected objects. """ commit = read_commit(root, commit_id) if commit is None: logger.warning( "⚠️ clone: commit %s not found after apply_pack — working tree not restored", commit_id[:8], ) return snap = read_snapshot(root, commit.snapshot_id) if snap is None: logger.warning( "⚠️ clone: snapshot %s not found after apply_pack — working tree not restored", commit.snapshot_id[:8], ) return apply_manifest(root, snap.manifest) def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``muse clone`` subcommand and all its flags.""" parser = subparsers.add_parser( "clone", help="Create a local copy of a remote Muse repository.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "url", help="URL of the remote Muse repository to clone.", ) parser.add_argument( "directory", nargs="?", default=None, help=( "Local directory to clone into. " "Defaults to the last path segment of the URL." ), ) parser.add_argument( "--branch", "-b", default=None, help="Branch to check out after cloning (default: remote default branch).", ) parser.add_argument( "--dry-run", "-n", action="store_true", default=False, dest="dry_run", help=( "Contact the remote and show what would be cloned without writing " "any files or creating any directories." ), ) parser.add_argument( "--no-checkout", action="store_true", default=False, dest="no_checkout", help="Skip restoring the working tree after cloning.", ) 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 run(args: argparse.Namespace) -> None: """Clone a remote Muse repository into a new local directory. Downloads the full commit history, snapshots, objects, and branch heads. Configures ``origin`` and upstream tracking. Checks out the default (or requested) branch unless ``--no-checkout`` is given. All progress and diagnostic messages go to **stderr**. ``--json`` (or ``--format json``) emits a single :class:`_CloneJson` object on stdout. On any error after the target directory has been partially created, the directory is removed to leave the filesystem clean. """ url: str = args.url directory: str | None = args.directory branch: str | None = args.branch dry_run: bool = args.dry_run no_checkout: bool = args.no_checkout fmt: str = args.format # clone does not need to be inside a Muse repo — it creates a new one. # Resolve the target name and path before any network I/O. target_name = directory or _infer_dir_name(url) target = pathlib.Path.cwd() / target_name if dry_run: print("(dry run — no files will be created)", file=sys.stderr) if (target / ".muse").exists(): msg = f"❌ '{sanitize_display(str(target))}' is already a Muse repository." print(msg, file=sys.stderr) if fmt == "json": out: _CloneJson = { "status": "already_exists", "url": url, "directory": str(target), "branch": branch or "", "commits_received": 0, "objects_written": 0, "head": None, "domain": "", "dry_run": dry_run, } print(json.dumps(out)) raise SystemExit(ExitCode.USER_ERROR) signing = get_signing_identity(remote_url=url) transport = make_transport(url) print( f"Cloning from {sanitize_display(url)} …", file=sys.stderr, ) try: info = transport.fetch_remote_info(url, signing=signing) except TransportError as exc: print(f"❌ Cannot reach remote: {exc}", file=sys.stderr) raise SystemExit(ExitCode.INTERNAL_ERROR) # Use "code" as the domain fallback — "midi" was the first plugin but is # not the canonical default domain for new repositories. remote_repo_id = info["repo_id"] or str(uuid.uuid4()) domain = info["domain"] or "code" default_branch = branch or info["default_branch"] or "main" if not info["branch_heads"]: print( "❌ Remote repository has no branches (empty repository).", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) default_commit_id = info["branch_heads"].get(default_branch) if default_commit_id is None: # Fall back to the first available branch rather than failing hard — # a user who requests a non-existent branch gets a clear warning. first_branch, default_commit_id = next(iter(info["branch_heads"].items())) print( f" ⚠️ Branch '{sanitize_display(default_branch)}' not found on remote; " f"checking out '{sanitize_display(first_branch)}' instead.", file=sys.stderr, ) default_branch = first_branch available = sorted(info["branch_heads"]) logger.debug( "Remote has %d branch(es): %s", len(available), ", ".join(sanitize_display(b) for b in available), ) # ── dry-run exits here — no filesystem changes after this point ────────── if dry_run: want_count = len(info["branch_heads"]) result: _CloneJson = { "status": "dry_run", "url": url, "directory": str(target), "branch": default_branch, "commits_received": 0, "objects_written": 0, "head": default_commit_id, "domain": domain, "dry_run": True, } if fmt == "json": print(json.dumps(result)) else: print( f"Would clone {sanitize_display(url)} → {sanitize_display(str(target))}", file=sys.stderr, ) print( f" branch={sanitize_display(default_branch)}, " f"domain={sanitize_display(domain)}, " f"{want_count} branch head(s) to fetch", file=sys.stderr, ) return # ── real clone ──────────────────────────────────────────────────────────── target.mkdir(parents=True, exist_ok=True) try: _init_muse_dir(target, remote_repo_id, domain, default_branch) except OSError as exc: print( f"❌ Failed to create repository at '{sanitize_display(str(target))}': {exc}", file=sys.stderr, ) shutil.rmtree(target, ignore_errors=True) raise SystemExit(ExitCode.INTERNAL_ERROR) # Fetch full pack — clone always starts with an empty have list because # there are no local objects yet. want = list(info["branch_heads"].values()) try: bundle = transport.fetch_pack(url, signing=signing, want=want, have=[]) except TransportError as exc: print(f"❌ Fetch failed: {exc}", file=sys.stderr) shutil.rmtree(target, ignore_errors=True) raise SystemExit(ExitCode.INTERNAL_ERROR) apply_result: ApplyResult = apply_pack(target, bundle) objects_written: int = apply_result["objects_written"] # ── Phase 2: fetch object bytes for the default checkout target ────────── # fetch_pack returns only VCS metadata (commits + snapshots). We fetch # only the objects needed for the default checkout to keep clone fast even # for repos with large historical object sets. default_commit = read_commit(target, default_commit_id) if default_commit and default_commit.snapshot_id: default_snap = read_snapshot(target, default_commit.snapshot_id) if default_snap: needed_oids = list(default_snap.manifest.values()) missing_oids = [oid for oid in needed_oids if not has_object(target, oid)] if missing_oids: logger.debug( "clone: requesting %d missing object(s) via /fetch/objects", len(missing_oids), ) try: fetched_objs = transport.fetch_objects(url, signing=signing, object_ids=missing_oids) except TransportError as exc: print(f"❌ Fetch objects failed: {exc}", file=sys.stderr) shutil.rmtree(target, ignore_errors=True) raise SystemExit(ExitCode.INTERNAL_ERROR) for obj in fetched_objs: oid = obj.get("object_id", "") raw = obj.get("content", b"") if oid and isinstance(raw, bytes): if write_object(target, oid, raw): objects_written += 1 # Write branch head refs for every remote branch atomically and record # the remote tracking pointer so future fetches can detect staleness. for b, cid in info["branch_heads"].items(): ref_file = target / ".muse" / "refs" / "heads" / b ref_file.parent.mkdir(parents=True, exist_ok=True) write_text_atomic(ref_file, cid) set_remote_head("origin", b, cid, target) # Configure origin remote and upstream tracking. set_remote("origin", url, target) set_upstream(default_branch, "origin", target) # Restore working tree unless the caller opted out. if not no_checkout: _restore_working_tree(target, default_commit_id) commits_received = apply_result["commits_written"] clone_result: _CloneJson = { "status": "cloned", "url": url, "directory": str(target), "branch": default_branch, "commits_received": commits_received, "objects_written": objects_written, "head": default_commit_id, "domain": domain, "dry_run": False, } if fmt == "json": print(json.dumps(clone_result)) else: print( f"✅ Cloned into '{sanitize_display(target_name)}' — " f"{commits_received} commit(s), {objects_written} object(s), " f"domain={sanitize_display(domain)}, " f"branch={sanitize_display(default_branch)} ({default_commit_id[:8]})", file=sys.stderr, ) logger.info( "✅ clone: %s → %s commits=%d objects=%d", url, target, commits_received, objects_written, )