fix(downloads): harden album bundle staging

Route torrent and Usenet album bundles through private per-batch staging so Auto-Import cannot race public staging or duplicate imports.

Expose album-bundle progress in batch status and render it on the Downloads page while the external client is still downloading.

Tighten release handoff safety by rejecting archive path traversal, ignoring torrent candidates without a usable URL, and skipping Soulseek source reuse for torrent/Usenet batches.

Tests: .venv/bin/python -m pytest tests/downloads/test_downloads_status.py tests/test_album_bundle_dispatch.py tests/downloads/test_downloads_staging.py tests/test_torrent_usenet_plugins.py

core/archive_pipeline.py

@@ -225,6 +225,12 @@ def _extract_7z(archive_path: Path, dest: Path) -> Optional[Path]:
         return None
     try:
         with py7zr.SevenZipFile(archive_path, 'r') as sz:
+            dest_resolved = dest.resolve()
+            for name in sz.getnames():
+                target = (dest_resolved / name).resolve()
+                if dest_resolved not in target.parents and target != dest_resolved:
+                    logger.error("archive_pipeline: refusing path-traversal 7z member %r", name)
+                    return None
             sz.extractall(path=dest)
         return dest
     except Exception as e:

core/download_plugins/torrent.py

@@ -465,7 +465,8 @@ class TorrentDownloadPlugin(DownloadSourcePlugin):
             result['error'] = f'Prowlarr search failed: {e}'
             return result
 
-        candidates = [r for r in search_results if r.protocol == 'torrent']
+        candidates = [r for r in search_results
+                      if r.protocol == 'torrent' and (r.magnet_uri or r.download_url)]
         if not candidates:
             result['error'] = f'No torrent results found for "{query}"'
             return result

core/downloads/album_bundle_dispatch.py

@@ -31,6 +31,7 @@ testable without touching live runtime state.
 from __future__ import annotations
 
 import logging
+from pathlib import Path
 from typing import Any, Callable, Optional, Protocol
 
 logger = logging.getLogger(__name__)
@@ -127,7 +128,11 @@ def try_dispatch(
         )
         return False
 
-    staging_dir = config_get('import.staging_path', './Staging') or './Staging'
+    staging_root = config_get(
+        'download_source.album_bundle_staging_path',
+        'storage/album_bundle_staging',
+    ) or 'storage/album_bundle_staging'
+    staging_dir = str(Path(staging_root) / _safe_batch_dirname(batch_id))
     logger.info(
         "[Album Bundle] Engaging %s album flow for '%s' by '%s' -> %s",
         mode, album_name, artist_name, staging_dir,
@@ -136,6 +141,8 @@ def try_dispatch(
         'phase': 'album_downloading',
         'album_bundle_state': 'searching',
         'album_bundle_source': mode,
+        'album_bundle_staging_path': staging_dir,
+        'album_bundle_private_staging': True,
     })
 
     def _emit(payload):
@@ -177,3 +184,8 @@ def try_dispatch(
     # task rows. Those tasks will hit try_staging_match and pull the
     # files we just staged.
     return False
+
+
+def _safe_batch_dirname(batch_id: str) -> str:
+    safe = ''.join(ch if ch.isalnum() or ch in ('-', '_') else '_' for ch in str(batch_id or 'batch'))
+    return safe or 'batch'

core/downloads/staging.py

@@ -160,6 +160,14 @@ def try_staging_match(task_id, batch_id, track, deps: StagingDeps):
             except Exception as _exc:
                 logger.debug("get_batch_field failed: %s", _exc)
         _provenance_username = _provenance_override or 'staging'
+        _private_album_bundle_staging = False
+        if batch_id and deps.get_batch_field is not None:
+            try:
+                _private_album_bundle_staging = bool(
+                    deps.get_batch_field(batch_id, 'album_bundle_private_staging')
+                )
+            except Exception as _exc:
+                logger.debug("get_batch_field failed: %s", _exc)
         with tasks_lock:
             if task_id in download_tasks:
                 download_tasks[task_id]['status'] = 'post_processing'
@@ -167,6 +175,15 @@ def try_staging_match(task_id, batch_id, track, deps: StagingDeps):
                 download_tasks[task_id]['username'] = _provenance_username
                 download_tasks[task_id]['staging_match'] = True
 
+        if _private_album_bundle_staging:
+            try:
+                os.remove(best_match['full_path'])
+                logger.debug("[Staging] Removed private album-bundle staging file: %s", best_match['full_path'])
+            except FileNotFoundError:
+                pass
+            except Exception as _exc:
+                logger.debug("[Staging] Could not remove private album-bundle staging file: %s", _exc)
+
         # Run post-processing (tagging, AcoustID verification, path building)
         context_key = f"staging_{task_id}"
         with tasks_lock:

core/downloads/status.py

@@ -247,6 +247,9 @@ def build_batch_status_data(batch_id: str, batch: dict, live_transfers_lookup: d
         "playlist_id": batch.get('playlist_id'),  # Include playlist_id for rehydration
         "playlist_name": batch.get('playlist_name'),  # Include playlist_name for reference
     }
+    album_bundle = _build_album_bundle_status(batch)
+    if album_bundle:
+        response_data['album_bundle'] = album_bundle
 
     if response_data["phase"] == 'analysis':
         response_data['analysis_progress'] = {
@@ -610,7 +613,7 @@ def build_unified_downloads_response(limit: int, deps: StatusDeps) -> dict:
         for bid, batch in download_batches.items():
             queue = batch.get('queue', [])
             statuses = [download_tasks[tid]['status'] for tid in queue if tid in download_tasks]
-            batch_summaries.append({
+            summary = {
                 'batch_id': bid,
                 'playlist_id': batch.get('playlist_id', ''),
                 'batch_name': batch.get('playlist_name') or batch.get('album_name') or '',
@@ -621,7 +624,11 @@ def build_unified_downloads_response(limit: int, deps: StatusDeps) -> dict:
                 'failed': sum(1 for s in statuses if s in ('failed', 'not_found', 'cancelled')),
                 'active': sum(1 for s in statuses if s in ('downloading', 'searching', 'post_processing')),
                 'queued': sum(1 for s in statuses if s in ('queued', 'pending')),
-            })
+            }
+            album_bundle = _build_album_bundle_status(batch)
+            if album_bundle:
+                summary['album_bundle'] = album_bundle
+            batch_summaries.append(summary)
 
     return {
         'success': True,
@@ -630,3 +637,38 @@ def build_unified_downloads_response(limit: int, deps: StatusDeps) -> dict:
         'batches': batch_summaries,
         'timestamp': time.time(),
     }
+
+
+def _build_album_bundle_status(batch: dict) -> dict:
+    """Return public batch-level status for torrent/Usenet album-bundle work."""
+    state = batch.get('album_bundle_state')
+    if not state:
+        return {}
+
+    status = {
+        'state': state,
+        'source': batch.get('album_bundle_source'),
+        'release': batch.get('album_bundle_release'),
+        'progress': batch.get('album_bundle_progress'),
+        'progress_percent': _album_bundle_progress_percent(
+            batch.get('album_bundle_progress')
+        ),
+        'speed': batch.get('album_bundle_speed'),
+        'downloaded': batch.get('album_bundle_downloaded'),
+        'size': batch.get('album_bundle_size'),
+        'seeders': batch.get('album_bundle_seeders'),
+        'grabs': batch.get('album_bundle_grabs'),
+        'count': batch.get('album_bundle_count'),
+    }
+    return {key: value for key, value in status.items() if value is not None}
+
+
+def _album_bundle_progress_percent(value: Any) -> int:
+    try:
+        progress = float(value)
+    except (TypeError, ValueError):
+        return 0
+
+    if progress <= 1:
+        progress *= 100
+    return max(0, min(100, int(round(progress))))

tests/downloads/test_downloads_staging.py

@@ -60,6 +60,7 @@ def _build_deps(
     transfer_path,
     staging_files=None,
     post_process_calls=None,
+    get_batch_field=None,
 ):
     post_process_calls = post_process_calls if post_process_calls is not None else []
     deps = ds.StagingDeps(
@@ -68,6 +69,7 @@ def _build_deps(
         get_staging_file_cache=lambda batch_id: staging_files or [],
         docker_resolve_path=lambda p: p,  # passthrough
         post_process_matched_download_with_verification=lambda *a, **kw: post_process_calls.append((a, kw)),
+        get_batch_field=get_batch_field,
     )
     deps._post_process_calls = post_process_calls
     return deps
@@ -157,6 +159,56 @@ def test_exact_match_copies_to_transfer_and_marks_post_processing(tmp_path):
     assert context_key == 'staging_t4'
 
 
+def test_private_album_bundle_staging_source_is_removed_after_claim(tmp_path):
+    src_file = tmp_path / 'private' / 'Hello.flac'
+    src_file.parent.mkdir()
+    src_file.write_bytes(b'fake audio')
+
+    transfer_dir = tmp_path / 'transfer'
+
+    def get_batch_field(_batch_id, field):
+        if field == 'album_bundle_source':
+            return 'torrent'
+        if field == 'album_bundle_private_staging':
+            return True
+        return None
+
+    deps = _build_deps(
+        transfer_path=str(transfer_dir),
+        staging_files=[
+            {'full_path': str(src_file), 'title': 'Hello', 'artist': 'Artist One'},
+        ],
+        get_batch_field=get_batch_field,
+    )
+    _seed_task('t_private')
+
+    result = ds.try_staging_match('t_private', 'b_private', _Track(), deps)
+
+    assert result is True
+    assert (transfer_dir / 'Hello.flac').exists()
+    assert not src_file.exists()
+    assert download_tasks['t_private']['username'] == 'torrent'
+
+
+def test_public_staging_source_is_kept_after_match(tmp_path):
+    src_file = tmp_path / 'staging' / 'Hello.flac'
+    src_file.parent.mkdir()
+    src_file.write_bytes(b'fake audio')
+
+    deps = _build_deps(
+        transfer_path=str(tmp_path / 'transfer'),
+        staging_files=[
+            {'full_path': str(src_file), 'title': 'Hello', 'artist': 'Artist One'},
+        ],
+    )
+    _seed_task('t_public')
+
+    result = ds.try_staging_match('t_public', 'b_public', _Track(), deps)
+
+    assert result is True
+    assert src_file.exists()
+
+
 def test_existing_file_in_transfer_gets_staging_suffix(tmp_path):
     """If destination already exists, suffix '_staging' added to avoid overwrite."""
     src_file = tmp_path / 'staging' / 'Hello.flac'

tests/downloads/test_downloads_status.py

@@ -85,6 +85,38 @@ def test_analysis_phase_includes_analysis_progress_and_results():
     assert out['analysis_results'] == [{'track_index': 0, 'found': True}]
 
 
+def test_album_downloading_phase_exposes_bundle_progress_without_task_safety_valve():
+    deps, _ = _build_deps(config=_FakeConfig({'soulseek.download_timeout': 1}))
+    download_tasks['t1'] = {
+        'track_index': 0,
+        'status': 'queued',
+        'track_info': {},
+        'status_change_time': 0,
+    }
+    batch = {
+        'phase': 'album_downloading',
+        'queue': ['t1'],
+        'album_bundle_state': 'downloading',
+        'album_bundle_source': 'torrent',
+        'album_bundle_release': 'GNX [FLAC]',
+        'album_bundle_progress': 0.42,
+        'album_bundle_speed': 2048,
+        'album_bundle_size': 4096,
+    }
+    out = st.build_batch_status_data('b1', batch, {}, deps)
+    assert out['album_bundle'] == {
+        'state': 'downloading',
+        'source': 'torrent',
+        'release': 'GNX [FLAC]',
+        'progress': 0.42,
+        'progress_percent': 42,
+        'speed': 2048,
+        'size': 4096,
+    }
+    assert 'tasks' not in out
+    assert download_tasks['t1']['status'] == 'queued'
+
+
 def test_complete_phase_includes_wishlist_summary_when_present():
     deps, _ = _build_deps()
     batch = {
@@ -340,6 +372,26 @@ def test_batched_status_no_filter_returns_all_batches():
     assert set(out['batches'].keys()) == {'b1', 'b2'}
 
 
+def test_unified_downloads_response_includes_album_bundle_summary():
+    deps, _ = _build_deps()
+    download_batches['b1'] = {
+        'phase': 'album_downloading',
+        'playlist_id': 'pl1',
+        'playlist_name': 'GNX',
+        'queue': [],
+        'album_bundle_state': 'downloading',
+        'album_bundle_source': 'torrent',
+        'album_bundle_progress': 75,
+    }
+    out = st.build_unified_downloads_response(300, deps)
+    assert out['batches'][0]['album_bundle'] == {
+        'state': 'downloading',
+        'source': 'torrent',
+        'progress': 75,
+        'progress_percent': 75,
+    }
+
+
 def test_batched_status_metadata_present():
     deps, _ = _build_deps()
     download_batches['b1'] = {'phase': 'unknown'}

tests/test_album_bundle_dispatch.py

@@ -187,14 +187,36 @@ def test_dispatch_success_returns_false_so_per_track_can_run() -> None:
     args = plugin.download_album_to_staging.call_args
     assert args.args[0] == 'GNX'
     assert args.args[1] == 'Kendrick Lamar'
-    assert args.args[2] == '/staging/path'
+    assert args.args[2].replace('\\', '/').endswith('storage/album_bundle_staging/b1')
     # Phase transitioned through searching → analysis.
     assert state.fields['phase'] == 'analysis'
     assert state.fields['album_bundle_state'] == 'staged'
     assert state.fields['album_bundle_source'] == 'torrent'
+    assert state.fields['album_bundle_private_staging'] is True
+    assert state.fields['album_bundle_staging_path'].replace('\\', '/').endswith('storage/album_bundle_staging/b1')
     assert state.failed_with == ''
 
 
+def test_dispatch_uses_configured_private_album_bundle_staging_root() -> None:
+    state = _FakeState()
+    plugin = MagicMock()
+    plugin.download_album_to_staging.return_value = {'success': True, 'files': ['/tmp/a.flac']}
+
+    try_dispatch(
+        batch_id='batch:with/slash', is_album=True,
+        album_context={'name': 'GNX'}, artist_context={'name': 'Kendrick Lamar'},
+        config_get=_config({
+            'download_source.mode': 'torrent',
+            'download_source.album_bundle_staging_path': '/private/staging',
+        }),
+        plugin_resolver=lambda _name: plugin, state=state,
+    )
+
+    staging_arg = plugin.download_album_to_staging.call_args.args[2].replace('\\', '/')
+    assert staging_arg == '/private/staging/batch_with_slash'
+    assert state.fields['album_bundle_staging_path'].replace('\\', '/') == staging_arg
+
+
 def test_dispatch_failure_returns_true_so_master_stops() -> None:
     state = _FakeState()
     plugin = MagicMock()

tests/test_torrent_usenet_plugins.py

@@ -12,7 +12,7 @@ from __future__ import annotations
 import asyncio
 from pathlib import Path
 from typing import Optional
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -452,6 +452,22 @@ def test_torrent_album_to_staging_short_circuits_when_not_configured() -> None:
     assert 'not configured' in outcome['error'].lower()
 
 
+def test_torrent_album_to_staging_ignores_candidates_without_download_url(tmp_path: Path) -> None:
+    plugin = TorrentDownloadPlugin()
+    fake_adapter = MagicMock()
+    fake_adapter.is_configured.return_value = True
+    with patch.object(plugin, 'is_configured', return_value=True), \
+         patch.object(plugin._prowlarr, 'search', new=AsyncMock(return_value=[
+             _make_torrent_result(download_url=None, magnet_uri=None),
+         ])), \
+         patch('core.download_plugins.torrent.get_active_torrent_adapter', return_value=fake_adapter):
+        outcome = plugin.download_album_to_staging('GNX', 'Kendrick Lamar', str(tmp_path))
+
+    assert outcome['success'] is False
+    assert 'No torrent results' in outcome['error']
+    fake_adapter.add_torrent.assert_not_called()
+
+
 def test_registry_includes_torrent_and_usenet() -> None:
     """The registry decides what shows up in the orchestrator's
     iteration helpers. If we forget to register a new plugin the

web_server.py

@@ -16463,9 +16463,10 @@ def _get_staging_file_cache(batch_id):
     """Scan staging folder once per batch and cache the results."""
     with _staging_cache_lock:
         if batch_id in _staging_cache:
-            return _staging_cache[batch_id]
+            cached = _staging_cache[batch_id]
+            return [f for f in cached if os.path.exists(f.get('full_path', ''))]
 
-    staging_path = get_staging_path()
+    staging_path = _get_album_bundle_staging_path(batch_id) or get_staging_path()
     if not os.path.isdir(staging_path):
         with _staging_cache_lock:
             _staging_cache[batch_id] = []
@@ -16496,6 +16497,17 @@ def _get_staging_file_cache(batch_id):
     return files
 
 
+def _get_album_bundle_staging_path(batch_id):
+    """Return the private album-bundle staging path for this batch, if any."""
+    if not batch_id:
+        return None
+    with tasks_lock:
+        row = download_batches.get(batch_id)
+        if isinstance(row, dict) and row.get('album_bundle_private_staging'):
+            return row.get('album_bundle_staging_path')
+    return None
+
+
 # Staging-folder match shortcut lives in core/downloads/staging.py.
 from core.downloads import staging as _downloads_staging
 
@@ -16659,7 +16671,7 @@ def _store_batch_source(batch_id, username, filename):
     """Browse the successful download's folder and store results on the batch for reuse."""
     _sr = source_reuse_logger
     _sr.info(f"_store_batch_source called: batch={batch_id}, user={username}, file={filename}")
-    if not batch_id or username in ('youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud', 'amazon'):
+    if not batch_id or username in ('youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud', 'amazon', 'torrent', 'usenet'):
         _sr.info(f"Skipped — no batch_id or streaming source ({username})")
         return
 

webui/static/pages-extra.js

@@ -2445,6 +2445,34 @@ function _adlEsc(str) {
     return String(str).replace(/&/g, '&amp;').replace(/</g, '&lt;').replace(/>/g, '&gt;').replace(/"/g, '&quot;');
 }
 
+function _adlBundleProgressPercent(bundle) {
+    if (!bundle) return 0;
+    const raw = bundle.progress_percent ?? bundle.progress ?? 0;
+    let progress = Number(raw);
+    if (!Number.isFinite(progress)) progress = 0;
+    if (progress <= 1) progress *= 100;
+    return Math.max(0, Math.min(100, Math.round(progress)));
+}
+
+function _adlFormatBytes(bytes) {
+    const value = Number(bytes);
+    if (!Number.isFinite(value) || value <= 0) return '';
+    const units = ['B', 'KB', 'MB', 'GB', 'TB'];
+    let size = value;
+    let unit = 0;
+    while (size >= 1024 && unit < units.length - 1) {
+        size /= 1024;
+        unit += 1;
+    }
+    const decimals = size >= 10 || unit === 0 ? 0 : 1;
+    return `${size.toFixed(decimals)} ${units[unit]}`;
+}
+
+function _adlFormatSpeed(bytesPerSecond) {
+    const formatted = _adlFormatBytes(bytesPerSecond);
+    return formatted ? `${formatted}/s` : '';
+}
+
 async function adlClearCompleted() {
     try {
         const resp = await fetch('/api/downloads/clear-completed', { method: 'POST' });
@@ -2504,12 +2532,16 @@ function _adlRenderBatchPanel() {
         const colorStyle = colorIdx >= 0 ? `border-left-color: rgba(var(--batch-color-${colorIdx}), 0.6)` : '';
         const isExpanded = _adlExpandedBatches.has(batch.batch_id);
         const isFiltered = _adlFilterBatchId === batch.batch_id;
+        const albumBundle = batch.album_bundle || null;
+        const bundleProgress = _adlBundleProgressPercent(albumBundle);
         const total = batch.total || 1;
         const done = batch.completed + batch.failed;
-        const pct = Math.round((done / total) * 100);
+        const pct = batch.phase === 'album_downloading'
+            ? bundleProgress
+            : Math.round((done / total) * 100);
         const hasFailed = batch.failed > 0;
         const isTerminal = batch.phase === 'complete' || batch.phase === 'cancelled' || batch.phase === 'error';
-        const isActive = batch.phase === 'downloading' && batch.active > 0;
+        const isActive = (batch.phase === 'downloading' && batch.active > 0) || batch.phase === 'album_downloading';
 
         // Fade progress for completing batches
         let fadeStyle = '';
@@ -2532,6 +2564,14 @@ function _adlRenderBatchPanel() {
         if (batch.phase === 'analysis') {
             phaseText = 'Analyzing...';
             phaseIcon = '<span class="adl-spinner" style="margin-right:4px"></span>';
+        } else if (batch.phase === 'album_downloading') {
+            const source = albumBundle && albumBundle.source ? `${albumBundle.source} ` : '';
+            const release = albumBundle && albumBundle.release ? ` - ${albumBundle.release}` : '';
+            const speed = _adlFormatSpeed(albumBundle && albumBundle.speed);
+            const size = _adlFormatBytes(albumBundle && albumBundle.size);
+            const detail = speed || size ? ` (${[speed, size].filter(Boolean).join(' of ')})` : '';
+            phaseText = `${source}album ${albumBundle && albumBundle.state ? albumBundle.state : 'downloading'} ${bundleProgress}%${release}${detail}`;
+            phaseIcon = '<span class="adl-spinner" style="margin-right:4px"></span>';
         } else if (batch.phase === 'downloading') {
             phaseText = `${batch.completed}/${total} tracks`;
             if (batch.active > 0) phaseIcon = '<span class="adl-spinner" style="margin-right:4px"></span>';

webui/static/style.css

@@ -58212,6 +58212,9 @@ body.reduce-effects *::after {
     font-size: 0.7rem;
     color: rgba(255, 255, 255, 0.4);
     margin-top: 2px;
+    white-space: nowrap;
+    overflow: hidden;
+    text-overflow: ellipsis;
 }
 
 .adl-batch-card-source {