"""``muse push`` — upload local commits, snapshots, and objects to a remote. Push protocol — presigned mpack --------------------------------- ``muse push`` uses a presigned mpack upload: **Phase 0 — ref discovery:** ``GET {url}/refs`` returns current branch heads. This cheap call establishes the ``have`` anchors used in commit deduplication. **Phase 1 — upload:** Client builds a wire MPack (``b"MUSE"`` binary format), calls ``POST {url}/push/mpack-presign`` to get a presigned R2 PUT URL, then PUTs the mpack bytes directly to R2 — bypassing Cloudflare's 100 MB body limit entirely. **Phase 2 — index:** Client calls ``POST {url}/push/unpack-mpack`` with the mpack content-address. Server reads the mpack from R2, indexes all commits/snapshots/objects, and advances the branch pointer. Fast-forward check ------------------ By default, ``muse push`` requires the remote branch to be an ancestor of the local branch (a fast-forward update). If the remote has diverged, the push is rejected with exit code 1. Pass ``--force`` to bypass this check. Upstream tracking ----------------- Pass ``-u`` / ``--set-upstream`` on first push to record the tracking relationship between the local branch and the remote branch so that future ``muse pull`` and ``muse push`` invocations can resolve the remote automatically. JSON output (``--format json`` / ``--json``) schema:: { "status": "pushed | up_to_date | dry_run | deleted", "remote": "", "branch": "", "head": " | null", "commits_sent": , "objects_sent": , "force": false, "dry_run": false } Exit codes:: 0 — success (pushed, up_to_date, deleted, or dry_run) 1 — remote not configured (error lists all configured remotes so the agent knows the exact names without a follow-up ``muse remote`` call), branch has no commits, push rejected, authentication failure, network error """ import argparse import json import logging import pathlib import sys import time as _time from collections.abc import Mapping from typing import TypedDict from muse.cli.config import ( delete_remote_head, get_signing_identity, get_remote, get_remote_head, get_config_value, list_remotes, set_remote_head, set_upstream, ) from muse.core.version_tags import list_version_tags from muse.core.types import split_id from muse.core.paths import remotes_dir as _remotes_dir from muse.core.envelope import EnvelopeJson, make_envelope from muse.core.errors import ExitCode from muse.core.mpack import ( PushResult, RemoteInfo, build_mpack_from_walk, build_wire_mpack, collect_blob_ids, walk_commits, ) from muse.core.types import blob_id as _blob_id from muse.core.refs import read_ref from muse.core.repo import require_repo from muse.core.timing import start_timer from muse.core.refs import ( get_head_commit_id, read_current_branch, ) from muse.core.commits import ( commit_exists, read_commit, ) from muse.core.snapshots import read_snapshot as _read_snapshot from muse.core.transport import ( MuseTransport, SigningIdentity, TransportError, _ignore_sigpipe, make_transport, ) from muse.core.validation import sanitize_display logger = logging.getLogger(__name__) import http.client as _http_client # noqa: E402 import ssl as _ssl # noqa: E402 import urllib.request as _urllib_req # noqa: E402 import urllib.error as _urllib_err # noqa: E402 import urllib.parse as _urllib_parse # noqa: E402 import asyncio as _asyncio # noqa: E402 from muse.core.transport import _TIMEOUT_SECONDS as _PUSH_TIMEOUT # noqa: E402 def _urllib_put(url: str, data: bytes, *, timeout: int = _PUSH_TIMEOUT, verify: bool = True) -> tuple[int, bytes]: """PUT *data* to a presigned URL using http.client directly. urllib.request adds Content-Type automatically which breaks S3/MinIO presigned URL signatures (signed with empty Content-Type). http.client gives us full control — we send only Content-Length. """ parsed = _urllib_parse.urlparse(url) host = parsed.netloc path = parsed.path + ("?" + parsed.query if parsed.query else "") if parsed.scheme == "https": ctx = _ssl.create_default_context() if verify else _ssl.SSLContext(_ssl.PROTOCOL_TLS_CLIENT) if not verify: ctx.check_hostname = False ctx.verify_mode = _ssl.CERT_NONE conn: "_http_client.HTTPConnection" = _http_client.HTTPSConnection( host, timeout=timeout, context=ctx ) else: conn = _http_client.HTTPConnection(host, timeout=timeout) try: conn.request("PUT", path, body=data, headers={"Content-Length": str(len(data))}) resp = conn.getresponse() return resp.status, resp.read() finally: conn.close() def _urllib_post(url: str, data: bytes, headers: Mapping[str, str], *, timeout: int = _PUSH_TIMEOUT, verify: bool = True) -> tuple[int, bytes]: """POST *data* to *url* using stdlib urllib. Returns (status_code, body).""" ctx = None if verify else _ssl.create_default_context() if ctx is not None: ctx.check_hostname = False ctx.verify_mode = _ssl.CERT_NONE req = _urllib_req.Request(url, data=data, headers=dict(headers), method="POST") try: with _urllib_req.urlopen(req, timeout=timeout, context=ctx) as resp: return resp.status, resp.read() except _urllib_err.HTTPError as exc: return exc.code, exc.read() class _PushJson(EnvelopeJson): """Stable JSON schema for push output.""" status: str remote: str branch: str head: str | None commits_sent: int objects_sent: int force: bool dry_run: bool class _PushWithTagsJson(_PushJson): """Extended push schema emitted when ``--tags`` is active.""" tags_pushed: int tags_skipped: int def _push_mpack( transport: MuseTransport, url: str, signing: SigningIdentity | None, root: pathlib.Path, local_head: str, have: list[str], branch: str, force: bool, branch_have: list[str] | None = None, ) -> tuple[PushResult, int, int]: """Push commits to a remote via the mpack presign protocol. Walks commits from *local_head* (excluding anything reachable from *branch_have*), builds a single mpack, PUTs it to MinIO via a presigned URL, then POSTs to push/unpack-mpack to index and advance the branch pointer. Returns: ``(PushResult, commits_sent, objects_sent)`` """ _t0 = _time.perf_counter() _branch_have = branch_have or [] # ── 1. BFS walk — client-side commit + object dedup ────────────────────── commit_walk = walk_commits(root, [local_head], have=_branch_have) _t1 = _time.perf_counter() all_commits_raw = commit_walk.get("commits") or [] commits_oldest_first = [ c.to_dict() if hasattr(c, "to_dict") else c for c in reversed(all_commits_raw) ] all_blob_ids: list[str] = commit_walk["all_blob_ids"] n_commits = len(commits_oldest_first) n_objects = len(all_blob_ids) print(f"[PUSH step 1] new_commits = {n_commits}", file=sys.stderr) # ── 2. Per-commit: snapshot delta + structured_delta + object collection ── _commits_oldest = list(reversed(all_commits_raw)) _snapshot_deltas_log = commit_walk.get("snapshot_deltas") or [] _blobs_to_send: set[str] = set(all_blob_ids) _have_blobs: set[str] = commit_walk.get("have_blobs") or set() _manifest_blobs: set[str] = commit_walk.get("manifest_blobs") or set() _accumulated_blobs: set[str] = set() # Seed _prev_manifest from the parent snapshot of the first new commit. _prev_manifest: dict[str, str] = {} if _commits_oldest: _first_commit = _commits_oldest[0] _first_parent_id = ( _first_commit.parent_commit_id if hasattr(_first_commit, "parent_commit_id") else (_first_commit.get("parent_commit_id") if isinstance(_first_commit, dict) else None) ) if _first_parent_id: _parent_commit = read_commit(root, _first_parent_id) if _parent_commit is not None: _parent_snap = _read_snapshot(root, _parent_commit.snapshot_id) if _parent_snap is not None: _prev_manifest = dict(_parent_snap.manifest) print(f"[PUSH step 2] have_blobs={len(_have_blobs)} manifest_blobs={len(_manifest_blobs)} blobs_to_send={len(_blobs_to_send)}", file=sys.stderr) for _commit_rec, _delta in zip(_commits_oldest, _snapshot_deltas_log): _snap_id_log = _delta.get("snapshot_id") or "" _child_snap = _read_snapshot(root, _snap_id_log) if _snap_id_log else None _child_manifest: dict[str, str] = dict(_child_snap.manifest) if _child_snap else {} _delta_upsert: dict[str, str] = { k: v for k, v in _child_manifest.items() if _prev_manifest.get(k) != v } _delta_remove: list[str] = [k for k in _prev_manifest if k not in _child_manifest] _added = {p: _delta_upsert[p] for p in _delta_upsert if p not in _prev_manifest} _modified = {p: _delta_upsert[p] for p in _delta_upsert if p in _prev_manifest} for _oid in set(_delta_upsert.values()): if _oid not in _accumulated_blobs and _oid in _blobs_to_send: _accumulated_blobs.add(_oid) _prev_manifest = _child_manifest _t4_start = _time.perf_counter() def _run_mpack_path() -> "PushResult": # Step 3: pack the walk result into a single mpack. mpack = build_mpack_from_walk(root, commit_walk, compress=True) _n_commits_mpack = len(mpack.get("commits", [])) _n_snapshots_mpack = len(mpack.get("snapshots", [])) _n_blobs_mpack = len(mpack.get("blobs", [])) wire_bytes = build_wire_mpack(mpack) mpack_key = _blob_id(wire_bytes) print(f"[PUSH step 3] pack into one mpack:", file=sys.stderr) print(f"[PUSH step 3] {{ commits: {_n_commits_mpack}, snapshots: {_n_snapshots_mpack}, blobs: {_n_blobs_mpack} }}", file=sys.stderr) print(f"[PUSH step 3] format=MUSE binary size={len(wire_bytes)} bytes", file=sys.stderr) print(f"[PUSH step 4] mpack_bytes = {len(wire_bytes)} bytes", file=sys.stderr) print(f"[PUSH step 4] mpack_key = {mpack_key}", file=sys.stderr) print(f"[PUSH step 4] size_bytes = {len(wire_bytes)}", file=sys.stderr) # Step 5: get presigned PUT URL print(f"[PUSH step 5] POST /push/mpack-presign {{ mpack_key, size_bytes }}", file=sys.stderr) presign = transport.push_mpack_presign(url, signing, wire_bytes) upload_url: str = presign["upload_url"] print(f"[PUSH step 5] → upload_url = {upload_url}", file=sys.stderr) # Step 6: PUT mpack to presigned URL _pre_put_hash = _blob_id(wire_bytes) _pre_put_match = _pre_put_hash == mpack_key print(f"[PUSH step 6] pre-PUT integrity check:", file=sys.stderr) print(f"[PUSH step 6] len(wire_bytes) = {len(wire_bytes)}", file=sys.stderr) print(f"[PUSH step 6] sha256(wire_bytes) = {_pre_put_hash}", file=sys.stderr) print(f"[PUSH step 6] mpack_key = {mpack_key}", file=sys.stderr) print(f"[PUSH step 6] match = {_pre_put_match} {'✓' if _pre_put_match else '✗ MISMATCH'}", file=sys.stderr) print(f"[PUSH step 6] PUT mpack_bytes → presigned_url", file=sys.stderr) transport.push_mpack_put(upload_url, wire_bytes, mpack_key) # Step 7: notify server to unpack print(f"[PUSH step 7] POST /push/unpack-mpack {{ mpack_key={mpack_key}, branch={branch}, head_commit_id={local_head}, force={force} }}", file=sys.stderr) unpack_data = transport.push_mpack_unpack( url, signing, mpack_key, branch=branch, head=local_head, commits_count=n_commits, blobs_count=n_objects, force=force, ) print( f"[PUSH step 7] → {{ blobs_written={unpack_data.get('blobs_in_mpack')}, " f"blobs_skipped=0, " f"commits_written={unpack_data.get('commits_in_mpack')}, " f"branch_head={unpack_data.get('head', '')} }}", file=sys.stderr, ) return PushResult(ok=True, message="ok", branch_heads={branch: local_head}) with _ignore_sigpipe(): result = _run_mpack_path() _t4 = _time.perf_counter() return result, n_commits, n_objects def _fetch_remote_info_safe( transport: MuseTransport, url: str, token: str | None, ) -> RemoteInfo | None: """Call ``GET /refs`` on the remote and return its current branch heads. Returns ``None`` only for network-level failures (``status_code == 0``, e.g. connection refused or DNS failure) so callers can fall back to locally-cached tracking refs. Re-raises ``TransportError`` for all HTTP error codes (404, 401, 409, 5xx) — those require user action and must not be silenced. """ try: return transport.fetch_remote_info(url, token) except TransportError as exc: if exc.status_code == 0: return None raise def _is_valid_commit_id(value: str) -> bool: """Return True if *value* is a well-formed sha256:<64-hex> commit ID. Filters out server sentinels like ``"init-sha256:"`` that signal an empty remote branch. These are not real commit IDs and must not be passed to ``commit_exists`` or included in the ``have`` list. """ try: split_id(value) return True except ValueError: return False def _all_known_have_anchors(root: pathlib.Path) -> list[str]: """Return every commit ID cached in any remote's tracking refs. When pushing a new branch (or to a remote with no local tracking cache), these commits are our best guess at what the remote already holds. Any remote the user has previously pushed to shares commit ancestry with other remotes — using all cached heads as ``have`` anchors ensures ``build_mpack`` only transmits the delta since the nearest shared ancestor. Symlinks inside ``.muse/remotes/`` are skipped rather than followed to prevent path-traversal attacks. Unreadable files are silently skipped so a corrupted tracking ref doesn't abort the push. """ remotes_dir = _remotes_dir(root) if not remotes_dir.is_dir(): return [] heads: list[str] = [] for ref_file in remotes_dir.rglob("*"): if ref_file.is_symlink() or not ref_file.is_file(): continue commit_id = read_ref(ref_file) if commit_id: heads.append(commit_id) return heads def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``muse push`` subcommand and all its flags.""" parser = subparsers.add_parser( "push", help="Upload local commits, snapshots, and objects to a remote.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "remote", nargs="?", default="origin", help="Remote name to push to (default: origin).", ) parser.add_argument( "branch_pos", nargs="?", default=None, metavar="BRANCH", help="Branch to push (default: current branch). Same as --branch.", ) parser.add_argument( "--branch", "-b", default=None, dest="branch_flag", help="Branch to push (default: current branch).", ) parser.add_argument( "-u", "--set-upstream", action="store_true", dest="set_upstream_flag", help="Record upstream tracking for this branch.", ) parser.add_argument( "--force", "-f", action="store_true", help="Force push even if the remote has diverged.", ) parser.add_argument( "--force-with-lease", action="store_true", dest="force_with_lease", help=( "Force push only if the remote branch has not advanced since the last fetch. " "Safer than --force: rejects the push if someone else has pushed in the meantime." ), ) parser.add_argument( "--delete", "-d", action="store_true", dest="delete_branch", help="Delete the named branch on the remote.", ) parser.add_argument( "-n", "--dry-run", action="store_true", help="Compute what would be pushed without sending any data.", ) parser.add_argument( "--workers", type=int, default=16, metavar="N", help="Number of parallel upload workers (default: 16).", ) parser.add_argument( "--tags", action="store_true", dest="push_tags", help=( "Push all local version tags to the remote after pushing commits. " "Also enabled automatically when push.tags=true in config." ), ) parser.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit machine-readable JSON instead of human text.", ) parser.add_argument( "--hub", default=None, metavar="URL", help=( "Resolve a hub URL to its configured remote name and use that remote. " "Equivalent to finding the remote whose URL matches and using it as " "the argument. " "Example: --hub https://staging.musehub.ai" ), ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Upload local commits, snapshots, and objects to a remote. Requires the remote to be a fast-forward of the local branch unless ``--force`` is specified. All progress and error messages go to **stderr**. ``--format json`` (or ``--json``) emits a single JSON object on stdout for agent pipelines. ``--dry-run`` computes the pack that *would* be sent using local tracking refs as have-anchors and exits 0 without connecting to the remote. JSON schema:: { "status": "pushed | up_to_date | dry_run | deleted", "remote": "", "branch": "", "head": " | null", "commits_sent": , "objects_sent": , "force": false, "dry_run": false } Exit codes:: 0 — success 1 — remote not configured, branch has no commits, push rejected, authentication failure, network error, invalid format """ elapsed = start_timer() # ── --hub URL: resolve to a configured remote name ─────────────────────── # Agents sometimes pass --hub (the muse hub subcommand pattern) to # muse push by mistake. Accept it gracefully: look up the remote whose URL # matches and use it, or fail with a clear, actionable message. # # Positional re-interpretation: when --hub is given, the positional # slot is irrelevant (the remote is determined by the URL). If only one # positional was provided it landed in args.remote but was intended as the # branch — move it to branch_pos so it is not silently discarded. json_out: bool = args.json_out hub_url: str | None = getattr(args, "hub", None) if hub_url is not None: root_for_hub = require_repo() all_remotes = list_remotes(root_for_hub) matched = next( (r for r in all_remotes if r["url"].rstrip("/") == hub_url.rstrip("/")), None, ) if matched is None: remote_list = ", ".join(r["name"] for r in all_remotes) if all_remotes else "(none configured)" if json_out: print(json.dumps({"error": "remote_not_found", "hub_url": hub_url, "configured_remotes": remote_list})) else: print( f"❌ No remote is configured for hub URL '{sanitize_display(hub_url)}'.\n" f" Configured remotes: {remote_list}\n" f" muse push uses remote names, not --hub URLs:\n" f" muse push ", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # If only one positional was supplied it consumed the slot but # is actually the branch name — rescue it before overwriting args.remote. if args.branch_pos is None and args.remote not in {r["name"] for r in all_remotes}: args.branch_pos = args.remote args.remote = matched["name"] remote: str = args.remote branch: str | None = ( getattr(args, "branch_flag", None) or getattr(args, "branch_pos", None) ) set_upstream_flag: bool = args.set_upstream_flag force: bool = args.force force_with_lease: bool = getattr(args, "force_with_lease", False) if force_with_lease and not force: # --force-with-lease implies force behaviour but with a safety check. force = True delete_branch: bool = getattr(args, "delete_branch", False) dry_run: bool = getattr(args, "dry_run", False) root = require_repo() url = get_remote(remote, root) if url is None: all_remotes = list_remotes(root) configured = ( ", ".join(r["name"] for r in all_remotes) if all_remotes else "(none)" ) if json_out: print(json.dumps({"error": "remote_not_configured", "remote": remote, "configured_remotes": configured})) else: print( f"❌ Remote '{sanitize_display(remote)}' is not configured.\n" f" Configured remotes: {configured}\n" f" Add one with: muse remote add {sanitize_display(remote)} ", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # ── DELETE MODE ─────────────────────────────────────────────────────────── if delete_branch: current_branch = read_current_branch(root) del_branch = branch or current_branch if not del_branch: if json_out: print(json.dumps({"error": "no_branch_specified", "message": "specify a branch to delete: muse push --delete "})) else: print( "❌ Specify a branch to delete: muse push --delete ", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if dry_run: msg = f"Would delete {sanitize_display(remote)}/{sanitize_display(del_branch)}" if json_out: print(json.dumps(_PushJson( **make_envelope(elapsed), status="dry_run", remote=remote, branch=del_branch, head=None, commits_sent=0, objects_sent=0, force=force, dry_run=True, ))) else: print(msg) return token = get_signing_identity(root, remote_url=url) transport = make_transport(url) print( f"Deleting {sanitize_display(remote)}/{sanitize_display(del_branch)} …", file=sys.stderr, ) already_gone = False try: transport.delete_branch_remote(url, token, del_branch) except TransportError as exc: if exc.status_code == 404: already_gone = True elif exc.status_code == 409: if json_out: print(json.dumps({"error": "default_branch", "branch": del_branch, "message": f"cannot delete the default branch '{del_branch}'"})) else: print(f"❌ Cannot delete the default branch '{sanitize_display(del_branch)}'.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) elif exc.status_code == 403: if json_out: print(json.dumps({"error": "permission_denied", "message": "only the repo owner may delete branches"})) else: print("❌ Permission denied — only the repo owner may delete branches.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) else: if json_out: print(json.dumps({"error": "delete_failed", "message": str(exc)})) else: print(f"❌ Delete failed: {sanitize_display(str(exc))}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) pruned = delete_remote_head(remote, del_branch, root) if json_out: print(json.dumps(_PushJson( **make_envelope(elapsed), status="deleted", remote=remote, branch=del_branch, head=None, commits_sent=0, objects_sent=0, force=force, dry_run=False, ))) else: if already_gone: note = " (already absent on remote)" if pruned: note += ", tracking ref pruned" print(f"✅ {sanitize_display(remote)}/{sanitize_display(del_branch)}{note}") else: prune_note = " (tracking ref pruned)" if pruned else "" print( f"✅ Deleted remote branch " f"{sanitize_display(remote)}/{sanitize_display(del_branch)}{prune_note}" ) return # ── NORMAL PUSH ────────────────────────────────────────────────────────── current_branch = read_current_branch(root) push_branch = branch or current_branch local_head = get_head_commit_id(root, push_branch) if local_head is None: if json_out: print(json.dumps({"error": "no_commits", "branch": push_branch, "message": f"branch '{push_branch}' has no commits to push"})) else: print(f"❌ Branch '{sanitize_display(push_branch)}' has no commits to push.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # ── DRY-RUN: compute pack locally, no network calls ─────────────────────── if dry_run: have: list[str] = [ c for c in _all_known_have_anchors(root) if c != local_head and commit_exists(root, c) ] dry_branch_have: list[str] = [] _cached = get_remote_head(remote, push_branch, root) if _cached and commit_exists(root, _cached): dry_branch_have = [_cached] dry_walk = walk_commits(root, [local_head], have=dry_branch_have) dry_commits = len(dry_walk["commits"]) dry_objects = len(collect_blob_ids(root, [local_head], have=have)) if json_out: print(json.dumps(_PushJson( **make_envelope(elapsed), status="dry_run", remote=remote, branch=push_branch, head=local_head, commits_sent=dry_commits, objects_sent=dry_objects, force=force, dry_run=True, ))) else: print( f"Would push {dry_commits} commit(s) and ~{dry_objects} object(s) " f"to {sanitize_display(remote)}/{sanitize_display(push_branch)} (dry run)" ) return transport = make_transport(url) token = get_signing_identity(root, remote_url=url) # ── STEP 0: discover remote state ──────────────────────────────────────── print(f"[PUSH step 0] GET {url.rstrip('/')}/refs", file=sys.stderr) try: remote_info = _fetch_remote_info_safe(transport, url, token) except TransportError as exc: if json_out: code_map = {404: "repository_not_found", 401: "auth_required"} print(json.dumps({"error": code_map.get(exc.status_code, "remote_error"), "status_code": exc.status_code, "remote": remote, "branch": push_branch, "message": str(exc)})) elif exc.status_code == 404: print( f"❌ Repository not found on remote '{sanitize_display(remote)}'.\n" f" Create it first: muse hub repo create --name \n" f" Then verify the remote URL: muse remote -v", file=sys.stderr, ) elif exc.status_code == 401: print( f"❌ Authentication required — remote '{sanitize_display(remote)}' returned 401.\n" f" Run: muse auth register", file=sys.stderr, ) else: print(f"❌ Remote error ({exc.status_code}): {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if remote_info is not None: remote_branch_heads = remote_info["branch_heads"] candidate_have = list(remote_branch_heads.values()) _remote_head = remote_branch_heads.get(push_branch) or None print(f"[PUSH step 0] → branch_heads = {remote_branch_heads}", file=sys.stderr) print(f"[PUSH step 0] remote_head = {_remote_head or 'null'}", file=sys.stderr) print(f"[PUSH step 0] have = {candidate_have}", file=sys.stderr) else: # live /refs unreachable (e.g. SSL failure on self-signed cert) — # fall back to locally cached tracking refs for THIS remote only. # Same-remote tracking refs are safe: they were written on the last # successful push to this exact remote, so they represent objects # the remote already holds. Cross-remote refs are NOT included. remote_branch_heads = {} remote_tracking_dir = _remotes_dir(root) / remote candidate_have = [] if remote_tracking_dir.is_dir(): for _ref_file in remote_tracking_dir.rglob("*"): if _ref_file.is_symlink() or not _ref_file.is_file(): continue _cached_id = read_ref(_ref_file) if _cached_id: candidate_have.append(_cached_id) # have-anchors are what the SERVER already has — not what we have locally. # commit_exists() must NOT be used here: if the remote has a commit we # don't have locally (e.g. after a force-push or diverged history), the # local walk stops naturally when it can't traverse past a missing commit. # Filtering by commit_exists() causes have=[] and re-sends the full history. have = [ c for c in candidate_have if c != local_head and _is_valid_commit_id(c) ] if remote_info is not None: remote_head: str | None = remote_branch_heads.get(push_branch) else: remote_head = get_remote_head(remote, push_branch, root) print(f"[PUSH step 1] walk local DAG → find commits not on remote (want - have)", file=sys.stderr) print(f"[PUSH step 1] want = {local_head}", file=sys.stderr) print(f"[PUSH step 1] remote_head = {remote_head or 'null'}", file=sys.stderr) print(f"[PUSH step 1] have = {have}", file=sys.stderr) # ── FORCE-WITH-LEASE safety check ──────────────────────────────────────── # Reject the push if the live remote HEAD has advanced beyond what we last # fetched — i.e., someone else pushed after our last fetch/pull. if force_with_lease and remote_head is not None: cached_head = get_remote_head(remote, push_branch, root) if cached_head != remote_head: if json_out: print(json.dumps({"error": "force_with_lease_rejected", "remote": remote, "branch": push_branch, "message": "remote has advanced since last fetch — run muse fetch first"})) else: print( f"❌ Push rejected (--force-with-lease): remote " f"'{sanitize_display(remote)}/{sanitize_display(push_branch)}' has advanced " f"since your last fetch.\n" f" Run 'muse fetch' to update your tracking refs, then retry.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if remote_head is None: print(f"[PUSH step 1] remote_head is null → new_commits = all commits reachable from local tip", file=sys.stderr) elif remote_head == local_head: print(f"[PUSH step 1] remote_head == local_tip → nothing to push", file=sys.stderr) else: print(f"[PUSH step 1] new_commits = commits reachable from local tip, not reachable from remote_head", file=sys.stderr) if remote_head == local_head: if json_out: print(json.dumps(_PushJson( **make_envelope(elapsed), status="up_to_date", remote=remote, branch=push_branch, head=local_head, commits_sent=0, objects_sent=0, force=force, dry_run=False, ))) else: print( f"Everything up to date. " f"Remote {sanitize_display(remote)}/{sanitize_display(push_branch)} " f"is already at {local_head}." ) return # `branch_have` is the commit-graph boundary for build_mpack. Use ALL # known remote branch heads unconditionally so the BFS stops at any commit # the server already holds — regardless of which branch it lives on. # # When pushing a second branch (e.g. dev after main), the remote has no # head for dev yet, so the old code set branch_have=[] and walked the # entire DAG back to root, re-sending every commit already on main. Using # all remote heads fixes this: the BFS stops at main's head and only sends # the dev-only commits above it. branch_have: list[str] = [ h for h in remote_branch_heads.values() if _is_valid_commit_id(h) ] try: result, commits_sent, objects_sent = _push_mpack( transport, url, token, root, local_head, have, push_branch, force, branch_have=branch_have, ) except TransportError as exc: if json_out: code_map = {404: "repo_not_found", 401: "auth_required", 409: "diverged"} print(json.dumps({"error": code_map.get(exc.status_code, "push_failed"), "status_code": exc.status_code, "remote": remote, "branch": push_branch, "message": str(exc)})) elif exc.status_code == 404: print( f"❌ Repository not found on remote '{sanitize_display(remote)}'.\n" f" The remote server returned 404 for this repo.\n" f"\n" f" If this is a new repo, create it on the server first:\n" f" muse hub repo create (if using MuseHub)\n" f"\n" f" If the repo exists, verify your remote URL:\n" f" muse remote -v\n" f"\n" f" If you are authenticated, run: muse auth whoami", file=sys.stderr, ) elif exc.status_code == 401: print( f"❌ Authentication required — remote '{sanitize_display(remote)}' returned 401.\n" f" Log in first: muse auth register", file=sys.stderr, ) elif exc.status_code == 409: print( f"❌ Push rejected — remote " f"'{sanitize_display(remote)}/{sanitize_display(push_branch)}' has diverged.\n" f" Pull first (muse pull) or use --force to override.", file=sys.stderr, ) else: print(f"❌ Push failed: {sanitize_display(str(exc))}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if not result["ok"]: if json_out: print(json.dumps({"error": "push_rejected", "remote": remote, "branch": push_branch, "message": result["message"]})) else: print(f"❌ Push rejected by remote: {sanitize_display(str(result['message']))}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) updated_head = result["branch_heads"].get(push_branch, local_head) set_remote_head(remote, push_branch, updated_head, root) if set_upstream_flag: set_upstream(push_branch, remote, root) print( f" Upstream set: {sanitize_display(push_branch)} → " f"{sanitize_display(remote)}/{sanitize_display(push_branch)}", file=sys.stderr, ) # ── OPTIONAL: push version tags ────────────────────────────────────────── # Enabled by --tags flag or push.tags=true in config. push_tags_flag: bool = getattr(args, "push_tags", False) if not push_tags_flag: try: cfg_val = get_config_value("push.tags", root) push_tags_flag = (cfg_val or "").lower() in ("true", "1", "yes") except Exception: pass tags_pushed = 0 tags_skipped = 0 if push_tags_flag: local_tags = list_version_tags(root) if local_tags: tag_dicts = [ { "tag_id": r.tag_id, "repo_id": r.repo_id, "tag": r.tag, "semver": { "major": r.semver["major"], "minor": r.semver["minor"], "patch": r.semver["patch"], "pre": r.semver["pre"], "build": r.semver["build"], }, "commit_id": r.commit_id, "created_at": r.created_at.isoformat(), "author": r.author, "message": r.message, } for r in local_tags ] try: tag_result = transport.push_version_tags(url, token, tag_dicts, force=force) tags_pushed = tag_result.get("stored", 0) tags_skipped = tag_result.get("skipped", 0) except TransportError as exc: logger.warning("Version tag push failed: %s", exc) if push_tags_flag: if json_out: print(json.dumps(_PushWithTagsJson( **make_envelope(elapsed), status="pushed", remote=remote, branch=push_branch, head=updated_head, commits_sent=commits_sent, objects_sent=objects_sent, force=force, dry_run=False, tags_pushed=tags_pushed, tags_skipped=tags_skipped, ))) else: print( f"✅ Pushed {commits_sent} commit(s), {objects_sent} object(s) " f"to {sanitize_display(remote)}/{sanitize_display(push_branch)} " f"({updated_head})" ) if tags_pushed or tags_skipped: print(f" Tags: {tags_pushed} pushed, {tags_skipped} already current") else: if json_out: print(json.dumps(_PushJson( **make_envelope(elapsed), status="pushed", remote=remote, branch=push_branch, head=updated_head, commits_sent=commits_sent, objects_sent=objects_sent, force=force, dry_run=False, ))) else: print( f"✅ Pushed {commits_sent} commit(s), {objects_sent} object(s) " f"to {sanitize_display(remote)}/{sanitize_display(push_branch)} " f"({updated_head})" ) # probe # probe2