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>

backend/app/api/admin.py

@@ -56,6 +56,14 @@ async def _run_dry_run_op(service_fn, **service_kwargs):
     return jsonify(result)
 
 
+def _queued(async_result):
+    """Standard 202 for an operator-triggered maintenance task: hand the UI the
+    Celery task id so it can tail /maintenance/task-result (or the activity
+    dashboard) for the summary. (trigger_vacuum stays bespoke — the UI doesn't
+    poll it, so it returns no task id.)"""
+    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
+
+
 @admin_bp.route("/artists/<slug>/cascade-delete", methods=["POST"])
 async def artist_cascade_delete(slug: str):
     body = await request.get_json(silent=True) or {}
@@ -295,7 +303,7 @@ async def tags_normalize():
     from ..tasks.admin import normalize_tags_task
 
     async_result = normalize_tags_task.delay()
-    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
+    return _queued(async_result)
 
 
 @admin_bp.route("/maintenance/db-stats", methods=["GET"])
@@ -352,7 +360,7 @@ async def trigger_reextract_archives():
     from ..tasks.admin import reextract_archive_attachments_task
 
     async_result = reextract_archive_attachments_task.delay()
-    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
+    return _queued(async_result)
 
 
 @admin_bp.route("/maintenance/prune-missing-files", methods=["POST"])
@@ -365,7 +373,7 @@ async def trigger_prune_missing_files():
     from ..tasks.admin import prune_missing_file_records_task
 
     async_result = prune_missing_file_records_task.delay()
-    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
+    return _queued(async_result)
 
 
 @admin_bp.route("/maintenance/dedup-videos", methods=["POST"])
@@ -381,7 +389,7 @@ async def trigger_dedup_videos():
     body = await request.get_json(silent=True) or {}
     dry_run = bool(body.get("dry_run", True))  # default to the SAFE preview
     async_result = dedup_videos_task.delay(dry_run=dry_run)
-    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
+    return _queued(async_result)
 
 
 @admin_bp.route("/maintenance/purge-gated-previews", methods=["POST"])
@@ -397,7 +405,7 @@ async def trigger_purge_gated_previews():
     body = await request.get_json(silent=True) or {}
     dry_run = bool(body.get("dry_run", True))  # default to the SAFE preview
     async_result = purge_gated_previews_task.delay(dry_run=dry_run)
-    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
+    return _queued(async_result)
 
 
 @admin_bp.route("/maintenance/task-result/<task_id>", methods=["GET"])

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 -------------------------------