refactor(admin): consolidate maintenance-trigger 202 responses onto _queued() (#753 Finding B)

DRY pass follow-up (note #1026). Five handlers returned the identical
jsonify({task_id, status:queued}), 202 shape; extract _queued(async_result).

Consumers routed through it: tags_normalize (live branch), trigger_reextract_archives,
trigger_prune_missing_files, trigger_dedup_videos, trigger_purge_gated_previews.
trigger_vacuum stays bespoke (returns no task_id — the UI doesn't poll it).

Added route-level tests for all five consumers (these trigger endpoints had no
route coverage before): 202 + task_id via _queued, and the dry_run flag threading
through to dedup/purge-gated. Behavior unchanged.

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

tests/test_api_admin.py

@@ -5,6 +5,7 @@ without actually queuing. Tier-B/A endpoints run synchronously
 through the real service.
 """
 import hashlib
+from types import SimpleNamespace
 
 import pytest
 
@@ -498,6 +499,87 @@ async def test_trigger_vacuum_queues_the_task(client, monkeypatch):
     assert calls == [1]
 
 
+# --- maintenance triggers route through _queued() (DRY Finding B, #753) ---
+# Each operator-triggered task returns 202 + the Celery task id via the shared
+# _queued() helper. These trigger endpoints had no route-level tests; cover every
+# helper consumer, and assert the dry_run flag threads through where applicable.
+
+
+def _fake_delay(captured):
+    def _delay(*args, **kwargs):
+        captured.append((args, kwargs))
+        return SimpleNamespace(id="task-xyz")
+    return _delay
+
+
+@pytest.mark.asyncio
+async def test_trigger_reextract_archives_queues(client, monkeypatch):
+    from backend.app.tasks import admin as admin_tasks
+
+    calls = []
+    monkeypatch.setattr(
+        admin_tasks.reextract_archive_attachments_task, "delay", _fake_delay(calls)
+    )
+    resp = await client.post("/api/admin/maintenance/reextract-archives")
+    assert resp.status_code == 202
+    assert (await resp.get_json())["task_id"] == "task-xyz"
+    assert len(calls) == 1
+
+
+@pytest.mark.asyncio
+async def test_trigger_prune_missing_files_queues(client, monkeypatch):
+    from backend.app.tasks import admin as admin_tasks
+
+    calls = []
+    monkeypatch.setattr(
+        admin_tasks.prune_missing_file_records_task, "delay", _fake_delay(calls)
+    )
+    resp = await client.post("/api/admin/maintenance/prune-missing-files")
+    assert resp.status_code == 202
+    assert (await resp.get_json())["task_id"] == "task-xyz"
+
+
+@pytest.mark.asyncio
+async def test_trigger_dedup_videos_queues_and_threads_dry_run(client, monkeypatch):
+    from backend.app.tasks import admin as admin_tasks
+
+    calls = []
+    monkeypatch.setattr(admin_tasks.dedup_videos_task, "delay", _fake_delay(calls))
+    resp = await client.post(
+        "/api/admin/maintenance/dedup-videos", json={"dry_run": False},
+    )
+    assert resp.status_code == 202
+    assert (await resp.get_json())["task_id"] == "task-xyz"
+    assert calls[0][1] == {"dry_run": False}  # flag threaded to the task
+
+
+@pytest.mark.asyncio
+async def test_trigger_purge_gated_previews_queues_and_threads_dry_run(client, monkeypatch):
+    from backend.app.tasks import admin as admin_tasks
+
+    calls = []
+    monkeypatch.setattr(
+        admin_tasks.purge_gated_previews_task, "delay", _fake_delay(calls)
+    )
+    resp = await client.post(
+        "/api/admin/maintenance/purge-gated-previews", json={"dry_run": True},
+    )
+    assert resp.status_code == 202
+    assert (await resp.get_json())["task_id"] == "task-xyz"
+    assert calls[0][1] == {"dry_run": True}
+
+
+@pytest.mark.asyncio
+async def test_trigger_normalize_live_queues(client, monkeypatch):
+    from backend.app.tasks import admin as admin_tasks
+
+    calls = []
+    monkeypatch.setattr(admin_tasks.normalize_tags_task, "delay", _fake_delay(calls))
+    resp = await client.post("/api/admin/tags/normalize", json={"dry_run": False})
+    assert resp.status_code == 202
+    assert (await resp.get_json())["task_id"] == "task-xyz"
+
+
 # --- Tier-A: POST /tags/reset-content -------------------------------