"""MWP-9 — fetch.mpack.prebuild dedup must cover 'running', not just 'pending'.

Root cause (found live, while running a bulk commit-signature repair against
staging.musehub.ai, during the muse#108 wire-protocol investigation):
``fetch.mpack.prebuild`` is explicitly self-coalescing by design --
``process_fetch_mpack_prebuild_job`` re-reads live branch tips at run time
rather than trusting its enqueue-time payload (see its own docstring). But
every enqueue site (the regular post-push path in ``enqueue_push_intel``,
and the repair path in ``wire_repair_commit`` -> ``_enqueue_repair_prebuild``)
only dedups against jobs with ``status == "pending"``.

For a repo whose full rebuild takes longer than the gap between two triggers
(confirmed live: ~1624 commits / 162MB took 30-100+s to rebuild under load),
the first job transitions pending -> running, and every subsequent trigger
(another push, another repair-commit call) sees no "pending" row and happily
enqueues a brand new one -- even though the currently-running job will,
by its own self-coalescing design, already produce a mpack covering
whatever triggered the new enqueue. Each of those redundant jobs is a full,
independent rebuild. Under rapid repeated triggers (a bulk repair script
issuing 180+ calls in a few minutes, or just several pushes in quick
succession) this floods the single-worker queue and starves ordinary
clone/fetch requests past their 120s retry budget -- confirmed live: 3 of 5
consecutive clone attempts hit 4x 503 retries and failed outright during a
repair batch run.

Fix: for this specific, explicitly self-coalescing job type, dedup against
both 'pending' and 'running' at both enqueue sites.
"""
from __future__ import annotations

import hashlib
from datetime import datetime, timedelta, timezone

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from musehub.core.genesis import compute_job_id
from musehub.db.musehub_jobs_models import MusehubBackgroundJob
from musehub.services.musehub_jobs import (
    _MAX_ATTEMPTS,
    enqueue_push_intel,
    reclaim_stale_jobs,
)
from musehub.services.musehub_wire_push import _enqueue_repair_prebuild
from tests.factories import create_branch, create_repo


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


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


async def _insert_prebuild_job(session: AsyncSession, repo_id: str, status: str) -> str:
    now = _now()
    job_id = compute_job_id(repo_id, "fetch.mpack.prebuild", now.isoformat() + status)
    session.add(MusehubBackgroundJob(
        job_id=job_id,
        repo_id=repo_id,
        job_type="fetch.mpack.prebuild",
        payload={"tip_commit_ids": [_fake_commit_id("a")]},
        status=status,
        created_at=now,
        attempt=0,
    ))
    await session.commit()
    return job_id


async def _count_prebuild_jobs(session: AsyncSession, repo_id: str) -> int:
    rows = (await session.execute(
        select(MusehubBackgroundJob.job_id).where(
            MusehubBackgroundJob.repo_id == repo_id,
            MusehubBackgroundJob.job_type == "fetch.mpack.prebuild",
        )
    )).scalars().all()
    return len(rows)


@pytest.mark.asyncio
async def test_mwp9_01_pending_prebuild_still_deduped_on_repair(db_session: AsyncSession) -> None:
    """GREEN guard: existing 'pending' dedup must keep working unchanged."""
    repo = await create_repo(db_session)
    await create_branch(db_session, repo.repo_id, name="main", head_commit_id=_fake_commit_id("tip"))
    await _insert_prebuild_job(db_session, repo.repo_id, status="pending")

    await _enqueue_repair_prebuild(db_session, repo.repo_id)
    await db_session.commit()

    assert await _count_prebuild_jobs(db_session, repo.repo_id) == 1


@pytest.mark.asyncio
async def test_mwp9_02_running_prebuild_deduped_on_repair(db_session: AsyncSession) -> None:
    """RED before the fix: a 'running' prebuild job is not covered by the
    current dedup, so repair-commit enqueues a redundant second job."""
    repo = await create_repo(db_session)
    await create_branch(db_session, repo.repo_id, name="main", head_commit_id=_fake_commit_id("tip"))
    await _insert_prebuild_job(db_session, repo.repo_id, status="running")

    await _enqueue_repair_prebuild(db_session, repo.repo_id)
    await db_session.commit()

    assert await _count_prebuild_jobs(db_session, repo.repo_id) == 1, (
        "repair-commit enqueued a redundant fetch.mpack.prebuild job while one "
        "was already running — the exact flooding mechanism that caused live "
        "clone failures during a bulk repair run"
    )


@pytest.mark.asyncio
async def test_mwp9_03_running_prebuild_deduped_on_regular_push(db_session: AsyncSession) -> None:
    """RED before the fix: the regular post-push path has the identical gap --
    confirms this isn't a repair-script-only issue, it's the baseline
    push-time behavior under any rapid repeated push load."""
    repo = await create_repo(db_session)
    await create_branch(db_session, repo.repo_id, name="main", head_commit_id=_fake_commit_id("tip"))
    await _insert_prebuild_job(db_session, repo.repo_id, status="running")

    await enqueue_push_intel(
        db_session, repo.repo_id, head=_fake_commit_id("tip"),
        domain_id="code", branch="main",
    )
    await db_session.commit()

    assert await _count_prebuild_jobs(db_session, repo.repo_id) == 1, (
        "a regular push enqueued a redundant fetch.mpack.prebuild job while "
        "one was already running"
    )


@pytest.mark.asyncio
async def test_mwp9_04_other_coarse_job_types_unaffected(db_session: AsyncSession) -> None:
    """Scope guard: the fix must be narrow to fetch.mpack.prebuild. A
    'running' gc job (not self-coalescing by any documented design) must
    still allow a fresh 'pending' gc job to be enqueued, exactly as before."""
    repo = await create_repo(db_session)
    await create_branch(db_session, repo.repo_id, name="main", head_commit_id=_fake_commit_id("tip"))
    now = _now()
    db_session.add(MusehubBackgroundJob(
        job_id=compute_job_id(repo.repo_id, "gc", now.isoformat()),
        repo_id=repo.repo_id, job_type="gc", payload={},
        status="running", created_at=now, attempt=0,
    ))
    await db_session.commit()

    await enqueue_push_intel(
        db_session, repo.repo_id, head=_fake_commit_id("tip"),
        domain_id="code", branch="main",
    )
    await db_session.commit()

    gc_rows = (await db_session.execute(
        select(MusehubBackgroundJob.job_id).where(
            MusehubBackgroundJob.repo_id == repo.repo_id,
            MusehubBackgroundJob.job_type == "gc",
        )
    )).scalars().all()
    assert len(gc_rows) == 2, (
        "gc's dedup window must remain pending-only -- broadening it to "
        "'running' for every coarse job type was explicitly out of scope"
    )


@pytest.mark.asyncio
async def test_mwp9_05_exhausted_stale_running_job_marked_failed_not_left_running(
    db_session: AsyncSession,
) -> None:
    """RED before the fix (found live on staging, 2026-10-21): two
    fetch.mpack.prebuild rows have sat at status='running', attempt=3
    (== _MAX_ATTEMPTS) since July and September -- a worker crash/restart
    mid-job never called fail_job, and reclaim_stale_jobs explicitly
    excludes attempt >= _MAX_ATTEMPTS from its reset-to-pending query, so
    these rows are permanently stuck in 'running'.

    This was invisible before MWP9_02/03 because the old dedup only ever
    looked at 'pending'. Once dedup also treats 'running' as coverage (the
    whole point of this file), an exhausted zombie 'running' row permanently
    blocks every future trigger for that job type on that repo -- staging's
    clone has been wedged on exactly this since this fix went live.

    reclaim_stale_jobs must terminally fail exhausted stale jobs (not just
    silently skip them) so they stop blocking dedup forever, while jobs
    still under budget keep retrying via the existing reset-to-pending path.
    """
    repo = await create_repo(db_session)
    await create_branch(db_session, repo.repo_id, name="main", head_commit_id=_fake_commit_id("tip"))

    stale_claimed_at = _now() - timedelta(minutes=30)
    job_id = compute_job_id(repo.repo_id, "fetch.mpack.prebuild", "zombie")
    db_session.add(MusehubBackgroundJob(
        job_id=job_id,
        repo_id=repo.repo_id,
        job_type="fetch.mpack.prebuild",
        payload={"tip_commit_ids": [_fake_commit_id("a")]},
        status="running",
        created_at=stale_claimed_at,
        claimed_at=stale_claimed_at,
        attempt=_MAX_ATTEMPTS,
    ))
    await db_session.commit()

    await reclaim_stale_jobs(db_session)
    await db_session.commit()

    row = (await db_session.execute(
        select(MusehubBackgroundJob).where(MusehubBackgroundJob.job_id == job_id)
    )).scalar_one()
    assert row.status == "failed", (
        "an exhausted stale job must be terminally failed, not left in "
        "'running' forever -- a permanent zombie that blocks dedup for this "
        "job type/repo indefinitely"
    )


@pytest.mark.asyncio
async def test_mwp9_06_stale_running_job_under_budget_still_resets_to_pending(
    db_session: AsyncSession,
) -> None:
    """GREEN guard: the existing retry behavior for stale jobs that still
    have attempts remaining must be unchanged by the exhausted-attempt fix."""
    repo = await create_repo(db_session)
    await create_branch(db_session, repo.repo_id, name="main", head_commit_id=_fake_commit_id("tip"))

    stale_claimed_at = _now() - timedelta(minutes=30)
    job_id = compute_job_id(repo.repo_id, "fetch.mpack.prebuild", "retryable")
    db_session.add(MusehubBackgroundJob(
        job_id=job_id,
        repo_id=repo.repo_id,
        job_type="fetch.mpack.prebuild",
        payload={"tip_commit_ids": [_fake_commit_id("a")]},
        status="running",
        created_at=stale_claimed_at,
        claimed_at=stale_claimed_at,
        attempt=_MAX_ATTEMPTS - 1,
    ))
    await db_session.commit()

    await reclaim_stale_jobs(db_session)
    await db_session.commit()

    row = (await db_session.execute(
        select(MusehubBackgroundJob).where(MusehubBackgroundJob.job_id == job_id)
    )).scalar_one()
    assert row.status == "pending"