"""TDD — Phase 1: Ground truth for #55 sibling-ref push negotiation.

Issue:   https://staging.musehub.ai/gabriel/muse/issues/55
Tracker: https://staging.musehub.ai/gabriel/muse/issues/62

Scenario under test (the #55 shape):
    remote has:  dev → D   (a linear chain root..D, full manifests on remote)
    local has:   dev → D   AND   main → M, where M = merge commit with parent D
    push main →  remote_branch_heads = {"dev": D}   (main absent on remote)
    expected:    commits_sent == 1 (just M),  objects_sent ≈ 0 (only M's net-new blobs)

Test IDs
--------
GT_01  helper builders (_chain, _merge_commit)
GT_02  commit-boundary truth — walk_commits stops at sibling tip D
GT_03  blob-dedup truth — all_blob_ids contains only M's net-new blob(s)
GT_04  run()-level live path truth — branch_have from all remote heads; commits_sent==1
GT_05  diverged / force edge — main present at old diverged tip M_old on remote

GROUND TRUTH VERDICT: CORRECT — Phase 2 Mode A applies.

All GT_02..GT_05 tests pass (9/9 green, 0.53 s, 2026-06-29).

GT_02  PASS — walk_commits(root, [M], have=[D])["commits"] == 1  ✓
GT_03  PASS — all_blob_ids contains only M's net-new blob; D's manifest excluded  ✓
GT_04  PASS — run() builds branch_have from all remote heads; D included; commits_sent==1  ✓
GT_05  PASS — diverged M_old case: branch_have=[D, M_old]; commits_sent==1  ✓

Conclusion: the live push path (push.py:819-822) correctly uses ALL remote branch
heads as the BFS boundary.  The RC-7 claim is empirically verified.  #55 is
closed at the logic level — staging evidence (Phase 3) is still required to
formally close the issue.  Phase 2 Mode A: promote GT tests to regression
assertions; run adjacent push corpus (LF_01..LF_03).
"""
from __future__ import annotations

import argparse
import datetime
import io
import json
import pathlib
from contextlib import redirect_stdout
from typing import Any
from unittest.mock import MagicMock, patch

import pytest

from muse.core.commits import read_commit, write_commit, CommitRecord
from muse.core.mpack import walk_commits, collect_blob_ids
from muse.core.object_store import write_object
from muse.core.paths import init_repo_dirs as init_repo
from muse.core.refs import write_branch_ref
from muse.core.snapshot import compute_commit_id, compute_snapshot_id
from muse.core.snapshots import read_snapshot, write_snapshot, SnapshotRecord
from muse.core.types import Manifest, blob_id

pytestmark = pytest.mark.wire

_TS = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)


# ---------------------------------------------------------------------------
# GT_01 — Helper builders
#
# Reuses patterns from:
#   tests/test_push_branch_have.py::_commit / _build_linear_chain
#   tests/test_push_have_filter.py::_make_commit
# ---------------------------------------------------------------------------

def _obj(root: pathlib.Path, content: bytes) -> str:
    """Write a blob to the object store and return its ID."""
    oid = blob_id(content)
    write_object(root, oid, content)
    return oid


def _snap(root: pathlib.Path, manifest: Manifest) -> str:
    """Write a snapshot for *manifest* and return its ID."""
    sid = compute_snapshot_id(manifest)
    write_snapshot(root, SnapshotRecord(snapshot_id=sid, manifest=manifest))
    return sid


def _chain(root: pathlib.Path, n: int) -> list[str]:
    """Build a linear commit chain of *n* commits; return IDs oldest first.

    Each commit adds exactly one unique file (file0.txt … fileN-1.txt) so
    every commit contributes one new blob and the full chain has n blobs.
    """
    commits: list[str] = []
    manifest: dict[str, str] = {}
    prev: str | None = None
    for i in range(n):
        content = f"chain-file-{i}".encode()
        oid = _obj(root, content)
        manifest = {**manifest, f"file{i}.txt": oid}
        sid = _snap(root, manifest)
        cid = compute_commit_id(
            parent_ids=[prev] if prev else [],
            snapshot_id=sid,
            message=f"commit {i}",
            committed_at_iso=_TS.isoformat(),
            author="test",
        )
        write_commit(root, CommitRecord(
            commit_id=cid,
            branch="dev",
            snapshot_id=sid,
            message=f"commit {i}",
            committed_at=_TS,
            parent_commit_id=prev,
            author="test",
        ))
        commits.append(cid)
        prev = cid
    return commits


def _merge_commit(
    root: pathlib.Path,
    parent_id: str,
    content: bytes = b"merge-file-content",
    branch: str = "main",
) -> str:
    """Create a commit whose manifest = parent's manifest + 1 new file.

    Models the merge-commit shape from #55: main advances by exactly one
    commit on top of dev's tip, adding one new blob.  Everything else in the
    manifest is shared with the parent, so the blob dedup in walk_commits
    will correctly exclude all pre-existing blobs when have=[parent].
    """
    parent_commit = read_commit(root, parent_id)
    assert parent_commit is not None, f"parent commit {parent_id[:16]} not found"
    parent_snap = read_snapshot(root, parent_commit.snapshot_id)
    assert parent_snap is not None, f"parent snapshot not found for {parent_id[:16]}"

    new_oid = _obj(root, content)
    manifest: dict[str, str] = {**parent_snap.manifest, "merge-file.txt": new_oid}
    sid = _snap(root, manifest)

    cid = compute_commit_id(
        parent_ids=[parent_id],
        snapshot_id=sid,
        message=f"merge: {branch} from dev",
        committed_at_iso=_TS.isoformat(),
        author="test",
    )
    write_commit(root, CommitRecord(
        commit_id=cid,
        branch=branch,
        snapshot_id=sid,
        message=f"merge: {branch} from dev",
        committed_at=_TS,
        parent_commit_id=parent_id,
        author="test",
    ))
    return cid


# ---------------------------------------------------------------------------
# GT_02 — Commit-boundary truth
# ---------------------------------------------------------------------------

class TestGT02CommitBoundary:
    """walk_commits(root, [M], have=[D]) yields exactly one commit: M."""

    def test_GT02_walk_stops_at_sibling_tip(self, tmp_path: pathlib.Path) -> None:
        """Assert len(walk_commits(root, [M], have=[D])["commits"]) == 1.

        Proves the BFS commit walk stops the moment it reaches the sibling
        branch tip D, leaving only the merge commit M in commits_to_send.
        This is the core invariant for the #55 scenario: the remote already
        holds every commit reachable from D (via the dev push), so the push
        of main should send only M.

        Mirrors test_push_branch_have.py::test_BH2 but framed for #55's
        merge-commit shape rather than a linear second-branch push.
        """
        init_repo(tmp_path)
        commits = _chain(tmp_path, 4)  # root, C1, C2, D
        D = commits[-1]
        M = _merge_commit(tmp_path, D)

        result = walk_commits(tmp_path, [M], have=[D])
        sent_ids = [c.commit_id for c in result["commits"]]

        assert len(sent_ids) == 1, (
            f"Expected commits_sent=1 (only M); got {len(sent_ids)}: "
            f"{[cid[:16] for cid in sent_ids]}"
        )
        assert M in sent_ids, "M (the merge commit) must be the one commit sent"
        assert D not in sent_ids, (
            "D must NOT be sent — it is already on the remote under dev"
        )

    def test_GT02_empty_have_sends_full_chain(self, tmp_path: pathlib.Path) -> None:
        """Baseline: have=[] causes walk to return all commits (no pruning)."""
        init_repo(tmp_path)
        commits = _chain(tmp_path, 4)
        D = commits[-1]
        M = _merge_commit(tmp_path, D)

        result = walk_commits(tmp_path, [M], have=[])
        # M + the 4 chain commits
        assert len(result["commits"]) == 5, (
            f"With have=[], all 5 commits should be walked; got {len(result['commits'])}"
        )


# ---------------------------------------------------------------------------
# GT_03 — Blob-dedup truth
# ---------------------------------------------------------------------------

class TestGT03BlobDedup:
    """walk_commits blob subtraction correctly excludes all blobs from D's manifest."""

    def test_GT03_only_merge_blob_in_all_blob_ids(self, tmp_path: pathlib.Path) -> None:
        """all_blob_ids contains only M's net-new blob; none of D's manifest blobs.

        Directly exercises have_blobs subtraction at mpack.py:330-362.
        The flat-manifest model means D's snapshot lists every file in the
        entire repo at that point.  Subtracting D's have_blobs from
        manifest_blobs leaves only the one blob M adds — merge-file.txt.
        """
        init_repo(tmp_path)
        commits = _chain(tmp_path, 4)
        D = commits[-1]
        merge_content = b"unique-merge-file-content-for-GT03"
        M = _merge_commit(tmp_path, D, content=merge_content)

        result = walk_commits(tmp_path, [M], have=[D])
        all_blob_ids: list[str] = result["all_blob_ids"]

        # M's new blob must be present.
        expected_oid = blob_id(merge_content)
        assert expected_oid in all_blob_ids, (
            f"M's new blob ({expected_oid[:16]}) must appear in all_blob_ids"
        )

        # No blob from D's manifest should appear — they are all in have_blobs.
        d_commit = read_commit(tmp_path, D)
        assert d_commit is not None
        d_snap = read_snapshot(tmp_path, d_commit.snapshot_id)
        assert d_snap is not None
        d_blobs = set(d_snap.manifest.values())

        stale_blobs = d_blobs & set(all_blob_ids)
        assert not stale_blobs, (
            "Blobs from D's manifest must NOT appear in all_blob_ids — they are "
            "in have_blobs and must be excluded from the push delta. "
            f"Leaking: {[b[:16] for b in stale_blobs]}"
        )

    def test_GT03_have_blobs_covers_entire_chain(self, tmp_path: pathlib.Path) -> None:
        """have_blobs includes every blob reachable from D's snapshot manifest.

        Because Muse snapshots are flat full-repo manifests, D's snapshot
        manifest lists every file from root through D.  Subtracting it from
        manifest_blobs leaves only M's additions.
        """
        init_repo(tmp_path)
        commits = _chain(tmp_path, 4)
        D = commits[-1]
        M = _merge_commit(tmp_path, D)

        result = walk_commits(tmp_path, [M], have=[D])
        have_blobs: set[str] = result["have_blobs"]

        d_commit = read_commit(tmp_path, D)
        assert d_commit is not None
        d_snap = read_snapshot(tmp_path, d_commit.snapshot_id)
        assert d_snap is not None

        for path, oid in d_snap.manifest.items():
            assert oid in have_blobs, (
                f"Blob {oid[:16]} for path {path!r} (from D's manifest) "
                f"must be in have_blobs — the flat-manifest model guarantees this"
            )

    def test_GT03_only_one_new_blob_in_merge_case(self, tmp_path: pathlib.Path) -> None:
        """all_blob_ids has exactly 1 entry: M's single net-new blob."""
        init_repo(tmp_path)
        commits = _chain(tmp_path, 4)
        D = commits[-1]
        M = _merge_commit(tmp_path, D, content=b"just-one-new-file")

        result = walk_commits(tmp_path, [M], have=[D])
        assert len(result["all_blob_ids"]) == 1, (
            f"Exactly 1 new blob (merge-file.txt); got {len(result['all_blob_ids'])}"
        )


# ---------------------------------------------------------------------------
# GT_04 — run()-level live path truth
# ---------------------------------------------------------------------------

def _make_push_ns(
    remote: str = "staging",
    branch: str = "main",
    force: bool = False,
    dry_run: bool = False,
    json_out: bool = False,
) -> argparse.Namespace:
    """Build an argparse.Namespace for push.run() with sensible defaults."""
    return argparse.Namespace(
        remote=remote,
        branch_pos=branch,
        branch_flag=None,
        force=force,
        force_with_lease=False,
        delete_branch=False,
        dry_run=dry_run,
        json_out=json_out,
        set_upstream_flag=False,
        workers=4,
        hub=None,
    )


def _write_tracking_ref(root: pathlib.Path, remote: str, branch: str, commit_id: str) -> None:
    """Write a local tracking ref to simulate a previous push of remote/branch.

    The file at .muse/remotes/<remote>/<branch> is what _all_known_have_anchors
    and get_remote_head read.  Writing it simulates "this branch was pushed
    to this remote at this commit" without doing a real push.
    """
    ref_file = root / ".muse" / "remotes" / remote / branch
    ref_file.parent.mkdir(parents=True, exist_ok=True)
    ref_file.write_text(commit_id)


def _run_dry_push(
    tmp_path: pathlib.Path,
    local_head: str,
    remote: str = "staging",
) -> dict[str, Any]:
    """Invoke push.run() in dry-run JSON mode; return parsed JSON from stdout.

    Does NOT mock _push_mpack (dry-run never reaches it).
    Tracking refs are read from disk — write them with _write_tracking_ref first.
    """
    out = io.StringIO()
    ns = _make_push_ns(remote=remote, dry_run=True, json_out=True)
    with (
        patch("muse.cli.commands.push.require_repo", return_value=tmp_path),
        patch("muse.cli.commands.push.read_current_branch", return_value="main"),
        patch("muse.cli.commands.push.get_head_commit_id", return_value=local_head),
        patch("muse.cli.commands.push.get_remote", return_value="https://staging.musehub.ai/repo"),
        redirect_stdout(out),
    ):
        from muse.cli.commands.push import run
        run(ns)
    return json.loads(out.getvalue())


def _spy_push_mpack(
    captured: list[dict[str, Any]],
) -> Any:
    """Return a spy function for _push_mpack that captures branch_have, commits_sent, objects_sent."""
    def _spy(
        transport: Any,
        url: str,
        signing: Any,
        root: pathlib.Path,
        local_head: str,
        have: list[str],
        branch: str,
        force: bool,
        branch_have: list[str] | None = None,
    ) -> None:
        # Reproduce the exact calculation _push_mpack performs, without network I/O.
        walk = walk_commits(root, [local_head], have=branch_have or [])
        captured.append({
            "branch_have": list(branch_have or []),
            "have": list(have),
            "commits_sent": len(walk["commits"]),
            "objects_sent": len(walk["all_blob_ids"]),
        })
        raise SystemExit(0)  # abort before any network I/O
    return _spy


class TestGT04LivePathTruth:
    """push.py::run() builds branch_have from ALL remote heads; commits_sent==1 for #55 shape."""

    def test_GT04_run_uses_sibling_dev_as_commit_boundary(
        self, tmp_path: pathlib.Path
    ) -> None:
        """In-process proxy for the staging #55 scenario.

        Transport returns branch_heads={"dev": D} — main is absent on the remote.
        Spy on _push_mpack to capture (branch_have, commits_sent) that run()
        computes.  Asserts:
          - branch_have contains D (dev's remote tip used as BFS boundary)
          - commits_sent == 1  (only M — the merge commit — is new)

        If commits_sent > 1 here, the live path (#55 claim) is NOT correct and
        Phase 2 Mode B must fix push.py:819-822.
        """
        init_repo(tmp_path)
        commits = _chain(tmp_path, 4)
        D = commits[-1]
        M = _merge_commit(tmp_path, D)
        write_branch_ref(tmp_path, "dev", D)
        write_branch_ref(tmp_path, "main", M)

        captured: list[dict[str, Any]] = []
        mock_transport = MagicMock()
        mock_transport.fetch_remote_info.return_value = {
            "branch_heads": {"dev": D},
            "domain": "code",
            "default_branch": "dev",
        }

        with (
            patch("muse.cli.commands.push.require_repo", return_value=tmp_path),
            patch("muse.cli.commands.push.read_current_branch", return_value="main"),
            patch("muse.cli.commands.push.get_head_commit_id", return_value=M),
            patch("muse.cli.commands.push.get_remote", return_value="https://staging.musehub.ai/repo"),
            patch("muse.cli.commands.push.get_signing_identity", return_value=None),
            patch("muse.cli.commands.push.make_transport", return_value=mock_transport),
            patch("muse.cli.commands.push._push_mpack", _spy_push_mpack(captured)),
        ):
            from muse.cli.commands.push import run
            try:
                run(_make_push_ns())
            except SystemExit:
                pass

        assert captured, (
            "_push_mpack spy was never called — run() aborted before reaching _push_mpack. "
            "Check mock setup or whether there is a new early-exit path."
        )
        c = captured[0]

        # branch_have must contain D (dev's remote tip).
        assert D in c["branch_have"], (
            f"branch_have must include D (dev's remote tip) so the BFS stops there. "
            f"Got: {[h[:16] for h in c['branch_have']]}. "
            f"If D is absent, run() is NOT using all remote heads as the walk boundary."
        )

        # commits_sent must be 1 — only M.
        assert c["commits_sent"] == 1, (
            f"Expected commits_sent=1 (only M, the merge commit). "
            f"Got {c['commits_sent']}. "
            f"If > 1, the #55 bug is present in the live path: run() is walking back "
            f"into commits already on the remote via dev, re-sending their full history."
        )

    def test_GT04_branch_have_excludes_local_head(
        self, tmp_path: pathlib.Path
    ) -> None:
        """branch_have must not contain local_head (M), only remote heads.

        Including local_head would cause walk_commits to prune immediately and
        report commits_sent=0, giving a false 'up_to_date' result.
        """
        init_repo(tmp_path)
        commits = _chain(tmp_path, 3)
        D = commits[-1]
        M = _merge_commit(tmp_path, D)
        write_branch_ref(tmp_path, "dev", D)
        write_branch_ref(tmp_path, "main", M)

        captured: list[dict[str, Any]] = []
        mock_transport = MagicMock()
        mock_transport.fetch_remote_info.return_value = {
            "branch_heads": {"dev": D},
            "domain": "code",
            "default_branch": "dev",
        }

        with (
            patch("muse.cli.commands.push.require_repo", return_value=tmp_path),
            patch("muse.cli.commands.push.read_current_branch", return_value="main"),
            patch("muse.cli.commands.push.get_head_commit_id", return_value=M),
            patch("muse.cli.commands.push.get_remote", return_value="https://staging.musehub.ai/repo"),
            patch("muse.cli.commands.push.get_signing_identity", return_value=None),
            patch("muse.cli.commands.push.make_transport", return_value=mock_transport),
            patch("muse.cli.commands.push._push_mpack", _spy_push_mpack(captured)),
        ):
            from muse.cli.commands.push import run
            try:
                run(_make_push_ns())
            except SystemExit:
                pass

        assert captured
        assert M not in captured[0]["branch_have"], (
            "M (local_head) must not appear in branch_have — "
            "including it would prune the walk at M itself and report commits_sent=0"
        )


# ---------------------------------------------------------------------------
# GT_05 — Diverged / force edge
# ---------------------------------------------------------------------------

class TestGT05DivergedEdge:
    """Remote has main at M_old (an older, diverged tip not on M's ancestry).

    This is the most likely place a residual #55 bug hides: the up-to-date
    short-circuit (push.py:789) and the remote_head / have split (push.py:749-757).

    With branch_have=[D, M_old] and M's parent = D:
      - walk_commits(root, [M], have=[D, M_old]) prunes at D → commits_sent=1
      - If the code wrongly uses branch_have=[M_old] only:
        walk does not stop at D → walks back through D, C, B → over-counts
    """

    def test_GT05_diverged_remote_main_sends_only_merge_commit(
        self, tmp_path: pathlib.Path
    ) -> None:
        """Push main when remote/main is at a diverged old tip M_old.

        branch_have must include BOTH D (dev's tip) AND M_old (main's old tip)
        so the BFS stops at D, yielding commits_sent=1.
        """
        init_repo(tmp_path)
        # Chain: commits[0]=root, commits[1]=A, commits[2]=B, commits[3]=C, commits[4]=D
        commits = _chain(tmp_path, 5)
        D = commits[-1]
        B = commits[2]

        # M = merge commit on top of D (the correct local main)
        M = _merge_commit(tmp_path, D, content=b"new-merge-content-GT05")
        # M_old = an old merge commit branching from B (diverged from D)
        M_old = _merge_commit(tmp_path, B, content=b"old-merge-content-GT05", branch="main")

        write_branch_ref(tmp_path, "dev", D)
        write_branch_ref(tmp_path, "main", M)

        captured: list[dict[str, Any]] = []
        mock_transport = MagicMock()
        # Remote knows both branches: dev at D, main at M_old (the old diverged tip).
        mock_transport.fetch_remote_info.return_value = {
            "branch_heads": {"dev": D, "main": M_old},
            "domain": "code",
            "default_branch": "dev",
        }

        with (
            patch("muse.cli.commands.push.require_repo", return_value=tmp_path),
            patch("muse.cli.commands.push.read_current_branch", return_value="main"),
            patch("muse.cli.commands.push.get_head_commit_id", return_value=M),
            patch("muse.cli.commands.push.get_remote", return_value="https://staging.musehub.ai/repo"),
            patch("muse.cli.commands.push.get_signing_identity", return_value=None),
            patch("muse.cli.commands.push.make_transport", return_value=mock_transport),
            patch("muse.cli.commands.push._push_mpack", _spy_push_mpack(captured)),
        ):
            from muse.cli.commands.push import run
            # force=True: M and M_old are diverged; without force the push would be rejected
            try:
                run(_make_push_ns(force=True))
            except SystemExit:
                pass

        assert captured, (
            "_push_mpack spy was never called. "
            "Check whether force=True triggers an unexpected early-exit path."
        )
        c = captured[0]

        # Both remote heads must be in branch_have.
        assert D in c["branch_have"], (
            f"D (dev's remote tip) must be in branch_have. "
            f"Got: {[h[:16] for h in c['branch_have']]}"
        )
        assert M_old in c["branch_have"], (
            f"M_old (main's old remote tip) must be in branch_have. "
            f"Got: {[h[:16] for h in c['branch_have']]}"
        )

        # With branch_have=[D, M_old] and M's parent = D:
        # walk_commits prunes at D → only M is sent.
        assert c["commits_sent"] == 1, (
            f"Expected commits_sent=1: M's parent D is in branch_have so walk stops there. "
            f"Got {c['commits_sent']}. "
            f"If > 1, there is a residual #55 bug in the diverged-tip handling path: "
            f"branch_have is not including all remote heads as the BFS boundary."
        )

    def test_GT05_main_absent_on_remote_also_sends_one(
        self, tmp_path: pathlib.Path
    ) -> None:
        """Control: main absent on remote (original #55 shape) → commits_sent=1.

        Repeats GT_04 as a cross-check under GT_05 naming to make the
        GT_05 diverged vs GT_04 absent comparison explicit.
        """
        init_repo(tmp_path)
        commits = _chain(tmp_path, 4)
        D = commits[-1]
        M = _merge_commit(tmp_path, D, content=b"GT05-control-content")
        write_branch_ref(tmp_path, "dev", D)
        write_branch_ref(tmp_path, "main", M)

        captured: list[dict[str, Any]] = []
        mock_transport = MagicMock()
        mock_transport.fetch_remote_info.return_value = {
            "branch_heads": {"dev": D},   # main absent
            "domain": "code",
            "default_branch": "dev",
        }

        with (
            patch("muse.cli.commands.push.require_repo", return_value=tmp_path),
            patch("muse.cli.commands.push.read_current_branch", return_value="main"),
            patch("muse.cli.commands.push.get_head_commit_id", return_value=M),
            patch("muse.cli.commands.push.get_remote", return_value="https://staging.musehub.ai/repo"),
            patch("muse.cli.commands.push.get_signing_identity", return_value=None),
            patch("muse.cli.commands.push.make_transport", return_value=mock_transport),
            patch("muse.cli.commands.push._push_mpack", _spy_push_mpack(captured)),
        ):
            from muse.cli.commands.push import run
            try:
                run(_make_push_ns())
            except SystemExit:
                pass

        assert captured
        assert c["commits_sent"] == 1 if (c := captured[0]) else True, (
            f"commits_sent should be 1 (only M); got {captured[0]['commits_sent']}"
        )


# ---------------------------------------------------------------------------
# Helpers for Phase 2 (LF_01, LF_02)
# ---------------------------------------------------------------------------

def _merge_commit_pure(
    root: pathlib.Path,
    parent_id: str,
    branch: str = "main",
) -> str:
    """Create a merge commit with the SAME manifest as the parent — zero new blobs.

    This models the common case where merging dev into main when main was
    directly behind dev produces a commit whose snapshot is identical to dev's
    tip.  objects_sent for this commit should be 0 (every blob is already on
    the remote via dev).
    """
    parent_commit = read_commit(root, parent_id)
    assert parent_commit is not None, f"parent commit {parent_id[:16]} not found"
    # Reuse the parent's snapshot_id — same manifest, same objects.
    sid = parent_commit.snapshot_id

    cid = compute_commit_id(
        parent_ids=[parent_id],
        snapshot_id=sid,
        message=f"merge: {branch} from dev (pure — no new files)",
        committed_at_iso=_TS.isoformat(),
        author="test",
    )
    write_commit(root, CommitRecord(
        commit_id=cid,
        branch=branch,
        snapshot_id=sid,
        message=f"merge: {branch} from dev (pure — no new files)",
        committed_at=_TS,
        parent_commit_id=parent_id,
        author="test",
    ))
    return cid


def _run_push_spy(
    tmp_path: pathlib.Path,
    local_head: str,
    remote_branch_heads: dict[str, str],
    force: bool = False,
) -> dict[str, Any]:
    """Invoke push.run() with a spy on _push_mpack; return the captured result dict."""
    captured: list[dict[str, Any]] = []
    mock_transport = MagicMock()
    mock_transport.fetch_remote_info.return_value = {
        "branch_heads": remote_branch_heads,
        "domain": "code",
        "default_branch": next(iter(remote_branch_heads), "dev"),
    }
    with (
        patch("muse.cli.commands.push.require_repo", return_value=tmp_path),
        patch("muse.cli.commands.push.read_current_branch", return_value="main"),
        patch("muse.cli.commands.push.get_head_commit_id", return_value=local_head),
        patch("muse.cli.commands.push.get_remote", return_value="https://staging.musehub.ai/repo"),
        patch("muse.cli.commands.push.get_signing_identity", return_value=None),
        patch("muse.cli.commands.push.make_transport", return_value=mock_transport),
        patch("muse.cli.commands.push._push_mpack", _spy_push_mpack(captured)),
    ):
        from muse.cli.commands.push import run
        try:
            run(_make_push_ns(force=force))
        except SystemExit:
            pass
    assert captured, "spy was never called — check mock setup"
    return captured[0]


# ---------------------------------------------------------------------------
# LF_01 — Permanent regression assertion for #55
# ---------------------------------------------------------------------------

class TestLF01RegressionAssertion:
    """Permanent contract: pushing main after dev sends ONLY the merge commit.

    This test was promoted from GT_04 and is the named regression anchor for
    muse#55 (sibling-ref push negotiation).  Its presence in the wire suite
    ensures the fix can never silently regress.

    Before the fix: commits_sent ≈ 1375, objects_sent ≈ 379 MB of redundant
    blobs (the entire chain already on the remote via dev).
    After verification: commits_sent == 1, objects_sent == 1 (one new blob).
    """

    def test_LF_push_main_after_dev_sends_only_merge_commit(
        self, tmp_path: pathlib.Path
    ) -> None:
        """Regression guard for muse#55 — the 1375-commit over-send is closed.

        Scenario: dev has been pushed (D and all ancestors are on the remote).
        Local main = D + one merge commit M.  push main → remote only needs M.

        Assert commits_sent == 1.  If this assertion ever fails, the sibling-ref
        push negotiation has regressed: run() is no longer using all remote
        branch heads as the BFS boundary (push.py:819-822).

        Historical data point from issue #55:
            BEFORE fix: commits_sent ≈ 1375, objects ≈ 379 MB (full re-send)
            AFTER fix:  commits_sent == 1,    objects == 1 new blob
        """
        init_repo(tmp_path)
        commits = _chain(tmp_path, 4)
        D = commits[-1]
        M = _merge_commit(tmp_path, D, content=b"LF01-merge-content")
        write_branch_ref(tmp_path, "dev", D)
        write_branch_ref(tmp_path, "main", M)

        c = _run_push_spy(tmp_path, M, remote_branch_heads={"dev": D})

        assert c["commits_sent"] == 1, (
            f"REGRESSION: commits_sent={c['commits_sent']} (expected 1). "
            f"muse#55 has regressed — push.py:819-822 is no longer using "
            f"all remote_branch_heads.values() as the commit-walk boundary."
        )
        assert D in c["branch_have"], (
            f"REGRESSION: D (dev's remote tip) not in branch_have. "
            f"Got: {[h[:16] for h in c['branch_have']]}"
        )

    def test_LF_large_chain_sends_only_new_commits(
        self, tmp_path: pathlib.Path
    ) -> None:
        """Scale variant: 17-commit dev chain (mirrors the #55 incident report).

        The original incident had dev at 17 commits / 46 blobs.  Verify that
        pushing a main that is dev + 1 merge commit still sends only 1 commit
        regardless of the chain length.
        """
        init_repo(tmp_path)
        commits = _chain(tmp_path, 17)
        D = commits[-1]
        M = _merge_commit(tmp_path, D, content=b"LF01-17-commit-merge")
        write_branch_ref(tmp_path, "dev", D)
        write_branch_ref(tmp_path, "main", M)

        c = _run_push_spy(tmp_path, M, remote_branch_heads={"dev": D})

        assert c["commits_sent"] == 1, (
            f"Expected commits_sent=1 for 17-commit chain; got {c['commits_sent']}. "
            f"The #55 regression is back."
        )
        # objects_sent == 1: just the merge blob (merge-file.txt is new)
        assert c["objects_sent"] == 1, (
            f"Expected objects_sent=1 (one new blob from M); got {c['objects_sent']}."
        )


# ---------------------------------------------------------------------------
# LF_02 — objects_sent parametrized: 0 (pure merge) and K (merge with new blobs)
# ---------------------------------------------------------------------------

class TestLF02ObjectsSent:
    """objects_sent reflects only blobs genuinely absent from the remote.

    Parametrized over two cases:
      n_new_blobs=0 (pure merge — M's manifest == D's manifest)
        → objects_sent == 0; every blob already on remote via dev
      n_new_blobs=1 (one new file added in the merge commit)
        → objects_sent == 1; one net-new blob not reachable from D
    """

    @pytest.mark.parametrize("n_new_blobs,expected_objects", [
        pytest.param(0, 0, id="pure_merge_zero_new_blobs"),
        pytest.param(1, 1, id="merge_with_one_new_blob"),
    ])
    def test_LF_no_net_new_blobs_when_sibling_has_all_objects(
        self,
        tmp_path: pathlib.Path,
        n_new_blobs: int,
        expected_objects: int,
    ) -> None:
        """objects_sent == expected_objects for the given merge shape.

        pure_merge_zero_new_blobs (n_new_blobs=0):
            M re-uses D's snapshot verbatim — no new files, no new blobs.
            All blobs are in have_blobs (via D) → objects_sent == 0.
            This is the ideal merge: when dev already contains everything
            and main fast-forwards by recording the merge, the push of main
            after dev is purely metadata (one commit, zero objects).

        merge_with_one_new_blob (n_new_blobs=1):
            M adds one file not in D's manifest → objects_sent == 1.
            Verifies we are not over-excluding (under-sending): the one
            genuinely new blob must still be transmitted.
        """
        init_repo(tmp_path)
        commits = _chain(tmp_path, 4)
        D = commits[-1]

        if n_new_blobs == 0:
            M = _merge_commit_pure(tmp_path, D)
        else:
            M = _merge_commit(tmp_path, D, content=b"LF02-one-new-blob")

        write_branch_ref(tmp_path, "dev", D)
        write_branch_ref(tmp_path, "main", M)

        c = _run_push_spy(tmp_path, M, remote_branch_heads={"dev": D})

        assert c["commits_sent"] == 1, (
            f"commits_sent must always be 1 (only M) for both parametrize cases; "
            f"got {c['commits_sent']}"
        )
        assert c["objects_sent"] == expected_objects, (
            f"objects_sent={c['objects_sent']} (expected {expected_objects}) "
            f"for n_new_blobs={n_new_blobs}. "
            + (
                "pure merge: M re-uses D's snapshot so all blobs are in have_blobs — "
                "objects_sent must be 0."
                if expected_objects == 0 else
                "merge with 1 new blob: that blob must NOT be excluded — "
                "objects_sent must be 1."
            )
        )


# ---------------------------------------------------------------------------
# DR_01 — Red repro of #32: dry-run over-counts commits when sibling exists
# ---------------------------------------------------------------------------

class TestDR01DryRunCommitCount:
    """DR_01: dry-run commits_sent must equal live commits_sent (==1) for #55 shape.

    Bug (#32): push.py:668-672 builds dry_branch_have from only the TARGET branch's
    cached tip (get_remote_head("staging", "main", root)).  When main has not yet
    been pushed, this is None → dry_branch_have=[] → walk from M touches ALL commits
    → over-counts.

    Fix: use have = _all_known_have_anchors(root) for the commit walk too (the same
    set already used for object dedup at push.py:664-667).

    Before fix: commits_sent == len(chain) + 1  (over-count, RED)
    After fix:  commits_sent == 1               (correct, GREEN)
    """

    def test_DR01_dry_run_overcounts_when_main_not_yet_pushed(
        self, tmp_path: pathlib.Path
    ) -> None:
        """Repro of #32: dry_branch_have=[] when remote/main tracking ref is absent.

        Scenario: dev has been pushed (tracking ref staging/dev = D exists).
        Local main = D + merge commit M.  Main has NOT been pushed yet.

        Current code: get_remote_head("staging", "main") → None → dry_branch_have=[]
            → walk_commits(root, [M], have=[]) walks ALL commits → over-counts.
        Fixed code:   have = _all_known_have_anchors(root) = [D]
            → walk_commits(root, [M], have=[D]) → commits_sent=1.

        MUST be RED against current push.py (push.py:668-672 not yet fixed).
        """
        init_repo(tmp_path)
        commits = _chain(tmp_path, 4)
        D = commits[-1]
        M = _merge_commit(tmp_path, D, content=b"DR01-merge-content")
        write_branch_ref(tmp_path, "dev", D)
        write_branch_ref(tmp_path, "main", M)
        # Simulate dev having been pushed: tracking ref staging/dev = D
        _write_tracking_ref(tmp_path, "staging", "dev", D)
        # staging/main does NOT exist (main not yet pushed)

        result = _run_dry_push(tmp_path, M)

        assert result["commits_sent"] == 1, (
            f"DR_01 REPRO (#32): dry-run commits_sent={result['commits_sent']} (expected 1). "
            f"Before fix: dry_branch_have=[] causes walk to return all "
            f"{result['commits_sent']} commits. "
            f"After fix (push.py:663-674): _all_known_have_anchors returns [D], "
            f"walk_commits(root, [M], have=[D]) yields commits_sent=1."
        )

    def test_DR01_large_chain_mirrors_incident_report(
        self, tmp_path: pathlib.Path
    ) -> None:
        """17-commit chain variant — mirrors the literal #32/#55 incident report.

        The #32 report observed commits_sent: 1271 for a push --dry-run that
        the live push completed in commits_sent: 0.  This test uses the same
        scale (17 commits, same as the #55 shape) to confirm the fix works at
        the incident's order of magnitude.

        Before fix: commits_sent == 18  (all 17 chain + merge commit)
        After fix:  commits_sent == 1
        """
        init_repo(tmp_path)
        commits = _chain(tmp_path, 17)
        D = commits[-1]
        M = _merge_commit(tmp_path, D, content=b"DR01-17-commit-merge")
        write_branch_ref(tmp_path, "dev", D)
        write_branch_ref(tmp_path, "main", M)
        _write_tracking_ref(tmp_path, "staging", "dev", D)

        result = _run_dry_push(tmp_path, M)

        assert result["commits_sent"] == 1, (
            f"DR_01 large-chain: dry-run commits_sent={result['commits_sent']} (expected 1). "
            f"Before fix: walk counts all 18 commits (17 dev + merge). "
            f"After fix: walk stops at D (tracking ref staging/dev), yielding commits_sent=1."
        )


# ---------------------------------------------------------------------------
# DR_02 — Object dimension: dry-run objects_sent already correct (document)
# ---------------------------------------------------------------------------

class TestDR02ObjectsSent:
    """DR_02: dry-run objects_sent already uses all tracking refs — document and guard.

    The dry-run block at push.py:664-667 correctly builds `have` from
    _all_known_have_anchors(root) for collect_blob_ids.  Only the commit walk
    (dry_branch_have) was broken.  DR_02 documents this and provides a permanent
    guard that objects parity is not accidentally regressed.

    After the DR_01 fix, both commit and object counts use the same `have` set.
    """

    def test_DR02_objects_sent_matches_live_merge_with_one_blob(
        self, tmp_path: pathlib.Path
    ) -> None:
        """Dry-run objects_sent==1 matches live push when merge adds one new blob.

        Already correct before the fix (the `have` variable already uses
        _all_known_have_anchors).  This test is a permanent guard — it must
        remain green after DR_03 and must never silently regress to over- or
        under-counting objects.
        """
        init_repo(tmp_path)
        commits = _chain(tmp_path, 4)
        D = commits[-1]
        M = _merge_commit(tmp_path, D, content=b"DR02-one-new-blob")
        write_branch_ref(tmp_path, "dev", D)
        write_branch_ref(tmp_path, "main", M)
        _write_tracking_ref(tmp_path, "staging", "dev", D)

        result = _run_dry_push(tmp_path, M)

        assert result["objects_sent"] == 1, (
            f"DR_02: dry-run objects_sent={result['objects_sent']} (expected 1). "
            f"The object dimension already uses _all_known_have_anchors (push.py:664-667) "
            f"so exactly the 1 new blob from the merge commit should be counted. "
            f"If this fails, collect_blob_ids dedup has regressed."
        )

    def test_DR02_objects_sent_zero_for_pure_merge(
        self, tmp_path: pathlib.Path
    ) -> None:
        """Pure merge commit (same manifest as parent) → dry-run objects_sent==0.

        The merge commit re-uses D's snapshot_id — all blobs already in have_blobs.
        After the fix, commits_sent==1 and objects_sent==0 for this case.
        """
        init_repo(tmp_path)
        commits = _chain(tmp_path, 4)
        D = commits[-1]
        M = _merge_commit_pure(tmp_path, D)
        write_branch_ref(tmp_path, "dev", D)
        write_branch_ref(tmp_path, "main", M)
        _write_tracking_ref(tmp_path, "staging", "dev", D)

        result = _run_dry_push(tmp_path, M)

        assert result["objects_sent"] == 0, (
            f"DR_02 pure merge: dry-run objects_sent={result['objects_sent']} (expected 0). "
            f"M re-uses D's snapshot — every blob is in have_blobs. "
            f"objects_sent must be 0 (no new blobs to send)."
        )


# ---------------------------------------------------------------------------
# RG_01 — End-to-end #32 scenario: local main == remote dev → commits_sent=0
# ---------------------------------------------------------------------------

class TestRG01UpToDate:
    """RG_01: the literal #32 report — 1271 → 0.

    In the #32 incident, the live push sent commits_sent=0 (up-to-date: remote dev
    already held every commit local main had).  The dry-run reported 1271, because
    dry_branch_have ignored sibling tracking refs.

    After the Phase 4 fix (use _all_known_have_anchors for the commit walk), we still
    need to include local_head itself in the have anchor when a sibling tracking ref
    equals local_head.  The Phase 4 fix filtered it out with `c != local_head`, which
    incorrectly yielded commits_sent=len(chain) instead of 0 for this exact scenario.

    Fix: remove the `c != local_head` guard — walk_commits prunes local_head on entry
    when it is already in have_set, cleanly reporting commits_sent=0.  This does NOT
    break the DR_01 scenario because there local_head=M != D, so D remains in have.
    """

    def test_RG01_dry_run_zero_when_local_main_equals_remote_dev(
        self, tmp_path: pathlib.Path
    ) -> None:
        """Push --dry-run main when main == dev tip that was already pushed.

        Scenario: dev was pushed at D.  Local main has been fast-forwarded to D
        (no new commits on top).  Push --dry-run main should report commits_sent=0
        and objects_sent=0 — identical to the live push's up-to-date result.

        Historical data point from issue #32:
            BEFORE fix: commits_sent ≈ 1271 (walks full history ignoring have)
            AFTER fix:  commits_sent == 0 (local_head D is in have_set; pruned immediately)
        """
        init_repo(tmp_path)
        commits = _chain(tmp_path, 4)
        D = commits[-1]
        # local main is at D — same commit the remote already holds via dev
        write_branch_ref(tmp_path, "dev", D)
        write_branch_ref(tmp_path, "main", D)
        _write_tracking_ref(tmp_path, "staging", "dev", D)

        result = _run_dry_push(tmp_path, D)

        assert result["commits_sent"] == 0, (
            f"RG_01 (#32): dry-run commits_sent={result['commits_sent']} (expected 0). "
            f"local main == remote dev (both at D); the remote already holds every commit. "
            f"After the full fix (c != local_head removed): walk_commits([D], have=[D]) "
            f"prunes D immediately → commits_sent=0. "
            f"If > 0, the `c != local_head` filter is still excluding D from have."
        )
        assert result["objects_sent"] == 0, (
            f"RG_01: objects_sent={result['objects_sent']} (expected 0). "
            f"All blobs already on remote via dev."
        )

    def test_RG01_dry_run_zero_when_main_tracking_ref_equals_local_head(
        self, tmp_path: pathlib.Path
    ) -> None:
        """Re-push dry-run: staging/main already points at local_head → 0 commits.

        Second variant: main was previously pushed to staging (staging/main = M).
        Local main is still at M (no new commits).  Dry-run should report 0,
        matching the live up-to-date short-circuit.
        """
        init_repo(tmp_path)
        commits = _chain(tmp_path, 3)
        D = commits[-1]
        M = _merge_commit(tmp_path, D, content=b"RG01-repush-variant")
        write_branch_ref(tmp_path, "dev", D)
        write_branch_ref(tmp_path, "main", M)
        _write_tracking_ref(tmp_path, "staging", "dev", D)
        # staging/main also points at M (it was pushed in a prior session)
        _write_tracking_ref(tmp_path, "staging", "main", M)

        result = _run_dry_push(tmp_path, M)

        assert result["commits_sent"] == 0, (
            f"RG_01 re-push variant: dry-run commits_sent={result['commits_sent']} "
            f"(expected 0). staging/main already points at M (local_head). "
            f"walk_commits([M], have=[M, D]) prunes M immediately → 0."
        )


# ---------------------------------------------------------------------------
# RG_02 — Non-zero control: genuinely new commits → correct count
# ---------------------------------------------------------------------------

class TestRG02NonZeroControl:
    """RG_02: non-zero control — new commits produce the correct count.

    Verifies that the fix (removing c != local_head) does NOT cause under-sending:
    when there ARE genuinely new commits not reachable from any tracking ref, they
    are all counted correctly.  Maps to #32's VL_02.
    """

    def test_RG02_two_new_commits_counted_correctly(
        self, tmp_path: pathlib.Path
    ) -> None:
        """Two commits beyond D → commits_sent=2.

        Topology:
            chain(4): root→C1→C2→D
            M  = merge commit on top of D (1 new file)
            N  = one more commit on top of M (1 new file)
            staging/dev = D

        push --dry-run main (at N) should report commits_sent=2 (M and N),
        objects_sent=2 (one blob per new commit).  Proves we are not under-counting
        because of the has-it-all-already shortcut.
        """
        init_repo(tmp_path)
        commits = _chain(tmp_path, 4)
        D = commits[-1]
        M = _merge_commit(tmp_path, D, content=b"RG02-blob-1")
        N = _merge_commit(tmp_path, M, content=b"RG02-blob-2", branch="main")
        write_branch_ref(tmp_path, "dev", D)
        write_branch_ref(tmp_path, "main", N)
        _write_tracking_ref(tmp_path, "staging", "dev", D)

        result = _run_dry_push(tmp_path, N)

        assert result["commits_sent"] == 2, (
            f"RG_02: dry-run commits_sent={result['commits_sent']} (expected 2). "
            f"N and M are both new (not in any tracking ref); "
            f"walk stops at D (staging/dev)."
        )
        assert result["objects_sent"] == 2, (
            f"RG_02: objects_sent={result['objects_sent']} (expected 2). "
            f"Two new blobs (one per new commit); D's blobs are in have_blobs."
        )

    def test_RG02_single_new_commit_matches_DR01_result(
        self, tmp_path: pathlib.Path
    ) -> None:
        """Single new merge commit → commits_sent=1 (matches DR_01).

        Cross-check that RG_02's result is consistent with DR_01's: one new
        commit above a pushed sibling always yields commits_sent=1.
        """
        init_repo(tmp_path)
        commits = _chain(tmp_path, 4)
        D = commits[-1]
        M = _merge_commit(tmp_path, D, content=b"RG02-single-new")
        write_branch_ref(tmp_path, "dev", D)
        write_branch_ref(tmp_path, "main", M)
        _write_tracking_ref(tmp_path, "staging", "dev", D)

        result = _run_dry_push(tmp_path, M)

        assert result["commits_sent"] == 1
        assert result["objects_sent"] == 1