perf(native-worker): stop shipping the full item payload on every poll

The 1s snapshot poll re-read and re-parsed the whole worker state file
— which retains every completed item's result (history row, possibly
full synced lyrics; 1.5-5 KB each) for the entire run — then
re-serialized it across the channel and re-iterated it on the Dart UI
isolate. Late in a 1000-track batch that is megabytes parsed several
times per second, on the Android main thread.

Pollers now echo back the state_serial of the last snapshot they fully
processed; when the state has not advanced, the native side serves a
cached compact header (no items, no results, no settings) merged with
the small progress delta — steady-state polls are O(1) regardless of
batch size, and the full payload is delivered exactly once per state
transition. The channel handler also reads and parses off the main
thread now, matching its neighbours.

lib/providers/download_queue_provider.dart

@@ -3809,6 +3809,17 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
     return version == DownloadRequestPayload.nativeWorkerContractVersion;
   }
 
+  /// Serial of the last fully-processed state snapshot, echoed back to the
+  /// native side so steady-state polls skip the per-item payload. Falls back
+  /// to the previous value for snapshots without the field.
+  int _snapshotStateSerial(Map<String, dynamic> snapshot, int previous) {
+    final serial = snapshot['state_serial'];
+    if (serial is num && serial > 0) {
+      return serial.toInt();
+    }
+    return previous;
+  }
+
   bool _isNativeWorkerSnapshotForRun(
     Map<String, dynamic> snapshot,
     String runId,
@@ -4019,9 +4030,12 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
     String runId,
   ) async {
     var deadServicePolls = 0;
+    var lastStateSerial = 0;
     try {
       while (true) {
-        final snapshot = await PlatformBridge.getNativeDownloadWorkerSnapshot();
+        final snapshot = await PlatformBridge.getNativeDownloadWorkerSnapshot(
+          sinceStateSerial: lastStateSerial,
+        );
         final matchesRun = _isNativeWorkerSnapshotForRun(snapshot, runId);
         if (!matchesRun || snapshot['is_running'] == true) {
           if (await _isNativeWorkerServiceAlive()) {
@@ -4054,6 +4068,7 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
           }
         }
         if (!matchesRun) {
+          lastStateSerial = 0;
           await Future<void>.delayed(const Duration(seconds: 1));
           continue;
         }
@@ -4068,6 +4083,7 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
           reconciledIds,
           settings,
         );
+        lastStateSerial = _snapshotStateSerial(snapshot, lastStateSerial);
         if (snapshot['is_running'] != true) {
           await _clearNativeWorkerRunId(runId);
           // Items may have been requeued during reconciliation (e.g. batch
@@ -4160,9 +4176,13 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
       );
 
       final runStartWait = Stopwatch()..start();
+      var lastStateSerial = 0;
       while (true) {
-        final snapshot = await PlatformBridge.getNativeDownloadWorkerSnapshot();
+        final snapshot = await PlatformBridge.getNativeDownloadWorkerSnapshot(
+          sinceStateSerial: lastStateSerial,
+        );
         if (!_isNativeWorkerSnapshotForRun(snapshot, runId)) {
+          lastStateSerial = 0;
           if (runStartWait.elapsed > const Duration(seconds: 30)) {
             throw _NativeWorkerStartupTimeout();
           }
@@ -4175,6 +4195,7 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
           reconciledIds,
           settings,
         );
+        lastStateSerial = _snapshotStateSerial(snapshot, lastStateSerial);
         if (snapshot['is_running'] != true) {
           await _clearNativeWorkerRunId(runId);
           break;

lib/services/platform_bridge.dart

@@ -1096,10 +1096,16 @@ class PlatformBridge {
     await _channel.invokeMethod('cancelNativeDownloadWorker');
   }
 
-  static Future<Map<String, dynamic>> getNativeDownloadWorkerSnapshot() async {
-    final result = await _channel.invokeMethod(
-      'getNativeDownloadWorkerSnapshot',
-    );
+  /// [sinceStateSerial]: pass the `state_serial` of the last fully-processed
+  /// snapshot; when the native state hasn't advanced past it, the heavy
+  /// per-item payload (which grows with every completed item) is omitted and
+  /// only compact progress fields are returned.
+  static Future<Map<String, dynamic>> getNativeDownloadWorkerSnapshot({
+    int sinceStateSerial = 0,
+  }) async {
+    final result = await _channel.invokeMethod('getNativeDownloadWorkerSnapshot', {
+      'since_state_serial': sinceStateSerial,
+    });
     return _decodeMapResult(result);
   }