fix(imports): quality/audio-guard quarantine no longer also marks task completed

A file quarantined for QUALITY (e.g. MP3-VBR rejected by a FLAC-only profile)
showed up in BOTH the Completed and Quarantine tabs. Cause: the verification
wrapper (post_process_matched_download_with_verification) handled the
_acoustid_quarantined / _integrity_failure_msg / _race_guard_failed markers but
NOT the quality marker _bitdepth_rejected (nor _silence_rejected). A quality
quarantine leaves no _final_processed_path, so the wrapper hit the
"no final path — assuming success" branch and marked the task Completed.

Unlike acoustid/integrity (retry driven by the wrapper), the inner pipeline
already owns the quality/audio-guard outcome — it quarantines then re-queues the
next-best candidate or marks the task failed. So the wrapper now just returns
when it sees _bitdepth_rejected/_silence_rejected, without marking completed
(which clobbered both the quarantine state AND any successful retry).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

core/imports/pipeline.py

@@ -1028,6 +1028,22 @@ def post_process_matched_download_with_verification(context_key, context, file_p
         if original_batch_id:
             context['batch_id'] = original_batch_id
 
+        # Quality / audio-guard quarantine. Unlike acoustid/integrity (whose
+        # retry+fail is driven by THIS wrapper below), the inner pipeline fully
+        # owns the quality and audio-guard outcome: it quarantines the file and
+        # then either re-queues the next-best candidate or marks the task failed
+        # and notifies. The wrapper must NOT continue to the "assume success"
+        # fall-through — that marked the quarantined file Completed, so the same
+        # track showed in BOTH Completed and Quarantine (e.g. an MP3-VBR rejected
+        # by a FLAC-only profile). It must also NOT mark it failed here, which
+        # would clobber a successful next-candidate retry. Just return.
+        if context.get('_bitdepth_rejected') or context.get('_silence_rejected'):
+            logger.info(
+                f"Task {task_id} quarantined by the quality/audio guard — inner "
+                f"pipeline already handled retry/fail; wrapper not marking completed"
+            )
+            return
+
         if context.get('_race_guard_failed'):
             logger.info(f"Race guard: source file gone for task {task_id} — marking as failed")
             with tasks_lock:

tests/test_integrity_failure_marks_task_failed.py

@@ -144,6 +144,60 @@ def test_race_guard_failure_marker_marks_task_failed(_isolate_state):
     assert ('b2', 't2', False) in completion_calls
 
 
+def test_quality_quarantine_does_not_mark_completed(_isolate_state):
+    """When the inner pipeline quality-quarantines the file (``_bitdepth_rejected``),
+    the wrapper must NOT fall through to "assume success" and mark the task
+    Completed — that made the quarantined file appear in BOTH Completed and
+    Quarantine. The inner pipeline already owns the retry/fail for quality, so
+    the wrapper just returns."""
+    completion_calls = []
+    runtime = _build_runtime(completion_calls)
+
+    _seed_task('t5', 'b5')
+
+    context = {'task_id': 't5', 'batch_id': 'b5', 'context_key': 'test::ctx5'}
+
+    def _inner(*a, **kw):
+        # Inner pipeline quarantined for quality and handled its own retry/fail;
+        # it sets the marker and leaves NO _final_processed_path.
+        context['_bitdepth_rejected'] = True
+
+    with patch.object(import_pipeline, 'post_process_matched_download', _inner), \
+         patch.object(import_pipeline, '_mark_task_completed',
+                      lambda task_id, ti: runtime_state.download_tasks[task_id].update(
+                          {'status': 'completed'})):
+        import_pipeline.post_process_matched_download_with_verification(
+            'test::ctx5', context, '/fake/source.mp3', 't5', 'b5', runtime,
+        )
+
+    assert runtime_state.download_tasks['t5']['status'] != 'completed'
+    assert ('b5', 't5', True) not in completion_calls
+
+
+def test_silence_quarantine_does_not_mark_completed(_isolate_state):
+    """Same contract for the audio guard (``_silence_rejected``)."""
+    completion_calls = []
+    runtime = _build_runtime(completion_calls)
+
+    _seed_task('t6', 'b6')
+
+    context = {'task_id': 't6', 'batch_id': 'b6', 'context_key': 'test::ctx6'}
+
+    def _inner(*a, **kw):
+        context['_silence_rejected'] = True
+
+    with patch.object(import_pipeline, 'post_process_matched_download', _inner), \
+         patch.object(import_pipeline, '_mark_task_completed',
+                      lambda task_id, ti: runtime_state.download_tasks[task_id].update(
+                          {'status': 'completed'})):
+        import_pipeline.post_process_matched_download_with_verification(
+            'test::ctx6', context, '/fake/source.flac', 't6', 'b6', runtime,
+        )
+
+    assert runtime_state.download_tasks['t6']['status'] != 'completed'
+    assert ('b6', 't6', True) not in completion_calls
+
+
 def test_no_failure_markers_still_assumes_success(_isolate_state):
     """The pre-existing "assume success" fallback must STILL fire when
     no failure markers are set — some legitimate flows complete without