"""``muse push`` — upload local commits, snapshots, and objects to a remote.

MWP push protocol
-----------------

``muse push`` uses the Muse Wire Protocol (MWP) for all remotes.

**Phase 0 — ref discovery:**
    ``GET {url}/refs`` returns current branch heads.  This cheap call also
    establishes the ``have`` anchors used in commit negotiation.

**Phase 1 — object deduplication (MWP):**
    ``POST {url}/filter-objects`` accepts the full list of object IDs the
    client intends to push.  The server returns only the *missing* subset.
    For incremental pushes this reduces the object payload to near-zero.

**Phase 2 — large-object presign (MWP):**
    Objects above :data:`~muse.core.transport.LARGE_OBJECT_THRESHOLD` (64 KB)
    are uploaded directly to S3/R2 via presigned PUT URLs — they never transit
    the API server.  ``local://`` remotes return all IDs in ``inline`` and
    fall back to the pack body automatically.

**Phase 3 — parallel object upload:**
    Remaining (small) objects are batched into chunks of
    :data:`~muse.core.transport.CHUNK_OBJECTS` and uploaded in parallel using
    ``concurrent.futures.ThreadPoolExecutor`` (``--workers``, default 16).

**Phase 4 — commit push:**
    A single ``POST {url}/push`` carries commits and snapshots with an empty
    ``objects`` list (blobs are already on the remote after Phases 1-3).

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":       "<name>",
      "branch":       "<branch>",
      "head":         "<sha256> | null",
      "commits_sent": <N>,
      "objects_sent": <N>,
      "force":        false,
      "dry_run":      false
    }

Exit codes::

    0 — success (pushed, up_to_date, deleted, or dry_run)
    1 — remote not configured, branch has no commits, push rejected,
        authentication failure, network error
"""

from __future__ import annotations

import argparse
import json
import logging
import pathlib
import platform
import subprocess
import sys
import urllib.error
import urllib.request
from concurrent.futures import Future, ThreadPoolExecutor, as_completed
from typing import TypedDict

from muse.cli.config import (
    delete_remote_head,
    get_signing_identity,
    get_remote,
    get_remote_head,
    get_remote_pack_origin,
    set_remote_head,
    set_upstream,
)
from muse.core.errors import ExitCode
from muse.core.object_store import read_object
from muse.core.pack import (
    ObjectPayload,
    PackBundle,
    PushResult,
    RemoteInfo,
    build_pack,
    build_pack_from_walk,
    collect_object_ids,
    collect_object_ids_from_walk,
    stream_object_chunks,
    walk_commits,
)
from muse.core.repo import require_repo
from muse.core.store import commit_exists, get_head_commit_id, read_commit, read_current_branch
from muse.core.compression import compress_zlib, compute_delta
from muse.core.transport import (
    CHUNK_COMMITS,
    CHUNK_OBJECTS,
    MuseTransport,
    TransportError,
    make_transport,
)
from muse.core.validation import sanitize_display

logger = logging.getLogger(__name__)

# Number of retries for transient presigned-URL failures (S3/R2 503, 429).
_PRESIGN_RETRIES: int = 3

# Objects larger than this threshold use the presigned PUT path (direct to R2).
# Objects at or below this threshold use the pack endpoint (batched API POSTs).
#
# Rationale: direct R2 PUT bypasses Cloudflare entirely, which is worthwhile
# when the body is large enough to saturate available bandwidth.  For small
# objects the overhead is dominated by TLS handshake latency (N handshakes for
# N objects), not transfer time.  The pack endpoint costs M = ceil(N/1000)
# handshakes instead of N — a 1 000× reduction for the common case.
#
# 1 MB is the crossover: below this, handshake cost dominates (use pack);
# above this, transfer time dominates (use presign).
PACK_OBJECT_THRESHOLD: int = 1_000_000   # 1 MB

# Maximum objects per pack POST and maximum total bytes per pack.
# Server-side limits are PACK_MAX_OBJECTS=1_000 and PACK_MAX_BYTES=50_000_000.
# The client uses a more conservative 250-object limit per pack so that each
# server-side batch of R2 puts (concurrent via asyncio.gather) completes well
# within the 60 s nginx upstream timeout.  Staging data: 342 objects ≈ 27 s,
# so 250 objects ≈ 20 s → 3× safety margin.  Larger repos simply use more packs.
PACK_MAX_OBJECTS: int = 250
PACK_MAX_BYTES: int = 50_000_000         # 50 MB

# Maximum concurrent pack POST requests.  All packs are submitted in parallel
# up to this limit.  8 workers is enough to cover a 2 000-file repo in one
# wave (8 × 250 = 2 000).  Higher values provide diminishing returns since
# the bottleneck shifts to the server-side R2 connection pool (100 slots).
_PACK_WORKERS: int = 8


class _PushJson(TypedDict):
    """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


def _upload_chunk(
    transport: MuseTransport,
    url: str,
    token: str | None,
    chunk: list[ObjectPayload],
    chunk_num: int,
    total_chunks: int,
) -> tuple[int, int]:
    """Upload one chunk of objects and return ``(stored, skipped)``.

    Progress messages go to **stderr** so they don't pollute JSON stdout.
    """
    print(
        f"  Uploading objects chunk {chunk_num}/{total_chunks} "
        f"({len(chunk)} object(s)) …",
        file=sys.stderr,
    )
    resp = transport.push_objects(url, token, chunk)
    return resp["stored"], resp["skipped"]


def _upload_presigned(oid: str, url: str, raw: bytes, retries: int = _PRESIGN_RETRIES) -> None:
    """PUT *raw* bytes directly to a presigned S3/R2 URL (MWP Phase 2).

    Dispatches to a curl-based implementation on macOS to work around an
    OpenSSL 3.6.x / Python 3.14 TLS 1.3 bug (SSLV3_ALERT_BAD_RECORD_MAC)
    that only affects Homebrew Python on macOS.  Linux (staging/prod Docker)
    uses urllib directly — different OpenSSL build, no bug.

    Retries ``retries`` times on transient errors (HTTP 429/5xx, network).
    Raises on the last exception if all retries are exhausted.
    """
    if platform.system() == "Darwin":
        _upload_presigned_curl(oid, url, raw, retries)
    else:
        _upload_presigned_urllib(oid, url, raw, retries)


def _upload_presigned_curl(oid: str, url: str, raw: bytes, retries: int) -> None:
    """macOS: PUT via system curl (SecureTransport/LibreSSL — not Homebrew OpenSSL)."""
    import time as _time

    last_exc: Exception | None = None
    for attempt in range(1, retries + 1):
        result = subprocess.run(
            [
                "curl", "--silent", "--show-error", "--fail-with-body",
                "--request", "PUT",
                "--data-binary", "@-",
                "--header", f"Content-Length: {len(raw)}",
                "--max-time", "300",
                url,
            ],
            input=raw,
            capture_output=True,
        )
        if result.returncode == 0:
            return
        # curl exit 22 = HTTP error (--fail-with-body); others = network/SSL
        err_msg = result.stderr.decode(errors="replace").strip()
        exc = RuntimeError(f"presigned upload failed for {oid[:8]}: {err_msg}")
        last_exc = exc
        if attempt < retries:
            _time.sleep(2 ** (attempt - 1))
        else:
            break

    raise last_exc or RuntimeError(f"presigned upload failed for {oid[:8]}")


def _upload_presigned_urllib(oid: str, url: str, raw: bytes, retries: int) -> None:
    """Linux: PUT via urllib (works correctly on Docker/prod OpenSSL builds)."""
    import time as _time

    last_exc: Exception | None = None
    for attempt in range(1, retries + 1):
        try:
            req = urllib.request.Request(url=url, data=raw, method="PUT")
            with urllib.request.urlopen(req, timeout=300):
                pass
            return
        except urllib.error.HTTPError as exc:
            last_exc = exc
            if exc.code not in (429, 500, 502, 503, 504):
                raise
            wait = 2 ** (attempt - 1)
            logger.warning(
                "⚠️ Presigned upload for %s returned %d — retry %d/%d in %ds",
                oid[:8], exc.code, attempt, retries, wait,
            )
            _time.sleep(wait)
        except urllib.error.URLError as exc:
            last_exc = exc
            if attempt < retries:
                _time.sleep(1)
            else:
                break

    raise last_exc or RuntimeError(f"presigned upload failed for {oid[:8]}")


def _encode_object(
    oid: str,
    raw: bytes,
    path: str,
    oid_to_base: dict[str, str],
    root: pathlib.Path,
) -> ObjectPayload:
    """Compress or delta-encode one object for wire transmission.

    Tries Tier 2 (delta+zlib) first if a base ID is available.  Falls back to
    Tier 1 (plain zlib) when the delta is not profitable or the base is absent.
    Always returns a valid :class:`ObjectPayload` with ``encoding`` set.

    Args:
        oid:          SHA-256 object ID of the target object.
        raw:          Raw (uncompressed) object bytes.
        path:         Repository path (used only for the payload's ``path`` field).
        oid_to_base:  Mapping of oid → base_oid from the server's filter-objects response.
        root:         Local repo root for reading base object bytes.

    Returns:
        :class:`ObjectPayload` with ``encoding``, ``path``, and optionally ``base_id`` set.
    """
    base_id = oid_to_base.get(oid)
    if base_id:
        base_raw = read_object(root, base_id)
        if base_raw is not None:
            try:
                delta = compute_delta(base_raw, raw)
                return ObjectPayload(
                    object_id=oid,
                    content=delta,
                    path=path,
                    encoding="delta+zlib",
                    base_id=base_id,
                )
            except ValueError:
                pass  # delta not profitable — fall through to plain zlib

    return ObjectPayload(
        object_id=oid,
        content=compress_zlib(raw),
        path=path,
        encoding="zlib",
    )


def _push_objects_as_packs(
    transport: MuseTransport,
    url: str,
    signing: object,
    object_ids: list[str],
    root: pathlib.Path,
    oid_to_path: dict[str, str] | None = None,
    oid_to_base: dict[str, str] | None = None,
    pack_origin: str | None = None,
) -> tuple[int, int]:
    """Upload *object_ids* via the pack endpoint in parallel (MWP Phase 2).

    Objects are zlib-compressed (Tier 1) or delta+zlib-encoded (Tier 2) before
    batching.  Batches of ≤ :data:`PACK_MAX_OBJECTS` items and ≤ :data:`PACK_MAX_BYTES`
    total are submitted simultaneously using a :class:`~concurrent.futures.ThreadPoolExecutor`
    with up to ``_PACK_WORKERS`` workers.

    GitHub-level approach: every pack flies concurrently rather than waiting
    for the previous one to complete.  For a 2 000-file repo with 8 packs of
    250 objects, the server processes all 2 000 objects simultaneously through
    the global R2 semaphore (100 slots), yielding ~1.5 s instead of ~10 s serial.

    Args:
        transport:    Active :class:`~muse.core.transport.MuseTransport` instance.
        url:          Remote repository URL (used for path extraction).
        signing:      Ed25519 signing identity (passed through to ``push_object_pack``).
        object_ids:   List of object IDs to upload (must all be ≤ 1 MB each).
        root:         Local repository root.
        oid_to_path:  ``{oid: path}`` map used to populate the ``path`` field on each
                      payload (from the walk result, optional).
        oid_to_base:  ``{oid: base_oid}`` map from the server's filter-objects response
                      for Tier 2 delta encoding (optional).
        pack_origin:  Optional origin override for pack uploads (e.g. a Cloudflare
                      Worker URL).  When set, the URL origin is replaced with this value
                      while preserving the ``/{owner}/{slug}`` path from *url*.

    Returns:
        ``(total_stored, total_skipped)`` accumulated across all pack requests.

    Raises:
        :class:`~muse.core.transport.TransportError` on HTTP 4xx/5xx or network failure.
    """
    import urllib.parse as _up
    if pack_origin:
        # Replace the origin (scheme + netloc) while keeping the path (/{owner}/{slug}).
        _parsed = _up.urlparse(url)
        _origin = _up.urlparse(pack_origin)
        pack_url = _up.urlunparse(_origin._replace(
            path=_parsed.path, query="", fragment=""
        ))
    else:
        pack_url = url

    _oid_to_path: dict[str, str] = oid_to_path or {}
    _oid_to_base: dict[str, str] = oid_to_base or {}

    # ── Step 1: build all packs in memory ────────────────────────────────────
    batches: list[list[ObjectPayload]] = []
    current_batch: list[ObjectPayload] = []
    current_bytes = 0

    for oid in object_ids:
        raw = read_object(root, oid) or b""
        payload = _encode_object(oid, raw, _oid_to_path.get(oid, ""), _oid_to_base, root)
        wire_size = len(payload["content"])
        if current_batch and (
            len(current_batch) >= PACK_MAX_OBJECTS
            or current_bytes + wire_size > PACK_MAX_BYTES
        ):
            batches.append(current_batch)
            current_batch = []
            current_bytes = 0
        current_batch.append(payload)
        current_bytes += wire_size

    if current_batch:
        batches.append(current_batch)

    if not batches:
        return 0, 0

    total_batches = len(batches)

    # ── Step 2: submit all packs in parallel ──────────────────────────────────
    pack_workers = min(_PACK_WORKERS, total_batches)

    def _send_pack(pack_num: int, pack: list[ObjectPayload]) -> tuple[int, int]:
        wire_bytes = sum(len(o["content"]) for o in pack)
        print(
            f"  Pack {pack_num}/{total_batches} ({len(pack)} object(s), "
            f"{wire_bytes // 1024} KiB) …",
            file=sys.stderr,
        )
        resp = transport.push_object_pack(pack_url, signing, pack)
        return resp["stored"], resp["skipped"]

    total_stored = 0
    total_skipped = 0

    with ThreadPoolExecutor(max_workers=pack_workers) as pool:
        pack_futures: list[Future[tuple[int, int]]] = [
            pool.submit(_send_pack, i + 1, pack)
            for i, pack in enumerate(batches)
        ]
        for fut in as_completed(pack_futures):
            stored, skipped = fut.result()  # re-raises TransportError on failure
            total_stored += stored
            total_skipped += skipped

    return total_stored, total_skipped


def _push_mwp(
    transport: MuseTransport,
    url: str,
    token: str | None,
    root: pathlib.Path,
    local_head: str,
    have: list[str],
    branch: str,
    force: bool,
    max_workers: int = 16,
    branch_have: list[str] | None = None,
    send_local_head: bool = True,
    pack_origin: str | None = None,
) -> tuple[PushResult, int, int]:
    """Full MWP push: dedup negotiation → presign → parallel upload → commit push.

    MWP phases executed here:

    - Phase 1 — ``POST /filter-objects``: discover which objects are missing.
    - Phase 2 — ``POST /presign``: get presigned PUT URLs for large objects.
    - Phase 3 — parallel direct-to-storage upload for large objects.
    - Phase 4 — parallel chunked ``POST /push/objects`` for small objects.
    - Phase 5 — ``POST /push`` with commits + snapshots, empty objects list.

    All progress messages go to **stderr** so ``--format json`` stdout stays clean.

    *have* is the wide set used for **object deduplication** (Phase 1) — it
    should include every ref the remote has on any branch so blobs are never
    re-uploaded unnecessarily.

    *branch_have* is the narrow set used for the **commit bundle boundary**
    (Phase 5).  It must contain only the remote HEAD of the target branch so
    the BFS walks all the way back to a commit whose parent is the remote
    branch tip.  Without this separation, an intermediate branch (e.g. dev)
    sitting between the target branch's remote HEAD and the local tip can
    short-circuit the BFS and produce a bundle the server cannot verify as a
    fast-forward.
    """
    # ── Phase 1: object deduplication ────────────────────────────────────────
    # Single BFS walk — result is shared with pack building in Phase 4 so we
    # never traverse the commit graph twice in the same push.
    # Use the full `have` list here — every remote ref the server might have
    # on any branch is a potential source of already-present blobs.
    import time as _pt
    _t0 = _pt.perf_counter()
    dedup_walk = walk_commits(root, [local_head], have=have)
    _t1 = _pt.perf_counter()
    all_candidate_ids = collect_object_ids_from_walk(dedup_walk)
    _t2 = _pt.perf_counter()
    print(f"  [perf] walk_commits: {_t1-_t0:.3f}s  collect_object_ids: {_t2-_t1:.3f}s  ({len(all_candidate_ids)} objects)", file=sys.stderr)
    try:
        filter_result = transport.filter_objects(
            url, token, all_candidate_ids, dedup_walk["oid_to_path"]
        )
        _t3 = _pt.perf_counter()
        print(f"  [perf] filter_objects: {_t3-_t2:.3f}s", file=sys.stderr)
        missing_ids: set[str] = set(filter_result["missing"])
        oid_to_base: dict[str, str] = filter_result["bases"]
        skipped_count = len(all_candidate_ids) - len(missing_ids)
        if skipped_count:
            print(
                f"  Object dedup: {skipped_count} already on remote, "
                f"{len(missing_ids)} to upload.",
                file=sys.stderr,
            )
        if oid_to_base:
            print(f"  Delta bases: {len(oid_to_base)} object(s) will use delta+zlib encoding.", file=sys.stderr)
    except TransportError as _filter_exc:
        if _filter_exc.status_code == 404:
            # Server predates filter-objects endpoint — send everything.
            logger.debug("filter-objects not supported (404), falling back to full upload")
            missing_ids = set(all_candidate_ids)
            oid_to_base = {}
        else:
            # Any other error (502 mid-startup, connection refused, timeout) must
            # not silently fall back to uploading every object — that causes
            # cascading OOM crashes on small servers.  Fail loudly instead.
            raise

    # ── Phase 2: split missing objects into large (presign) and small (pack) ──
    #
    # Design goal: web-scale with millions of agents per second, no bottlenecks.
    #
    # OLD approach: presign ALL N objects → N TLS handshakes from client to R2.
    # For 7 000 objects this was 281 s on macOS (OpenSSL 3.6.x TLS 1.3 bug
    # multiplied by N connections).
    #
    # NEW approach:
    #   Small objects (≤ 1 MB): pack endpoint → M = ceil(N/1000) batched POSTs
    #     through Cloudflare → MuseHub API.  7 000 objects = 7 POSTs.  No
    #     direct R2 connections so the macOS TLS bug does not apply.
    #   Large objects (> 1 MB): presigned PUT → direct to R2.  Per-object
    #     handshake overhead is negligible vs body transfer time at this size.
    #
    # Fallback hierarchy:
    #   1. pack endpoint → if 404 (old server), fall back to presign for all.
    #   2. presign → if not supported (local://), inline → push_objects.
    presigned_ids: list[str] = []   # objects uploaded directly to R2
    inline_ids: set[str] = set()    # objects that must go through push_objects
    total_stored = 0
    total_skipped = 0

    _t_p2 = _pt.perf_counter()
    all_missing_list = list(missing_ids)

    # Read sizes to classify: large → presign, small → pack.
    # Objects are read once here; presign reads them again (acceptable trade-off
    # — avoids holding all bytes in RAM simultaneously).
    large_ids: list[str] = []   # > PACK_OBJECT_THRESHOLD — direct R2 PUT
    small_ids: list[str] = []   # ≤ PACK_OBJECT_THRESHOLD — pack endpoint POST
    for oid in all_missing_list:
        raw = read_object(root, oid)
        if raw is None or len(raw) <= PACK_OBJECT_THRESHOLD:
            small_ids.append(oid)
        else:
            large_ids.append(oid)

    # ── Phase 2a: pack endpoint for small objects ─────────────────────────────
    # M = ceil(N / 1000) round-trips replaces N individual TLS handshakes.
    pack_fallback = False   # True if server lacks pack endpoint (→ use presign)
    if small_ids:
        print(
            f"  Uploading {len(small_ids)} small object(s) via pack endpoint …",
            file=sys.stderr,
        )
        try:
            _t_pack = _pt.perf_counter()
            pack_stored, pack_skipped = _push_objects_as_packs(
                transport, url, token, small_ids, root,
                oid_to_path=dedup_walk["oid_to_path"],
                oid_to_base=oid_to_base,
                pack_origin=pack_origin,
            )
            _t_pack_end = _pt.perf_counter()
            total_stored += pack_stored
            total_skipped += pack_skipped
            print(
                f"  [perf] push/object-pack: {_t_pack_end-_t_pack:.3f}s  "
                f"({len(small_ids)} objects, {pack_stored} stored, {pack_skipped} skipped)",
                file=sys.stderr,
            )
        except TransportError as _pack_exc:
            if _pack_exc.status_code == 404:
                # Server predates push/object-pack (old deployment) — treat all
                # small objects as large and fall back to the presign path.
                logger.debug(
                    "push/object-pack not supported (404), "
                    "falling back to presign for all objects"
                )
                large_ids = all_missing_list  # send everything via presign
                small_ids = []
                pack_fallback = True
            else:
                raise

    # ── Phase 2b: presign large objects (direct R2 PUT) ──────────────────────
    # When the backend is S3/R2, large objects get presigned PUT URLs for
    # direct client→R2 upload, bypassing Cloudflare entirely.
    # When the backend is local://, presign returns everything as inline.
    presigned_map: dict[str, str] = {}
    if large_ids:
        try:
            for batch_start in range(0, max(1, len(large_ids)), CHUNK_OBJECTS):
                batch = large_ids[batch_start : batch_start + CHUNK_OBJECTS]
                presign_resp = transport.presign_objects(url, token, batch, "put")
                presigned_map.update(presign_resp["presigned"])
                inline_ids.update(presign_resp["inline"])
        except TransportError:
            logger.debug("presign not supported — all objects go through push_objects")
            presigned_map = {}
            inline_ids = set(large_ids)

    if presigned_map:
        # ── Phase 3: parallel direct-to-R2 uploads ────────────────────────────
        presigned_ids = [oid for oid in large_ids if oid in presigned_map]
        print(
            f"  Uploading {len(presigned_ids)} large object(s) directly to R2 …",
            file=sys.stderr,
        )
        sizes: dict[str, int] = {}
        presign_first_exc: Exception | None = None

        # Process in batches of CHUNK_OBJECTS to bound peak RAM usage.
        with ThreadPoolExecutor(max_workers=max_workers) as pool:
            for batch_start in range(0, max(1, len(presigned_ids)), CHUNK_OBJECTS):
                batch_ids = presigned_ids[batch_start : batch_start + CHUNK_OBJECTS]
                batch_futs: dict[Future[None], str] = {}
                for oid in batch_ids:
                    raw = read_object(root, oid) or b""
                    sizes[oid] = len(raw)
                    batch_futs[pool.submit(
                        _upload_presigned, oid, presigned_map[oid], raw
                    )] = oid
                for fut in as_completed(batch_futs):
                    try:
                        fut.result()
                    except Exception as exc:
                        logger.error("❌ Presigned upload failed for %s: %s",
                                     batch_futs[fut][:8], exc)
                        if presign_first_exc is None:
                            presign_first_exc = exc

        if presign_first_exc is not None:
            raise presign_first_exc

        # ── Phase 3b: confirm uploaded objects with server ─────────────────────
        total_registered = 0
        total_conf_skipped = 0
        for batch_start in range(0, max(1, len(presigned_ids)), CHUNK_OBJECTS):
            batch_ids = presigned_ids[batch_start : batch_start + CHUNK_OBJECTS]
            batch_sizes = {oid: sizes[oid] for oid in batch_ids if oid in sizes}
            conf = transport.confirm_objects(url, token, batch_ids, batch_sizes)
            total_registered += conf["registered"]
            total_conf_skipped += conf["skipped"]
        logger.info(
            "✅ Confirmed %d large object(s) with server (%d already registered)",
            total_registered, total_conf_skipped,
        )

    _t_p2_end = _pt.perf_counter()
    print(
        f"  [perf] object upload total: {_t_p2_end-_t_p2:.3f}s  "
        f"({len(small_ids)} via pack, {len(presigned_ids)} via R2, {len(inline_ids)} inline)",
        file=sys.stderr,
    )

    # ── Build pack (commits + snapshots; no object bytes) ─────────────────────
    # The bundle carries commits + snapshots only; object bytes are already
    # on the remote (via pack endpoint or presign) or are in inline_ids (local://).
    # Pass only inline_ids as the object boundary so bytes are not double-sent.
    _branch_have = branch_have or []
    _t_bp = _pt.perf_counter()
    if set(_branch_have) == set(have):
        bundle = build_pack_from_walk(root, dedup_walk, only_objects=inline_ids)
    else:
        branch_walk = walk_commits(root, [local_head], have=_branch_have)
        bundle = build_pack_from_walk(root, branch_walk, only_objects=inline_ids)
    _t_bp_end = _pt.perf_counter()
    print(f"  [perf] build_pack: {_t_bp_end-_t_bp:.3f}s", file=sys.stderr)

    # ── Phase 4: push_objects fallback (local:// backend only) ───────────────
    # Only runs when presign returned all objects as inline (no presign support).
    # On S3/R2 backends inline_ids is empty after Phase 2b and this is a no-op.
    # On local:// backends all objects come back as inline regardless of size.
    inline_object_ids: list[str] = sorted(inline_ids)
    total_inline = len(inline_object_ids)
    chunk_num = 0

    _t_upload = _pt.perf_counter()
    if total_inline > 0:
        total_chunks = (total_inline + CHUNK_OBJECTS - 1) // CHUNK_OBJECTS

        with ThreadPoolExecutor(max_workers=max_workers) as pool:
            futures: dict[Future[tuple[int, int]], int] = {}
            first_exc: Exception | None = None

            for chunk in stream_object_chunks(root, inline_object_ids, CHUNK_OBJECTS):
                chunk_num += 1
                fut = pool.submit(
                    _upload_chunk, transport, url, token, chunk, chunk_num, total_chunks
                )
                futures[fut] = chunk_num

            for fut in as_completed(futures):
                try:
                    stored, skipped = fut.result()
                    total_stored += stored
                    total_skipped += skipped
                except Exception as exc:
                    logger.error("❌ Chunk upload failed: %s", exc)
                    if first_exc is None:
                        first_exc = exc

        if first_exc is not None:
            raise first_exc

        logger.info(
            "✅ push/objects complete: %d stored, %d skipped across %d chunk(s)",
            total_stored, total_skipped, chunk_num,
        )

    _t_upload_end = _pt.perf_counter()
    print(
        f"  [perf] push_objects (inline fallback): {_t_upload_end-_t_upload:.3f}s"
        f"  ({total_inline} objects)",
        file=sys.stderr,
    )

    # ── Phase 5: push commits + snapshots (chunked) ───────────────────────────
    # Sending 900+ commits with their snapshot manifests in one HTTP request
    # body can easily exceed 50 MB, causing server-side body-read timeouts.
    # We chunk commits (oldest-first) so each request stays well under 5 MB.
    # Intermediate chunks use force=True (safe — they are known ancestors of
    # the final HEAD we already verified). Only the last chunk uses the caller's
    # force flag and sends local_head so the server finalises the branch ref.
    all_commits = bundle.get("commits") or []
    all_snapshots = bundle.get("snapshots") or []
    branch_heads = bundle.get("branch_heads")

    # Build snapshot lookup by snapshot_id for fast per-chunk assembly.
    snap_by_id: dict[str, object] = {}
    for s in all_snapshots:
        sid = s.snapshot_id if hasattr(s, "snapshot_id") else (s.get("snapshot_id") if isinstance(s, dict) else None)
        if sid:
            snap_by_id[sid] = s

    # walk_commits returns newest-first (BFS from HEAD). Reverse so we push
    # oldest commits first — each chunk is a valid fast-forward of the previous.
    commits_oldest_first = list(reversed(all_commits))
    commits_count = len(commits_oldest_first)

    print(
        f"  Pushing {commits_count} commit(s) "
        f"and {len(all_snapshots)} snapshot(s) …",
        file=sys.stderr,
    )

    total_chunks = max(1, (commits_count + CHUNK_COMMITS - 1) // CHUNK_COMMITS)
    result: PushResult = "pushed"
    _t_pp = _pt.perf_counter()

    for chunk_idx in range(0, max(1, commits_count), CHUNK_COMMITS):
        chunk_commits = commits_oldest_first[chunk_idx : chunk_idx + CHUNK_COMMITS]
        is_last = (chunk_idx + CHUNK_COMMITS >= commits_count)

        # Include only snapshots referenced by commits in this chunk.
        chunk_snap_ids: set[str] = set()
        for c in chunk_commits:
            sid = c.snapshot_id if hasattr(c, "snapshot_id") else (c.get("snapshot_id") if isinstance(c, dict) else None)
            if sid:
                chunk_snap_ids.add(sid)
        chunk_snaps = [snap_by_id[sid] for sid in chunk_snap_ids if sid in snap_by_id]

        chunk_bundle: PackBundle = {
            "commits": chunk_commits,
            "snapshots": chunk_snaps,
            "objects": [],
        }
        if branch_heads and is_last:
            chunk_bundle["branch_heads"] = branch_heads

        chunk_num = chunk_idx // CHUNK_COMMITS + 1
        if total_chunks > 1:
            print(f"  Chunk {chunk_num}/{total_chunks} ({len(chunk_commits)} commits) …", file=sys.stderr)

        chunk_force = True if not is_last else force
        chunk_local_head = (local_head if send_local_head else None) if is_last else None
        result = transport.push_pack(url, token, chunk_bundle, branch, chunk_force, local_head=chunk_local_head)

    _t_pp_end = _pt.perf_counter()
    print(f"  [perf] push_pack ({total_chunks} chunk(s)): {_t_pp_end-_t_pp:.3f}s  ({commits_count} commits, {len(all_snapshots)} snapshots)", file=sys.stderr)
    print(f"  [perf] TOTAL: {_t_pp_end-_t0:.3f}s", file=sys.stderr)
    total_objects = len(presigned_ids) + total_inline
    return result, commits_count, total_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`` on any transport error so callers can fall back
    gracefully instead of aborting the whole push.
    """
    try:
        return transport.fetch_remote_info(url, token)
    except TransportError:
        return None


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_pack``
    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 = root / ".muse" / "remotes"
    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
        try:
            commit_id = ref_file.read_text(encoding="utf-8", errors="ignore").strip()
        except OSError:
            continue
        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", 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(
        "--format", "-f", default="text", dest="fmt",
        help="Output format: text or json.",
    )
    parser.add_argument(
        "--json", action="store_const", const="json", dest="fmt",
        help="Shorthand for --format json.",
    )
    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":       "<remote_name>",
          "branch":       "<branch>",
          "head":         "<sha256> | null",
          "commits_sent": <N>,
          "objects_sent": <N>,
          "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
    """
    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)
    max_workers: int = max(1, getattr(args, "workers", 16))
    fmt: str = getattr(args, "fmt", "text")

    if fmt not in ("text", "json"):
        print(
            f"❌ Unknown --format '{sanitize_display(fmt)}'. Choose text or json.",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    root = require_repo()

    url = get_remote(remote, root)
    pack_origin = get_remote_pack_origin(remote, root)
    if url is None:
        print(
            f"❌ Remote '{sanitize_display(remote)}' is not configured.\n"
            f"  Add it with: muse remote add {sanitize_display(remote)} <url>",
            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:
            print(
                "❌ Specify a branch to delete: muse push <remote> --delete <branch>",
                file=sys.stderr,
            )
            raise SystemExit(ExitCode.USER_ERROR)

        if dry_run:
            msg = f"Would delete {sanitize_display(remote)}/{sanitize_display(del_branch)}"
            if fmt == "json":
                print(json.dumps(_PushJson(
                    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:
                print(
                    f"❌ Cannot delete the default branch '{sanitize_display(del_branch)}'.",
                    file=sys.stderr,
                )
                raise SystemExit(ExitCode.USER_ERROR)
            elif exc.status_code == 403:
                print(
                    "❌ Permission denied — only the repo owner may delete branches.",
                    file=sys.stderr,
                )
                raise SystemExit(ExitCode.USER_ERROR)
            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 fmt == "json":
            print(json.dumps(_PushJson(
                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:
        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_bundle = build_pack(root, [local_head], have=dry_branch_have)
        dry_commits = len(dry_bundle.get("commits") or [])
        dry_objects = len(collect_object_ids(root, [local_head], have=have))
        if fmt == "json":
            print(json.dumps(_PushJson(
                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)

    # GET /refs: live authoritative have-anchors + up-to-date check.
    remote_info = _fetch_remote_info_safe(transport, url, token)
    remote_branch_heads = remote_info["branch_heads"] if remote_info else {}

    # Resolve effective pack_origin: server-advertised value takes priority
    # over the local config override so clients need zero configuration when
    # the server has a Worker deployed.
    pack_origin = (
        remote_info.get("pack_origin") if remote_info else None
    ) or pack_origin

    candidate_have = list(remote_branch_heads.values()) + _all_known_have_anchors(root)
    have = [
        c for c in candidate_have
        if c != local_head and commit_exists(root, 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)

    # ── 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:
            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 == local_head:
        if fmt == "json":
            print(json.dumps(_PushJson(
                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[:8]}."
            )
        return

    print(
        f"Pushing {sanitize_display(push_branch)} → "
        f"{sanitize_display(remote)}/{sanitize_display(push_branch)} …",
        file=sys.stderr,
    )

    # `branch_have` is the commit-graph boundary for build_pack (Phase 5).
    # Normally it contains only the target branch's remote HEAD: using broader
    # remote refs risks stopping the BFS at an "intermediate" branch that sits
    # between the target's remote HEAD and the local tip, producing a bundle
    # the server cannot verify as a fast-forward.
    #
    # Merge commits are the exception.  The merge commit's first parent IS the
    # target branch's remote HEAD (one hop), so there is nothing "between" it
    # and local HEAD on P1's chain.  The second parent (P2) may have a long
    # ancestry that the server already has on another branch.  Without stop
    # anchors on P2's chain, the BFS walks every ancestor of P2 — potentially
    # thousands of commits the remote already stores — overwhelming the server.
    #
    # Fix: for merge commits, add all known remote branch heads to branch_have.
    # The BFS stops at the nearest already-remote commit on P2's chain instead
    # of walking back to the root of the repository.
    branch_have: list[str] = (
        [remote_head] if remote_head and commit_exists(root, remote_head) else []
    )
    local_commit = read_commit(root, local_head)
    if local_commit and local_commit.parent2_commit_id:
        # No commit_exists() guard here: the BFS stops when a parent commit_id
        # is in `seen` — it never tries to read a commit whose ID is already
        # in the seen set.  So the anchor works even if the commit isn't in
        # the local object store (e.g. a remote branch that was never fetched).
        for ref_head in remote_branch_heads.values():
            if ref_head not in branch_have:
                branch_have.append(ref_head)

    try:
        result, commits_sent, objects_sent = _push_mwp(
            transport, url, token, root, local_head, have, push_branch, force,
            max_workers=max_workers,
            branch_have=branch_have,
            pack_origin=pack_origin,
        )
    except TransportError as exc:
        if 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"]:
        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,
        )

    if fmt == "json":
        print(json.dumps(_PushJson(
            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[:8]})"
        )
# probe
# probe2