feat(import): /api/import/clear-stuck endpoint + Clear stuck UI button — escape hatch for the autoretry-loop case the automatic sweep can't break

Operator hit 3 large PNGs stuck in 'processing' for 2 days 2026-05-25:
the existing recover_interrupted_tasks flips processing > 5min back to
queued + .delay(), but if the underlying file is unfixably broken (e.g.,
PIL OSError, also patched in 68cffce), the loop never terminates and the
'Scanning...' banner sticks at 0/0 forever blocking new scans.

/api/import/clear-stuck:
- Flips every task in pending/queued/processing to 'failed' with a clear
  marker error message
- Finalizes any 'running' ImportBatch that has no remaining active children
- Idempotent + non-destructive: rows survive, can be retried once the
  underlying cause is resolved

UI button 'Clear stuck...' sits next to 'Retry failed' / 'Clear completed'
with a warning-tonal alert in the confirm dialog explaining what it does
and recommending Retry failed once the cause is fixed.

Tests: clears mixed non-terminal states, untouches complete rows,
finalizes orphan batch, no-op when nothing stuck.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

tests/test_api_import_admin.py

@@ -86,6 +86,61 @@ async def test_clear_completed(client, db):
     assert body["deleted"] == 1
 
 
+@pytest.mark.asyncio
+async def test_clear_stuck_fails_non_terminal_and_finalizes_orphan_batch(client, db):
+    """Operator-flagged 2026-05-25: 3 large PNGs got stuck in 'processing'
+    for 2 days, the active ImportBatch never finalized, and the UI's
+    'Scanning...' banner persisted with 0/0 files. /api/import/clear-stuck
+    is the escape hatch to break the autoretry loop manually."""
+    from sqlalchemy import select as _select
+
+    batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick")
+    db.add(batch)
+    await db.flush()
+    # Three stuck rows in mixed non-terminal states.
+    db.add(ImportTask(
+        batch_id=batch.id, source_path="/p1", task_type="media", status="processing",
+    ))
+    db.add(ImportTask(
+        batch_id=batch.id, source_path="/p2", task_type="media", status="queued",
+    ))
+    db.add(ImportTask(
+        batch_id=batch.id, source_path="/p3", task_type="media", status="pending",
+    ))
+    # One already-complete row should be untouched.
+    db.add(ImportTask(
+        batch_id=batch.id, source_path="/done", task_type="media",
+        status="complete", finished_at=datetime.now(UTC),
+    ))
+    await db.commit()
+
+    resp = await client.post("/api/import/clear-stuck")
+    body = await resp.get_json()
+    assert resp.status_code == 200
+    assert body["tasks_failed"] == 3
+    assert body["batches_finalized"] == 1
+
+    statuses = {
+        row.status for row in
+        (await db.execute(_select(ImportTask).where(ImportTask.batch_id == batch.id)))
+        .scalars().all()
+    }
+    assert statuses == {"failed", "complete"}
+
+    batch_status = (await db.execute(
+        _select(ImportBatch.status).where(ImportBatch.id == batch.id)
+    )).scalar_one()
+    assert batch_status == "complete"
+
+
+@pytest.mark.asyncio
+async def test_clear_stuck_no_op_when_nothing_stuck(client, db):
+    resp = await client.post("/api/import/clear-stuck")
+    body = await resp.get_json()
+    assert resp.status_code == 200
+    assert body == {"tasks_failed": 0, "batches_finalized": 0}
+
+
 @pytest.mark.asyncio
 async def test_trigger_accepts_deep(client, monkeypatch):
     # Stub the task dispatch — assert the API accepts 'deep' and forwards