feat(plex): Surprise-me shows only unseen titles, capped 3 movies + 3 shows (C)

The Surprise rail drew a random 20+20 sample including already-watched titles. Its point is
to surface something to WATCH, so exclude seen titles (a watched movie, a fully-watched show
— a partially-watched or never-started show still counts as unseen) and cap at 3 movies + 3
shows (6 total), interleaved.

Verified live: /surprise returns 6 items, 3+3, none watched.

Gate: backend ruff clean, 12 discover tests (+ a watched-exclusion/cap assertion).

backend/app/routes/plex/discover.py

@@ -38,6 +38,7 @@ router = APIRouter(prefix="/api/plex", tags=["plex"])
 
 RAIL_LIMIT = 20
 GENRE_TILES = 12
+SURPRISE_PER_KIND = 3  # Surprise-me: at most this many movies + this many shows (6 total)
 
 
 def _scope_libs(db: Session, scope: str) -> tuple[list[int], list[int]]:
@@ -309,42 +310,62 @@ def discover(
 def surprise(
     scope: str = "both",
     lang: str = "en",
-    limit: int = Query(default=RAIL_LIMIT, ge=1, le=40),
+    limit: int = Query(default=2 * SURPRISE_PER_KIND, ge=1, le=40),
     f: LibraryFilters = Depends(),
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
     """A fresh random sample for the reshuffling Surprise-me rail (a random draw each call), narrowed by
-    the active filters. Movies and shows are drawn separately then interleaved so a lopsided library
-    doesn't crowd out one type."""
+    the active filters. Only titles the user has NOT seen — a watched movie / a fully-watched show is
+    excluded (the point is to surface something to watch). Capped at SURPRISE_PER_KIND movies +
+    SURPRISE_PER_KIND shows, drawn separately then interleaved so a lopsided library doesn't crowd out
+    one type."""
     lang = _norm_lang(lang)
     p = f.as_dict()
     movie_ids, show_ids = _scope_libs(db, scope)
-    movies = (
-        db.execute(
+    per = min(limit, SURPRISE_PER_KIND)
+    movies = []
+    if movie_ids:
+        # Exclude movies this user has watched.
+        watched_movie = (
+            select(PlexState.id)
+            .where(
+                PlexState.item_id == PlexItem.id,
+                PlexState.user_id == user.id,
+                PlexState.status == "watched",
+            )
+            .exists()
+        )
+        movies = db.execute(
             _movie_meta(
                 select(PlexItem).where(
-                    PlexItem.kind == "movie", PlexItem.library_id.in_(movie_ids)
+                    PlexItem.kind == "movie",
+                    PlexItem.library_id.in_(movie_ids),
+                    ~watched_movie,
                 ),
                 p, lang, f,
             )
             .order_by(func.random())
-            .limit(limit)
+            .limit(per)
         ).scalars().all()
-        if movie_ids
-        else []
-    )
-    shows = (
-        db.execute(
+    shows = []
+    if show_ids:
+        # Exclude fully-watched shows (LEFT JOIN so a never-started show — no rollup row — still counts
+        # as unseen). `total IS NULL / == 0 / watched < total` == "not fully watched".
+        prog = rollup.agg_subquery(db, user.id).subquery()
+        not_fully_watched = or_(
+            prog.c.total.is_(None), prog.c.total == 0, prog.c.watched < prog.c.total
+        )
+        shows = db.execute(
             apply_meta_filters(
-                select(PlexShow).where(PlexShow.library_id.in_(show_ids)), PlexShow, p, lang
+                select(PlexShow)
+                .outerjoin(prog, prog.c.show_id == PlexShow.id)
+                .where(PlexShow.library_id.in_(show_ids), not_fully_watched),
+                PlexShow, p, lang,
             )
             .order_by(func.random())
-            .limit(limit)
+            .limit(per)
         ).scalars().all()
-        if show_ids
-        else []
-    )
     mcards = _movie_cards(db, user.id, movies)
     scards = _show_cards(db, user.id, shows)
     # Interleave movie/show cards, then trim to `limit`.

backend/tests/test_plex_discover.py

@@ -204,6 +204,18 @@ def test_surprise_scope_and_bound(db, monkeypatch):
     assert all(c["type"] == "movie" for c in items)  # scope=movie excludes shows
 
 
+def test_surprise_excludes_watched_and_caps_per_kind(db, monkeypatch):
+    # Surprise surfaces only UNSEEN titles: the watched movie (m1) is out; the unwatched (m2), the
+    # in-progress movie (m3) and the partially-watched show (sh1) are eligible. Capped at 3 per kind.
+    user = _seed(db)
+    items = _get(db, user, monkeypatch, "/api/plex/surprise?scope=both").json()["items"]
+    ids = _ids(items)
+    assert "m1" not in ids  # watched → excluded
+    assert set(ids) <= {"m2", "m3", "sh1"}  # only unseen titles
+    assert len([c for c in items if c["type"] == "movie"]) <= 3
+    assert len([c for c in items if c["type"] == "show"]) <= 3
+
+
 def test_discover_scope_movie_excludes_shows(db, monkeypatch):
     user = _seed(db)
     d = _get(db, user, monkeypatch, "/api/plex/discover?scope=movie").json()