diff --git a/README.md b/README.md
index 5ea8faeded..7da6b3f4e7 100644
--- a/README.md
+++ b/README.md
@@ -44,6 +44,7 @@ have a file containing those on your harddrive, you can not use this project.
   - [Setup](#setup)
   - [Configuration](#configuration)
     - [Libraries](#libraries)
+    - [Plex Discover Sync](#plex-discover-sync)
     - [Per server configuration](#per-server-configuration)
     - [Logging](#logging)
   - [Commands](#commands)
@@ -405,6 +406,14 @@ Enabled 2 libraries in Plex Server:
  - 2: TV Shows
 ```
 
+### Plex Discover Sync
+
+PlexTraktSync can sync watched status for movies and TV episodes that are not in your local Plex library but are available in Plex Discover (Plex's cloud database). This allows you to maintain watched history for content you've deleted from your server or never added.
+
+To enable this feature, set `plex_online: true` in the sync section of your config. Note that this only syncs watched status from Trakt to Plex (not the reverse), and requires ["Sync My Watch State and Ratings"](https://support.plex.tv/articles/sync-watch-state-and-ratings/) to be enabled in your Plex account settings.
+
+This feature is useful for preserving watch history for old content or maintaining consistency across devices.
+
 ### Per server configuration
 
 If you want to specify your config per server you can do so inside of
diff --git a/plextraktsync/config.default.yml b/plextraktsync/config.default.yml
index 7d25f349cc..5c52e45b2a 100644
--- a/plextraktsync/config.default.yml
+++ b/plextraktsync/config.default.yml
@@ -53,6 +53,8 @@ logging:
 
 # settings for 'sync' command
 sync:
+  # Enable syncing watched status with Plex Discover (cloud items not in local library)
+  plex_online: false
   # Setting for whether ratings from one platform should have priority.
   # Valid values are trakt, plex or none. (default: plex)
   # none - No rating priority. Existing ratings are not overwritten.
diff --git a/plextraktsync/config/SyncConfig.py b/plextraktsync/config/SyncConfig.py
index 468f9bbb91..c8d03af553 100644
--- a/plextraktsync/config/SyncConfig.py
+++ b/plextraktsync/config/SyncConfig.py
@@ -101,6 +101,10 @@ def sync_watchlists(self):
             ]
         )
 
+    @property
+    def plex_online(self):
+        return self.config.get("plex_online", False)
+
     @cached_property
     def need_library_walk(self):
         return any(
diff --git a/plextraktsync/media/MediaFactory.py b/plextraktsync/media/MediaFactory.py
index 81134902c1..778932a61c 100644
--- a/plextraktsync/media/MediaFactory.py
+++ b/plextraktsync/media/MediaFactory.py
@@ -84,7 +84,10 @@ def resolve_guid(self, guid: PlexGuid, show: Media = None):
 
     def resolve_trakt(self, tm: TraktItem) -> Media:
         """Find Plex media from Trakt id using Plex Search and Discover"""
-        result = self.plex.search_online(tm.item.title, tm.type)
+        title = tm.item.title
+        if hasattr(tm.item, "year") and tm.item.year:
+            title = f"{title} {tm.item.year}"
+        result = self.plex.search_online(title, tm.type)
         pm = self._guid_match(result, tm)
         return self.make_media(pm, tm.item)
 
diff --git a/plextraktsync/sync/Sync.py b/plextraktsync/sync/Sync.py
index 9c36b0c58a..2a18d87675 100644
--- a/plextraktsync/sync/Sync.py
+++ b/plextraktsync/sync/Sync.py
@@ -5,6 +5,7 @@
 
 from plextraktsync.factory import logging
 from plextraktsync.trakt.TraktUserListCollection import TraktUserListCollection
+from plextraktsync.trakt.TraktWatchedCollection import TraktWatchedCollection
 
 if TYPE_CHECKING:
     from plextraktsync.config.SyncConfig import SyncConfig
@@ -52,4 +53,40 @@ async def sync(self, walker: Walker, dry_run=False):
             async for episode in walker.find_episodes():
                 await pm.ahook.walk_episode(episode=episode, dry_run=dry_run)
 
+        if self.config.plex_online:
+            await self.sync_online(dry_run)
+
         await pm.ahook.fini(walker=walker, dry_run=dry_run)
+
+    async def sync_online(self, dry_run: bool):
+        """
+        Sync watched status from Trakt to Plex Discover (cloud items)
+        """
+        logger = logging.getLogger(__name__)
+        logger.info("Syncing watched status with Plex Discover")
+        watched_collection = TraktWatchedCollection(self.trakt)
+
+        for media_type in ["movies", "episodes"]:
+            try:
+                watched_items = watched_collection[media_type]
+            except Exception as e:
+                logger.error(f"Failed to fetch watched {media_type} from Trakt: {e}")
+                continue
+            logger.info(f"Processing {len(watched_items)} watched {media_type}")
+
+            trakt_items = watched_items.values()
+            async for m in self.walker.media_from_traktlist(trakt_items):
+                if not m.plex or not m.plex.is_discover:
+                    if not m.plex:
+                        logger.warning(f"Could not resolve {m.trakt_item} in Plex Discover")
+                    continue  # Skip if not in discover
+
+                # Sync watched status from Trakt to Plex
+                if not m.watched_on_plex and m.watched_on_trakt:
+                    logger.info(f"Marking {m} as watched on Plex")
+                    if not dry_run:
+                        try:
+                            m.mark_watched_plex()
+                        except Exception as e:
+                            logger.error(f"Failed to mark {m} as watched on Plex: {e}")
+                            # Continue processing other items
diff --git a/plextraktsync/trakt/TraktWatchedCollection.py b/plextraktsync/trakt/TraktWatchedCollection.py
new file mode 100644
index 0000000000..0fc4c5f068
--- /dev/null
+++ b/plextraktsync/trakt/TraktWatchedCollection.py
@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from plextraktsync.decorators.flatten import flatten_dict
+
+if TYPE_CHECKING:
+    from plextraktsync.trakt.TraktApi import TraktApi
+
+
+class TraktWatchedCollection(dict):
+    """
+    Lazy-loaded mapping:
+    ["movies", "episodes"] => {trakt_id: TraktMedia}
+    (values are the raw pytrakt Movie/TVEpisode objects returned by the API).
+    """
+
+    def __init__(self, trakt: TraktApi):
+        super().__init__()
+        self.trakt = trakt
+
+    def __missing__(self, media_type: str):
+        self[media_type] = items = self.watched_items(media_type)
+
+        return items
+
+    @flatten_dict
+    def watched_items(self, media_type: str):
+        if media_type == "movies":
+            for movie in self.trakt.me.watched_movies:
+                yield movie.trakt, movie
+        elif media_type == "episodes":
+            for episode in self.trakt.me.watched_episodes:
+                yield episode.trakt, episode
+        else:
+            raise ValueError(f"Unsupported media type: {media_type}")
diff --git a/tests/conftest.py b/tests/conftest.py
index 6179cef285..aea3c3ed5c 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -6,6 +6,7 @@
 from os.path import dirname
 from os.path import join as join_path
 
+import pytest
 from trakt.tv import TVShow
 
 from plextraktsync.factory import Factory
@@ -35,3 +36,8 @@ def make(cls=None, **kwargs) -> TVShow:
     cls = cls if cls is not None else "object"
     # https://stackoverflow.com/a/2827726/2314626
     return type(cls, (object,), kwargs)
+
+
+@pytest.fixture
+def trakt_api():
+    return factory.trakt_api
diff --git a/tests/test_trakt_watched_collection.py b/tests/test_trakt_watched_collection.py
new file mode 100644
index 0000000000..9e877dbfeb
--- /dev/null
+++ b/tests/test_trakt_watched_collection.py
@@ -0,0 +1,114 @@
+#!/usr/bin/env python3 -m pytest
+"""Unit tests for TraktWatchedCollection with mocked Trakt API responses.
+
+These tests use MagicMock to simulate Trakt API responses without making
+real API calls. They run in CI and provide fast, reliable test coverage.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from plextraktsync.trakt.TraktWatchedCollection import TraktWatchedCollection
+
+
+def test_trakt_watched_collection_movies_with_sample_data():
+    """Test movies are correctly indexed by trakt ID using sample Trakt API data."""
+    # Mock Trakt API with real-world sample watched movies
+    # Sample structure: Trakt API returns watched movies with trakt ID and metadata
+    mock_trakt = MagicMock()
+
+    # Create mock movies using real-world-like structure
+    mock_movie1 = MagicMock()
+    mock_movie1.trakt = 278  # The Shawshank Redemption
+    mock_movie1.title = "The Shawshank Redemption"
+    mock_movie1.year = 1994
+
+    mock_movie2 = MagicMock()
+    mock_movie2.trakt = 278945  # Inception
+    mock_movie2.title = "Inception"
+    mock_movie2.year = 2010
+
+    # Configure mock API to return these movies
+    mock_trakt.me.watched_movies = [mock_movie1, mock_movie2]
+
+    collection = TraktWatchedCollection(mock_trakt)
+    movies = collection["movies"]
+
+    # Verify structure and content match expected behavior
+    assert isinstance(movies, dict)
+    assert len(movies) == 2
+    assert 278 in movies
+    assert 278945 in movies
+    assert movies[278] is mock_movie1
+    assert movies[278945] is mock_movie2
+
+
+def test_trakt_watched_collection_episodes_with_sample_data():
+    """Test episodes are correctly indexed by trakt ID using sample Trakt API data."""
+    # Mock Trakt API with real-world sample watched episodes
+    mock_trakt = MagicMock()
+
+    # Create mock episodes using real-world-like structure
+    # Episodes typically have: trakt ID, show info, season, number, etc.
+    mock_episode1 = MagicMock()
+    mock_episode1.trakt = 73641  # Game of Thrones S01E01
+    mock_episode1.show = MagicMock()
+    mock_episode1.show.title = "Game of Thrones"
+    mock_episode1.season = 1
+    mock_episode1.number = 1
+
+    mock_episode2 = MagicMock()
+    mock_episode2.trakt = 73642  # Game of Thrones S01E02
+    mock_episode2.show = MagicMock()
+    mock_episode2.show.title = "Game of Thrones"
+    mock_episode2.season = 1
+    mock_episode2.number = 2
+
+    # Configure mock API to return these episodes
+    mock_trakt.me.watched_episodes = [mock_episode1, mock_episode2]
+
+    collection = TraktWatchedCollection(mock_trakt)
+    episodes = collection["episodes"]
+
+    # Verify structure and content match expected behavior
+    assert isinstance(episodes, dict)
+    assert len(episodes) == 2
+    assert 73641 in episodes
+    assert 73642 in episodes
+    assert episodes[73641] is mock_episode1
+    assert episodes[73642] is mock_episode2
+
+
+def test_trakt_watched_collection_empty():
+    """Test handling of empty watched collections."""
+    # Mock Trakt API with no watched items
+    mock_trakt = MagicMock()
+    mock_trakt.me.watched_movies = []
+    mock_trakt.me.watched_episodes = []
+
+    collection = TraktWatchedCollection(mock_trakt)
+    movies = collection["movies"]
+    episodes = collection["episodes"]
+
+    # Verify empty dicts are returned correctly
+    assert isinstance(movies, dict)
+    assert len(movies) == 0
+    assert isinstance(episodes, dict)
+    assert len(episodes) == 0
+
+
+def test_trakt_watched_collection_invalid_media_type():
+    """Test that unsupported media types raise appropriate errors."""
+    mock_trakt = MagicMock()
+    collection = TraktWatchedCollection(mock_trakt)
+
+    # Verify only "movies" and "episodes" are supported
+    with pytest.raises(ValueError, match="Unsupported media type: invalid"):
+        _ = collection["invalid"]
+
+    # Verify other unsupported types also fail
+    with pytest.raises(ValueError, match="Unsupported media type: shows"):
+        _ = collection["shows"]
diff --git a/tests/test_trakt_watched_collection_integration.py b/tests/test_trakt_watched_collection_integration.py
new file mode 100644
index 0000000000..821c3ba09d
--- /dev/null
+++ b/tests/test_trakt_watched_collection_integration.py
@@ -0,0 +1,31 @@
+#!/usr/bin/env python3 -m pytest
+"""Integration tests for TraktWatchedCollection using real Trakt API.
+
+These tests validate TraktWatchedCollection against real Trakt API responses.
+They are skipped in CI to avoid dependency on external API availability.
+Run locally with: pytest tests/test_trakt_watched_collection_integration.py -v
+"""
+
+from __future__ import annotations
+
+import os
+
+import pytest
+
+from plextraktsync.trakt.TraktWatchedCollection import TraktWatchedCollection
+
+
+@pytest.mark.skipif(os.environ.get("CI"), reason="Requires real Trakt API")
+def test_trakt_watched_collection_movies(trakt_api):
+    """Validate that movies are correctly loaded from real Trakt API."""
+    collection = TraktWatchedCollection(trakt_api)
+    movies = collection["movies"]
+    assert isinstance(movies, dict)
+
+
+@pytest.mark.skipif(os.environ.get("CI"), reason="Requires real Trakt API")
+def test_trakt_watched_collection_episodes(trakt_api):
+    """Validate that episodes are correctly loaded from real Trakt API."""
+    collection = TraktWatchedCollection(trakt_api)
+    episodes = collection["episodes"]
+    assert isinstance(episodes, dict)