fix(downloads): profile-scoped ownership and watch-sync integrity

lib/database/app_database.dart

@@ -631,6 +631,11 @@ class AppDatabase extends _$AppDatabase {
     await (delete(syncRules)..where((t) => t.profileId.equals(profileId))).go();
   }
 
+  /// Drop every sync rule (full logout).
+  Future<void> clearAllSyncRules() async {
+    await delete(syncRules).go();
+  }
+
   /// Get all downloaded media items (for syncing watch states)
   Future<List<DownloadedMediaItem>> getAllDownloadedMetadata() {
     return (select(downloadedMedia)..where((t) => t.status.equals(DownloadStatus.completed.index))).get();

lib/database/download_operations.dart

@@ -65,12 +65,33 @@ extension DownloadDatabaseOperations on AppDatabase {
   }
 
   /// Claim pre-v17 shared download rows for [profileId]. Rows that already
-  /// have any owner are left untouched so later profiles do not inherit them.
+  /// have any valid owner are left untouched so later profiles do not
+  /// inherit them.
+  ///
+  /// Runs on every profile switch — validity context is computed once and
+  /// applied in memory instead of the per-download full-table rescan
+  /// `getDownloadOwnerCount` would do.
   Future<void> adoptLegacyDownloadsForProfile(String profileId) async {
     if (profileId.isEmpty) return;
     final rows = await select(downloadedMedia).get();
+    if (rows.isEmpty) return;
+
+    final owners = await select(downloadOwners).get();
+    final localProfileIds = (await select(profiles).get()).map((row) => row.id).toSet();
+    final connectionIds = (await select(connections).get()).map((row) => row.id).toSet();
+    bool valid(DownloadOwnerItem owner) {
+      if (localProfileIds.contains(owner.profileId)) return true;
+      final plexHome = parsePlexHomeProfileId(owner.profileId);
+      if (plexHome != null) return connectionIds.contains(plexHome.accountConnectionId);
+      return localProfileIds.isEmpty;
+    }
+
+    final ownedKeys = <String>{
+      for (final owner in owners)
+        if (valid(owner)) owner.globalKey,
+    };
     for (final row in rows) {
-      if (await getDownloadOwnerCount(row.globalKey) == 0) {
+      if (!ownedKeys.contains(row.globalKey)) {
         await addDownloadOwner(profileId: profileId, globalKey: row.globalKey);
       }
     }

lib/main.dart

@@ -780,7 +780,11 @@ class _MainAppState extends State<MainApp> with WidgetsBindingObserver {
             final activeProfile = context.read<ActiveProfileProvider>();
             _offlineWatchSyncService.setActiveProfileId(
               activeProfile.activeId,
-              availableProfileCount: activeProfile.profiles.length,
+              // Legacy-adoption gate: only trust the count once the provider
+              // has hydrated (locals + cached home users) — a transient
+              // count of 1 mid-load would permanently mis-adopt pre-profile
+              // watch actions.
+              availableProfileCount: activeProfile.isInitialized ? activeProfile.profiles.length : null,
             );
 
             // Offline-sync drain replays a batch of queued watch actions without
@@ -818,7 +822,10 @@ class _MainAppState extends State<MainApp> with WidgetsBindingObserver {
           },
           update: (_, activeProfile, previous) {
             final provider = previous ?? _offlineWatchSyncService;
-            provider.setActiveProfileId(activeProfile.activeId, availableProfileCount: activeProfile.profiles.length);
+            provider.setActiveProfileId(
+              activeProfile.activeId,
+              availableProfileCount: activeProfile.isInitialized ? activeProfile.profiles.length : null,
+            );
             return provider;
           },
         ),

lib/providers/download_provider.dart

@@ -165,10 +165,13 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
 
   bool _ownsProgressEntry(MapEntry<String, DownloadProgress> entry) => _ownsDownloadKey(entry.key);
 
-  Future<bool> _claimDownloadForActiveProfile(String globalKey) async {
-    final profileId = _requireActiveProfileId();
-    if (_ownedDownloadKeys.contains(globalKey)) return false;
+  /// Claim [globalKey] for an explicit [profileId] — sync rules claim for
+  /// the RULE'S owner, not whoever is active when the pass lands, so a
+  /// mid-run profile switch can't leak ownership across profiles.
+  Future<bool> _claimDownloadForProfile(String globalKey, String profileId) async {
+    if (_activeProfileId == profileId && _ownedDownloadKeys.contains(globalKey)) return false;
     await _database.addDownloadOwner(profileId: profileId, globalKey: globalKey);
+    // _ownedDownloadKeys mirrors only the active profile's rows.
     if (_activeProfileId != profileId) return false;
     _ownedDownloadKeys.add(globalKey);
     return true;
@@ -469,15 +472,24 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
     // (`_loadPersistedDownloads` rehydrates `_metadata` from the cache).
     if (shouldPersistToCache) {
       unawaited(
-        ApiCache.forBackend(base.backend)
-            .applyWatchState(
-              serverId: ServerId(event.cacheServerId ?? event.serverId),
-              itemId: event.itemId,
-              isWatched: isWatched,
-            )
-            .catchError((Object e) {
-              appLogger.w('Failed to apply watch state to cache for $globalKey', error: e);
-            }),
+        () async {
+          // Jellyfin cache rows are per-user (cacheServerId embeds the user);
+          // Plex rows are keyed by server only, so persisting one user's flip
+          // into a download SHARED with another profile would surface as that
+          // profile's watch state too. Skip the shared case — each profile's
+          // own queued watch actions still re-apply its state on reload.
+          if (base.backend == MediaBackend.plex &&
+              await _database.hasDownloadOwner(globalKey, excludingProfileId: _activeProfileId)) {
+            return;
+          }
+          await ApiCache.forBackend(base.backend).applyWatchState(
+            serverId: ServerId(event.cacheServerId ?? event.serverId),
+            itemId: event.itemId,
+            isWatched: isWatched,
+          );
+        }().catchError((Object e) {
+          appLogger.w('Failed to apply watch state to cache for $globalKey', error: e);
+        }),
       );
     }
     safeNotifyListeners();
@@ -968,21 +980,22 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
     int mediaIndex = 0,
     DownloadVersionConfig? versionConfig,
     _RelatedMetadataDownloadContext? relatedContext,
+    String? claimForProfileId,
   }) async {
     if (!_downloadManager.downloadsSupported) return false;
 
-    _requireActiveProfileId();
+    final ownerProfileId = claimForProfileId ?? _requireActiveProfileId();
     final globalKey = metadata.globalKey;
 
     // Don't duplicate the physical download. If another profile already owns
-    // the shared row, claiming it makes it visible for the active profile.
+    // the shared row, claiming it makes it visible for the owning profile.
     if (_downloads.containsKey(globalKey)) {
       final existing = _downloads[globalKey]!;
       if (existing.status == DownloadStatus.downloading ||
           existing.status == DownloadStatus.completed ||
           existing.status == DownloadStatus.queued ||
           existing.status == DownloadStatus.paused) {
-        final claimed = await _claimDownloadForActiveProfile(globalKey);
+        final claimed = await _claimDownloadForProfile(globalKey, ownerProfileId);
         if (claimed) safeNotifyListeners();
         return claimed;
       }
@@ -1044,7 +1057,7 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
     // Store full metadata for display
     _metadata[globalKey] = metadataToStore;
 
-    await _claimDownloadForActiveProfile(globalKey);
+    await _claimDownloadForProfile(globalKey, ownerProfileId);
 
     // Update local state immediately for UI feedback
     _downloads[globalKey] = DownloadProgress(globalKey: globalKey, status: DownloadStatus.queued);
@@ -1681,8 +1694,19 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
       serverManager: serverManager,
       downloads: downloads,
       metadata: Map.unmodifiable(_metadata),
-      queueSingleDownload: (episode, client, {int mediaIndex = 0}) =>
-          _queueSingleDownload(episode, client, mediaIndex: mediaIndex, relatedContext: relatedContext),
+      queueSingleDownload: (episode, client, {int mediaIndex = 0}) async {
+        // A profile switch mid-pass must not keep queueing the old
+        // profile's rules; whatever does get queued is claimed for the
+        // rule's owner, never the new active profile.
+        if (_activeProfileId != profileId) return false;
+        return _queueSingleDownload(
+          episode,
+          client,
+          mediaIndex: mediaIndex,
+          relatedContext: relatedContext,
+          claimForProfileId: profileId,
+        );
+      },
       isOffline: _offlineSource?.isOffline ?? false,
       force: force,
     );
@@ -1709,8 +1733,16 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
       serverManager: serverManager,
       downloads: downloads,
       metadata: Map.unmodifiable(_metadata),
-      queueSingleDownload: (episode, client, {int mediaIndex = 0}) =>
-          _queueSingleDownload(episode, client, mediaIndex: mediaIndex, relatedContext: relatedContext),
+      queueSingleDownload: (episode, client, {int mediaIndex = 0}) async {
+        if (_activeProfileId != profileId) return false;
+        return _queueSingleDownload(
+          episode,
+          client,
+          mediaIndex: mediaIndex,
+          relatedContext: relatedContext,
+          claimForProfileId: profileId,
+        );
+      },
       isOffline: _offlineSource?.isOffline ?? false,
     );
   }

lib/screens/profile/profile_teardown.dart

@@ -240,6 +240,11 @@ Future<void> logoutAllProfiles(BuildContext context) async {
   await scope.storage.clearActiveProfileId();
   await scope.storage.clearAllProfileLastUsed();
   await scope.storage.clearAllUserScopedPreferences();
+  // Queued watch actions and sync rules are keyed by the profiles that just
+  // ceased to exist; left behind they'd strand forever (or worse, replay
+  // through the next sign-in's clients).
+  await scope.database.clearAllWatchActions();
+  await scope.database.clearAllSyncRules();
   // The API cache is app-global and Plex rows are keyed by server only, so
   // a later sign-in as a different user must not inherit them.
   await ApiCache.instance.clearVolatile();

lib/services/offline_watch_sync_service.dart

@@ -372,9 +372,14 @@ class OfflineWatchSyncService extends ChangeNotifier {
     try {
       await _adoptLegacyWatchActionsForActiveProfile();
       final profileId = _activeProfileId;
-      final pendingActions = profileId == null || profileId.isEmpty
-          ? await _database.getPendingWatchActions()
-          : await _database.getPendingWatchActions(profileId: profileId);
+      if (profileId == null || profileId.isEmpty) {
+        // No active profile: dropping the filter would replay EVERY
+        // profile's queued actions through whatever clients happen to be
+        // bound — the wrong user's account. Actions stay queued.
+        appLogger.d('No active profile — deferring pending watch sync');
+        return;
+      }
+      final pendingActions = await _database.getPendingWatchActions(profileId: profileId);
 
       if (pendingActions.isEmpty) {
         appLogger.d('No pending watch actions to sync');
@@ -384,6 +389,14 @@ class OfflineWatchSyncService extends ChangeNotifier {
       appLogger.i('Syncing ${pendingActions.length} pending watch actions');
 
       for (final action in pendingActions) {
+        if (_activeProfileId != profileId) {
+          // A profile switch mid-loop rebinds server clients to the NEW
+          // user's tokens under the same server ids — replaying the rest
+          // would write this profile's watch history to another account.
+          // Remaining actions stay queued for the next sync.
+          appLogger.i('Active profile changed mid-sync — requeueing remaining watch actions');
+          return;
+        }
         if (action.syncAttempts >= maxSyncAttempts) {
           appLogger.w(
             'Skipping action ${action.id} - exceeded retry limit '

lib/services/sync_rule_executor.dart

@@ -39,7 +39,10 @@ class SyncRuleResult {
 class SyncRuleExecutor {
   final AppDatabase _database;
   bool _isExecuting = false;
-  DateTime? _lastFullRunAt;
+
+  /// Per profile: one profile's background pass must not consume another
+  /// profile's cooldown window after a switch.
+  final Map<String, DateTime> _lastFullRunAtByProfile = {};
 
   static const Duration _cooldownWifi = Duration(minutes: 30);
   static const Duration _cooldownCellular = Duration(hours: 3);
@@ -85,11 +88,12 @@ class SyncRuleExecutor {
       return [];
     }
 
-    if (!force && _lastFullRunAt != null) {
+    final lastFullRunAt = _lastFullRunAtByProfile[profileId];
+    if (!force && lastFullRunAt != null) {
       final hasWifi =
           connectivity.contains(ConnectivityResult.wifi) || connectivity.contains(ConnectivityResult.ethernet);
       final cooldown = hasWifi ? _cooldownWifi : _cooldownCellular;
-      final elapsed = DateTime.now().difference(_lastFullRunAt!);
+      final elapsed = DateTime.now().difference(lastFullRunAt);
       if (elapsed < cooldown) {
         appLogger.d(
           'Sync rules cooldown active (${elapsed.inMinutes}m < ${cooldown.inMinutes}m, hasWifi=$hasWifi) — skipping',
@@ -124,7 +128,7 @@ class SyncRuleExecutor {
         }
       }
 
-      _lastFullRunAt = DateTime.now();
+      _lastFullRunAtByProfile[profileId] = DateTime.now();
       return results;
     } finally {
       _isExecuting = false;

test/services/offline_watch_sync_service_test.dart

@@ -346,6 +346,7 @@ void main() {
         mgr.dispose();
         await db.close();
       });
+      svc.setActiveProfileId('p1');
 
       final client = _RecordingMediaClient(serverId: ServerId('srv'), backend: MediaBackend.plex);
       mgr.debugRegisterClientForTesting(client);
@@ -371,6 +372,7 @@ void main() {
         mgr.dispose();
         await db.close();
       });
+      svc.setActiveProfileId('p1');
 
       final client = _RecordingMediaClient(serverId: ServerId('srv'), backend: MediaBackend.plex);
       mgr.debugRegisterClientForTesting(client);
@@ -395,6 +397,7 @@ void main() {
         mgr.dispose();
         await db.close();
       });
+      svc.setActiveProfileId('p1');
 
       final client = _RecordingMediaClient(serverId: ServerId('srv'), backend: MediaBackend.plex);
       mgr.debugRegisterClientForTesting(client);
@@ -421,6 +424,7 @@ void main() {
         mgr.dispose();
         await db.close();
       });
+      svc.setActiveProfileId('p1');
 
       final client = _RecordingMediaClient(serverId: ServerId('srv'), backend: MediaBackend.jellyfin);
       mgr.debugRegisterClientForTesting(client);
@@ -443,6 +447,69 @@ void main() {
   // of getLocalWatchStatus / getLocalViewOffset).
   // ============================================================
 
+  group('syncPendingItems profile scoping', () {
+    test('defers entirely when no profile is active', () async {
+      final (svc: svc, db: db, mgr: mgr) = _makeService();
+      addTearDown(() async {
+        svc.dispose();
+        mgr.dispose();
+        await db.close();
+      });
+
+      svc.setActiveProfileId('p1');
+      final client = _RecordingMediaClient(serverId: ServerId('srv'), backend: MediaBackend.plex);
+      mgr.debugRegisterClientForTesting(client);
+      await svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '42');
+
+      // Active profile cleared (sign-out/teardown window): replaying the
+      // queue through whatever clients are bound would hit the wrong user.
+      svc.setActiveProfileId(null);
+      await svc.syncPendingItems();
+
+      expect(client.watched, isEmpty);
+      expect(await db.getPendingSyncCount(), 1);
+    });
+
+    test('requeues remaining actions when the active profile changes mid-sync', () async {
+      final (svc: svc, db: db, mgr: mgr) = _makeService();
+      addTearDown(() async {
+        svc.dispose();
+        mgr.dispose();
+        await db.close();
+      });
+
+      svc.setActiveProfileId('p1');
+      final posts = <String>[];
+      final client = JellyfinClient.forTesting(
+        connection: _jellyfinConnection('user-a'),
+        httpClient: MockClient((request) async {
+          if (request.method == 'GET' && request.url.path.startsWith('/Users/user-a/Items/')) {
+            final id = request.url.pathSegments.last;
+            return http.Response('{"Id":"$id","Type":"Movie","Name":"Movie $id"}', 200);
+          }
+          if (request.method == 'POST' && request.url.path.startsWith('/UserPlayedItems/')) {
+            posts.add(request.url.path);
+            // The switch lands while action 1 is mid-flight — the binder
+            // would now be rebinding this server id to another user.
+            svc.setActiveProfileId('p2');
+            return http.Response('', 204);
+          }
+          return http.Response('not found', 404);
+        }),
+      );
+      addTearDown(client.close);
+      mgr.debugRegisterJellyfinClientForTesting(client);
+
+      await svc.queueMarkWatched(serverId: ServerId('jf-machine'), itemId: 'item-1');
+      await svc.queueMarkWatched(serverId: ServerId('jf-machine'), itemId: 'item-2');
+
+      await svc.syncPendingItems();
+
+      expect(posts, hasLength(1));
+      expect(await db.getPendingSyncCount(), 1);
+    });
+  });
+
   group('queueProgressUpdate', () {
     test('persists a progress row with shouldMarkWatched=false below threshold', () async {
       final (svc: svc, db: db, mgr: mgr) = _makeService();
@@ -881,6 +948,7 @@ void main() {
         mgr.dispose();
         await db.close();
       });
+      svc.setActiveProfileId('p1');
 
       final pathsByUser = <String, List<String>>{'user-a': [], 'user-b': []};