feat(phash): import task maps 'superseded' to complete + re-queues ML/thumbnail

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

backend/app/tasks/import_file.py

@@ -24,6 +24,18 @@ def _sync_session_factory():
 IMAGES_ROOT = Path("/images")
 
 
+def _map_result_to_status(result):
+    """(ImportTask.status, should_requeue_ml_and_thumb) for an ImportResult.
+    'superseded' = the kept row's file/ML changed → complete + re-derive."""
+    if result.status == "imported":
+        return ("complete", True)
+    if result.status == "superseded":
+        return ("complete", True)
+    if result.status == "skipped":
+        return ("skipped", False)
+    return ("failed", False)
+
+
 @celery.task(name="backend.app.tasks.import_file.import_media_file", bind=True)
 def import_media_file(self, import_task_id: int) -> dict:
     """Returns a dict so the eager-mode tests can assert without DB."""
@@ -65,7 +77,7 @@ def import_media_file(self, import_task_id: int) -> dict:
             session.commit()
             raise
 
-        if result.status == "imported":
+        if result.status in ("imported", "superseded"):
             task.status = "complete"
             task.result_image_id = result.image_id
             counter_col_name = "imported"
@@ -95,8 +107,12 @@ def import_media_file(self, import_task_id: int) -> dict:
         session.add(task)
         session.commit()
 
-        # Enqueue the thumbnail + ML tasks for newly imported images.
+        # Enqueue thumbnail + ML for newly imported AND superseded images
-        if result.status == "imported" and result.image_id is not None:
+        # (a superseded row has cleared ML + no thumbnail).
+        if (
+            result.status in ("imported", "superseded")
+            and result.image_id is not None
+        ):
             from .ml import tag_and_embed
             from .thumbnail import generate_thumbnail
 

tests/test_phash_dedup.py

@@ -136,3 +136,21 @@ def test_threshold_controls_match(importer, import_layout):
     _write(b, (104, 100, 100), (900, 900))  # tiny tweak
     r = importer.import_one(b)
     assert r.status == "imported"  # threshold 0 → not treated as near-dup
+
+
+def test_import_task_maps_superseded_to_complete_and_requeues():
+    from backend.app.services.importer import ImportResult
+    from backend.app.tasks.import_file import _map_result_to_status
+
+    assert _map_result_to_status(
+        ImportResult(status="superseded", image_id=5)
+    ) == ("complete", True)
+    assert _map_result_to_status(
+        ImportResult(status="imported", image_id=5)
+    ) == ("complete", True)
+    assert _map_result_to_status(
+        ImportResult(status="skipped", skip_reason=SkipReason.duplicate_phash)
+    ) == ("skipped", False)
+    assert _map_result_to_status(
+        ImportResult(status="failed", error="boom")
+    ) == ("failed", False)