"""musehub#113 Phase 0 (RED) — all_oids must union every want-tip's manifest.

Sub-ticket: musehub#113 — https://staging.musehub.ai/gabriel/musehub/issues/113
Master:     muse#63     — https://staging.musehub.ai/gabriel/muse/issues/63 (MWP-9)

Root cause: ``wire_fetch_mpack`` (``musehub_wire_fetch.py:555-559``) picks a
single "the tip" snapshot via::

    select(MusehubCommitGraph.snapshot_id)
    .where(MusehubCommitGraph.commit_id.in_(_needed_cids))
    .order_by(MusehubCommitGraph.generation.desc())
    .limit(1)

and builds ``all_oids`` (the full set of blobs to include in the response,
lines 612-648) from **only that one snapshot's manifest**. When a ``want``
set spans more than one branch tip — which every clone/fetch request does,
confirmed by live testing that even a plain ``muse clone`` with no
``--branch`` flag sends every known tip — every tip except the single
globally-deepest one has its manifest silently ignored. The resulting mpack
has correct commits/snapshots metadata but is missing blobs unique to every
non-winning tip. The client reports ``exit_code: 0`` / ``"cloned"`` with an
incomplete working tree; ``muse verify`` reports ``all_ok: true``.

Confirmed live (against a freshly-restarted local musehub, not a stale
worker — see musehub#113's retraction note) via a direct
``musehub_commit_graph`` query and three real clone scenarios:

  MTU_01 — two sibling branches at the SAME generation (a tie): the
           tie-losing sibling's unique blob is dropped.
  MTU_02 — two sibling branches at DIFFERENT generations (no tie, the
           shallower one loses deterministically): the shallower sibling's
           unique blob is dropped.
  MTU_03 — the DEFAULT branch itself loses its own newest content merely
           because some other branch in the repo is deeper. This is the
           realistic "just run `muse clone <url>`" case, not an edge case.

Expected Phase 0 status: MTU_01, MTU_02, MTU_03 all RED — each currently
fails because the losing tip's blob is absent from the fetch result.

The fix (Phase 1) makes ``all_oids`` the union of every want-tip's own
manifest, resolved from ``MusehubCommit.snapshot_id`` (already loaded into
``commit_rows``, already the authoritative source per the existing MWP1_13
guard's own comment: "MusehubCommit always has the correct snapshot_id") —
not a single CommitGraph-generation-ranked pick.
"""
from __future__ import annotations

import datetime
from unittest.mock import patch

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from muse.core.ids import hash_snapshot
from muse.core.mpack import build_wire_mpack, parse_wire_mpack
from muse.core.types import blob_id
from musehub.core.genesis import compute_identity_id
from musehub.services.musehub_repository import create_repo
from musehub.services.musehub_wire_fetch import wire_fetch_mpack
from musehub.services.musehub_wire_push import wire_push_unpack_mpack

_OWNER = "gabriel"
_IDENTITY_ID = compute_identity_id(b"gabriel")


# ---------------------------------------------------------------------------
# Helpers — mirrors test_mwp1_generation_authority.py / test_mwp2_walk_fallback.py
# ---------------------------------------------------------------------------


def _cid(seed: str) -> str:
    """Deterministic sha256: commit id from a seed string."""
    return blob_id(f"mtu-commit-{seed}".encode())


def _now() -> datetime.datetime:
    return datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)


def _raw_commit(cid: str, parent: str | None, snap_id: str, *, branch: str = "main") -> dict:
    """A raw commit dict in the shape wire_push_unpack_mpack consumes."""
    return {
        "commit_id": cid,
        "branch": branch,
        "message": f"commit {cid[:12]}",
        "author": _OWNER,
        "committed_at": _now().isoformat(),
        "parent_commit_id": parent,
        "parent2_commit_id": None,
        "snapshot_id": snap_id,
        "agent_id": "",
        "model_id": "",
        "toolchain_id": "",
        "sem_ver_bump": "none",
        "breaking_changes": [],
        "signature": "",
        "signer_key_id": "",
        "signer_public_key": "",
        "prompt_hash": "",
    }


class _InMemBackend:
    """In-memory storage backend for E2E push->fetch tests."""

    def __init__(self) -> None:
        self._mpacks: dict[str, bytes] = {}

    async def put_mpack(self, mpack_id: str, data: bytes) -> None:
        self._mpacks[mpack_id] = data

    async def get_mpack(self, mpack_id: str) -> bytes | None:
        return self._mpacks.get(mpack_id)

    async def presign_mpack_get(self, mpack_id: str, ttl: int = 3600) -> str:
        return f"inmem://{mpack_id}"

    async def put(self, oid: str, data: bytes) -> str:
        return oid

    async def get(self, oid: str) -> bytes | None:
        return None

    async def presign_get(self, oid: str, ttl: int = 3600) -> str:
        return f"inmem://{oid}"

    async def quarantine_mpack(self, *args: object, **kwargs: object) -> None:
        pass


async def _push_full_manifest(
    session: AsyncSession,
    backend: _InMemBackend,
    repo_id: str,
    branch: str,
    cid: str,
    parent: str | None,
    manifest: dict[str, str],
    blobs: dict[str, bytes],
) -> None:
    """Push one commit whose snapshot's manifest is the FULL (non-delta) file set.

    ``parent_snapshot_id=None`` + a complete ``delta_upsert`` is a valid,
    self-contained root snapshot representation — avoids relying on
    delta-chain reconstruction for this test's purposes.
    """
    snap_id = hash_snapshot(manifest)
    mpack_bytes = build_wire_mpack({
        "commits": [_raw_commit(cid, parent, snap_id, branch=branch)],
        "snapshots": [{
            "snapshot_id": snap_id,
            "parent_snapshot_id": None,
            "delta_upsert": manifest,
            "delta_remove": [],
            "directories": [],
        }],
        "blobs": [{"object_id": oid, "content": data} for oid, data in blobs.items()],
        "tags": [],
    })
    mpack_key = blob_id(mpack_bytes)
    backend._mpacks[mpack_key] = mpack_bytes
    with patch("musehub.services.musehub_wire.get_backend", return_value=backend), \
         patch("musehub.services.musehub_wire_push.get_backend", return_value=backend), \
         patch("musehub.storage.backends.get_backend", return_value=backend):
        await wire_push_unpack_mpack(
            session,
            repo_id,
            mpack_key,
            pusher_id=_OWNER,
            branch=branch,
            head_commit_id=cid,
            commits_count=1,
            blobs_count=len(blobs),
            force=True,
        )
    await session.flush()


async def _fetch(
    session: AsyncSession,
    backend: _InMemBackend,
    repo_id: str,
    want: list[str],
    have: list[str] | None = None,
) -> dict:
    with patch("musehub.services.musehub_wire.get_backend", return_value=backend), \
         patch("musehub.services.musehub_wire_fetch.get_backend", return_value=backend), \
         patch("musehub.storage.backends.get_backend", return_value=backend):
        result = await wire_fetch_mpack(session, repo_id, want=want, have=have or [], force_build=True)
    if not result.get("mpack_id"):
        return {"commits": [], "snapshots": [], "blobs": []}
    mpack_bytes = backend._mpacks[result["mpack_id"]]
    return parse_wire_mpack(mpack_bytes)


# ---------------------------------------------------------------------------
# MTU_01 — sibling branches at the SAME generation (a tie)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_mtu_01_sibling_tie_both_manifests_included(db_session: AsyncSession) -> None:
    """RED: feature-a and feature-b are both gen-1 children of main (a tie).

    want=[feature-a tip, feature-b tip] must include BOTH a.txt and b.txt.
    Today only whichever tip wins the arbitrary tie-break is included.
    """
    repo = await create_repo(
        db_session, name="mtu-01", owner=_OWNER, owner_user_id=_IDENTITY_ID,
        visibility="public", initialize=False,
    )
    await db_session.commit()
    backend = _InMemBackend()

    oid_shared = blob_id(b"mtu01-shared")
    oid_a = blob_id(b"mtu01-a")
    oid_b = blob_id(b"mtu01-b")
    m1 = _cid("01-main")
    fa1 = _cid("01-feature-a")
    fb1 = _cid("01-feature-b")

    await _push_full_manifest(
        db_session, backend, repo.repo_id, "main", m1, None,
        {"shared.txt": oid_shared}, {oid_shared: b"shared"},
    )
    await _push_full_manifest(
        db_session, backend, repo.repo_id, "feature-a", fa1, m1,
        {"shared.txt": oid_shared, "a.txt": oid_a}, {oid_a: b"aa"},
    )
    await _push_full_manifest(
        db_session, backend, repo.repo_id, "feature-b", fb1, m1,
        {"shared.txt": oid_shared, "b.txt": oid_b}, {oid_b: b"bb"},
    )

    parsed = await _fetch(db_session, backend, repo.repo_id, want=[fa1, fb1])
    blob_oids = {b["object_id"] for b in parsed.get("blobs", [])}

    assert oid_a in blob_oids, (
        f"feature-a's unique blob must be present; blobs found: {[o[:16] for o in blob_oids]}"
    )
    assert oid_b in blob_oids, (
        f"feature-b's unique blob must be present; blobs found: {[o[:16] for o in blob_oids]}"
    )


# ---------------------------------------------------------------------------
# MTU_02 — sibling branches at DIFFERENT generations (no tie)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_mtu_02_shallower_sibling_manifest_included(db_session: AsyncSession) -> None:
    """RED: feature-deep (gen 3) and feature-shallow (gen 1) share base main.

    want=[feature-deep tip, feature-shallow tip] must include shallow.txt.
    Today feature-deep unambiguously wins ORDER BY generation DESC LIMIT 1,
    so feature-shallow's manifest — and shallow.txt — is never consulted.
    """
    repo = await create_repo(
        db_session, name="mtu-02", owner=_OWNER, owner_user_id=_IDENTITY_ID,
        visibility="public", initialize=False,
    )
    await db_session.commit()
    backend = _InMemBackend()

    oid_shared = blob_id(b"mtu02-shared")
    oid_d1, oid_d2, oid_d3 = (blob_id(f"mtu02-d{i}".encode()) for i in (1, 2, 3))
    oid_shallow = blob_id(b"mtu02-shallow")

    m1 = _cid("02-main")
    fd1, fd2, fd3 = _cid("02-fd1"), _cid("02-fd2"), _cid("02-fd3")
    fs1 = _cid("02-fs1")

    await _push_full_manifest(
        db_session, backend, repo.repo_id, "main", m1, None,
        {"shared.txt": oid_shared}, {oid_shared: b"shared"},
    )
    await _push_full_manifest(
        db_session, backend, repo.repo_id, "feature-deep", fd1, m1,
        {"shared.txt": oid_shared, "d1.txt": oid_d1}, {oid_d1: b"d1"},
    )
    await _push_full_manifest(
        db_session, backend, repo.repo_id, "feature-deep", fd2, fd1,
        {"shared.txt": oid_shared, "d1.txt": oid_d1, "d2.txt": oid_d2}, {oid_d2: b"d2"},
    )
    await _push_full_manifest(
        db_session, backend, repo.repo_id, "feature-deep", fd3, fd2,
        {"shared.txt": oid_shared, "d1.txt": oid_d1, "d2.txt": oid_d2, "d3.txt": oid_d3},
        {oid_d3: b"d3"},
    )
    await _push_full_manifest(
        db_session, backend, repo.repo_id, "feature-shallow", fs1, m1,
        {"shared.txt": oid_shared, "shallow.txt": oid_shallow}, {oid_shallow: b"shallow"},
    )

    parsed = await _fetch(db_session, backend, repo.repo_id, want=[fd3, fs1])
    blob_oids = {b["object_id"] for b in parsed.get("blobs", [])}

    assert oid_shallow in blob_oids, (
        f"feature-shallow's unique blob must be present even though feature-deep "
        f"is unambiguously deeper (no tie); blobs found: {[o[:16] for o in blob_oids]}"
    )


# ---------------------------------------------------------------------------
# MTU_03 — the DEFAULT branch loses its own content to a deeper sibling
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_mtu_03_default_branch_keeps_its_own_newest_content(db_session: AsyncSession) -> None:
    """RED: main gets its own new commit; feature-deep is deeper elsewhere.

    want=[main tip, feature-deep tip] (mirrors a real clone, which always
    requests every known tip) must include main's own newest blob. This is
    the realistic "just run `muse clone <url>`" case, not a contrived edge
    case: any repo with an active feature branch deeper than main will
    silently corrupt clones of main itself today.
    """
    repo = await create_repo(
        db_session, name="mtu-03", owner=_OWNER, owner_user_id=_IDENTITY_ID,
        visibility="public", initialize=False,
    )
    await db_session.commit()
    backend = _InMemBackend()

    oid_shared = blob_id(b"mtu03-shared")
    oid_main_unique = blob_id(b"mtu03-main-unique")
    oid_d1, oid_d2, oid_d3 = (blob_id(f"mtu03-d{i}".encode()) for i in (1, 2, 3))

    m1 = _cid("03-main1")
    m2 = _cid("03-main2")
    fd1, fd2, fd3 = _cid("03-fd1"), _cid("03-fd2"), _cid("03-fd3")

    await _push_full_manifest(
        db_session, backend, repo.repo_id, "main", m1, None,
        {"shared.txt": oid_shared}, {oid_shared: b"shared"},
    )
    await _push_full_manifest(
        db_session, backend, repo.repo_id, "feature-deep", fd1, m1,
        {"shared.txt": oid_shared, "d1.txt": oid_d1}, {oid_d1: b"d1"},
    )
    await _push_full_manifest(
        db_session, backend, repo.repo_id, "feature-deep", fd2, fd1,
        {"shared.txt": oid_shared, "d1.txt": oid_d1, "d2.txt": oid_d2}, {oid_d2: b"d2"},
    )
    await _push_full_manifest(
        db_session, backend, repo.repo_id, "feature-deep", fd3, fd2,
        {"shared.txt": oid_shared, "d1.txt": oid_d1, "d2.txt": oid_d2, "d3.txt": oid_d3},
        {oid_d3: b"d3"},
    )
    # main gets its own new commit — pushed directly to the default branch,
    # never inherited by feature-deep.
    await _push_full_manifest(
        db_session, backend, repo.repo_id, "main", m2, m1,
        {"shared.txt": oid_shared, "main_unique.txt": oid_main_unique},
        {oid_main_unique: b"main-only-content"},
    )

    parsed = await _fetch(db_session, backend, repo.repo_id, want=[m2, fd3])
    blob_oids = {b["object_id"] for b in parsed.get("blobs", [])}

    assert oid_main_unique in blob_oids, (
        f"main's own newest blob must be present in its own clone regardless of "
        f"any other branch's depth; blobs found: {[o[:16] for o in blob_oids]}"
    )


# ---------------------------------------------------------------------------
# MTU_04 — the mirror bug: `have_oids` also only consults ONE have-tip
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_mtu_04_have_oids_unions_every_have_tip(db_session: AsyncSession) -> None:
    """RED: have_oids must union every tip in `have`, not pick one.

    Client already fully has BOTH have-shallow (adds file_a.txt, generation 1)
    and have-deep (two commits deeper, generation 3 — deterministically wins
    any generation-ranked pick, no tie involved). A new push, `target`, is a
    sibling of have-shallow that reuses have-shallow's exact file_a.txt
    content (same oid) plus adds its own new file_target.txt.

    want=[target], have=[have-shallow tip, have-deep tip]. Correct
    behaviour: the response must exclude file_a.txt's oid (the client
    already has it via have-shallow) and include only file_target.txt's
    oid as genuinely new.

    Today `have_oids` is built from a single CommitGraph generation-ranked
    pick across the whole `have` set — have-deep unambiguously wins, so
    have-shallow's manifest (and file_a.txt's oid) is never consulted, and
    file_a.txt is wrongly treated as new. Not a data-loss bug like
    MTU_01-03 — a wasted-bandwidth bug: the server redundantly re-sends
    content the client already has via a sibling it correctly reported
    having.
    """
    repo = await create_repo(
        db_session, name="mtu-04", owner=_OWNER, owner_user_id=_IDENTITY_ID,
        visibility="public", initialize=False,
    )
    await db_session.commit()
    backend = _InMemBackend()

    oid_shared = blob_id(b"mtu04-shared")
    oid_a = blob_id(b"mtu04-file-a")
    oid_d1, oid_d2 = blob_id(b"mtu04-deep1"), blob_id(b"mtu04-deep2")
    oid_target = blob_id(b"mtu04-file-target")

    m1 = _cid("04-main")
    hs1 = _cid("04-have-shallow")
    hd1, hd2 = _cid("04-have-deep-1"), _cid("04-have-deep-2")
    t1 = _cid("04-target")

    await _push_full_manifest(
        db_session, backend, repo.repo_id, "main", m1, None,
        {"shared.txt": oid_shared}, {oid_shared: b"shared"},
    )
    await _push_full_manifest(
        db_session, backend, repo.repo_id, "have-shallow", hs1, m1,
        {"shared.txt": oid_shared, "file_a.txt": oid_a}, {oid_a: b"file-a-content"},
    )
    await _push_full_manifest(
        db_session, backend, repo.repo_id, "have-deep", hd1, m1,
        {"shared.txt": oid_shared, "deep1.txt": oid_d1}, {oid_d1: b"deep1"},
    )
    await _push_full_manifest(
        db_session, backend, repo.repo_id, "have-deep", hd2, hd1,
        {"shared.txt": oid_shared, "deep1.txt": oid_d1, "deep2.txt": oid_d2}, {oid_d2: b"deep2"},
    )
    # target reuses file_a's exact oid (client already has it via
    # have-shallow) and adds one genuinely new file.
    await _push_full_manifest(
        db_session, backend, repo.repo_id, "target", t1, m1,
        {"shared.txt": oid_shared, "file_a.txt": oid_a, "file_target.txt": oid_target},
        {oid_a: b"file-a-content", oid_target: b"file-target-content"},
    )

    parsed = await _fetch(db_session, backend, repo.repo_id, want=[t1], have=[hs1, hd2])
    blob_oids = {b["object_id"] for b in parsed.get("blobs", [])}

    assert oid_target in blob_oids, (
        f"file_target.txt is genuinely new and must be sent; blobs found: {[o[:16] for o in blob_oids]}"
    )
    assert oid_a not in blob_oids, (
        f"file_a.txt must NOT be re-sent — the client already has it via "
        f"have-shallow, one of the two tips in `have`. Its presence here means "
        f"have_oids only consulted have-deep's manifest (the deterministic "
        f"generation winner), not the union of both have-tips. "
        f"blobs found: {[o[:16] for o in blob_oids]}"
    )