refactor(queue): extract shared payload, history, and polling helpers

- _buildDownloadRequestPayload/_historyItemFromResult shared by the
  inline and native-worker paths; _buildOutputDir now derives from
  _buildRelativeOutputDir
- shouldAbortWork extended (deleteFileOnAbort, null-item check) and
  used at all ten guard sites; _purgeAlbumRgEntry replaces six copies
- ProgressStreamPoller shared by download queue and local library;
  openVerificationAndAwaitGrant shared by queue and track provider

lib/utils/extension_auth_launcher.dart

@@ -97,6 +97,65 @@ Timer? scheduleExtensionVerificationHelpDialog(
   });
 }
 
+/// Opens a pending extension verification challenge and waits (up to 5
+/// minutes) for its grant result, returning whether it succeeded.
+///
+/// [awaitForeground], if given, is awaited first — passed the trimmed
+/// [extensionId] — before opening the challenge; used by callers that must
+/// defer verification until the app is foregrounded (launching a browser
+/// from the background is blocked by the OS).
+Future<bool> openVerificationAndAwaitGrant(
+  String extensionId, {
+  required String browserMode,
+  Future<void> Function(String extensionId)? awaitForeground,
+}) async {
+  final normalizedExtensionId = extensionId.trim();
+  if (normalizedExtensionId.isEmpty) return false;
+
+  if (awaitForeground != null) {
+    await awaitForeground(normalizedExtensionId);
+  }
+
+  final grantCompleter = Completer<ExtensionSessionGrantEvent>();
+  late final StreamSubscription<ExtensionSessionGrantEvent> grantSub;
+  grantSub = PlatformBridge.extensionSessionGrantEvents()
+      .where((event) => event.extensionId.trim() == normalizedExtensionId)
+      .listen((event) {
+        if (!grantCompleter.isCompleted) {
+          grantCompleter.complete(event);
+        }
+      });
+
+  Uri? authUri;
+  Timer? helpDialogTimer;
+
+  try {
+    final opened = await openPendingExtensionVerification(
+      normalizedExtensionId,
+      browserMode: browserMode,
+      onAuthUri: (uri) => authUri = uri,
+    );
+    if (!opened) return false;
+
+    helpDialogTimer = scheduleExtensionVerificationHelpDialog(
+      normalizedExtensionId,
+      authUri,
+      browserMode: browserMode,
+    );
+
+    final event = await grantCompleter.future.timeout(
+      const Duration(minutes: 5),
+    );
+    return event.success;
+  } on TimeoutException {
+    _log.w('Timed out waiting for verification grant: $normalizedExtensionId');
+    return false;
+  } finally {
+    helpDialogTimer?.cancel();
+    await grantSub.cancel();
+  }
+}
+
 Future<bool> showExtensionVerificationHelpDialog(
   String extensionId,
   Uri authUri, {

lib/utils/progress_stream_poller.dart

@@ -0,0 +1,165 @@
+import 'dart:async';
+
+/// Shared scaffold for consuming a native progress stream (EventChannel-backed)
+/// with a bootstrap-timeout fallback to polling, an in-flight guard against
+/// re-entrant processing, and an error-count throttle that suppresses log
+/// spam after repeated consecutive failures.
+///
+/// Callers own all logging text (via the `on*Error`/`onStreamTimeout`/
+/// `onStreamFailed` callbacks) and platform gating (via [start]'s [useStream]
+/// argument), so behavior stays identical to each call site's prior bespoke
+/// implementation.
+class ProgressStreamPoller<T> {
+  ProgressStreamPoller({
+    required this.streamProvider,
+    required this.pollProvider,
+    required this.onProgress,
+    required this.pollingInterval,
+    required this.bootstrapTimeout,
+    required this.onStreamProcessingError,
+    required this.onStreamFailed,
+    required this.onStreamTimeout,
+    required this.onPollError,
+    this.shouldPollTick,
+  });
+
+  final Stream<T> Function() streamProvider;
+  final Future<T> Function() pollProvider;
+  final Future<void> Function(T progress) onProgress;
+  final Duration pollingInterval;
+  final Duration bootstrapTimeout;
+
+  /// Optional gate checked on every periodic poll tick (stream events are
+  /// never gated), after the in-flight guard; returning false skips that
+  /// tick without fetching or affecting the error counter. Used for
+  /// idle-cadence throttling.
+  final bool Function()? shouldPollTick;
+
+  /// Called when a stream event's [onProgress] throws, only while the error
+  /// count is within the log-spam threshold.
+  final void Function(Object error) onStreamProcessingError;
+
+  /// Called when the stream itself errors, before falling back to polling.
+  final void Function(Object error) onStreamFailed;
+
+  /// Called when no stream event arrived before [bootstrapTimeout].
+  final void Function() onStreamTimeout;
+
+  /// Called when a poll tick's [onProgress] throws, only while the error
+  /// count is within the log-spam threshold.
+  final void Function(Object error) onPollError;
+
+  static const _errorLogThreshold = 3;
+
+  Timer? _timer;
+  Timer? _bootstrapTimer;
+  StreamSubscription<T>? _sub;
+  bool _hasReceivedStreamEvent = false;
+  bool _usingStream = false;
+  bool _inFlight = false;
+  int _errorCount = 0;
+
+  bool get usingStream => _usingStream;
+
+  /// (Re)starts progress consumption. When [useStream] is true, attaches the
+  /// stream (falling back to polling on timeout/error); otherwise starts
+  /// polling immediately.
+  void start({required bool useStream}) {
+    _timer?.cancel();
+    _bootstrapTimer?.cancel();
+    _bootstrapTimer = null;
+    _sub?.cancel();
+    _sub = null;
+    _hasReceivedStreamEvent = false;
+    _usingStream = false;
+
+    if (useStream) {
+      _attachStream();
+      return;
+    }
+    startPollingTimer();
+  }
+
+  void _attachStream() {
+    _sub = streamProvider().listen(
+      (progress) async {
+        _hasReceivedStreamEvent = true;
+        _usingStream = true;
+        _bootstrapTimer?.cancel();
+        _bootstrapTimer = null;
+        await _runGuarded(() async => progress, onStreamProcessingError);
+      },
+      onError: (Object error, StackTrace stackTrace) {
+        if (_usingStream) {
+          onStreamFailed(error);
+        }
+        _sub?.cancel();
+        _sub = null;
+        _usingStream = false;
+        _bootstrapTimer?.cancel();
+        _bootstrapTimer = null;
+        startPollingTimer();
+      },
+      cancelOnError: false,
+    );
+
+    _bootstrapTimer = Timer(bootstrapTimeout, () {
+      if (_hasReceivedStreamEvent) return;
+      onStreamTimeout();
+      _sub?.cancel();
+      _sub = null;
+      _usingStream = false;
+      startPollingTimer();
+    });
+  }
+
+  /// (Re)starts the fallback polling timer.
+  void startPollingTimer() {
+    _timer?.cancel();
+    _timer = Timer.periodic(pollingInterval, (_) async {
+      await _runGuarded(pollProvider, onPollError, gate: shouldPollTick);
+    });
+  }
+
+  /// One-shot immediate fetch reusing the same in-flight guard and error
+  /// counter as the periodic poll, with its own error callback.
+  Future<void> pollOnce(void Function(Object error) onError) =>
+      _runGuarded(pollProvider, onError);
+
+  Future<void> _runGuarded(
+    Future<T> Function() fetch,
+    void Function(Object error) onError, {
+    bool Function()? gate,
+  }) async {
+    if (_inFlight) return;
+    _inFlight = true;
+    try {
+      if (gate != null && !gate()) return;
+      final progress = await fetch();
+      await onProgress(progress);
+      _errorCount = 0;
+    } catch (e) {
+      _errorCount++;
+      if (_errorCount <= _errorLogThreshold) {
+        onError(e);
+      }
+    } finally {
+      _inFlight = false;
+    }
+  }
+
+  /// Cancels timers/subscription and resets guard/counter state, without
+  /// releasing the poller for reuse (matches each caller's `_stop*` reset).
+  void stop() {
+    _timer?.cancel();
+    _bootstrapTimer?.cancel();
+    _sub?.cancel();
+    _timer = null;
+    _bootstrapTimer = null;
+    _sub = null;
+    _errorCount = 0;
+    _inFlight = false;
+    _hasReceivedStreamEvent = false;
+    _usingStream = false;
+  }
+}