Fix import normalization and task completion locking

- Promote legacy _source into source during import normalization.
- Keep the normalized import context neutral after stripping aliases.
- Avoid re-entering tasks_lock when marking completed download tasks.

core/imports/context.py

@@ -45,16 +45,20 @@ def normalize_import_context(context: Optional[Dict[str, Any]]) -> Dict[str, Any
     if not isinstance(context, dict):
         return {}
 
+    source = context.get("source") or context.get("_source") or ""
     artist = _as_dict(context.get("artist") or context.get("spotify_artist"))
     album = _as_dict(context.get("album") or context.get("spotify_album"))
     track_info = _as_dict(context.get("track_info"))
     original_search = _as_dict(context.get("original_search_result"))
     search_result = _as_dict(context.get("search_result"))
 
+    if source:
+        context["source"] = source
     context["artist"] = artist
     context["album"] = album
     context["track_info"] = track_info
     context["original_search_result"] = original_search or search_result
+    context.pop("_source", None)
     context.pop("spotify_artist", None)
     context.pop("spotify_album", None)
 

core/runtime_state.py

@@ -51,15 +51,17 @@ def add_activity_item(icon, title, subtitle, time_ago="Now", show_toast=True):
 
 
 def mark_task_completed(task_id: str, track_info: Optional[Dict[str, Any]] = None) -> bool:
-    """Mark a download task as completed in the shared task registry."""
-    with tasks_lock:
-        task = download_tasks.get(task_id)
-        if not task:
-            return False
+    """Mark a download task as completed.
 
-        task["status"] = "completed"
-        task["stream_processed"] = True
-        task["status_change_time"] = time.time()
-        if track_info is not None:
-            task["track_info"] = track_info
-        return True
+    Callers must already hold `tasks_lock`.
+    """
+    task = download_tasks.get(task_id)
+    if not task:
+        return False
+
+    task["status"] = "completed"
+    task["stream_processed"] = True
+    task["status_change_time"] = time.time()
+    if track_info is not None:
+        task["track_info"] = track_info
+    return True

tests/imports/test_import_context.py

@@ -63,6 +63,22 @@ def test_normalize_import_context_promotes_neutral_fields_without_legacy_aliases
     assert get_import_has_full_metadata(normalized) is False
 
 
+def test_normalize_import_context_promotes_legacy_source_alias():
+    context = {
+        "_source": "spotify",
+        "artist": {"name": "Artist One", "id": "artist-1"},
+        "album": {"name": "Album One", "id": "album-1"},
+        "track_info": {"name": "Song One", "id": "track-1"},
+        "original_search_result": {"title": "Song One"},
+    }
+
+    normalized = normalize_import_context(context)
+
+    assert normalized["source"] == "spotify"
+    assert "_source" not in normalized
+    assert get_import_source(normalized) == "spotify"
+
+
 def test_neutral_import_context_helpers_work_without_legacy_aliases():
     context = {
         "source": "deezer",

web_server.py

@@ -141,7 +141,7 @@ from core.runtime_state import (
     download_tasks,
     matched_context_lock,
     matched_downloads_context,
-    mark_task_completed as _core_mark_task_completed,
+    mark_task_completed,
     processed_download_ids,
     set_activity_toast_emitter,
     tasks_lock,
@@ -2483,7 +2483,7 @@ def _mark_task_completed(task_id, track_info=None):
     Assumes task_id exists in download_tasks (should be called within tasks_lock).
     """
     global session_completed_downloads
-    _core_mark_task_completed(task_id, track_info)
+    mark_task_completed(task_id, track_info)
 
     # Increment session counter (matches dashboard.py behavior)
     with session_stats_lock: