fix(plex): Restart only where an episode has a resume point; the Surprise draw is re-read, not redrawn

Show cards carry resumable (ShowState, from the same rollup) so a part-watched
show whose next episode is untouched offers Play, not a Continue and a Restart
that do the same thing; plexStarted is the one rule for cards and menus. The
Surprise rail's key sits under the discover root: whatever refreshes the rails
re-reads the draw (keep=) with its watch state now, kept past the 5-minute gc,
and only the reshuffle deals. Closing the player refreshes the Discover rails.
A show card's play acts on the last click only, and only where it was made.

backend/app/discover/pins.py

@@ -24,7 +24,7 @@ from sqlalchemy.orm import Session
 from app.discover import cards, queries
 from app.models import Channel, ComfortPin, PlexItem, PlexShow, Playlist, User, Video
 from app.plex import rollup
-from app.plex.cards import leaf_card, show_card
+from app.plex.cards import SHOW_NEW, leaf_card, show_card
 from app.plex.lookups import by_rating_key
 
 # kind -> the ComfortPin column that holds it. The client's vocabulary and the schema's, in one place.
@@ -203,8 +203,8 @@ def _shelf_card(db: Session, user: User, pin: ComfortPin, batch: _Batch, lang: s
         sh = db.get(PlexShow, pin.plex_show_id)
         if sh is None:
             return None
-        status = rollup.state_map(db, user.id, [sh.id]).get(sh.id, "new")
-        card = cards.from_plex(show_card(sh, status, lang=lang))
+        state = rollup.state_map(db, user.id, [sh.id]).get(sh.id, SHOW_NEW)
+        card = cards.from_plex(show_card(sh, state, lang=lang))
         ep = queries.next_episode(db, user.id, sh.id, lang=lang)
         nxt = cards.from_plex(ep) if ep else None
 

backend/app/discover/queries.py

@@ -40,7 +40,7 @@ from app.models import (
     VideoState,
 )
 from app.plex import rollup
-from app.plex.cards import display_title, leaf_card, movie_card, show_card
+from app.plex.cards import ShowState, display_title, leaf_card, movie_card, show_card
 from app.plex.query import apply_show_state_filter, content_libs
 
 # How many rows a single rail may carry — Home's rails and the Plex module's Discover rails alike
@@ -256,16 +256,19 @@ def plex_in_progress(
     if show_libs:
         prog = rollup.agg_subquery(db, uid, include_resumed=True).subquery()
         sq = apply_show_state_filter(
-            select(PlexShow, prog.c.resumed_at).join(prog, prog.c.show_id == PlexShow.id),
+            select(PlexShow, prog.c.resumed_at, prog.c.inprog).join(
+                prog, prog.c.show_id == PlexShow.id
+            ),
             prog.c.total,
             prog.c.watched,
             prog.c.inprog,
             "in_progress",
         ).where(PlexShow.library_id.in_(show_libs))
-        for sh, resumed_at in db.execute(
+        for sh, resumed_at, inprog in db.execute(
             sq.order_by(prog.c.resumed_at.desc().nullslast()).limit(limit)
         ).all():
-            out.append((resumed_at, show_card(sh, "in_progress", lang=lang)))
+            state = ShowState("in_progress", (inprog or 0) > 0)
+            out.append((resumed_at, show_card(sh, state, lang=lang)))
 
     return out
 

backend/app/plex/cards.py

@@ -3,9 +3,24 @@ playlists, and detail views render. Extracted from routes/plex/browse.py so the
 lives beside the other Plex read-model modules (detail.py, query.py) rather than inside the
 router. Pure functions — no DB, no request state."""
 
+from typing import NamedTuple
+
 from app.models import PlexItem, PlexShow, PlexState
 
 
+class ShowState(NamedTuple):
+    """A show's watch state rolled up from its episodes, as its card carries it. `resumable`: an unwatched
+    episode has a resume point, the one the card's Continue resumes and its Restart replays from 0. A
+    show part watched with its next episode untouched has none: Continue and Restart would both play
+    that episode from its start, so the card offers Play alone."""
+
+    status: str
+    resumable: bool
+
+
+SHOW_NEW = ShowState("new", False)
+
+
 # A card's wide backdrop, for a rail's landscape tile or a crossfading hero — but ONLY when the item
 # actually has art. `thumb` is always emitted (its endpoint placeholders a missing poster), but an
 # `art` URL for an item with no `art_key` would resolve to a broken/blank image on every card that
@@ -54,7 +69,7 @@ def movie_card(it: PlexItem, st: PlexState | None, *, lang: str) -> dict:
     }
 
 
-def show_card(sh: PlexShow, status: str = "new", *, lang: str) -> dict:
+def show_card(sh: PlexShow, state: ShowState = SHOW_NEW, *, lang: str) -> dict:
     return {
         "id": sh.rating_key,
         "type": "show",
@@ -66,7 +81,8 @@ def show_card(sh: PlexShow, status: str = "new", *, lang: str) -> dict:
         "art": _art_url(sh.rating_key, sh.art_key),
         "season_count": sh.child_count,
         # Aggregate watch-state across the show's episodes (new | in_progress | watched) → grid badge.
-        "status": status,
+        "status": state.status,
+        "resumable": state.resumable,
     }
 
 

backend/app/plex/rollup.py

@@ -9,6 +9,7 @@ from sqlalchemy import and_, case, func, or_, select
 from sqlalchemy.orm import Session, aliased
 
 from app.models import PlexItem, PlexState
+from app.plex.cards import ShowState
 
 # Sort sentinel for a rail row whose timestamp is missing, so `key=lambda r: r[0] or NO_TS` reads the
 # same at every rail that merges two type-homogeneous lists in Python.
@@ -32,6 +33,11 @@ def show_status(total: int, watched: int, inprog: int) -> str:
     return "new"
 
 
+def show_state(total: int, watched: int, inprog: int) -> ShowState:
+    """`show_status` plus whether an episode holds a resume point (the card's `ShowState`)."""
+    return ShowState(show_status(total, watched, inprog), (inprog or 0) > 0)
+
+
 def rollup(cards: list[dict]) -> dict:
     """From ORDERED episode cards → the aggregate watch status, the "on deck" episode to Resume (the
     last in-progress one, else the first unwatched), the first episode (for Play from the start), and
@@ -92,11 +98,11 @@ def agg_subquery(db: Session, user_id: int, include_resumed: bool = False):
     )
 
 
-def state_map(db: Session, user_id: int, show_ids: list[int]) -> dict[int, str]:
+def state_map(db: Session, user_id: int, show_ids: list[int]) -> dict[int, ShowState]:
     if not show_ids:
         return {}
     rows = db.execute(agg_subquery(db, user_id).where(PlexItem.show_id.in_(show_ids))).all()
-    return {r.show_id: show_status(r.total, r.watched, r.inprog) for r in rows}
+    return {r.show_id: show_state(r.total, r.watched, r.inprog) for r in rows}
 
 
 def item_state_map(db: Session, user_id: int, item_ids: list[int]) -> dict[int, PlexState]:

backend/app/routes/plex/browse.py

@@ -164,6 +164,7 @@ def unified_library(
                 cast(null(), Integer).label("season_count"),
                 func.coalesce(st.position_seconds, 0).label("position_seconds"),
                 m_status.label("status"),
+                literal(False).label("resumable"),
                 # "last watched" sort key: when the movie was last resumed, else when it was marked
                 # watched. NULL for never-touched movies → nullslast sinks them (see the sort dict).
                 func.coalesce(st.progress_updated_at, st.watched_at).label("last_watched"),
@@ -211,6 +212,8 @@ def unified_library(
                 PlexShow.child_count.label("season_count"),
                 literal(0).label("position_seconds"),
                 s_status.label("status"),
+                # An episode holds a resume point (`ShowState.resumable`, the card's Continue + Restart).
+                (inp > 0).label("resumable"),
                 # Show leg's "last watched": max(resume/watched time) across the show's episodes (see
                 # rollup.agg_subquery include_resumed). NULL for a never-watched show → nullslast.
                 agg.c.resumed_at.label("last_watched"),
@@ -276,6 +279,7 @@ def unified_library(
                         "thumb": f"/api/plex/image/{r.rk}",
                         "season_count": r.season_count,
                         "status": r.status,
+                        "resumable": r.resumable,
                     }
                 )
 

backend/app/routes/plex/collections.py

@@ -24,7 +24,7 @@ from app.models import (
 )
 from app.plex import rollup
 from app.plex import sync as plex_sync
-from app.plex.cards import display_title, movie_card, norm_lang, show_card
+from app.plex.cards import SHOW_NEW, display_title, movie_card, norm_lang, show_card
 from app.plex.client import PlexClient, PlexError, PlexNotConfigured
 from app.plex.lookups import by_rating_key
 from app.plex.query import content_libs, runtime_by_key, title_col, top_art_keys
@@ -318,7 +318,7 @@ def collection_detail(
         if shows
         else []
     )
-    sstate = {r.show_id: rollup.show_status(r.total, r.watched, r.inprog) for r in show_rows}
+    sstate = {r.show_id: rollup.show_state(r.total, r.watched, r.inprog) for r in show_rows}
     sresumed = {r.show_id: r.resumed_at for r in show_rows}
 
     def _movie_last_watched(st: PlexState | None) -> datetime | None:
@@ -359,7 +359,7 @@ def collection_detail(
     items = [
         movie_card(m.obj, mstate.get(m.obj.id), lang=lang)
         if m.kind == "movie"
-        else show_card(m.obj, sstate.get(m.obj.id, "new"), lang=lang)
+        else show_card(m.obj, sstate.get(m.obj.id, SHOW_NEW), lang=lang)
         for m in members
     ]
     years = [m.year for m in members if m.year is not None]

backend/app/routes/plex/detail.py

@@ -29,7 +29,7 @@ from app.plex import people as plex_people
 from app.plex import rollup, subtitles
 from app.plex.query import title_col
 from app.plex import tmdb as plex_tmdb
-from app.plex.cards import episode_card, movie_card, show_card
+from app.plex.cards import SHOW_NEW, episode_card, movie_card, show_card
 from app.plex.client import PlexClient, PlexError, PlexNotConfigured
 from app.plex.lookups import by_rating_key
 from app.routes.plex._shared import _norm_lang, item_or_404
@@ -279,7 +279,7 @@ def _relatedshow_cards(
         if sh is None or rk in seen:
             continue
         seen.add(rk)
-        out.append(show_card(sh, statuses.get(sh.id, "new"), lang=lang))
+        out.append(show_card(sh, statuses.get(sh.id, SHOW_NEW), lang=lang))
         if len(out) >= 20:
             break
     return out

backend/app/routes/plex/discover.py

@@ -23,7 +23,14 @@ from app.auth import current_user
 from app.db import get_db
 from app.models import PlexItem, PlexShow, PlexState, User
 from app.plex import rollup
-from app.plex.cards import display_title, leaf_card, movie_card, show_card
+from app.plex.cards import (
+    SHOW_NEW,
+    ShowState,
+    display_title,
+    leaf_card,
+    movie_card,
+    show_card,
+)
 from app.plex.query import (
     LibraryFilters,
     apply_meta_filters,
@@ -69,7 +76,7 @@ def _movie_cards(db: Session, uid: int, items: list[PlexItem], lang: str) -> lis
 
 def _show_cards(db: Session, uid: int, shows: list[PlexShow], lang: str) -> list[dict]:
     statuses = rollup.state_map(db, uid, [s.id for s in shows])
-    return [show_card(s, statuses.get(s.id, "new"), lang=lang) for s in shows]
+    return [show_card(s, statuses.get(s.id, SHOW_NEW), lang=lang) for s in shows]
 
 
 def _recently_added(
@@ -232,7 +239,9 @@ def _continue_watching(
         # `apply_show_state_filter` the grid uses, so the rail can't drift from the grid's definition.
         prog = rollup.agg_subquery(db, uid, include_resumed=True).subquery()
         sq = apply_show_state_filter(
-            select(PlexShow, prog.c.resumed_at).join(prog, prog.c.show_id == PlexShow.id),
+            select(PlexShow, prog.c.resumed_at, prog.c.inprog).join(
+                prog, prog.c.show_id == PlexShow.id
+            ),
             prog.c.total,
             prog.c.watched,
             prog.c.inprog,
@@ -243,8 +252,9 @@ def _continue_watching(
             .order_by(prog.c.resumed_at.desc().nullslast())
             .limit(RAIL_LIMIT)
         ).all()
-        for sh, resumed_at in srows:
-            rows.append((resumed_at, show_card(sh, "in_progress", lang=lang)))
+        for sh, resumed_at, inprog in srows:
+            state = ShowState("in_progress", (inprog or 0) > 0)
+            rows.append((resumed_at, show_card(sh, state, lang=lang)))
 
     rows.sort(key=lambda r: r[0] or rollup.NO_TS, reverse=True)
     return [card for _, card in rows[:RAIL_LIMIT]]
@@ -420,6 +430,43 @@ def _draw(db: Session, stmt, key_col, per: int, exclude: list[str]) -> list:
     return [*fresh, *rest.all()]
 
 
+def _kept(
+    db: Session, uid: int, movie_ids: list[int], show_ids: list[int], keep: list[str], lang: str
+) -> list[dict]:
+    """The `keep` titles' cards in their given order, as they stand now: a title watched since the draw
+    stays in it, marked watched, and one gone from the scope's libraries drops out."""
+    movies = (
+        db.execute(
+            select(PlexItem).where(
+                PlexItem.kind == "movie",
+                PlexItem.library_id.in_(movie_ids),
+                PlexItem.rating_key.in_(keep),
+            )
+        )
+        .scalars()
+        .all()
+        if movie_ids
+        else []
+    )
+    shows = (
+        db.execute(
+            select(PlexShow).where(PlexShow.library_id.in_(show_ids), PlexShow.rating_key.in_(keep))
+        )
+        .scalars()
+        .all()
+        if show_ids
+        else []
+    )
+    cards = {
+        c["id"]: c
+        for c in [
+            *_movie_cards(db, uid, list(movies), lang),
+            *_show_cards(db, uid, list(shows), lang),
+        ]
+    }
+    return [cards[k] for k in keep if k in cards]
+
+
 @router.get("/surprise")
 def surprise(
     scope: str = "both",
@@ -427,6 +474,9 @@ def surprise(
     limit: int = Query(default=2 * SURPRISE_PER_KIND, ge=1, le=40),
     # The previous draw's rating keys: a reshuffle deals other titles first.
     exclude: list[str] = Query(default=[], max_length=40),
+    # The draw on screen, read again rather than redrawn: its titles' watch state moves (a play, a
+    # toggle) while the draw itself stands until the ↻.
+    keep: list[str] = Query(default=[], max_length=40),
     f: LibraryFilters = Depends(),
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
@@ -439,6 +489,8 @@ def surprise(
     lang = _norm_lang(lang)
     p = f.as_dict()
     movie_ids, show_ids = _scope_libs(db, scope)
+    if keep:
+        return {"scope": scope, "items": _kept(db, user.id, movie_ids, show_ids, keep, lang)}
     per = min(limit, SURPRISE_PER_KIND)
     movies = []
     if movie_ids:

backend/app/routes/plex/people.py

@@ -19,7 +19,7 @@ from app.models import (
 )
 from app.plex import rollup
 from app.plex import tmdb as plex_tmdb
-from app.plex.cards import movie_card, norm_lang, show_card
+from app.plex.cards import SHOW_NEW, movie_card, norm_lang, show_card
 from app.plex.query import content_libs, split_csv, title_col
 
 router = APIRouter(prefix="/api/plex", tags=["plex"])
@@ -57,7 +57,7 @@ def _resolve_known_for(
         # A credit resolves against its OWN media_type's bucket only.
         if k.get("media_type") == "tv":
             if g in shows:
-                out.append(show_card(shows[g], show_statuses.get(shows[g].id, "new"), lang=lang))
+                out.append(show_card(shows[g], show_statuses.get(shows[g].id, SHOW_NEW), lang=lang))
         elif g in movies:
             out.append(movie_card(movies[g], m_states.get(movies[g].id), lang=lang))
     return out[:_KNOWN_FOR_SHOWN]
@@ -245,7 +245,7 @@ def person_detail(
             .all()
         )
         statuses = rollup.state_map(db, user.id, [s.id for s in show_rows])
-        shows = [show_card(sh, statuses.get(sh.id, "new"), lang=lang) for sh in show_rows]
+        shows = [show_card(sh, statuses.get(sh.id, SHOW_NEW), lang=lang) for sh in show_rows]
 
     known_for = _resolve_known_for(
         db, person.known_for if person else None, user.id, movie_lib_ids, show_lib_ids, lang

backend/tests/test_plex_discover.py

@@ -331,6 +331,20 @@ def test_a_reshuffle_deals_other_titles_first(db, monkeypatch, override_dep):
     assert set(movies("/api/plex/surprise?scope=movie&exclude=m2&exclude=m3")) == {"m2", "m3"}
 
 
+def test_a_kept_draw_is_read_again_not_redrawn(db, monkeypatch, override_dep):
+    # The rail's draw stands until the ↻, but its cards' watch state moves: `keep` answers the titles on
+    # screen, in their order and as they stand now, a watched one included; the scope still bounds it.
+    user = _seed(db)
+    url = "/api/plex/surprise?scope=both&keep=m3&keep=sh1&keep=m1&keep=nope"
+    items = _get(db, user, monkeypatch, override_dep, url).json()["items"]
+    assert _ids(items) == ["m3", "sh1", "m1"]
+    by_id = {c["id"]: c for c in items}
+    assert by_id["m1"]["status"] == "watched"
+    assert by_id["m3"]["position_seconds"] > 0
+    movie_only = "/api/plex/surprise?scope=movie&keep=sh1&keep=m2"
+    assert _ids(_get(db, user, monkeypatch, override_dep, movie_only).json()["items"]) == ["m2"]
+
+
 def test_discover_scope_movie_excludes_shows(db, monkeypatch, override_dep):
     user = _seed(db)
     d = _get(db, user, monkeypatch, override_dep, "/api/plex/discover?scope=movie").json()

backend/tests/test_plex_query_conversion.py

@@ -18,6 +18,7 @@ from app.db import get_db
 from app.main import app
 from app.models import PlexItem, PlexLibrary, PlexSeason, PlexShow, PlexState, User
 from app.plex import rollup, sync
+from app.plex.cards import ShowState
 
 
 def _dt(day: int) -> datetime:
@@ -184,21 +185,50 @@ class TestRollup:
     def test_state_map_new_inprogress_watched(self, db):
         s = _seed(db)
         show, user = s["show"], s["user"]
-        # e1 in-progress, e2 new -> in_progress
-        assert rollup.state_map(db, user.id, [show.id]) == {show.id: "in_progress"}
+        # e1 in-progress, e2 new -> in_progress, and e1's resume point makes it resumable
+        assert rollup.state_map(db, user.id, [show.id]) == {show.id: ShowState("in_progress", True)}
         # empty input short-circuits
         assert rollup.state_map(db, user.id, []) == {}
         # a fresh user with no state -> new
         other = User(email="rollup-other@example.com")
         db.add(other)
         db.commit()
-        assert rollup.state_map(db, other.id, [show.id]) == {show.id: "new"}
-        # mark every episode watched for `user` -> watched
+        assert rollup.state_map(db, other.id, [show.id]) == {show.id: ShowState("new", False)}
+        # e1 finished, e2 untouched: part watched, but no episode to resume — Continue and Restart
+        # would both play e2 from its start
         db.execute(delete(PlexState).where(PlexState.user_id == user.id))
-        for e in db.execute(select(PlexItem).where(PlexItem.kind == "episode")).scalars():
-            db.add(PlexState(user_id=user.id, item_id=e.id, status="watched"))
+        e1, e2 = db.execute(
+            select(PlexItem).where(PlexItem.kind == "episode").order_by(PlexItem.episode_number)
+        ).scalars()
+        db.add(PlexState(user_id=user.id, item_id=e1.id, status="watched"))
         db.commit()
-        assert rollup.state_map(db, user.id, [show.id]) == {show.id: "watched"}
+        assert rollup.state_map(db, user.id, [show.id]) == {
+            show.id: ShowState("in_progress", False)
+        }
+        # mark every episode watched for `user` -> watched
+        db.add(PlexState(user_id=user.id, item_id=e2.id, status="watched"))
+        db.commit()
+        assert rollup.state_map(db, user.id, [show.id]) == {show.id: ShowState("watched", False)}
+
+    def test_the_grid_card_says_whether_a_show_has_an_episode_to_resume(
+        self, db, monkeypatch, override_dep
+    ):
+        s = _seed(db)
+        user = s["user"]
+        cards = {
+            i["id"]: i
+            for i in _get(db, user, monkeypatch, override_dep, "/api/plex/library").json()["items"]
+        }
+        assert cards["s1"]["resumable"] is True
+        db.execute(
+            delete(PlexState).where(PlexState.user_id == user.id, PlexState.position_seconds > 0)
+        )
+        db.commit()
+        cards = {
+            i["id"]: i
+            for i in _get(db, user, monkeypatch, override_dep, "/api/plex/library").json()["items"]
+        }
+        assert cards["s1"]["resumable"] is False
 
 
 def _fake_paginate(by_type):