Fix/prevent intentional conversions from creating false quality upgrades

core/imports/file_ops.py

@@ -568,14 +568,15 @@ def probe_audio_quality(file_path: str):
                 sample_rate=audio.info.sample_rate,
             )
 
-        if ext in ('m4a', 'aac', 'mp4'):
+        if ext in ('m4a', 'aac', 'mp4', 'alac'):
             from mutagen.mp4 import MP4
             audio = MP4(file_path)
             # .m4a can carry AAC (lossy) OR ALAC (lossless) — only the real
-            # codec tells them apart, which is why extension-based classification
-            # defaults to 'aac' and we correct it here from the probed file.
+            # codec tells them apart.  The explicit .alac extension is already
+            # unambiguous and is accepted by the lossy-copy path, so it must be
+            # probed here as ALAC as well to preserve acquisition provenance.
             codec = (getattr(audio.info, 'codec', '') or '').lower()
-            if 'alac' in codec:
+            if ext == 'alac' or 'alac' in codec:
                 return AudioQuality(
                     format='alac',
                     bitrate=audio.info.bitrate // 1000 if audio.info.bitrate else None,

core/imports/pipeline.py

@@ -17,6 +17,7 @@
     downsample_hires_flac,
     get_audio_quality_string,
     get_quality_tier_from_extension,
+    probe_audio_quality,
     safe_move_file,
 )
 from core.imports.context import (
@@ -432,6 +433,58 @@ def _persist_verification_status(context, final_path):
         logger.debug(f"verification-status persist skipped: {_vs_err}")
 
 
+def _apply_profile_output_transforms(final_path: str, context: dict,
+                                     profile: dict) -> str:
+    """Apply downsample/lossy retention while preserving acquisition truth."""
+    acquired_quality = probe_audio_quality(final_path)
+    if acquired_quality is not None:
+        context['_acquired_audio_quality'] = acquired_quality.to_dict()
+
+    downsampled_path = downsample_hires_flac(
+        final_path, context, enabled=profile.get('downsample_enabled'))
+    if downsampled_path:
+        final_path = downsampled_path
+        context['_final_processed_path'] = final_path
+        retained_quality = probe_audio_quality(final_path)
+        context.setdefault('_retention_transforms', []).append({
+            'type': 'downsample_hires_flac',
+            'source_replaced': True,
+            'target_bit_depth': 16,
+            'target_sample_rate': 44100,
+            'output_quality': retained_quality.to_dict() if retained_quality else None,
+        })
+
+    _persist_verification_status(context, final_path)
+
+    lossy_path = create_lossy_copy(final_path, settings={
+        'enabled': profile.get('lossy_copy_enabled'),
+        'codec': profile.get('lossy_copy_codec'),
+        'bitrate': profile.get('lossy_copy_bitrate'),
+        'delete_original': profile.get('lossy_copy_delete_original'),
+    } if profile else None)
+    if not lossy_path:
+        return final_path
+
+    source_retained = os.path.isfile(final_path)
+    lossy_quality = probe_audio_quality(lossy_path)
+    context.setdefault('_retention_transforms', []).append({
+        'type': 'lossy_copy',
+        'source_replaced': not source_retained,
+        'codec': profile.get('lossy_copy_codec'),
+        'bitrate': profile.get('lossy_copy_bitrate'),
+        'output_quality': lossy_quality.to_dict() if lossy_quality else None,
+    })
+    if source_retained:
+        companions = context.setdefault('_companion_file_paths', [])
+        if lossy_path not in companions:
+            companions.append(lossy_path)
+        context['_final_processed_path'] = final_path
+        return final_path
+
+    context['_final_processed_path'] = lossy_path
+    return lossy_path
+
+
 def post_process_matched_download(context_key, context, file_path, runtime, metadata_runtime=None):
     on_download_completed = getattr(runtime, "on_download_completed", None)
     automation_engine = getattr(runtime, "automation_engine", None)
@@ -1298,23 +1351,8 @@ def _notify_download_completed(batch_id, task_id, success=True):
                 pp_logger.debug(f"ReplayGain analysis skipped: {rg_err}")
 
         _qp_post = _resolve_context_quality_profile(context)
-        downsampled_path = downsample_hires_flac(
-            final_path, context,
-            enabled=_qp_post.get('downsample_enabled'))
-        if downsampled_path:
-            final_path = downsampled_path
-            context['_final_processed_path'] = final_path
-
-        _persist_verification_status(context, final_path)
-
-        blasphemy_path = create_lossy_copy(final_path, settings={
-            'enabled': _qp_post.get('lossy_copy_enabled'),
-            'codec': _qp_post.get('lossy_copy_codec'),
-            'bitrate': _qp_post.get('lossy_copy_bitrate'),
-            'delete_original': _qp_post.get('lossy_copy_delete_original'),
-        } if _qp_post else None)
-        if blasphemy_path:
-            context['_final_processed_path'] = blasphemy_path
+        final_path = _apply_profile_output_transforms(
+            final_path, context, _qp_post)
 
         downloads_path = docker_resolve_path(config_manager.get('soulseek.download_path', './downloads'))
         cleanup_empty_directories(downloads_path, file_path)

core/imports/side_effects.py

@@ -50,6 +50,22 @@ def _stable_soulsync_id(text: str) -> str:
     return str(abs(int(hashlib.md5(text.encode("utf-8", errors="replace")).hexdigest(), 16)) % (10 ** 9))
 
 
+def _retention_provenance_json(context: Dict[str, Any]) -> tuple[str | None, str | None]:
+    """Serialize acquisition/retention truth for either persistence path."""
+    from core.quality.model import AudioQuality
+    from core.quality.retention import quality_json, transforms_json
+
+    acquired_value = context.get("_acquired_audio_quality")
+    try:
+        acquired_quality = AudioQuality.from_dict(acquired_value) if acquired_value else None
+    except (TypeError, ValueError):
+        acquired_quality = None
+    return (
+        quality_json(acquired_quality),
+        transforms_json(context.get("_retention_transforms")),
+    )
+
+
 # Tiny SQL allowlist for the fill-empty helpers — prevents accidental
 # SQL injection through the f-string column-name interpolation. Only
 # columns the soulsync library write path ever updates are listed.
@@ -348,6 +364,7 @@ def _embedded(*keys):
         audiodb_id = _embedded("AUDIODB_TRACK_ID")
         soul_id = _embedded("SOUL_ID")
         isrc = context.get("_isrc")
+        acquired_quality_json, retention_json = _retention_provenance_json(context)
 
         db = get_database()
         db.record_track_download(
@@ -372,6 +389,8 @@ def _embedded(*keys):
             audiodb_id=audiodb_id,
             soul_id=soul_id,
             isrc=isrc,
+            acquired_quality_json=acquired_quality_json,
+            retention_json=retention_json,
         )
     except Exception as e:
         logger.debug("record_download_provenance failed: %s", e)
@@ -667,18 +686,22 @@ def record_soulsync_library_entry(context: Dict[str, Any], artist_context: Dict[
             # Quality Upgrade Finder passes re-resolve the track against the
             # default profile instead of the one it was actually imported under.
             track_quality_profile_id = track_info.get("quality_profile_id")
+            acquired_quality_json, retention_json = _retention_provenance_json(context)
+            try:
+                track_columns = {
+                    column[1] for column in cursor.execute(
+                        "PRAGMA table_info(tracks)").fetchall()
+                }
+            except Exception:  # pragma is best-effort for non-SQLite test doubles
+                track_columns = set()
+            has_retention_columns = {
+                "acquired_quality_json", "retention_json"
+            }.issubset(track_columns)
 
             cursor.execute("SELECT id FROM tracks WHERE file_path = ?", (final_path,))
-            if not cursor.fetchone():
-                cursor.execute(
-                    """
-                    INSERT INTO tracks (id, album_id, artist_id, title, track_number,
-                                        duration, file_path, bitrate, file_size, track_artist,
-                                        musicbrainz_recording_id, isrc, quality_profile_id, server_source,
-                                        created_at, updated_at)
-                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'soulsync', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
-                    """,
-                    (
+            existing_track = cursor.fetchone()
+            if not existing_track:
+                base_values = (
                         track_id,
                         album_id,
                         artist_id,
@@ -692,8 +715,32 @@ def record_soulsync_library_entry(context: Dict[str, Any], artist_context: Dict[
                         track_mbid,
                         track_isrc,
                         track_quality_profile_id,
-                    ),
                 )
+                if has_retention_columns:
+                    cursor.execute(
+                        """
+                        INSERT INTO tracks (id, album_id, artist_id, title, track_number,
+                                            duration, file_path, bitrate, file_size, track_artist,
+                                            musicbrainz_recording_id, isrc, quality_profile_id,
+                                            acquired_quality_json, retention_json, server_source,
+                                            created_at, updated_at)
+                        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
+                                'soulsync', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
+                        """,
+                        base_values + (acquired_quality_json, retention_json),
+                    )
+                else:
+                    cursor.execute(
+                        """
+                        INSERT INTO tracks (id, album_id, artist_id, title, track_number,
+                                            duration, file_path, bitrate, file_size, track_artist,
+                                            musicbrainz_recording_id, isrc, quality_profile_id, server_source,
+                                            created_at, updated_at)
+                        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
+                                'soulsync', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
+                        """,
+                        base_values,
+                    )
                 track_source_col = source_columns.get("track")
                 if track_source_col and track_source_id:
                     try:
@@ -709,6 +756,18 @@ def record_soulsync_library_entry(context: Dict[str, Any], artist_context: Dict[
                             )
                     except Exception as e:
                         logger.debug("track source-id update failed: %s", e)
+            elif has_retention_columns:
+                # A repeated/import-resume write refreshes transformation
+                # provenance for the existing physical row.  Never retain an
+                # old destructive-policy claim after a clean untransformed
+                # import of that same path.
+                cursor.execute(
+                    """UPDATE tracks
+                          SET acquired_quality_json=?, retention_json=?,
+                              updated_at=CURRENT_TIMESTAMP
+                        WHERE id=?""",
+                    (acquired_quality_json, retention_json, existing_track[0]),
+                )
 
             conn.commit()
             logger.info("[SoulSync Library] Added: %s / %s / %s", artist_name, album_name, track_name)

core/library/duplicate_cleaner.py

@@ -9,7 +9,9 @@
 
 from core.settings import config_manager
 from core.runtime_state import add_activity_item
+from core.library.duplicate_rules import is_lossy_companion_pair, lossy_companion_exts
 from core.repair_jobs.base import skip_deleted_quarantine
+from database.music_database import get_database
 
 logger = logging.getLogger(__name__)
 
@@ -136,6 +138,19 @@ def _run_duplicate_cleaner():
             '.mp3': 4, '.wma': 4    # Lower quality lossy
         }
 
+        # The lossy-copy feature deliberately writes a same-folder,
+        # same-stem MP3/Opus/M4A next to its lossless source.  The catalogue
+        # detector already protects that pair; the filesystem cleaner must
+        # apply the same shared rule instead of quarantining the intended copy.
+        try:
+            database = get_database()
+        except Exception as e:  # noqa: BLE001 - filesystem scan stays usable
+            logger.debug("lossy companion database unavailable: %s", e)
+            database = None
+        companion_exts = lossy_companion_exts(
+            config_manager, database, logger=logger,
+        )
+
         duplicates_found = 0
         deleted_count = 0
         space_freed = 0
@@ -146,9 +161,6 @@ def _run_duplicate_cleaner():
                 if len(file_versions) <= 1:
                     continue
 
-                duplicates_found += len(file_versions) - 1  # Count all but the one we keep
-                logger.warning(f"[Duplicate Cleaner] Found {len(file_versions)} versions of '{filename}' in {directory}")
-
                 # Sort by priority: best format first, then largest size
                 def sort_key(f):
                     priority = format_priority.get(f['extension'], 999)
@@ -157,12 +169,29 @@ def sort_key(f):
 
                 sorted_versions = sorted(file_versions, key=sort_key)
 
-                # Keep the first one (best quality), delete the rest
+                # Keep the first one (best quality).  Configured lossy
+                # companions of that lossless winner are intentional; only
+                # the remaining versions are actionable duplicates.
                 best_version = sorted_versions[0]
+                duplicate_versions = [
+                    version for version in sorted_versions[1:]
+                    if not is_lossy_companion_pair(
+                        best_version['full_path'], version['full_path'],
+                        companion_exts,
+                    )
+                ]
+                if not duplicate_versions:
+                    continue
+
+                duplicates_found += len(duplicate_versions)
+                logger.warning(
+                    f"[Duplicate Cleaner] Found {len(duplicate_versions) + 1} "
+                    f"versions of '{filename}' in {directory}"
+                )
                 logger.warning(f"[Duplicate Cleaner] Keeping: {os.path.basename(best_version['full_path'])} "
                       f"({best_version['extension']}, {best_version['size']} bytes)")
 
-                for duplicate_file in sorted_versions[1:]:
+                for duplicate_file in duplicate_versions:
                     try:
                         # Move to deleted folder with relative path preserved
                         relative_path = os.path.relpath(duplicate_file['full_path'], transfer_folder)