"""DownloadService orchestrator tests.

GalleryDLService is mocked to return a canned DownloadResult; tests
provide on-disk files for the importer to attach_in_place.
"""

from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock

import pytest
from sqlalchemy import select

from backend.app.models import Artist, DownloadEvent, ImportSettings, Source
from backend.app.services.credential_crypto import CredentialCrypto
from backend.app.services.credential_service import CredentialService
from backend.app.services.thumbnailer import Thumbnailer

pytestmark = pytest.mark.integration


@pytest.fixture
async def seed_artist_and_source(db, db_sync):
    artist = Artist(name="Alice", slug="alice")
    db.add(artist)
    await db.flush()
    source = Source(
        artist_id=artist.id, platform="patreon",
        url="https://patreon.com/alice", enabled=True,
        config_overrides={},
    )
    db.add(source)
    await db.commit()
    return artist, source


def _make_jpg(path: Path, split: str = "h"):
    """Write a structured JPEG so phash isn't degenerate.

    Solid-color images all phash-collapse (DCT of a flat image is zero).
    Use `split='h'` (top/bottom halves) or `split='v'` (left/right) to
    give each fixture file a distinct phash.
    """
    from PIL import Image
    path.parent.mkdir(parents=True, exist_ok=True)
    im = Image.new("RGB", (64, 64), (255, 255, 255))
    px = im.load()
    if split == "h":
        for y in range(32):
            for x in range(64):
                px[x, y] = (0, 0, 0)
    else:  # 'v'
        for y in range(64):
            for x in range(32):
                px[x, y] = (0, 0, 0)
    im.save(path, "JPEG")


def _make_fake_dl_result(
    *, success=True, written_paths=None, quarantined_paths=None,
    files_downloaded=0, error_type=None, error_message=None,
    stdout="", stderr="", cursor=None, run_stats=None, posts_processed=0,
    relink_source_paths=None,
):
    return SimpleNamespace(
        relink_source_paths=relink_source_paths or [],
        success=success,
        url="https://patreon.com/alice",
        artist_slug="alice",
        platform="patreon",
        files_downloaded=files_downloaded,
        files_quarantined=len(quarantined_paths or []),
        quarantined_paths=quarantined_paths or [],
        written_paths=written_paths or [],
        stdout=stdout,
        stderr=stderr,
        return_code=0 if success else 1,
        error_type=error_type,
        error_message=error_message,
        duration_seconds=1.23,
        started_at="2026-05-20T14:00:00+00:00",
        completed_at="2026-05-20T14:01:00+00:00",
        # plan #704: native ingester now returns the cursor + run_stats
        # structurally (no more stderr `Cursor:` scraping).
        cursor=cursor,
        run_stats=run_stats,
        posts_processed=posts_processed,
    )


def _fake_gdl_with_result(result):
    fake = MagicMock()
    fake.download = AsyncMock(return_value=result)
    # Real values for the attrs run_download's native branch reads off the gdl
    # (the native pacing config, plan #703) — a MagicMock would break the
    # arithmetic (max(0.5, rate_limit/4)).
    fake._rate_limit = 3.0
    fake._validate_files = True
    fake._compute_run_stats = lambda *a, **k: {
        "exit_code": 0, "downloaded_count": len(result.written_paths),
        "skipped_count": 0, "per_item_failures": 0,
        "warning_count": 0, "tier_gated_count": 0,
    }
    fake._extract_errors_warnings = lambda *a, **k: ""
    fake._truncate_log = lambda x, **k: x
    return fake


def _stub_patreon_ingester(svc, result, resolved_campaign_id=None):
    """Route the phase-2 dispatch (download_backends.run_download via
    DownloadService._run_download, plan #707 A5) to a canned DownloadResult so
    these tests exercise download_service's phase-2-result→phase-3 handling
    (status mapping, backfill lifecycle, health) without the real ingester's
    network/DB. The ingester itself is covered by test_patreon_ingester.py; the
    native construction/resolution by the run_download tests below. Returns a list
    capturing (ctx, source_config, skip_value, mode) per call so a test can assert
    mode / resume_cursor threading."""
    calls = []

    async def fake(*, ctx, source_config, skip_value, mode):
        calls.append({
            "ctx": ctx, "source_config": source_config,
            "skip_value": skip_value, "mode": mode,
        })
        return result, resolved_campaign_id

    svc._run_download = fake
    return calls


@pytest.mark.asyncio
async def test_download_source_attaches_written_files(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    from backend.app.services.download_service import DownloadService
    from backend.app.services.importer import Importer

    _artist, source = seed_artist_and_source

    images_root = tmp_path / "images"
    f1 = images_root / "alice" / "patreon" / "post" / "a.jpg"
    f2 = images_root / "alice" / "patreon" / "post" / "b.jpg"
    # Structured h/v split → distinct sha256 AND distinct phash;
    # also pin phash_threshold=0 so even tiny perceptual similarity
    # between the two doesn't trigger duplicate_phash.
    _make_jpg(f1, split="h")
    _make_jpg(f2, split="v")

    result = _make_fake_dl_result(
        success=True,
        written_paths=[str(f1), str(f2)],
        files_downloaded=2,
        stdout=f"{f1}\n{f2}\n",
    )
    fake_gdl = _fake_gdl_with_result(result)

    sync_settings = db_sync.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    sync_settings.phash_threshold = 0  # only exact phash collisions dedup

    importer = Importer(
        session=db_sync, images_root=images_root,
        import_root=images_root,
        thumbnailer=Thumbnailer(images_root=images_root),
        settings=sync_settings,
    )
    crypto = CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True)
    cred_service = CredentialService(db, crypto)

    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=fake_gdl, importer=importer, cred_service=cred_service,
    )
    _stub_patreon_ingester(svc, result)
    event_id = await svc.download_source(source.id)

    ev = (await db.execute(
        select(DownloadEvent).where(DownloadEvent.id == event_id)
    )).scalar_one()
    assert ev.status == "ok"
    assert ev.files_count == 2
    assert ev.metadata_["import_summary"]["attached"] == 2
    assert ev.metadata_["run_stats"]["downloaded_count"] == 2


@pytest.mark.asyncio
async def test_download_source_skipped_when_disabled(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    from backend.app.services.download_service import DownloadService
    from backend.app.services.importer import Importer

    _artist, source = seed_artist_and_source
    source.enabled = False
    await db.commit()

    fake_gdl = MagicMock()
    fake_gdl.download = AsyncMock()

    sync_settings = db_sync.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    importer = Importer(
        session=db_sync, images_root=tmp_path,
        import_root=tmp_path,
        thumbnailer=Thumbnailer(images_root=tmp_path),
        settings=sync_settings,
    )
    cred_service = CredentialService(db, MagicMock())

    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=fake_gdl, importer=importer, cred_service=cred_service,
    )
    event_id = await svc.download_source(source.id)

    ev = (await db.execute(
        select(DownloadEvent).where(DownloadEvent.id == event_id)
    )).scalar_one()
    assert ev.status == "skipped"
    assert ev.error == "source disabled"
    fake_gdl.download.assert_not_called()


@pytest.mark.asyncio
async def test_in_flight_idempotency_returns_existing_event(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    from backend.app.services.download_service import DownloadService

    _artist, source = seed_artist_and_source
    running = DownloadEvent(source_id=source.id, status="running")
    db.add(running)
    await db.commit()
    await db.refresh(running)

    fake_gdl = MagicMock()
    fake_gdl.download = AsyncMock()
    importer_stub = MagicMock()
    cred_service = CredentialService(db, MagicMock())

    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=fake_gdl, importer=importer_stub, cred_service=cred_service,
    )
    event_id = await svc.download_source(source.id)
    assert event_id == running.id
    fake_gdl.download.assert_not_called()


@pytest.mark.asyncio
async def test_patreon_resolved_campaign_id_is_cached(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """plan #697: the native ingester resolves a Patreon vanity → campaign id
    on the fly (replacing gallery-dl's reactive campaign-id retry). When phase 2
    reports a freshly-resolved id, phase 3 caches it on the source so later runs
    skip the lookup. Stub the ingester to report a resolved id; assert it lands
    in config_overrides."""
    from backend.app.services.download_service import DownloadService
    from backend.app.services.importer import Importer

    _artist, source = seed_artist_and_source

    sync_settings = db_sync.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    importer = Importer(
        session=db_sync, images_root=tmp_path,
        import_root=tmp_path,
        thumbnailer=Thumbnailer(images_root=tmp_path),
        settings=sync_settings,
    )
    cred_service = CredentialService(db, MagicMock())

    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=_fake_gdl_with_result(_make_fake_dl_result(success=True)),
        importer=importer, cred_service=cred_service,
    )
    _stub_patreon_ingester(
        svc, _make_fake_dl_result(success=True, written_paths=[]),
        resolved_campaign_id="99",
    )
    await svc.download_source(source.id)

    overrides = db_sync.execute(
        select(Source.config_overrides).where(Source.id == source.id)
    ).scalar_one()
    assert overrides.get("patreon_campaign_id") == "99"


@pytest.mark.asyncio
async def test_run_download_native_resolves_vanity_and_runs(
    db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
):
    """run_download (native branch, plan #707 A5): with no cached campaign id,
    resolve the vanity and pass the resolved id straight to the ingester; report
    it back so phase 3 can cache it."""
    from backend.app.services import download_backends as db_mod
    from backend.app.services.gallery_dl import SourceConfig

    _artist, source = seed_artist_and_source

    monkeypatch.setattr(
        db_mod, "resolve_campaign_id_for_source",
        AsyncMock(return_value=("4242", "4242")),
    )
    run_kwargs = {}

    class _FakeIngester:
        def __init__(self, **kw):
            pass

        def run(self, **kw):
            run_kwargs.update(kw)
            return _make_fake_dl_result(success=True, written_paths=[])

    monkeypatch.setattr(db_mod, "PatreonIngester", _FakeIngester)

    ctx = {
        "platform": "patreon",
        "source_id": source.id, "url": "https://patreon.com/alice",
        "artist_slug": "alice", "cookies_path": None, "auth_token": None,
        "config_overrides": {},
    }
    result, resolved = await db_mod.run_download(
        ctx=ctx, source_config=SourceConfig.from_dict({}), skip_value=True,
        mode="tick",
        gdl=_fake_gdl_with_result(_make_fake_dl_result(success=True)),
        sync_session_factory=MagicMock(),
    )
    assert result.success is True
    assert resolved == "4242"
    assert run_kwargs["campaign_id"] == "4242"
    assert run_kwargs["mode"] == "tick"


@pytest.mark.asyncio
async def test_run_download_native_unresolvable_fails_loud(
    db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
):
    """A campaign id we can't resolve is a loud NOT_FOUND failure, never a
    silent empty success."""
    from backend.app.services import download_backends as db_mod
    from backend.app.services.gallery_dl import ErrorType, SourceConfig

    _artist, source = seed_artist_and_source

    monkeypatch.setattr(
        db_mod, "resolve_campaign_id_for_source",
        AsyncMock(return_value=(None, None)),
    )
    ctx = {
        "platform": "patreon",
        "source_id": source.id, "url": "https://patreon.com/alice",
        "artist_slug": "alice", "cookies_path": None, "auth_token": None,
        "config_overrides": {},
    }
    result, resolved = await db_mod.run_download(
        ctx=ctx, source_config=SourceConfig.from_dict({}), skip_value=True,
        mode="tick", gdl=MagicMock(), sync_session_factory=MagicMock(),
    )
    assert result.success is False
    assert result.error_type == ErrorType.NOT_FOUND
    assert resolved is None


# --- FC-3d: finalize hook updates Source health columns -------------------


async def _seed_source_with_health(db, *, failures=0, last_error=None, suffix="fz"):
    artist = Artist(name=f"alice-{suffix}", slug=f"alice-{suffix}")
    db.add(artist)
    await db.flush()
    source = Source(
        artist_id=artist.id, platform="patreon",
        url=f"https://patreon.com/alice-{suffix}", enabled=True,
        consecutive_failures=failures, last_error=last_error,
    )
    db.add(source)
    await db.flush()
    event = DownloadEvent(source_id=source.id, status="running")
    db.add(event)
    await db.commit()
    return source.id, event.id


@pytest.mark.asyncio
async def test_finalize_ok_resets_failures_and_clears_error(db):
    from backend.app.services.download_service import DownloadService

    source_id, event_id = await _seed_source_with_health(
        db, failures=3, last_error="prev", suffix="ok",
    )
    svc = DownloadService(
        async_session=db, sync_session=None,
        gdl=None, importer=None, cred_service=None,
    )
    await svc._finalize_event_and_source(
        event_id=event_id, source_id=source_id, status="ok",
        error_message=None,
    )

    row = (await db.execute(
        select(
            Source.consecutive_failures, Source.last_error, Source.last_checked_at,
        ).where(Source.id == source_id)
    )).one()
    assert row.consecutive_failures == 0
    assert row.last_error is None
    assert row.last_checked_at is not None


@pytest.mark.asyncio
async def test_finalize_error_increments_failures_and_sets_error(db):
    from backend.app.services.download_service import DownloadService

    source_id, event_id = await _seed_source_with_health(
        db, failures=2, last_error=None, suffix="err",
    )
    svc = DownloadService(
        async_session=db, sync_session=None,
        gdl=None, importer=None, cred_service=None,
    )
    await svc._finalize_event_and_source(
        event_id=event_id, source_id=source_id, status="error",
        error_message="boom",
    )

    row = (await db.execute(
        select(
            Source.consecutive_failures, Source.last_error, Source.last_checked_at,
        ).where(Source.id == source_id)
    )).one()
    assert row.consecutive_failures == 3
    assert row.last_error == "boom"
    assert row.last_checked_at is not None


@pytest.mark.asyncio
async def test_rate_limited_cooldown_honors_retry_after(db, monkeypatch):
    """plan #708 B1: a RATE_LIMITED result with a Retry-After stamps the platform
    cooldown at that duration, clamped to [60, 3600]; no hint → the flat default."""
    from unittest.mock import AsyncMock, MagicMock

    from backend.app.services import download_service as dl_mod
    from backend.app.services.download_service import DownloadService

    source_id, _event_id = await _seed_source_with_health(db, suffix="rl")
    cooldown = AsyncMock()
    monkeypatch.setattr(dl_mod, "set_platform_cooldown", cooldown)
    svc = DownloadService(
        async_session=db, sync_session=None,
        gdl=MagicMock(), importer=MagicMock(), cred_service=MagicMock(),
    )

    async def _health(retry):
        cooldown.reset_mock()
        await svc._update_source_health(
            source_id=source_id, status="error", error_message="rate limited",
            error_type="rate_limited", retry_after_seconds=retry,
        )

    await _health(120.0)
    assert cooldown.await_args.kwargs["seconds"] == 120  # honored as-is
    await _health(5.0)
    assert cooldown.await_args.kwargs["seconds"] == 60  # floored
    await _health(99999.0)
    assert cooldown.await_args.kwargs["seconds"] == 3600  # capped
    await _health(None)
    assert "seconds" not in cooldown.await_args.kwargs  # default cooldown


@pytest.mark.asyncio
async def test_finalize_skipped_preserves_failures_clears_error(db):
    from backend.app.services.download_service import DownloadService

    source_id, event_id = await _seed_source_with_health(
        db, failures=4, last_error="stale", suffix="skip",
    )
    svc = DownloadService(
        async_session=db, sync_session=None,
        gdl=None, importer=None, cred_service=None,
    )
    await svc._finalize_event_and_source(
        event_id=event_id, source_id=source_id, status="skipped",
        error_message=None,
    )

    row = (await db.execute(
        select(
            Source.consecutive_failures, Source.last_error, Source.last_checked_at,
        ).where(Source.id == source_id)
    )).one()
    assert row.consecutive_failures == 4
    assert row.last_error is None
    assert row.last_checked_at is not None


# --- Plan #693: backfill state machine (time-boxed chunks, run-until-done) ---


def _backfill_svc(db, db_sync, tmp_path, result):
    """DownloadService wired with a real importer (so empty written_paths just
    attaches nothing) whose Patreon phase-2 branch is stubbed to return `result`.

    The seeded source is Patreon, so phase 2 routes to the native ingester
    (plan #697); the stub returns the canned result + captures the per-call
    (ctx, source_config, mode). Returns (svc, ingester_calls)."""
    from backend.app.services.download_service import DownloadService
    from backend.app.services.importer import Importer

    images_root = tmp_path / "images"
    sync_settings = db_sync.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    importer = Importer(
        session=db_sync, images_root=images_root, import_root=images_root,
        thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings,
    )
    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True))
    fake_gdl = _fake_gdl_with_result(result)
    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=fake_gdl, importer=importer, cred_service=cred_service,
    )
    calls = _stub_patreon_ingester(svc, result)
    return svc, calls


@pytest.mark.asyncio
async def test_backfill_chunk_progress_advances_cursor(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """A running backfill chunk that didn't finish but advanced (new cursor)
    stays 'running', checkpoints the cursor, bumps the chunk counter, and
    spends one safety-cap chunk."""
    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running"}
    source.backfill_runs_remaining = 5
    await db.commit()

    svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
        success=False, written_paths=[], files_downloaded=0,
        cursor="03:PAGE2:xyz",
    ))
    await svc.download_source(source.id)

    remaining, co = (await db.execute(
        select(Source.backfill_runs_remaining, Source.config_overrides)
        .where(Source.id == source.id)
    )).one()
    assert co.get("_backfill_state") == "running"
    assert co.get("_backfill_cursor") == "03:PAGE2:xyz"
    assert co.get("_backfill_chunks") == 1
    assert remaining == 4


@pytest.mark.asyncio
async def test_backfill_state_running_selects_backfill_mode_and_resumes(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """state=='running' drives backfill mode (chunk budget) and threads the
    stored cursor to the native ingester as the resume point."""
    from backend.app.services.gallery_dl import BACKFILL_CHUNK_SECONDS

    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running", "_backfill_cursor": "03:RESUME:here"}
    source.backfill_runs_remaining = 5
    await db.commit()

    svc, calls = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
        success=False, written_paths=[],
        cursor="03:RESUME2:next",
    ))
    await svc.download_source(source.id)

    assert len(calls) == 1
    assert calls[0]["mode"] == "backfill"
    assert calls[0]["source_config"].resume_cursor == "03:RESUME:here"
    assert calls[0]["source_config"].timeout == BACKFILL_CHUNK_SECONDS

    co = (await db.execute(
        select(Source.config_overrides).where(Source.id == source.id)
    )).scalar_one()
    assert co.get("_backfill_cursor") == "03:RESUME2:next"


@pytest.mark.asyncio
async def test_backfill_clean_exit_marks_complete(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """A clean rc=0 chunk = gallery-dl reached the bottom → state 'complete',
    cursor cleared, returns to tick mode."""
    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running", "_backfill_cursor": "03:NEAR:bottom"}
    source.backfill_runs_remaining = 5
    await db.commit()

    svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
        success=True, written_paths=[], files_downloaded=0,
    ))
    await svc.download_source(source.id)

    remaining, co = (await db.execute(
        select(Source.backfill_runs_remaining, Source.config_overrides)
        .where(Source.id == source.id)
    )).one()
    assert co.get("_backfill_state") == "complete"
    assert "_backfill_cursor" not in co
    assert remaining == 0


@pytest.mark.asyncio
async def test_backfill_cap_exhaustion_stalls(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """A progressing chunk that spends the last safety-cap chunk without
    reaching the bottom pauses as 'stalled' (doesn't loop forever)."""
    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running"}
    source.backfill_runs_remaining = 1
    await db.commit()

    svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
        success=False, written_paths=[],
        cursor="03:MORE:left",
    ))
    await svc.download_source(source.id)

    remaining, co = (await db.execute(
        select(Source.backfill_runs_remaining, Source.config_overrides)
        .where(Source.id == source.id)
    )).one()
    assert co.get("_backfill_state") == "stalled"
    assert remaining == 0


@pytest.mark.asyncio
async def test_backfill_stuck_guard_stalls_after_two_no_advance(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """Two consecutive chunks that fail to advance the cursor → 'stalled',
    cursor cleared, so a wedged walk can't re-strand forever."""
    from unittest.mock import MagicMock

    from backend.app.services.download_service import DownloadService

    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running", "_backfill_cursor": "stuck"}
    source.backfill_runs_remaining = 5
    await db.commit()

    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=MagicMock(), importer=MagicMock(), cred_service=MagicMock(),
    )
    ctx = {
        "source_id": source.id, "platform": "patreon",
        "config_overrides": {"_backfill_state": "running", "_backfill_cursor": "stuck"},
        "backfill_runs_remaining": 5,
    }
    stuck = _make_fake_dl_result(success=False, cursor="stuck")

    await svc._apply_backfill_lifecycle(ctx, stuck)
    await db.commit()
    co = (await db.execute(
        select(Source.config_overrides).where(Source.id == source.id)
    )).scalar_one()
    assert co.get("_backfill_state") == "running"
    assert co.get("_backfill_cursor_stalls") == 1

    ctx["config_overrides"] = dict(co)
    await svc._apply_backfill_lifecycle(ctx, stuck)
    await db.commit()
    co2 = (await db.execute(
        select(Source.config_overrides).where(Source.id == source.id)
    )).scalar_one()
    assert co2.get("_backfill_state") == "stalled"
    assert "_backfill_cursor" not in co2


@pytest.mark.asyncio
async def test_backfill_lifecycle_leaves_posts_to_ingester(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """plan #704 (#5), revised: _backfill_posts is OWNED by the ingester now (it
    writes a monotonic absolute live mid-walk), so the post-chunk lifecycle must
    NOT touch it — it carries the ingester's committed value forward unchanged."""
    from unittest.mock import MagicMock

    from backend.app.services.download_service import DownloadService

    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running", "_backfill_posts": 12}
    source.backfill_runs_remaining = 5
    await db.commit()

    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=MagicMock(), importer=MagicMock(), cred_service=MagicMock(),
    )
    ctx = {
        "source_id": source.id, "platform": "patreon",
        "config_overrides": {"_backfill_state": "running", "_backfill_posts": 12},
        "backfill_runs_remaining": 5,
    }
    await svc._apply_backfill_lifecycle(
        ctx, _make_fake_dl_result(success=False, cursor="C1", posts_processed=10),
    )
    await db.commit()
    co = (await db.execute(
        select(Source.config_overrides).where(Source.id == source.id)
    )).scalar_one()
    assert co.get("_backfill_posts") == 12  # untouched — the ingester owns it


@pytest.mark.asyncio
async def test_backfill_timeout_chunk_reclassified_to_ok(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """A backfill chunk that hit its time-box (TIMEOUT) but advanced is
    reclassified to PARTIAL → event status 'ok' (progress, not error), and
    consecutive_failures is not bumped."""
    from backend.app.services.gallery_dl import ErrorType

    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running"}
    source.backfill_runs_remaining = 5
    await db.commit()

    svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
        success=False, files_downloaded=3,
        error_type=ErrorType.TIMEOUT,
        error_message="Download timed out",
        cursor="03:NEXT:page",
    ))
    event_id = await svc.download_source(source.id)

    ev = (await db.execute(
        select(DownloadEvent).where(DownloadEvent.id == event_id)
    )).scalar_one()
    assert ev.status == "ok"
    failures = (await db.execute(
        select(Source.consecutive_failures).where(Source.id == source.id)
    )).scalar_one()
    assert failures == 0


@pytest.mark.asyncio
async def test_tick_mode_when_not_running_leaves_state_untouched(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """No _backfill_state → not backfilling; the lifecycle is a no-op and the
    counter/state aren't mutated."""
    _artist, source = seed_artist_and_source
    assert (source.config_overrides or {}).get("_backfill_state") is None

    svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
        success=True, written_paths=[], files_downloaded=0,
    ))
    await svc.download_source(source.id)

    co = (await db.execute(
        select(Source.config_overrides).where(Source.id == source.id)
    )).scalar_one()
    assert (co or {}).get("_backfill_state") is None


@pytest.mark.asyncio
async def test_recovery_mode_selected_and_flag_cleared_on_complete(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """plan #697: `_backfill_bypass_seen` alongside a running backfill selects
    recovery mode (ingester bypasses the seen-ledger). A clean rc=0 walk
    completes the shared lifecycle AND clears the bypass flag so the next tick
    honors the ledger again."""
    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running", "_backfill_bypass_seen": True}
    source.backfill_runs_remaining = 5
    await db.commit()

    svc, calls = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
        success=True, written_paths=[], files_downloaded=0,
    ))
    await svc.download_source(source.id)

    assert calls[0]["mode"] == "recovery"
    co = (await db.execute(
        select(Source.config_overrides).where(Source.id == source.id)
    )).scalar_one()
    assert co.get("_backfill_state") == "complete"
    assert "_backfill_bypass_seen" not in co


@pytest.mark.asyncio
async def test_recapture_mode_selected_and_flag_cleared_on_complete(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """#830: `_backfill_recapture` alongside a running backfill selects recapture
    mode. A clean rc=0 walk completes the shared lifecycle AND clears the
    recapture flag so the next tick is a plain tick again."""
    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running", "_backfill_recapture": True}
    source.backfill_runs_remaining = 5
    await db.commit()

    svc, calls = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
        success=True, written_paths=[], files_downloaded=0,
    ))
    await svc.download_source(source.id)

    assert calls[0]["mode"] == "recapture"
    co = (await db.execute(
        select(Source.config_overrides).where(Source.id == source.id)
    )).scalar_one()
    assert co.get("_backfill_state") == "complete"
    assert "_backfill_recapture" not in co


@pytest.mark.asyncio
async def test_partial_error_type_maps_to_ok_status(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """A non-zero gallery-dl exit accompanied by PARTIAL error_type
    (real files were downloaded before the run was cut short) is treated
    as status=ok by the orchestrator — next tick continues."""
    from backend.app.services.download_service import DownloadService
    from backend.app.services.gallery_dl import ErrorType
    from backend.app.services.importer import Importer

    _artist, source = seed_artist_and_source

    images_root = tmp_path / "images"
    f1 = images_root / "alice" / "patreon" / "post" / "a.jpg"
    _make_jpg(f1, split="h")

    fake_result = _make_fake_dl_result(
        success=False, written_paths=[str(f1)], files_downloaded=1,
        error_type=ErrorType.PARTIAL,
        error_message="Downloaded 1 file; run did not complete in budget",
        stdout=f"{f1}\n",
    )
    fake_gdl = _fake_gdl_with_result(fake_result)

    sync_settings = db_sync.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    importer = Importer(
        session=db_sync, images_root=images_root, import_root=images_root,
        thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings,
    )
    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True))
    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=fake_gdl, importer=importer, cred_service=cred_service,
    )
    _stub_patreon_ingester(svc, fake_result)
    event_id = await svc.download_source(source.id)

    ev = (await db.execute(
        select(DownloadEvent).where(DownloadEvent.id == event_id)
    )).scalar_one()
    assert ev.status == "ok"
    assert ev.error is None
    # And the source's failure counter isn't bumped — PARTIAL isn't a failure.
    src_after = (await db.execute(
        select(Source.consecutive_failures).where(Source.id == source.id)
    )).scalar_one()
    assert src_after == 0


@pytest.mark.asyncio
async def test_download_enqueues_thumbnail_and_ml_per_attached_image(
    db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
):
    """Operator-flagged 2026-06-01: downloaded images stayed at
    thumbnail_path=NULL until periodic backfill swept them up, surfacing
    as broken-thumbnail tiles in the gallery. Importer.attach_in_place
    deliberately skips inline generation; the calling code MUST enqueue
    the thumbnail + ML tasks per attached image (matching the pattern in
    tasks/import_file.py)."""
    from backend.app.services.download_service import DownloadService
    from backend.app.services.importer import Importer

    _artist, source = seed_artist_and_source

    images_root = tmp_path / "images"
    f1 = images_root / "alice" / "patreon" / "post" / "a.jpg"
    f2 = images_root / "alice" / "patreon" / "post" / "b.jpg"
    _make_jpg(f1, split="h")
    _make_jpg(f2, split="v")

    result = _make_fake_dl_result(
        success=True, written_paths=[str(f1), str(f2)],
        files_downloaded=2, stdout=f"{f1}\n{f2}\n",
    )
    fake_gdl = _fake_gdl_with_result(result)

    sync_settings = db_sync.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    sync_settings.phash_threshold = 0
    importer = Importer(
        session=db_sync, images_root=images_root, import_root=images_root,
        thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings,
    )
    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True))

    # Capture the IDs that the orchestrator hands off to each Celery task.
    # The .delay() shim runs inside DownloadService._phase3_persist (lazy
    # imports under ..tasks.thumbnail / ..tasks.ml), so monkeypatch the
    # symbols on those modules.
    thumb_calls: list[int] = []
    ml_calls: list[int] = []
    from backend.app.tasks import ml as ml_mod
    from backend.app.tasks import thumbnail as thumb_mod
    monkeypatch.setattr(
        thumb_mod.generate_thumbnail, "delay",
        lambda image_id: thumb_calls.append(image_id),
    )
    monkeypatch.setattr(
        ml_mod.embed_image, "delay",
        lambda image_id: ml_calls.append(image_id),
    )

    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=fake_gdl, importer=importer, cred_service=cred_service,
    )
    _stub_patreon_ingester(svc, result)
    await svc.download_source(source.id)

    # Two files attached → two thumbnail enqueues + two ML enqueues, IDs
    # match the actually-imported records (not ad-hoc — drawn from the
    # importer's returned image_id so future supersede paths stay covered).
    assert len(thumb_calls) == 2
    assert len(ml_calls) == 2
    assert sorted(thumb_calls) == sorted(ml_calls)


@pytest.mark.asyncio
async def test_releases_db_connections_before_subprocess(
    db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
):
    """Phase 1's DB connections must be released BEFORE the (multi-minute)
    phase-2 fetch, so they don't idle-die and strand phase 3 with
    ConnectionDoesNotExistError (Anduo #40014). Spy on the async session close
    and assert it happened before the phase-2 work (here the native Patreon
    ingester) runs; phase 3 must still finalize the event."""
    from backend.app.services.download_service import DownloadService
    from backend.app.services.importer import Importer

    _artist, source = seed_artist_and_source

    closed = {"async": False}
    orig_close = db.close

    async def spy_close():
        closed["async"] = True
        await orig_close()

    monkeypatch.setattr(db, "close", spy_close)

    seen = {}
    fake_gdl = _fake_gdl_with_result(
        _make_fake_dl_result(success=True, written_paths=[], stdout="")
    )

    sync_settings = db_sync.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    importer = Importer(
        session=db_sync, images_root=tmp_path, import_root=tmp_path,
        thumbnailer=Thumbnailer(images_root=tmp_path), settings=sync_settings,
    )
    cred_service = CredentialService(
        db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True)
    )

    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=fake_gdl, importer=importer, cred_service=cred_service,
    )

    async def fake_ingest(*, ctx, source_config, skip_value, mode):
        seen["closed_before_phase2"] = closed["async"]
        return _make_fake_dl_result(success=True, written_paths=[], stdout=""), None

    svc._run_download = fake_ingest
    event_id = await svc.download_source(source.id)

    assert seen["closed_before_phase2"] is True
    ev = (await db.execute(
        select(DownloadEvent).where(DownloadEvent.id == event_id)
    )).scalar_one()
    assert ev.status == "ok"


@pytest.mark.asyncio
async def test_async_task_engine_uses_nullpool():
    """The per-task async engine must use NullPool so phase 3 re-acquires a
    fresh connection instead of a pooled one the server reaped during the
    long subprocess (Anduo #40014)."""
    from sqlalchemy.pool import NullPool

    from backend.app.tasks._async_session import async_session_factory

    _factory, engine = async_session_factory()
    try:
        assert isinstance(engine.sync_engine.pool, NullPool)
    finally:
        await engine.dispose()