"""MWP-4 — Gate fetch.mpack.prebuild behind mpack.index (fixes RC-4).

Sub-ticket: musehub#109 — https://staging.musehub.ai/gabriel/musehub/issues/109
Master:     muse#58     — https://staging.musehub.ai/gabriel/muse/issues/58
Predecessors:
  musehub#106 (MWP-1, generation authority) — closed.
  musehub#107 (MWP-2, walk fallback) — closed.
  musehub#108 (MWP-3, job-enqueue idempotency) — closed.

Root cause (RC-4): ``claim_next_job`` orders only by ``created_at`` with no
tiebreaker.  When ``mpack.index`` and ``fetch.mpack.prebuild`` share the same
``created_at`` (the normal case from ``enqueue_push_intel``), or the prebuild
lands with an earlier timestamp due to cross-push interleaving, PostgreSQL may
return the prebuild first.  The prebuild then calls ``wire_fetch_mpack`` with
``force_build=True``, hits the index-coverage check at
``musehub_wire_fetch.py:681–694``, and raises ``FetchNotIndexedError`` because
``MusehubMPackIndex`` is still empty.  The broad ``except Exception`` at
``musehub_wire_fetch.py:1164–1167`` swallows the error silently: ``tips_built``
stays ``0``, no ``MusehubFetchMPackCache`` row is written, and the next clone
returns 503.

Phase 0 — RED reproduction (no production code changes):
  MWP4_01 RED  — ``claim_next_job`` returns the prebuild before the index
                  when the prebuild has an earlier ``created_at``.  Asserts
                  the desired correct behavior → FAILS before Phase 2 fix.
  MWP4_02      — handler-level proof: prebuild run before index swallows
                  FetchNotIndexedError and writes no cache row.  Asserts the
                  broken outcome → PASSES before fix (proves the bug).
                  Must be updated in Phase 3 when the swallow is narrowed.
"""

from __future__ import annotations

import hashlib
from datetime import datetime, timedelta, timezone

import pytest
from sqlalchemy import delete, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from muse.core.ids import hash_snapshot
from muse.core.mpack import build_wire_mpack
from muse.core.types import blob_id
from musehub.core.genesis import compute_job_id
from musehub.db.musehub_jobs_models import MusehubBackgroundJob
from musehub.db.musehub_repo_models import MusehubFetchMPackCache, MusehubMPackIndex
from musehub.services.musehub_jobs import claim_next_job, complete_job, enqueue_push_intel
from musehub.services.musehub_wire_fetch import process_fetch_mpack_prebuild_job
from musehub.services.musehub_wire_push import process_mpack_index_job, wire_push_unpack_mpack
from musehub.services.musehub_wire_shared import FetchNotIndexedError
from tests.factories import create_repo


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _fake_commit_id(seed: str) -> str:
    return "sha256:" + hashlib.sha256(seed.encode()).hexdigest()


def _now() -> datetime:
    return datetime.now(tz=timezone.utc)


def _job_row(
    repo_id: str,
    job_type: str,
    payload: dict,
    created_at: datetime,
    status: str = "pending",
) -> MusehubBackgroundJob:
    """Build a MusehubBackgroundJob row ready to be added to a session."""
    job_id = compute_job_id(repo_id, job_type, created_at.isoformat())
    return MusehubBackgroundJob(
        job_id=job_id,
        repo_id=repo_id,
        job_type=job_type,
        payload=payload,
        status=status,
        created_at=created_at,
        attempt=0,
    )


_E2E_OWNER = "gabriel"


def _e2e_raw_commit(
    cid: str,
    parent: str | None,
    snap_id: str,
    branch: str = "main",
) -> dict:
    return {
        "commit_id": cid,
        "branch": branch,
        "message": f"commit {cid[:12]}",
        "author": _E2E_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": "",
    }


async def _e2e_push(
    session: AsyncSession,
    repo_id: str,
    cid: str,
    parent: str | None,
    snap_id: str,
    fpath: str,
    oid: str,
    data: bytes,
    branch: str = "main",
) -> str:
    """Build a single-commit mpack, store it in the conftest MemoryBackend, push it.

    The conftest autouse fixture patches ``musehub.storage.backends.get_backend``
    to a ``MemoryBackend`` instance shared across all modules.  Pre-seeding
    ``backend._mpacks[key]`` is equivalent to a MinIO PUT — both
    ``wire_push_unpack_mpack`` and ``process_mpack_index_job`` call
    ``get_backend()`` and see the same bytes.

    Returns the ``mpack_key`` so callers can locate the enqueued index job.
    """
    import musehub.storage.backends as _b_mod
    backend = _b_mod.get_backend()

    mpack_bytes = build_wire_mpack({
        "commits": [_e2e_raw_commit(cid, parent, snap_id, branch=branch)],
        "snapshots": [{
            "snapshot_id": snap_id,
            "parent_snapshot_id": None,
            "delta_upsert": {fpath: oid},
            "delta_remove": [],
            "directories": [],
        }],
        "blobs": [{"object_id": oid, "content": data}],
        "tags": [],
    })
    mpack_key = blob_id(mpack_bytes)
    backend._mpacks[mpack_key] = mpack_bytes

    await wire_push_unpack_mpack(
        session,
        repo_id,
        mpack_key,
        pusher_id=_E2E_OWNER,
        branch=branch,
        head_commit_id=cid,
        commits_count=1,
        blobs_count=1,
        force=True,
    )
    await session.commit()
    return mpack_key


# ---------------------------------------------------------------------------
# MWP4_01 — claim_next_job returns prebuild before a pending index (RED)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_mwp4_01_claim_returns_prebuild_before_index(
    db_session: AsyncSession,
) -> None:
    """MWP4_01 RED: claim_next_job must NOT return fetch.mpack.prebuild while
    an mpack.index for the same repo is pending.

    Setup: insert a fetch.mpack.prebuild with created_at = now-5s (earlier
    than the mpack.index at now).  This simulates the cross-push race where
    push A's prebuild is enqueued before push B's index job arrives.

    Pre-fix (expected to FAIL):
      claim_next_job orders solely by created_at; the prebuild (earlier
      timestamp) is returned ahead of the index.  The assertion
      ``claimed.job_type == "mpack.index"`` therefore fails.

    Post-fix (Phase 2 — correlated NOT EXISTS barrier):
      The barrier prevents fetch.mpack.prebuild from being claimed while any
      mpack.index for the same repo_id is pending or running.  The index is
      returned instead, and the prebuild waits until the index drains.
    """
    repo = await create_repo(db_session, owner="gabriel")
    repo_id = str(repo.repo_id)

    now = _now()
    earlier = now - timedelta(seconds=5)

    # Insert prebuild FIRST with an earlier timestamp to force it ahead in the
    # ORDER BY created_at ordering — this is the race we're fixing.
    prebuild = _job_row(
        repo_id,
        "fetch.mpack.prebuild",
        {"tip_commit_ids": [_fake_commit_id("mwp4-01-tip")]},
        created_at=earlier,
    )
    index = _job_row(
        repo_id,
        "mpack.index",
        {
            "mpack_key": "sha256:" + hashlib.sha256(b"mwp4-01-mpack").hexdigest(),
            "head": _fake_commit_id("mwp4-01-head"),
            "branch": "main",
        },
        created_at=now,
    )
    db_session.add(prebuild)
    db_session.add(index)
    await db_session.commit()

    # Claim the next job.
    # Pre-fix:  returns the prebuild (earliest created_at wins) → assertion fails.
    # Post-fix: barrier skips the prebuild; returns the index instead.
    claimed = await claim_next_job(db_session)
    await db_session.commit()

    assert claimed is not None, "Expected a job to be claimed — queue has two pending rows."
    assert claimed.job_type == "mpack.index", (
        f"claim_next_job must skip fetch.mpack.prebuild while mpack.index is "
        f"pending for the same repo_id, but returned '{claimed.job_type}'. "
        f"The prebuild must be gated behind all pending index jobs (RC-4 barrier)."
    )


# ---------------------------------------------------------------------------
# MWP4_02 — prebuild run before index swallows FetchNotIndexedError (bug proof)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_mwp4_02_prebuild_swallows_fetch_not_indexed_error(
    db_session: AsyncSession,
) -> None:
    """MWP4_02 (updated for Phase 3): process_fetch_mpack_prebuild_job raises
    FetchNotIndexedError when MusehubMPackIndex is empty — no longer swallowed.

    Setup: push one commit so MusehubObject, MusehubCommit, and MusehubBranch
    rows exist.  wire_push_unpack_mpack writes MusehubMPackIndex rows at step 7d
    synchronously, so we explicitly delete them after the push to simulate the
    race state: DB has commits/snapshots but MusehubMPackIndex is empty (as if
    the mpack.index job hasn't run yet and has also terminally failed, bypassing
    the Phase 2 barrier).

    Phase 0 behavior (pre-Phase 3): FetchNotIndexedError was silently swallowed
    by the broad ``except Exception``; tips_built==0, no cache row.

    Phase 3 behavior: FetchNotIndexedError is caught specifically and re-raised
    so the worker's fail_job records it as a failed job (retryable).  The broad
    ``except Exception`` still handles genuinely unexpected errors.

    With the Phase 2 barrier in place this path is only reachable when an
    mpack.index has terminally failed — exactly the case that should surface as
    a failed job, not a silent empty cache.
    """
    repo = await create_repo(db_session, owner="gabriel")
    repo_id = str(repo.repo_id)

    data = b"mwp4-02-blob-content"
    oid = blob_id(data)
    snap_id = hash_snapshot({"file.txt": oid})
    cid = _fake_commit_id("mwp4-02-c1")

    # Push one commit — creates MusehubObject, MusehubCommit, MusehubBranch,
    # enqueues mpack.index + fetch.mpack.prebuild, AND writes MusehubMPackIndex
    # rows at step 7d of wire_push_unpack_mpack (synchronous inline indexing).
    mpack_key = await _e2e_push(db_session, repo_id, cid, None, snap_id, "file.txt", oid, data)

    # Simulate the RC-4 race: delete the inline MusehubMPackIndex rows that the
    # push wrote.  This replicates the state where the prebuild job is claimed
    # before the mpack.index job has had a chance to run — MusehubCommit,
    # MusehubSnapshot, and MusehubBranch are populated, but MusehubMPackIndex
    # is empty for this mpack_key.
    await db_session.execute(
        delete(MusehubMPackIndex).where(MusehubMPackIndex.mpack_id == mpack_key)
    )
    await db_session.commit()

    # Locate the prebuild job enqueued by wire_push_unpack_mpack.
    prebuild_job = (await db_session.execute(
        select(MusehubBackgroundJob).where(
            MusehubBackgroundJob.repo_id == repo_id,
            MusehubBackgroundJob.job_type == "fetch.mpack.prebuild",
            MusehubBackgroundJob.status == "pending",
        ).limit(1)
    )).scalar_one_or_none()
    assert prebuild_job is not None, (
        "wire_push_unpack_mpack must enqueue a fetch.mpack.prebuild job after a push."
    )

    # Phase 3: FetchNotIndexedError must propagate (no longer swallowed).
    # The worker's fail_job path will record this as a retryable failure.
    with pytest.raises(FetchNotIndexedError):
        await process_fetch_mpack_prebuild_job(db_session, prebuild_job.job_id)
    await db_session.rollback()


# ---------------------------------------------------------------------------
# MWP4_03 — fetch.mpack.prebuild created_at is strictly after mpack.index
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_mwp4_03_prebuild_created_at_after_index(
    db_session: AsyncSession,
) -> None:
    """MWP4_03: enqueue_push_intel must assign fetch.mpack.prebuild a created_at
    strictly greater than mpack.index and strictly less than intel.code.* subtypes.

    This is the Phase 1 defense-in-depth stagger:

        t+0s   mpack.index, intel.code, intel.structural, gc, ...  (foundation)
        t+1s   fetch.mpack.prebuild                                 (NEW middle tier)
        t+2s   intel.code.* subtypes                                (subtype tier)

    Pre-fix: FAILS — fetch.mpack.prebuild and mpack.index share the same
    created_at (both are in _FOUNDATION_TYPES at t+0s).

    Post-fix (Phase 1): PASSES — fetch.mpack.prebuild gets created_at_prebuild
    (t+1s) while mpack.index stays at the foundation tier (t+0s).
    """
    repo = await create_repo(db_session, owner="gabriel")
    repo_id = str(repo.repo_id)
    fake_head = _fake_commit_id("mwp4-03-head")
    fake_mpack_key = "sha256:" + hashlib.sha256(b"mwp4-03-mpack").hexdigest()

    await enqueue_push_intel(
        db_session,
        repo_id=repo_id,
        head=fake_head,
        domain_id=None,   # triggers code-domain subtypes (intel.code.*)
        branch="main",
        owner=None,
        mpack_key=fake_mpack_key,
    )
    await db_session.commit()

    # Fetch created_at for mpack.index
    index_row = (await db_session.execute(
        select(MusehubBackgroundJob.created_at).where(
            MusehubBackgroundJob.repo_id == repo_id,
            MusehubBackgroundJob.job_type == "mpack.index",
        ).limit(1)
    )).scalar_one_or_none()
    assert index_row is not None, "mpack.index job must be enqueued by enqueue_push_intel"

    # Fetch created_at for fetch.mpack.prebuild
    prebuild_row = (await db_session.execute(
        select(MusehubBackgroundJob.created_at).where(
            MusehubBackgroundJob.repo_id == repo_id,
            MusehubBackgroundJob.job_type == "fetch.mpack.prebuild",
        ).limit(1)
    )).scalar_one_or_none()
    assert prebuild_row is not None, "fetch.mpack.prebuild job must be enqueued by enqueue_push_intel"

    # Fetch created_at for any intel.code.* subtype (the t+2s tier)
    subtype_row = (await db_session.execute(
        select(MusehubBackgroundJob.created_at).where(
            MusehubBackgroundJob.repo_id == repo_id,
            MusehubBackgroundJob.job_type == "intel.code.coupling",
        ).limit(1)
    )).scalar_one_or_none()
    assert subtype_row is not None, (
        "intel.code.coupling job must be enqueued when domain_id=None — "
        "needed to verify the three-tier ordering"
    )

    index_at = index_row
    prebuild_at = prebuild_row
    subtype_at = subtype_row

    # Tier 1 → Tier 2: prebuild strictly after index
    assert prebuild_at > index_at, (
        f"fetch.mpack.prebuild created_at ({prebuild_at.isoformat()}) must be "
        f"strictly greater than mpack.index created_at ({index_at.isoformat()}). "
        f"Pre-fix: they share the same timestamp (both in _FOUNDATION_TYPES at t+0s). "
        f"Phase 1 fix: assign fetch.mpack.prebuild created_at_prebuild = t+1s."
    )

    # Tier 2 → Tier 3: prebuild strictly before intel.code.* subtypes
    assert prebuild_at < subtype_at, (
        f"fetch.mpack.prebuild created_at ({prebuild_at.isoformat()}) must be "
        f"strictly less than intel.code.* subtype created_at ({subtype_at.isoformat()}). "
        f"The three-tier ordering must be: foundation < prebuild < subtypes."
    )


# ---------------------------------------------------------------------------
# Phase 2 — claim-layer dependency barrier (MWP4_04 – MWP4_09)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_mwp4_04_prebuild_not_claimed_while_index_pending(
    db_session: AsyncSession,
) -> None:
    """MWP4_04: claim_next_job must NOT return fetch.mpack.prebuild while any
    mpack.index for the same repo is pending — even when the prebuild has an
    earlier created_at (the exact cross-push race from MWP4_01).

    Pre-fix: FAILS — claim returns the prebuild (earlier timestamp wins).
    Post-fix (Phase 2 barrier): PASSES — the correlated NOT EXISTS filter
    blocks the prebuild; claim returns the mpack.index instead.
    """
    repo = await create_repo(db_session, owner="gabriel")
    repo_id = str(repo.repo_id)

    now = _now()
    earlier = now - timedelta(seconds=5)

    prebuild = _job_row(
        repo_id, "fetch.mpack.prebuild",
        {"tip_commit_ids": [_fake_commit_id("mwp4-04-tip")]},
        created_at=earlier,
    )
    index = _job_row(
        repo_id, "mpack.index",
        {"mpack_key": "sha256:" + hashlib.sha256(b"mwp4-04").hexdigest(), "head": _fake_commit_id("h"), "branch": "main"},
        created_at=now,
    )
    db_session.add(prebuild)
    db_session.add(index)
    await db_session.commit()

    claimed = await claim_next_job(db_session)
    await db_session.commit()

    assert claimed is not None, "A claimable job must exist (mpack.index is pending)"
    assert claimed.job_type == "mpack.index", (
        f"claim_next_job must skip fetch.mpack.prebuild while mpack.index is "
        f"pending for the same repo — the Phase 2 barrier. Got: {claimed.job_type}"
    )


@pytest.mark.asyncio
async def test_mwp4_05_prebuild_claimable_after_index_done(
    db_session: AsyncSession,
) -> None:
    """MWP4_05: once mpack.index is in terminal state 'done', the prebuild
    must be claimable on the next claim_next_job call.

    Pre-fix: PASSES (no barrier means prebuild is always claimable by timestamp).
    Post-fix: must still PASS — 'done' is not a blocking state.
    """
    repo = await create_repo(db_session, owner="gabriel")
    repo_id = str(repo.repo_id)

    now = _now()
    index_done = _job_row(
        repo_id, "mpack.index",
        {"mpack_key": "sha256:" + hashlib.sha256(b"mwp4-05").hexdigest(), "head": _fake_commit_id("h"), "branch": "main"},
        created_at=now - timedelta(seconds=10),
        status="done",
    )
    prebuild = _job_row(
        repo_id, "fetch.mpack.prebuild",
        {"tip_commit_ids": [_fake_commit_id("mwp4-05-tip")]},
        created_at=now,
    )
    db_session.add(index_done)
    db_session.add(prebuild)
    await db_session.commit()

    claimed = await claim_next_job(db_session)
    await db_session.commit()

    assert claimed is not None, "fetch.mpack.prebuild must be claimable when mpack.index is done"
    assert claimed.job_type == "fetch.mpack.prebuild", (
        f"mpack.index in 'done' state must not block the prebuild. Got: {claimed.job_type}"
    )


@pytest.mark.asyncio
async def test_mwp4_06_prebuild_blocked_while_index_running(
    db_session: AsyncSession,
) -> None:
    """MWP4_06: an mpack.index in 'running' state (already claimed by another
    worker) must also block the prebuild — the barrier covers both pending and
    running.

    Setup: mpack.index is running (claimed), fetch.mpack.prebuild is pending.
    Expected: claim_next_job returns None (the only pending job is blocked).

    Pre-fix: FAILS — prebuild is returned (no barrier).
    Post-fix: PASSES — running mpack.index blocks the prebuild.
    """
    repo = await create_repo(db_session, owner="gabriel")
    repo_id = str(repo.repo_id)

    now = _now()
    index_running = _job_row(
        repo_id, "mpack.index",
        {"mpack_key": "sha256:" + hashlib.sha256(b"mwp4-06").hexdigest(), "head": _fake_commit_id("h"), "branch": "main"},
        created_at=now - timedelta(seconds=10),
        status="running",
    )
    prebuild = _job_row(
        repo_id, "fetch.mpack.prebuild",
        {"tip_commit_ids": [_fake_commit_id("mwp4-06-tip")]},
        created_at=now,
    )
    db_session.add(index_running)
    db_session.add(prebuild)
    await db_session.commit()

    claimed = await claim_next_job(db_session)
    await db_session.commit()

    assert claimed is None, (
        f"claim_next_job must return None when the only pending job is a "
        f"fetch.mpack.prebuild blocked by a running mpack.index. Got: "
        f"{claimed.job_type if claimed else None}"
    )


@pytest.mark.asyncio
async def test_mwp4_07_terminal_index_does_not_block_prebuild(
    db_session: AsyncSession,
) -> None:
    """MWP4_07: mpack.index rows in terminal states (failed, quarantined) must
    NOT block the prebuild.  A permanently-failed index should surface via
    fail_job, not silently hold up the prebuild forever.

    Pre-fix: PASSES (no barrier means prebuild is always claimable).
    Post-fix: must still PASS — only pending/running are blocking states.
    """
    repo = await create_repo(db_session, owner="gabriel")
    repo_id = str(repo.repo_id)

    now = _now()
    for i, status in enumerate(("failed", "quarantined")):
        index_terminal = _job_row(
            repo_id, "mpack.index",
            {"mpack_key": "sha256:" + hashlib.sha256(f"mwp4-07-{status}".encode()).hexdigest(),
             "head": _fake_commit_id(f"h-{status}"), "branch": "main"},
            # Distinct created_at per row to avoid PK collision in compute_job_id
            created_at=now - timedelta(seconds=10 + i),
            status=status,
        )
        db_session.add(index_terminal)

    prebuild = _job_row(
        repo_id, "fetch.mpack.prebuild",
        {"tip_commit_ids": [_fake_commit_id("mwp4-07-tip")]},
        created_at=now,
    )
    db_session.add(prebuild)
    await db_session.commit()

    claimed = await claim_next_job(db_session)
    await db_session.commit()

    assert claimed is not None, (
        "fetch.mpack.prebuild must be claimable when all mpack.index rows are terminal"
    )
    assert claimed.job_type == "fetch.mpack.prebuild", (
        f"Terminal mpack.index states (failed/quarantined) must not gate the "
        f"prebuild. Got: {claimed.job_type}"
    )


@pytest.mark.asyncio
async def test_mwp4_08_barrier_is_per_repo(
    db_session: AsyncSession,
) -> None:
    """MWP4_08: the barrier is scoped per repo_id.  A pending mpack.index in
    repo X must not block a fetch.mpack.prebuild in repo Y.

    Pre-fix: PASSES (no barrier means per-repo isolation is trivially true).
    Post-fix: must still PASS — the correlated subquery joins on repo_id.
    """
    repo_x = await create_repo(db_session, owner="gabriel")
    repo_y = await create_repo(db_session, owner="gabriel")
    repo_x_id = str(repo_x.repo_id)
    repo_y_id = str(repo_y.repo_id)

    now = _now()
    # Repo X: mpack.index pending (should block repo X's prebuild, not repo Y's)
    index_x = _job_row(
        repo_x_id, "mpack.index",
        {"mpack_key": "sha256:" + hashlib.sha256(b"mwp4-08-x").hexdigest(), "head": _fake_commit_id("hx"), "branch": "main"},
        created_at=now - timedelta(seconds=5),
    )
    # Repo Y: fetch.mpack.prebuild with no pending mpack.index — must be claimable
    prebuild_y = _job_row(
        repo_y_id, "fetch.mpack.prebuild",
        {"tip_commit_ids": [_fake_commit_id("mwp4-08-y-tip")]},
        created_at=now,
    )
    db_session.add(index_x)
    db_session.add(prebuild_y)
    await db_session.commit()

    claimed = await claim_next_job(db_session)
    await db_session.commit()

    # index_x has an earlier created_at than prebuild_y; without the per-repo
    # scope, index_x would be claimed. With the scope, index_x is also claimable
    # (it's not a prebuild), so it gets claimed first — that's fine. The key
    # assertion: claim must not return None (prebuild_y must not be spuriously
    # blocked by repo X's index).
    assert claimed is not None, (
        "A job must be claimable — repo Y's prebuild must not be blocked by "
        "repo X's pending mpack.index"
    )
    # Either index_x (earlier timestamp) or prebuild_y — both are acceptable.
    # The important thing is that prebuild_y is NOT blocked by a different repo's index.
    assert claimed.job_type in ("mpack.index", "fetch.mpack.prebuild"), (
        f"Unexpected job_type claimed: {claimed.job_type}"
    )
    assert claimed.repo_id != repo_x_id or claimed.job_type == "mpack.index", (
        "If repo X's job is claimed it must be the index, not any blocked prebuild"
    )


@pytest.mark.asyncio
async def test_mwp4_09_barrier_does_not_affect_other_job_types(
    db_session: AsyncSession,
) -> None:
    """MWP4_09: the barrier filters only fetch.mpack.prebuild.  A pending
    mpack.index is itself claimable, and other job types (e.g. intel.structural)
    are unaffected by the barrier.

    Pre-fix: PASSES (no barrier — all types always claimable by timestamp).
    Post-fix: must still PASS — the filter condition is
    job_type == 'fetch.mpack.prebuild' AND EXISTS(pending index for same repo).
    """
    repo = await create_repo(db_session, owner="gabriel")
    repo_id = str(repo.repo_id)

    now = _now()
    # A pending mpack.index — must still be claimable (barrier does not block index itself)
    index_job = _job_row(
        repo_id, "mpack.index",
        {"mpack_key": "sha256:" + hashlib.sha256(b"mwp4-09").hexdigest(), "head": _fake_commit_id("h"), "branch": "main"},
        created_at=now - timedelta(seconds=10),
    )
    # intel.structural — unrelated type, must be claimable regardless
    intel_job = _job_row(
        repo_id, "intel.structural",
        {"head": _fake_commit_id("h2"), "branch": "main"},
        created_at=now - timedelta(seconds=5),
    )
    db_session.add(index_job)
    db_session.add(intel_job)
    await db_session.commit()

    # Claim twice — both jobs must be claimable (neither is a blocked prebuild)
    first = await claim_next_job(db_session)
    await db_session.commit()
    second = await claim_next_job(db_session)
    await db_session.commit()

    assert first is not None, "First claim must succeed"
    assert second is not None, "Second claim must succeed — barrier must not block non-prebuild types"
    claimed_types = {first.job_type, second.job_type}
    assert "mpack.index" in claimed_types, "mpack.index must be claimable"
    assert "intel.structural" in claimed_types, "intel.structural must be claimable"


# ---------------------------------------------------------------------------
# Phase 3 — stop masking the index gap (MWP4_10)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_mwp4_10_fetch_not_indexed_error_propagates(
    db_session: AsyncSession,
) -> None:
    """MWP4_10: when wire_fetch_mpack raises FetchNotIndexedError inside
    process_fetch_mpack_prebuild_job, the handler must re-raise it instead of
    returning a success result with tips_built=0.

    This is the Phase 3 narrowing of the broad except Exception at
    musehub_wire_fetch.py:1164-1167.  After the fix, FetchNotIndexedError is
    caught specifically, logged, and re-raised so the worker's fail_job path
    records it as a retryable failure.  The broad except Exception still handles
    genuinely unexpected errors — it must NOT catch FetchNotIndexedError.

    Setup: push one commit (creates MusehubCommit, MusehubBranch, MusehubMPackIndex),
    then delete MusehubMPackIndex rows to simulate the state where a terminally-
    failed mpack.index left the index gap (the Phase 2 barrier allows the prebuild
    to run only after index work drains — terminal failures bypass the barrier).

    Pre-fix (Phase 0 behavior): FAILS — process_fetch_mpack_prebuild_job swallows
    FetchNotIndexedError and returns tips_built=0 (no propagation).

    Post-fix (Phase 3): PASSES — FetchNotIndexedError propagates to the caller.
    """
    repo = await create_repo(db_session, owner="gabriel")
    repo_id = str(repo.repo_id)

    data = b"mwp4-10-blob-content"
    oid = blob_id(data)
    snap_id = hash_snapshot({"file.txt": oid})
    cid = _fake_commit_id("mwp4-10-c1")

    mpack_key = await _e2e_push(db_session, repo_id, cid, None, snap_id, "file.txt", oid, data)

    # Delete the inline MusehubMPackIndex rows to simulate an empty index
    # (replicates the state after a terminally-failed mpack.index job).
    await db_session.execute(
        delete(MusehubMPackIndex).where(MusehubMPackIndex.mpack_id == mpack_key)
    )
    await db_session.commit()

    prebuild_job = (await db_session.execute(
        select(MusehubBackgroundJob).where(
            MusehubBackgroundJob.repo_id == repo_id,
            MusehubBackgroundJob.job_type == "fetch.mpack.prebuild",
            MusehubBackgroundJob.status == "pending",
        ).limit(1)
    )).scalar_one_or_none()
    assert prebuild_job is not None, "A fetch.mpack.prebuild job must be enqueued after push"

    # Phase 3: FetchNotIndexedError must propagate — not swallowed.
    # The worker calls fail_job on this exception, giving it retry chances.
    with pytest.raises(FetchNotIndexedError):
        await process_fetch_mpack_prebuild_job(db_session, prebuild_job.job_id)
    await db_session.rollback()


# ---------------------------------------------------------------------------
# Phase 4 — integration + stress (real worker claim loop)
# ---------------------------------------------------------------------------


async def _drain_via_claim_loop(
    session: AsyncSession,
    repo_id: str,
    max_iters: int = 60,
) -> tuple[list[str], int]:
    """Drive claim_next_job → dispatch → complete_job, mirroring _process_one.

    Returns:
        claim_log: job_types in the order they were claimed
        fetch_not_indexed_count: number of FetchNotIndexedError raised during drain
    """
    claim_log: list[str] = []
    fetch_not_indexed_count = 0

    for _ in range(max_iters):
        job = await claim_next_job(session)
        await session.commit()
        if job is None:
            break

        jid = job.job_id
        jtype = job.job_type
        claim_log.append(jtype)

        if jtype == "mpack.index":
            await process_mpack_index_job(session, jid)
            await session.commit()
        elif jtype == "fetch.mpack.prebuild":
            try:
                await process_fetch_mpack_prebuild_job(session, jid)
                await session.commit()
            except FetchNotIndexedError:
                fetch_not_indexed_count += 1
                await session.rollback()
        # all other types: pass (disabled in the worker)

        await complete_job(session, jid)
        await session.commit()

    return claim_log, fetch_not_indexed_count


@pytest.mark.asyncio
async def test_mwp4_11_e2e_claim_loop_orders_index_before_prebuild(
    db_session: AsyncSession,
) -> None:
    """MWP4_11 E2E: one push then real claim loop — mpack.index claimed and
    completed before fetch.mpack.prebuild is ever claimed.  A
    MusehubFetchMPackCache row must land at the pushed tip after drain.

    This mirrors the exact sequence musehub/worker.py::_process_one runs:
      claim_next_job → dispatch → complete_job, repeat until queue empty.

    The Phase 2 claim-layer barrier is the mechanism that enforces the ordering.
    Without it, PostgreSQL is free to return the prebuild first (RC-4).

    Assertions:
      - mpack.index appears in the claim log before any fetch.mpack.prebuild
      - A MusehubFetchMPackCache row exists for the pushed tip commit
    """
    repo = await create_repo(db_session, owner="gabriel")
    repo_id = str(repo.repo_id)

    data = b"mwp4-11-blob-content"
    oid = blob_id(data)
    snap_id = hash_snapshot({"file.txt": oid})
    cid = _fake_commit_id("mwp4-11-c1")

    await _e2e_push(db_session, repo_id, cid, None, snap_id, "file.txt", oid, data)

    claim_log, fetch_not_indexed_count = await _drain_via_claim_loop(db_session, repo_id)

    assert "mpack.index" in claim_log, (
        "mpack.index must be claimed during the drain"
    )
    assert "fetch.mpack.prebuild" in claim_log, (
        "fetch.mpack.prebuild must be claimed during the drain"
    )

    first_prebuild_pos = claim_log.index("fetch.mpack.prebuild")
    last_index_pos = max(i for i, t in enumerate(claim_log) if t == "mpack.index")
    assert last_index_pos < first_prebuild_pos, (
        f"Every mpack.index claim must complete before the first fetch.mpack.prebuild "
        f"is claimed. claim_log={claim_log}"
    )

    assert fetch_not_indexed_count == 0, (
        f"Zero FetchNotIndexedError must occur — the barrier ensures the prebuild "
        f"only runs after the index drains. Got {fetch_not_indexed_count} error(s)."
    )

    cache_count = (await db_session.execute(
        select(func.count()).select_from(MusehubFetchMPackCache).where(
            MusehubFetchMPackCache.repo_id == repo_id,
            MusehubFetchMPackCache.tip_commit_id == cid,
        )
    )).scalar_one()
    assert cache_count >= 1, (
        f"MusehubFetchMPackCache must have at least one row for tip={cid[:20]} "
        f"after the prebuild runs. Got {cache_count} rows."
    )


@pytest.mark.asyncio
async def test_mwp4_12_stress_n10_pushes_drain_with_no_ordering_violation(
    db_session: AsyncSession,
) -> None:
    """MWP4_12 stress: N=10 back-to-back pushes with no intermediate drain, then
    claim-loop drain.  Three assertions mirror the musehub#109 acceptance criteria:

    (a) Every fetch.mpack.prebuild claim happens only when zero mpack.index jobs
        are pending or running for the same repo.
    (b) Zero FetchNotIndexedError raised during the entire drain.
    (c) Final clone cache covers the latest tip: all cache rows point to the same
        mpack_id (len == 1) and a row exists for the latest pushed commit.

    N=10 creates N distinct mpack.index jobs (per-key dedup allows them all) and
    a single coalesced fetch.mpack.prebuild (coarse dedup folds them to one row).
    The MWP-3 self-coalescing fix makes the prebuild re-read live branch tips at
    run time, so it always targets the latest tip regardless of which push enqueued
    it.  RC-4 closes the remaining gap: the prebuild cannot be claimed until every
    mpack.index for the repo has drained.
    """
    repo = await create_repo(db_session, owner="gabriel")
    repo_id = str(repo.repo_id)

    N = 10
    cids = [_fake_commit_id(f"mwp4-12-c{i}") for i in range(N)]

    for i in range(N):
        data = f"mwp4-12-blob-{i}-unique-content".encode()
        oid = blob_id(data)
        snap_id = hash_snapshot({f"file_{i}.txt": oid})
        parent = cids[i - 1] if i > 0 else None
        await _e2e_push(
            db_session, repo_id, cids[i], parent,
            snap_id, f"file_{i}.txt", oid, data,
        )

    # Verify N mpack.index jobs and exactly 1 fetch.mpack.prebuild were enqueued.
    index_pending = (await db_session.execute(
        select(func.count()).select_from(MusehubBackgroundJob).where(
            MusehubBackgroundJob.repo_id == repo_id,
            MusehubBackgroundJob.job_type == "mpack.index",
            MusehubBackgroundJob.status == "pending",
        )
    )).scalar_one()
    assert index_pending == N, (
        f"Expected {N} pending mpack.index jobs after {N} pushes, got {index_pending}. "
        f"Per-key dedup (MWP-3) must allow each distinct mpack_key through."
    )

    prebuild_pending = (await db_session.execute(
        select(func.count()).select_from(MusehubBackgroundJob).where(
            MusehubBackgroundJob.repo_id == repo_id,
            MusehubBackgroundJob.job_type == "fetch.mpack.prebuild",
            MusehubBackgroundJob.status == "pending",
        )
    )).scalar_one()
    assert prebuild_pending == 1, (
        f"Coarse dedup must collapse {N} prebuild enqueues to 1 pending row. "
        f"Got {prebuild_pending}."
    )

    # Drive the claim loop and check barrier invariant at each prebuild claim.
    prebuild_barrier_violations: list[int] = []

    for _ in range(N + 30):  # N indexes + 1 prebuild + ~20 intel.* subtypes
        job = await claim_next_job(db_session)
        await db_session.commit()
        if job is None:
            break

        jid = job.job_id
        jtype = job.job_type

        if jtype == "fetch.mpack.prebuild":
            # (a) At the moment of claim, no mpack.index must be pending/running
            # for this repo.  claim_next_job committed above — DB state is stable.
            blocking = (await db_session.execute(
                select(func.count()).select_from(MusehubBackgroundJob).where(
                    MusehubBackgroundJob.repo_id == repo_id,
                    MusehubBackgroundJob.job_type == "mpack.index",
                    MusehubBackgroundJob.status.in_(("pending", "running")),
                )
            )).scalar_one()
            if blocking > 0:
                prebuild_barrier_violations.append(blocking)

            try:
                await process_fetch_mpack_prebuild_job(db_session, jid)
                await db_session.commit()
            except FetchNotIndexedError:
                await db_session.rollback()

        elif jtype == "mpack.index":
            await process_mpack_index_job(db_session, jid)
            await db_session.commit()

        # all other types: pass (mirrors the worker's disabled-jobs path)

        await complete_job(db_session, jid)
        await db_session.commit()

    # (a) No prebuild was ever claimed while mpack.index was pending/running
    assert not prebuild_barrier_violations, (
        f"The RC-4 barrier was violated: fetch.mpack.prebuild was claimed while "
        f"{prebuild_barrier_violations} mpack.index job(s) were still pending/running. "
        f"The Phase 2 correlated NOT EXISTS barrier must prevent this."
    )

    # (b) Zero FetchNotIndexedError raised during drain
    # Re-verify by checking all prebuild jobs are 'done' (not 'failed')
    failed_prebuild = (await db_session.execute(
        select(func.count()).select_from(MusehubBackgroundJob).where(
            MusehubBackgroundJob.repo_id == repo_id,
            MusehubBackgroundJob.job_type == "fetch.mpack.prebuild",
            MusehubBackgroundJob.status == "failed",
        )
    )).scalar_one()
    assert failed_prebuild == 0, (
        f"{failed_prebuild} fetch.mpack.prebuild job(s) ended in 'failed' state. "
        f"The barrier must ensure the prebuild runs only after all index work drains, "
        f"producing zero FetchNotIndexedError occurrences."
    )

    # (c) Final clone cache covers the latest tip
    latest_tip = cids[-1]
    cache_for_tip = (await db_session.execute(
        select(func.count()).select_from(MusehubFetchMPackCache).where(
            MusehubFetchMPackCache.repo_id == repo_id,
            MusehubFetchMPackCache.tip_commit_id == latest_tip,
        )
    )).scalar_one()
    assert cache_for_tip >= 1, (
        f"MusehubFetchMPackCache must have a row for the latest tip={latest_tip[:20]} "
        f"after a full N={N} drain. Got {cache_for_tip} rows."
    )

    all_mpack_ids = (await db_session.execute(
        select(MusehubFetchMPackCache.mpack_id).where(
            MusehubFetchMPackCache.repo_id == repo_id,
        )
    )).scalars().all()
    assert len(all_mpack_ids) >= 1, "At least one cache row must exist after full drain"
    unique_mpacks = set(all_mpack_ids)
    assert len(unique_mpacks) == 1, (
        f"All MusehubFetchMPackCache rows for the repo must point to the same mpack_id "
        f"after a full drain (single combined mpack covers all tips). "
        f"Got {len(unique_mpacks)} distinct mpack_ids."
    )