FabledCurator/backend/app/tasks/maintenance.py

"""Periodic maintenance: recover stuck import tasks, garbage-collect old finished tasks."""

import logging
import os
import subprocess
from datetime import UTC, datetime, timedelta
from pathlib import Path

from PIL import Image
from sqlalchemy import Integer, and_, cast, delete, func, or_, select, update

from ..celery_app import celery
from ..models import (
    BackupRun,
    DownloadEvent,
    HeadAutoApplyRun,
    HeadTrainingRun,
    ImageRecord,
    ImportBatch,
    ImportSettings,
    ImportTask,
    LibraryAuditRun,
    Source,
    TaskRun,
)
from ..utils.phash import compute_phash
from ._sync_engine import get_sync_engine
from ._sync_engine import sync_session_factory as _sync_session_factory

log = logging.getLogger(__name__)

# High-churn tables whose dead-tuple bloat matters: the TABLESAMPLE showcase
# reads physical blocks (bloat slows it directly), and the periodic
# prune/backfill/recovery tasks generate dead tuples faster than autovacuum
# always keeps up with. VACUUM reclaims them; ANALYZE refreshes planner stats.
# Allowlist ONLY — names are interpolated into VACUUM, so they must never come
# from request input.
VACUUM_TABLES = (
    "image_record",
    "image_provenance",
    "post_attachment",
    "download_event",
    "task_run",
    "import_task",
    "import_batch",
)

STUCK_THRESHOLD_MINUTES = 5
# Archive ImportTasks run the per-member pipeline inline for every
# member (import_archive_file: soft=30min/hard=35min). The ImportTask
# 'processing' recovery sweep must give them a longer threshold or it
# re-queues a legitimately-running archive mid-import (double-process).
# 40 min = 5-min buffer past the archive task's hard kill.
# Operator-flagged 2026-05-28 (target 1645019, a big archive).
ARCHIVE_STUCK_THRESHOLD_MINUTES = 40

# Poison-pill cap. After being recovered (re-queued from a stuck
# 'processing' state) MAX_RECOVERY_ATTEMPTS-1 times, the next sweep
# marks the row 'failed' instead of looping. 3 = two recoveries then
# give up. A row reaches this only if it leaves NO terminal flip each
# run — i.e. it hard-crashes the worker (OOM/segfault/SIGKILL), the
# signature of a corrupt or oversized input. Caught exceptions already
# flip to terminal 'failed' and never enter this loop.
MAX_RECOVERY_ATTEMPTS = 3
ORPHAN_PENDING_THRESHOLD_MINUTES = 30

# DownloadEvent (pending|running) recovery threshold. download_source has
# time_limit=1500s (25 min, DOWNLOAD_HARD_TIME_LIMIT); 30 min is 5 min past
# that, so a legitimately-running task is hard-killed before the sweep ever
# touches it — the sweep only catches events whose worker died without
# finalizing. Operator-confirmed 2026-05-29 after 43 sources stranded at
# "last check never" by the in-flight guard; budget bumped 2026-06-03 with
# the soft/hard limit raise (Anduo #39912).
DOWNLOAD_STALL_THRESHOLD_MINUTES = 30

OLD_TASK_DAYS = 7
PHASH_PAGE = 500
VERIFY_PAGE = 200
# Title-based WIP backfill (task #1458): posts scanned per keyset page.
WIP_BACKFILL_PAGE = 500
FFPROBE_TIMEOUT_SECONDS = 10
TASK_RUN_KEEP_OK_SECONDS = 24 * 3600       # 24 h
TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600  # 7 days
# Orphaned staging files: downloads/imports stage into <name>.part / <name>.partial
# then os.replace() into place (importer / external_fetch / native_ingest_common /
# attachment_store), so a kill mid-write leaves a discardable temp, never a corrupt
# final. cleanup_orphaned_temp_files sweeps ones left behind; the min-age guard
# keeps it from deleting an in-flight download's staging file mid-write.
IMAGES_ROOT = Path("/images")
TEMP_STAGING_SUFFIXES = (".part", ".partial")
ORPHAN_TEMP_MIN_AGE_HOURS = 6

# Audit 2026-06-02: per-entity recovery sweep thresholds. Each must be
# > the entity's longest legitimate runtime (its task's time_limit + a
# small buffer) so the sweep never flags in-flight work.
#
# Backups: images backup has time_limit=23400s (6.5h). 7h covers it
# with a 30-min buffer.
BACKUP_STALL_THRESHOLD_MINUTES = 7 * 60
# DB backup/restore is seconds-to-minutes (35-min hard limit). It must NOT share
# the images' 7h window — a DB backup wedged on NFS would otherwise sit "running"
# for 7 hours holding the concurrency-1 maintenance_long lane (operator-flagged
# 2026-06-07). 40 min gives a small buffer over the hard limit.
BACKUP_DB_STALL_THRESHOLD_MINUTES = 40
# Library audit: scan_library_for_rule has time_limit=7500s (2h5m).
# 2h15m gives a 10-min buffer.
LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES = 135
# head training (#114) has a 60-min soft limit; flag no-progress past 75.
HEAD_TRAINING_STALL_THRESHOLD_MINUTES = 75
HEAD_TRAINING_KEEP_RUNS = 20
# head auto-apply (#114) shares the 60-min soft limit; flag past 75.
HEAD_AUTO_APPLY_STALL_THRESHOLD_MINUTES = 75
HEAD_AUTO_APPLY_KEEP_RUNS = 20
# Import batches finalize only after every child ImportTask hits a
# terminal state. The recovery sweep targets the case where every
# task is done but the batch never got its closing UPDATE
# (orchestrator crashed at the wrong instant). 2h is well past any
# realistic single-batch import.
IMPORT_BATCH_STALL_THRESHOLD_MINUTES = 120

# Retention windows (terminal rows older than these get deleted by
# the daily prune sweeps). 30 days = operator-flagged "useful for
# triage for a few weeks, then noise."
LIBRARY_AUDIT_KEEP_DAYS = 30
IMPORT_BATCH_KEEP_DAYS = 30

# Overrides for recover_stalled_task_runs (the TaskRun 'running' sweep).
# Tasks/queues that legitimately run longer than the default 5-min
# threshold need their own larger value, else the sweep marks in-flight
# work 'error' before it finishes. Each value MUST be ≥ the relevant
# task.time_limit + a small buffer. task_name overrides take precedence
# over queue overrides.
#
# ml queue: embed_image video branch (≈20 GPU ops); time_limit=1200.
# import_archive_file: shares the 'import' queue with the fast
#   single-file import_media_file, so it needs a task-name override
#   (the import queue itself stays at the 5-min default for single
#   files); time_limit=2100.
QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
    "ml": 25,
    # download_source legitimately walks 5-25 min (Patreon/gallery-dl
    # deep creators); its hard time_limit is DOWNLOAD_HARD_TIME_LIMIT
    # (1500s = 25m). The 5-min default flagged healthy in-flight walks as
    # phantom 'RecoverySweep' failures (System Activity showed errors the
    # Subscriptions view correctly didn't — the download finished ok and
    # reset the source). 30 clears the 25-min limit with buffer and lines
    # up with DOWNLOAD_STALL_THRESHOLD_MINUTES (30) so a genuine hard kill
    # is swept by the task-run AND event sweeps together. Audit 2026-06-10.
    "download": 30,
    # Audit 2026-06-02 — maintenance/scan queues run tasks that
    # legitimately exceed the 5-min default (verify_integrity at 70m
    # hard, scan_directory at 70m hard, backfill_phash at 35m hard).
    # 75 min lives above the longest of those and the per-task
    # overrides below cover the outliers (backups, library audit).
    "maintenance": 75,
    "scan": 75,
}
TASK_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
    "backend.app.tasks.import_file.import_archive_file": 40,
    # Backup images runs hours, not minutes (6.5h hard limit). The
    # task-name override beats the queue's 75-min default so a
    # legitimately-running backup isn't flagged.
    "backend.app.tasks.backup.backup_images_task": 420,
    "backend.app.tasks.backup.restore_images_task": 420,
    # Library audit scans the full library — 2h hard limit.
    "backend.app.tasks.library_audit.scan_library_for_rule": 130,
    # External file-host fetches (mega/gdrive/film packs) can run to the task's
    # 60-min hard limit (time_limit=3600) — the fetcher's own read/total budgets
    # (external_fetch) cap a single fetch below that, but this stays the outer
    # backstop. Without an override these healthy in-flight fetches were
    # phantom-flagged 'RecoverySweep' before their own timeout/error could
    # surface (operator-flagged 2026-06-17, target 414 swept at 6.6min). A
    # task-name override beats the queue threshold whatever queue the row records
    # (it recorded 'default' before the celery_signals fix → download). 65 = 60+5.
    "backend.app.tasks.external.fetch_external_link": 65,
    # Attachment reclaim walks the whole sha-addressed store; the service caps
    # itself at a 900s budget and reports partial, but the task's hard limit is
    # 25 min for a wedged filesystem (NFS stall). Same phantom-flag class as the
    # external-fetch entry above — without an override a healthy in-flight walk
    # is swept 'RecoverySweep' at the bare 5-min default. 30 = 25 + 5.
    "backend.app.tasks.admin.reclaim_orphaned_attachments_task": 30,
}


@celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks")
def recover_interrupted_tasks() -> int:
    """Recover stuck ImportTask rows. Two distinct stuck states:

    1. 'processing' too long — worker crash mid-import. Re-queue via
       enqueue_import (routing media vs archive) and let the import
       retry. Threshold is task-type-aware: media files are sub-second
       and capped at the 5-min soft limit, so STUCK_THRESHOLD_MINUTES
       (5) means a confirmed crash; archives run the per-member
       pipeline inline (import_archive_file, 35-min hard limit) so they
       get ARCHIVE_STUCK_THRESHOLD_MINUTES (40) to avoid re-queueing a
       still-running archive. (Media was tightened from 30 min to 5
       2026-05-24 after a 2224-row zombie pile; archive split out
       2026-05-28.)

    2. 'pending' or 'queued' > 30 min — enqueue-phase crash. scan_directory
       creates rows with status='pending' (commit), then in a second pass
       transitions to 'queued' and calls .delay() (commit). If the scanner
       crashes between those two commits, rows are orphaned in 'pending'
       (never enqueued) with no recovery path — invisible to the
       'processing' sweep above. Flagged 2026-05-25 by operator hitting a
       5490-row orphan pile. Flip these to 'failed' (not re-enqueue) so
       the operator drains them via /api/import/retry-failed at their own
       pace; bulk-re-enqueueing 5000+ rows would thundering-herd the
       import worker.

    Returns total rows touched (recovered + marked failed).
    """
    SessionLocal = _sync_session_factory()
    now = datetime.now(UTC)
    media_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
    archive_cutoff = now - timedelta(minutes=ARCHIVE_STUCK_THRESHOLD_MINUTES)
    orphan_cutoff = now - timedelta(minutes=ORPHAN_PENDING_THRESHOLD_MINUTES)
    with SessionLocal() as session:
        # Both sweeps used to be SELECT ids → UPDATE WHERE id IN (...) which
        # blew past psycopg's 65535-parameter ceiling once a sweep covered
        # tens of thousands of rows (operator hit it 2026-05-26 after the
        # /import deep scan piled up orphans). Folding the SELECT into the
        # UPDATE eliminates the IN-list entirely. RETURNING gives us back
        # exactly the (id, task_type) pairs that flipped so the requeue
        # can route media vs archive correctly.
        #
        # Media + archive get separate cutoffs: a single media file is
        # sub-second so 5 min means crash; an archive runs the per-member
        # pipeline inline and can legitimately take up to its 35-min hard
        # limit, so it gets ARCHIVE_STUCK_THRESHOLD_MINUTES (40) to avoid
        # re-queueing a still-running archive.
        stuck_predicate = and_(
            ImportTask.status == "processing",
            or_(
                and_(ImportTask.task_type != "archive",
                     ImportTask.started_at < media_cutoff),
                and_(ImportTask.task_type == "archive",
                     ImportTask.started_at < archive_cutoff),
            ),
        )

        # POISON-PILL CIRCUIT BREAKER (Layer 1, 2026-05-28). A row that
        # leaves no terminal flip (hard worker crash: OOM/segfault/SIGKILL
        # on a corrupt or oversized input) gets re-queued by this sweep —
        # and would loop forever, re-crashing the worker each pass,
        # without a cap. Once a row has already been recovered
        # MAX_RECOVERY_ATTEMPTS-1 times, stop re-queueing it and mark it
        # 'failed' with a diagnostic so the operator can find + replace
        # the offending file. This UPDATE runs FIRST so the rows it
        # claims drop out of 'processing' before the re-queue pass.
        poison_result = session.execute(
            update(ImportTask)
            .where(stuck_predicate)
            .where(ImportTask.recovery_count >= MAX_RECOVERY_ATTEMPTS - 1)
            .values(
                status="failed",
                finished_at=now,
                error=(
                    f"crashed or stalled the worker {MAX_RECOVERY_ATTEMPTS} "
                    f"times without completing — likely a corrupt or "
                    f"oversized input. Not re-queued. Inspect/replace the "
                    f"file, then retry via /api/import/retry-failed."
                ),
            )
            .returning(ImportTask.id)
        )
        poison_ids = [r[0] for r in poison_result.all()]

        # Re-queue the remaining stuck rows (under the cap) and bump
        # their recovery_count. RETURNING (id, task_type) so the requeue
        # routes media vs archive correctly.
        stuck_result = session.execute(
            update(ImportTask)
            .where(stuck_predicate)
            .where(ImportTask.recovery_count < MAX_RECOVERY_ATTEMPTS - 1)
            .values(
                status="queued",
                started_at=None,
                recovery_count=ImportTask.recovery_count + 1,
                error="recovered from stuck state",
            )
            .returning(ImportTask.id, ImportTask.task_type)
        )
        stuck = stuck_result.all()

        orphan_result = session.execute(
            update(ImportTask)
            .where(ImportTask.status.in_(["pending", "queued"]))
            .where(ImportTask.created_at < orphan_cutoff)
            .values(
                status="failed",
                # Without finished_at, cleanup_old_tasks (`WHERE
                # finished_at < cutoff`) never reaps these rows —
                # orphan-swept rows would become permanent table
                # tenants. Audit 2026-06-02.
                finished_at=now,
                error=(
                    "orphan pending/queued swept by recover_interrupted_tasks "
                    "(scanner likely crashed mid-enqueue); retry via "
                    "/api/import/retry-failed"
                ),
            )
        )
        orphan_count = orphan_result.rowcount or 0

        session.commit()

        if stuck:
            from .import_file import enqueue_import
            for tid, task_type in stuck:
                enqueue_import(tid, task_type)

        # Layer-2 auto re-download (env-gated, default OFF). For each
        # poison-pill row that resolves to a pollable Source, delete the
        # bad file and trigger ONE source re-check to fetch a fresh
        # copy. Bounded by ImportTask.refetched so source-side
        # corruption can't loop. The 'failed' row stays as history; the
        # re-downloaded file re-imports as a fresh task on the next scan.
        if poison_ids and os.environ.get("FC_AUTO_REFETCH_CORRUPT", "0") == "1":
            from ..models import ImportSettings
            from ..services.refetch_service import attempt_refetch
            import_root = Path(session.execute(
                select(ImportSettings.import_scan_path)
                .where(ImportSettings.id == 1)
            ).scalar_one())
            for pid in poison_ids:
                ptask = session.get(ImportTask, pid)
                if ptask is None:
                    continue
                try:
                    attempt_refetch(session, ptask, import_root)
                except Exception as exc:  # noqa: BLE001 — best-effort
                    log.warning("auto-refetch failed for task %s: %s", pid, exc)

        return len(stuck) + len(poison_ids) + orphan_count


@celery.task(name="backend.app.tasks.maintenance.cleanup_old_tasks")
def cleanup_old_tasks() -> int:
    """Delete completed/skipped/failed ImportTask rows older than 7 days.

    Why 7 days: long enough to debug an issue an operator only notices days
    later; short enough that the task table stays a useful operational view
    rather than an archive. Matches IR's default.
    """
    SessionLocal = _sync_session_factory()
    cutoff = datetime.now(UTC) - timedelta(days=OLD_TASK_DAYS)
    with SessionLocal() as session:
        result = session.execute(
            delete(ImportTask)
            .where(ImportTask.status.in_(["complete", "skipped", "failed"]))
            .where(ImportTask.finished_at < cutoff)
        )
        session.commit()
        return result.rowcount or 0


@celery.task(name="backend.app.tasks.maintenance.cleanup_orphaned_temp_files")
def cleanup_orphaned_temp_files() -> int:
    """Delete orphaned .part/.partial staging files under the images root, left by
    a download/import killed mid-write. Only removes files older than
    ORPHAN_TEMP_MIN_AGE_HOURS so an in-flight download's staging file is never
    pulled out from under it. Returns the count removed."""
    if not IMAGES_ROOT.is_dir():
        return 0
    cutoff = datetime.now(UTC).timestamp() - ORPHAN_TEMP_MIN_AGE_HOURS * 3600
    removed = 0
    for path in IMAGES_ROOT.rglob("*"):
        if path.suffix not in TEMP_STAGING_SUFFIXES or not path.is_file():
            continue
        try:
            if path.stat().st_mtime >= cutoff:
                continue          # too fresh — may be an active download
            path.unlink()
            removed += 1
        except OSError as exc:
            log.warning("cleanup_orphaned_temp_files: %s: %s", path, exc)
    if removed:
        log.info(
            "cleanup_orphaned_temp_files: removed %d orphaned staging file(s)",
            removed,
        )
    return removed


@celery.task(name="backend.app.tasks.maintenance.recover_stalled_task_runs")
def recover_stalled_task_runs() -> int:
    """Flip task_run rows stuck in 'running' past their queue-specific
    threshold to 'error'. FC-3i.

    A row gets stuck when the worker dies without emitting
    task_postrun / task_failure (e.g. OOM, container restart between
    signals, signal handler raised+logged). The default 5-min threshold
    fits short-lived queues (import/thumbnail/download); queues that
    legitimately run longer tasks (ml-video, deep scans) get their
    own larger threshold via QUEUE_STUCK_THRESHOLD_MINUTES so the
    sweep doesn't preempt them.

    Runs once per distinct threshold value: each pass updates rows
    whose queue maps to that threshold.
    """
    SessionLocal = _sync_session_factory()
    now = datetime.now(UTC)
    override_tasks = set(TASK_STUCK_THRESHOLD_MINUTES.keys())
    override_queues = set(QUEUE_STUCK_THRESHOLD_MINUTES.keys())
    total = 0

    def _flag(minutes, *extra_where):
        cutoff = now - timedelta(minutes=minutes)
        stmt = (
            update(TaskRun)
            .where(TaskRun.status == "running")
            .where(TaskRun.started_at < cutoff)
            .values(
                status="error",
                error_type="RecoverySweep",
                error_message=(
                    f"no completion signal received within {minutes} min"
                ),
                finished_at=now,
                # Matches celery_signals.finalize's
                # int((now - started_at).total_seconds() * 1000)
                # — sweep-closed rows now carry duration like
                # normally-finalized rows. Audit 2026-06-02.
                duration_ms=cast(
                    func.extract("epoch", now - TaskRun.started_at) * 1000,
                    Integer,
                ),
            )
        )
        for w in extra_where:
            stmt = stmt.where(w)
        return session.execute(stmt).rowcount or 0

    with SessionLocal() as session:
        # Precedence: task_name override → queue override → default.
        # Each pass excludes rows claimed by a higher-precedence pass so
        # every row is touched at most once.

        # 1. Per-task-name overrides (e.g. import_archive_file, which
        #    shares the 'import' queue with fast single-file imports).
        for task_name, minutes in TASK_STUCK_THRESHOLD_MINUTES.items():
            total += _flag(minutes, TaskRun.task_name == task_name)

        # 2. Per-queue overrides, excluding the override task-names.
        for queue, minutes in QUEUE_STUCK_THRESHOLD_MINUTES.items():
            wheres = [TaskRun.queue == queue]
            if override_tasks:
                wheres.append(TaskRun.task_name.notin_(override_tasks))
            total += _flag(minutes, *wheres)

        # 3. Default — everything not claimed above.
        default_wheres = []
        if override_queues:
            default_wheres.append(TaskRun.queue.notin_(override_queues))
        if override_tasks:
            default_wheres.append(TaskRun.task_name.notin_(override_tasks))
        total += _flag(STUCK_THRESHOLD_MINUTES, *default_wheres)

        session.commit()
        return total


@celery.task(name="backend.app.tasks.maintenance.prune_task_runs")
def prune_task_runs() -> dict:
    """Daily retention for task_run rows. FC-3i.

    - 'ok' rows: deleted after TASK_RUN_KEEP_OK_SECONDS (24h default).
      Success is high-volume, not interesting after a day.
    - 'error' / 'timeout' rows: deleted after TASK_RUN_KEEP_FAILURE_SECONDS
      (7 days default). Failures are operationally interesting longer.
    - 'running' rows: NEVER deleted by this task. The recovery sweep
      (recover_stalled_task_runs) is the mechanism that flips them to
      terminal state; prune doesn't touch in-flight state.
    - 'retry' rows: treated as failures (>7d).

    Returns dict of how many rows were deleted in each bucket.
    """
    SessionLocal = _sync_session_factory()
    now = datetime.now(UTC)
    ok_cutoff = now - timedelta(seconds=TASK_RUN_KEEP_OK_SECONDS)
    fail_cutoff = now - timedelta(seconds=TASK_RUN_KEEP_FAILURE_SECONDS)
    with SessionLocal() as session:
        ok_deleted = session.execute(
            delete(TaskRun)
            .where(TaskRun.status == "ok")
            .where(TaskRun.finished_at < ok_cutoff)
        ).rowcount or 0
        fail_deleted = session.execute(
            delete(TaskRun)
            .where(TaskRun.status.in_(["error", "timeout", "retry"]))
            .where(TaskRun.finished_at < fail_cutoff)
        ).rowcount or 0
        session.commit()
        return {"ok_deleted": ok_deleted, "failures_deleted": fail_deleted}


@celery.task(
    name="backend.app.tasks.maintenance.backfill_phash",
    # Audit 2026-06-02 — keyset-paginated phash recompute over the whole
    # library; legitimately runs >5 min on large libraries.
    soft_time_limit=1800, time_limit=2100,
)
def backfill_phash() -> int:
    """Recompute phash for stored images that have none (imported before
    FC-2d-i+ii). Keyset-paginated by id (restart-safe), NULL-only fill,
    idempotent. Videos legitimately keep phash NULL. A missing/unreadable
    file is logged and left NULL — never fails the task."""
    SessionLocal = _sync_session_factory()
    updated = 0
    last_id = 0
    with SessionLocal() as session:
        while True:
            rows = session.execute(
                select(ImageRecord)
                .where(ImageRecord.id > last_id)
                .where(ImageRecord.phash.is_(None))
                .where(ImageRecord.mime.like("image/%"))
                .order_by(ImageRecord.id.asc())
                .limit(PHASH_PAGE)
            ).scalars().all()
            if not rows:
                break
            for rec in rows:
                try:
                    with Image.open(rec.path) as im:
                        ph = compute_phash(im)
                except Exception as exc:
                    log.warning(
                        "backfill_phash: unreadable %s: %s", rec.path, exc
                    )
                    ph = None
                if ph is not None and rec.phash is None:
                    rec.phash = ph
                    updated += 1
            session.commit()
            last_id = rows[-1].id
    return updated


def _verify_one(path: Path, expected_sha: str, mime: str, sha_fn) -> str:
    """Compute the integrity verdict for one file. Status precedence:
    failed_verification (can't run) > corrupt (sha mismatch / decode
    fails) > ok (passes both). Never raises."""
    try:
        if not path.is_file():
            return "failed_verification"
        try:
            actual_sha = sha_fn(path)
        except (OSError, PermissionError):
            return "failed_verification"
        if actual_sha != expected_sha:
            return "corrupt"
        if mime and mime.startswith("image/"):
            try:
                with Image.open(path) as im:
                    im.verify()
            except Exception:
                return "corrupt"
            return "ok"
        if mime and mime.startswith("video/"):
            try:
                proc = subprocess.run(
                    ["ffprobe", "-v", "error", "-i", str(path)],
                    capture_output=True,
                    timeout=FFPROBE_TIMEOUT_SECONDS,
                )
            except FileNotFoundError:
                # ffprobe binary missing — environment problem, not file.
                return "failed_verification"
            except subprocess.TimeoutExpired:
                return "corrupt"
            return "ok" if proc.returncode == 0 else "corrupt"
        # Unknown mime — sha matched already; trust that.
        return "ok"
    except Exception as exc:
        log.warning("verify_integrity unexpected error for %s: %s", path, exc)
        return "failed_verification"


@celery.task(
    name="backend.app.tasks.maintenance.verify_integrity",
    # Audit 2026-06-02 — full library sha256 + decode probe; on 100k-image
    # libraries this runs an hour or more. Match the maintenance queue's
    # recovery threshold (75 min) with 30s buffer below.
    soft_time_limit=3600, time_limit=4200,
)
def verify_integrity() -> int:
    """Verify every ImageRecord file: sha256 recompute + decode/probe
    (PIL for images; ffprobe for videos). Writes integrity_status
    (always — the column reflects the most recent verdict).
    Keyset-paginated, fail-soft per row, idempotent. Returns the total
    count verified."""
    from ..services.importer import _sha256_of  # reuse the importer's helper

    SessionLocal = _sync_session_factory()
    total = 0
    counts = {"ok": 0, "corrupt": 0, "failed_verification": 0}
    last_id = 0
    with SessionLocal() as session:
        while True:
            rows = session.execute(
                select(ImageRecord)
                .where(ImageRecord.id > last_id)
                .order_by(ImageRecord.id.asc())
                .limit(VERIFY_PAGE)
            ).scalars().all()
            if not rows:
                break
            for rec in rows:
                rec.integrity_status = _verify_one(
                    Path(rec.path), rec.sha256, rec.mime, _sha256_of
                )
                counts[rec.integrity_status] = (
                    counts.get(rec.integrity_status, 0) + 1
                )
                total += 1
            session.commit()
            last_id = rows[-1].id
    log.info("verify_integrity verdicts: %s (total %d)", counts, total)
    return total


@celery.task(
    name="backend.app.tasks.maintenance.triage_gpu_errors",
    # Bounded small-set probe (only errored images, once each), but a single
    # large original's sha256 over NFS can run tens of seconds — same quick-lane
    # tolerance rationale as verify_integrity above.
    soft_time_limit=600, time_limit=900,
)
def triage_gpu_errors() -> dict:
    """Failure triage (#125): probe each errored GPU job's file once and write
    the verdicts (ImageRecord.integrity_status + GpuJob.triage_status) — see
    services/ml/gpu_triage.py. Time-boxed + resumable; no-op when every errored
    job is already triaged."""
    from ..services.ml.gpu_triage import triage_errored_jobs

    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        summary = triage_errored_jobs(session, time_budget_seconds=300.0)
    if summary["probed"]:
        log.info("triage_gpu_errors: %s", summary)
    return summary


@celery.task(name="backend.app.tasks.maintenance.recover_stalled_download_events")
def recover_stalled_download_events() -> int:
    """Recover DownloadEvent rows stuck pending/running past the worker hard kill.

    The scan tick (scheduler_service.select_due_sources →
    tasks.scan._tick_due_sources_async) inserts DownloadEvent(status='pending')
    and fires download_source.delay(). If that task dies before finalizing the
    event — worker OOM/SIGKILL, lost task, or a gallery-dl that didn't unwind
    on the 1500s hard time_limit — the event stays in-flight forever. The next
    tick then skips that source because of the in-flight guard (scan.py:168)
    and Source.last_checked_at never updates; the operator sees "last check
    never" in the Subscriptions health column, permanently.

    This sweep flips matching events to 'error', stamps each affected Source's
    last_checked_at + last_error and bumps consecutive_failures (once per
    source, not per event — backoff is exponential on that count so an N-event
    bump would inflate the next interval by 2^N for no reason). The source
    becomes re-queueable on the next tick and the health dot goes amber.

    Operator-confirmed 2026-05-29 (43-row strand pile in production).
    """
    SessionLocal = _sync_session_factory()
    now = datetime.now(UTC)
    cutoff = now - timedelta(minutes=DOWNLOAD_STALL_THRESHOLD_MINUTES)
    msg = "stranded by recovery sweep (no terminal status after time_limit)"
    with SessionLocal() as session:
        # UPDATE...RETURNING the source_ids in one round trip — keeps us off
        # the psycopg 65535-param ceiling that SELECT-then-UPDATE-WHERE-IN
        # would hit on a large strand pile.
        result = session.execute(
            update(DownloadEvent)
            .where(DownloadEvent.status.in_(["pending", "running"]))
            .where(DownloadEvent.started_at < cutoff)
            .values(status="error", finished_at=now, error=msg)
            .returning(DownloadEvent.source_id)
        )
        returned = result.all()
        if not returned:
            session.commit()
            return 0
        events_recovered = len(returned)
        source_ids = list({row.source_id for row in returned})
        session.execute(
            update(Source)
            .where(Source.id.in_(source_ids))
            .values(
                consecutive_failures=Source.consecutive_failures + 1,
                last_error=msg,
                last_checked_at=now,
            )
        )
        session.commit()
        log.info(
            "recover_stalled_download_events: recovered %d events across %d sources",
            events_recovered, len(source_ids),
        )
        return events_recovered


@celery.task(name="backend.app.tasks.maintenance.recover_stalled_backup_runs")
def recover_stalled_backup_runs() -> int:
    """Flip BackupRun rows stuck in running/restoring past the hard limit
    to error. Audit 2026-06-02.

    prune_backups (FC-3h) used to claim the FC-3i task_run sweep handled
    these — but that sweep only flips TaskRun rows, not the BackupRun
    artifact rows. A SIGKILL'd backup left BackupRun stuck forever
    (dashboard showed phantom in-flight backups, keep_last_n offset
    arithmetic skewed because zombies sat outside the ok/error window).
    """
    SessionLocal = _sync_session_factory()
    now = datetime.now(UTC)
    db_cutoff = now - timedelta(minutes=BACKUP_DB_STALL_THRESHOLD_MINUTES)
    slow_cutoff = now - timedelta(minutes=BACKUP_STALL_THRESHOLD_MINUTES)
    msg = "stranded by recovery sweep (no terminal status within the stall window)"
    with SessionLocal() as session:
        result = session.execute(
            update(BackupRun)
            .where(BackupRun.status.in_(["running", "restoring"]))
            # db backups/restores are fast (40-min window); images run hours (7h).
            .where(
                or_(
                    and_(BackupRun.kind == "db", BackupRun.started_at < db_cutoff),
                    and_(BackupRun.kind != "db", BackupRun.started_at < slow_cutoff),
                )
            )
            .values(status="error", finished_at=now, error=msg)
        )
        session.commit()
        recovered = result.rowcount or 0
        if recovered:
            log.info("recover_stalled_backup_runs: recovered %d rows", recovered)
        return recovered


@celery.task(name="backend.app.tasks.maintenance.recover_stalled_library_audit_runs")
def recover_stalled_library_audit_runs() -> int:
    """Flip LibraryAuditRun rows stuck in running past the hard limit
    to error. Audit 2026-06-02.

    LibraryAuditRun.status='running' was protected by an exclusive
    guard in start_audit_run — a SIGKILL'd run would block all future
    audits until manual DB surgery. (The guard is now age-aware, but
    this sweep is what makes that work in practice.)

    Measures staleness from last_progress_at (alembic 0039), NOT started_at:
    a chunked scan stays 'running' across many re-enqueued chunks and can
    legitimately run for hours on a big library — only flag one that hasn't
    made progress in the threshold window (a dead chunk that never re-enqueued).
    Falls back to started_at for pre-0039 / never-ticked rows.
    """
    SessionLocal = _sync_session_factory()
    now = datetime.now(UTC)
    cutoff = now - timedelta(minutes=LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES)
    msg = (
        f"stranded by recovery sweep (no progress for "
        f"{LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES} min)"
    )
    with SessionLocal() as session:
        result = session.execute(
            update(LibraryAuditRun)
            .where(LibraryAuditRun.status == "running")
            .where(
                func.coalesce(
                    LibraryAuditRun.last_progress_at,
                    LibraryAuditRun.started_at,
                ) < cutoff
            )
            .values(status="error", finished_at=now, error=msg)
        )
        session.commit()
        recovered = result.rowcount or 0
        if recovered:
            log.info(
                "recover_stalled_library_audit_runs: recovered %d rows", recovered,
            )
        return recovered


def _recover_stalled_runs(model, *, stall_minutes: int, keep_runs: int, label: str) -> int:
    """Shared recovery + retention sweep for the head run-tracking tables
    (HeadTrainingRun / HeadAutoApplyRun, which share the
    status/last_progress_at/started_at/finished_at/error/id columns): flip 'running'
    rows with no progress past `stall_minutes` to 'error', then prune to the last
    `keep_runs` (rule 89). Returns the number recovered. NOTE the two other recover
    tasks are deliberately NOT folded in — library-audit has no prune tail and
    backup uses a single started_at cutoff."""
    SessionLocal = _sync_session_factory()
    now = datetime.now(UTC)
    cutoff = now - timedelta(minutes=stall_minutes)
    with SessionLocal() as session:
        result = session.execute(
            update(model)
            .where(model.status == "running")
            .where(func.coalesce(model.last_progress_at, model.started_at) < cutoff)
            .values(
                status="error", finished_at=now,
                error=f"stranded by recovery sweep (no progress for {stall_minutes} min)",
            )
        )
        keep = session.execute(
            select(model.id).order_by(model.id.desc()).limit(keep_runs)
        ).scalars().all()
        if keep:
            session.execute(delete(model).where(model.id.not_in(keep)))
        session.commit()
        recovered = result.rowcount or 0
        if recovered:
            log.info("%s: recovered %d rows", label, recovered)
        return recovered


@celery.task(name="backend.app.tasks.maintenance.recover_stalled_head_training_runs")
def recover_stalled_head_training_runs() -> int:
    """Flip HeadTrainingRun rows stuck in 'running' past the stall threshold to
    'error', and prune old runs to the last HEAD_TRAINING_KEEP_RUNS (retention,
    rule 89). Runs every 5 min on the maintenance lane; no-op when idle."""
    return _recover_stalled_runs(
        HeadTrainingRun,
        stall_minutes=HEAD_TRAINING_STALL_THRESHOLD_MINUTES,
        keep_runs=HEAD_TRAINING_KEEP_RUNS,
        label="recover_stalled_head_training_runs",
    )


@celery.task(name="backend.app.tasks.maintenance.recover_stalled_head_auto_apply_runs")
def recover_stalled_head_auto_apply_runs() -> int:
    """Flip stalled HeadAutoApplyRun 'running' rows to 'error' + prune to the
    last HEAD_AUTO_APPLY_KEEP_RUNS (retention, rule 89). 5-min maintenance lane."""
    return _recover_stalled_runs(
        HeadAutoApplyRun,
        stall_minutes=HEAD_AUTO_APPLY_STALL_THRESHOLD_MINUTES,
        keep_runs=HEAD_AUTO_APPLY_KEEP_RUNS,
        label="recover_stalled_head_auto_apply_runs",
    )


# Keep ~6 months of daily head-metric snapshots (enough to see tuning trends).
HEAD_METRICS_SNAPSHOT_RETENTION_DAYS = 180


@celery.task(name="backend.app.tasks.maintenance.snapshot_head_metrics")
def snapshot_head_metrics() -> int:
    """Daily per-concept observability point (#114): record each head-bearing
    concept's auto-applied volume, cumulative misfires/under-fires, and the
    head's measured quality — the time-series the operator tunes from. Prunes
    points older than the retention window."""
    from ..models import (
        HeadMetric,
        HeadMetricsSnapshot,
        Tag,
        TagHead,
    )
    from ..models.tag import image_tag

    SessionLocal = _sync_session_factory()
    now = datetime.now(UTC)
    with SessionLocal() as session:
        heads = {
            r.tag_id: r for r in session.execute(
                select(
                    TagHead.tag_id, TagHead.ap, TagHead.precision_cv,
                    TagHead.recall, TagHead.n_pos,
                )
            )
        }
        metrics = {
            r.tag_id: r for r in session.execute(
                select(
                    HeadMetric.tag_id, HeadMetric.n_misfires, HeadMetric.n_underfires
                )
            )
        }
        # .all() first: dict() of a bare Result tries the mapping protocol (a
        # Result exposes .keys()) and subscripts it, which fails.
        applied = dict(
            session.execute(
                select(image_tag.c.tag_id, func.count())
                .where(image_tag.c.source == "head_auto")
                .group_by(image_tag.c.tag_id)
            ).all()
        )
        tag_ids = set(heads) | set(metrics)
        if not tag_ids:
            return 0
        names = dict(
            session.execute(
                select(Tag.id, Tag.name).where(Tag.id.in_(tag_ids))
            ).all()
        )
        for tid in tag_ids:
            h = heads.get(tid)
            m = metrics.get(tid)
            session.add(HeadMetricsSnapshot(
                tag_id=tid, name=names.get(tid, str(tid)),
                snapshot_at=now,
                n_auto_applied=applied.get(tid, 0),
                n_misfires=m.n_misfires if m else 0,
                n_underfires=m.n_underfires if m else 0,
                ap=h.ap if h else None,
                precision_cv=h.precision_cv if h else None,
                recall=h.recall if h else None,
                n_pos=h.n_pos if h else None,
            ))
        session.execute(
            delete(HeadMetricsSnapshot).where(
                HeadMetricsSnapshot.snapshot_at
                < now - timedelta(days=HEAD_METRICS_SNAPSHOT_RETENTION_DAYS)
            )
        )
        session.commit()
        return len(tag_ids)


@celery.task(name="backend.app.tasks.maintenance.recover_stalled_import_batches")
def recover_stalled_import_batches() -> int:
    """Finalize ImportBatch rows stuck in running past the hard limit
    when NO outstanding ImportTask remains. Audit 2026-06-02.

    A batch row finalizes only after every child task hits a terminal
    state. The orphan case: scanner crashed between the last task's
    completion and the batch's closing UPDATE. The
    `/api/import/status` route then surfaces the batch as 'active'
    indefinitely while `/api/system/stats` (which uses the same
    EXISTS predicate we apply below) correctly returns null.
    """
    SessionLocal = _sync_session_factory()
    now = datetime.now(UTC)
    cutoff = now - timedelta(minutes=IMPORT_BATCH_STALL_THRESHOLD_MINUTES)
    with SessionLocal() as session:
        # Batches still 'running' past the cutoff whose tasks are all
        # terminal — there's no outstanding work, so flip the batch
        # too. Mirrors the EXISTS predicate the active-batch surfaces use.
        result = session.execute(
            update(ImportBatch)
            .where(ImportBatch.status == "running")
            .where(ImportBatch.started_at < cutoff)
            .where(
                ~select(ImportTask.id)
                .where(
                    ImportTask.batch_id == ImportBatch.id,
                    ImportTask.status.in_(["pending", "queued", "processing"]),
                )
                .exists()
            )
            .values(status="complete", finished_at=now)
        )
        session.commit()
        recovered = result.rowcount or 0
        if recovered:
            log.info(
                "recover_stalled_import_batches: finalized %d zombie batches",
                recovered,
            )
        return recovered


@celery.task(name="backend.app.tasks.maintenance.prune_library_audit_runs")
def prune_library_audit_runs() -> int:
    """Daily retention: delete terminal LibraryAuditRun rows older than
    LIBRARY_AUDIT_KEEP_DAYS. Never touches 'running'. Audit 2026-06-02.

    Audit rows carry matched_ids JSONB blobs that can hold tens of
    thousands of ids; without retention these accumulate.
    """
    SessionLocal = _sync_session_factory()
    cutoff = datetime.now(UTC) - timedelta(days=LIBRARY_AUDIT_KEEP_DAYS)
    with SessionLocal() as session:
        result = session.execute(
            delete(LibraryAuditRun)
            .where(LibraryAuditRun.status.in_(["ready", "applied", "cancelled", "error"]))
            .where(LibraryAuditRun.finished_at < cutoff)
        )
        session.commit()
        return result.rowcount or 0


@celery.task(name="backend.app.tasks.maintenance.prune_import_batches")
def prune_import_batches() -> int:
    """Daily retention: delete terminal ImportBatch rows older than
    IMPORT_BATCH_KEEP_DAYS. Cascade-deletes child ImportTask rows via
    the model relationship. Never touches 'running'. Audit 2026-06-02.
    """
    SessionLocal = _sync_session_factory()
    cutoff = datetime.now(UTC) - timedelta(days=IMPORT_BATCH_KEEP_DAYS)
    with SessionLocal() as session:
        # ORM-level delete here (not Core delete) so the
        # ImportBatch->tasks cascade fires; Core delete would skip it.
        old_batches = session.execute(
            select(ImportBatch)
            .where(ImportBatch.status.in_(["complete", "cancelled"]))
            .where(ImportBatch.finished_at < cutoff)
        ).scalars().all()
        for batch in old_batches:
            session.delete(batch)
        session.commit()
        return len(old_batches)


@celery.task(name="backend.app.tasks.maintenance.cleanup_old_download_events")
def cleanup_old_download_events() -> int:
    """FC-3d: delete terminal DownloadEvent rows older than the configured
    retention window. Never touches pending/running rows.

    Why terminal-only: pending/running rows represent in-flight work whose
    owning task may still be alive; deleting them would orphan the task.
    Retention days comes from ImportSettings.download_event_retention_days
    so the operator can tune without a code change.
    """
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        settings = ImportSettings.load_sync(session)
        retention_days = settings.download_event_retention_days
        cutoff = datetime.now(UTC) - timedelta(days=retention_days)
        result = session.execute(
            delete(DownloadEvent)
            .where(DownloadEvent.status.in_(["ok", "error", "skipped"]))
            .where(DownloadEvent.started_at < cutoff)
        )
        session.commit()
        return result.rowcount or 0


def _backfill_wip_tier(session, tag_id, prefilter, matcher, source) -> int:
    """One keyset-paginated pass over posts whose title matches a WIP tier, applying
    `tag_id` (stamped `source`) to their images. Shared by the hard + soft tiers
    (#1458 / #1474). Coarse `prefilter` (ILIKE superset) narrows the scan; the precise
    `matcher` confirms. Idempotent-additive (ON CONFLICT DO NOTHING). Returns the row
    count newly applied."""
    from ..models import Post
    from ..models.image_provenance import ImageProvenance
    from ..services.wip_title import apply_wip_image_tags

    applied = 0
    last_id = 0
    while True:
        rows = session.execute(
            select(Post.id, Post.post_title)
            .where(Post.id > last_id)
            .where(Post.post_title.is_not(None))
            .where(or_(*[Post.post_title.ilike(p) for p in prefilter]))
            .order_by(Post.id.asc())
            .limit(WIP_BACKFILL_PAGE)
        ).all()
        if not rows:
            break
        last_id = rows[-1][0]
        match_ids = [pid for pid, title in rows if matcher(title)]
        if match_ids:
            image_ids = session.execute(
                select(ImageProvenance.image_record_id)
                .where(ImageProvenance.post_id.in_(match_ids))
            ).scalars().all()
            applied += apply_wip_image_tags(session, image_ids, tag_id, source=source)
            session.commit()
    return applied


@celery.task(
    name="backend.app.tasks.maintenance.backfill_wip_title_tags",
    # Coarse-prefiltered scan over posts; the candidate set is small on a typical
    # library, but bound it like the other full-library sweeps.
    soft_time_limit=1800, time_limit=2100,
)
def backfill_wip_title_tags() -> int:
    """Scan EXISTING posts for WIP titles and apply the `wip` system tag to their
    images — the operator-triggered back-catalogue catch-up (task #1458 hard tier +
    #1474 soft tier). New imports are tagged live by the importer; this covers the
    existing library.

    HARD tier ("WIP"/"work in progress") always runs (the operator triggered the
    scan); the SOFT tier (sketch/doodle, provisional source) runs only when
    wip_soft_title_tagging_enabled, AFTER hard so a title matching both keeps the
    trained hard tag (ON CONFLICT DO NOTHING). Keyset-paginated, restart-safe.

    Deliberately NOT scheduled as a beat: a periodic re-run would re-apply to matching
    posts and silently undo a manual WIP removal, so it stays an explicit operator
    action (Settings → "Scan existing posts for WIP titles"). Returns rows applied.
    """
    from ..models import ImportSettings
    from ..services.wip_title import (
        SOFT_WIP_TITLE_SQL_PREFILTER,
        WIP_TITLE_SOFT_SOURCE,
        WIP_TITLE_SOURCE,
        WIP_TITLE_SQL_PREFILTER,
        matches_soft_wip_title,
        matches_wip_title,
        resolve_wip_tag_id,
    )

    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        tag_id = resolve_wip_tag_id(session)
        if tag_id is None:
            log.warning(
                "backfill_wip_title_tags: no `wip` system tag present; nothing to do"
            )
            return 0
        settings = ImportSettings.load_sync(session)
        applied = _backfill_wip_tier(
            session, tag_id, WIP_TITLE_SQL_PREFILTER, matches_wip_title,
            WIP_TITLE_SOURCE,
        )
        if settings.wip_soft_title_tagging_enabled:
            applied += _backfill_wip_tier(
                session, tag_id, SOFT_WIP_TITLE_SQL_PREFILTER, matches_soft_wip_title,
                WIP_TITLE_SOFT_SOURCE,
            )
    if applied:
        log.info("backfill_wip_title_tags: applied wip to %d image(s)", applied)
    return applied


@celery.task(name="backend.app.tasks.maintenance.vacuum_analyze")
def vacuum_analyze() -> dict:
    """Periodic VACUUM (ANALYZE) over the high-churn tables (VACUUM_TABLES) to
    reclaim dead-tuple bloat and refresh planner statistics. VACUUM cannot run
    inside a transaction block, so it runs on an AUTOCOMMIT connection.
    Scheduled weekly; also operator-triggerable from Settings → Maintenance.
    """
    engine = get_sync_engine()
    done = []
    with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
        for table in VACUUM_TABLES:
            conn.exec_driver_sql(f"VACUUM (ANALYZE) {table}")
            done.append(table)
    log.info("vacuum_analyze complete: %s", done)
    return {"vacuumed": done}