fix(plex): resolve movies stuck "in progress" on the 15-min incremental sync

The incremental watch-sync (run_plex_watch_sync) learned a movie's watched
transition only from the history feed + onDeck. When a movie is finished on
Plex (commonly by scrubbing through the end credits) Plex sets viewCount>0,
clears viewOffset and DROPS it from Continue-Watching, so onDeck no longer
lists it (and onDeck can only ever ADD an in-progress row, never resolve one),
while a scrub-to-end frequently emits no history session row. The stale local
status='new', position>0 row then survived until the DAILY full reconcile —
which is why movies kept showing "in progress".

Add _reconcile_inprogress_movies: for each local in-progress MOVIE that Plex's
onDeck no longer lists, read its current per-account viewCount/viewOffset with
one metadata() call and apply it via the shared _pull_apply, flipping to
watched when Plex says so. Bounded by a probe cap (100), most-recently-touched
first; episodes are left to onDeck/history. No-signal rows are preserved (an
un-watch stays the daily reconcile's job).

Tests: watched-flip, onDeck+episode skip, and no-signal-preserved.

backend/tests/test_plex_watch_sync.py

@@ -9,11 +9,12 @@ Two independent pieces:
 """
 
 from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
 
 from sqlalchemy import select
 
 from app.models import PlexItem, PlexLibrary, PlexState, User
-from app.plex.watch_sync import _pull_apply
+from app.plex.watch_sync import _pull_apply, _reconcile_inprogress_movies
 from app.routes.plex.playback import (
     _PUSH_THROTTLE_S,
     _resume_push_superseded,
@@ -33,6 +34,16 @@ def _seed_item_and_user(db):
     return item.id, user.id
 
 
+class _FakePlex:
+    """Minimal PlexClient stand-in for the movie-probe tests: metadata(rating_key) → a leaf dict."""
+
+    def __init__(self, by_rk: dict[str, dict]):
+        self._by_rk = by_rk
+
+    def metadata(self, rating_key: str, markers: bool = False):
+        return self._by_rk.get(str(rating_key))
+
+
 def test_pull_plex_wins_over_diverging_synced_local_resume(db):
     """The core S2 change: a SYNCED local row (its state already reached Plex) that later diverges from
     Plex = a real external Plex-side change, so the incremental pull resolves to PLEX's position — Plex
@@ -198,6 +209,130 @@ def test_pull_settles_flag_on_matching_dirty_row(db):
     assert st.synced_to_plex is True, "the dirty flag is settled so it isn't re-pushed endlessly"
 
 
+def test_movie_probe_flips_stale_in_progress_to_watched(db):
+    """S1 core: a local in-progress MOVIE that Plex has since finished (viewCount>0) and dropped from
+    Continue-Watching flips to watched on the incremental pass — without waiting for the daily reconcile.
+    This is the "movie stuck in progress after scrubbing the credits" bug."""
+    item_id, user_id = _seed_item_and_user(db)
+    now = datetime.now(timezone.utc)
+    db.add(
+        PlexState(
+            user_id=user_id,
+            item_id=item_id,
+            status="new",
+            position_seconds=300,
+            progress_updated_at=now,
+            synced_to_plex=True,  # was created by a prior pull; the user then watched it on Plex
+        )
+    )
+    db.flush()
+
+    # Plex now reports the movie watched (viewCount>0, no offset) and it is NOT on the deck.
+    plex = _FakePlex({"m1": {"viewCount": 1, "lastViewedAt": int(now.timestamp())}})
+    out = _reconcile_inprogress_movies(
+        db, plex, SimpleNamespace(user_id=user_id), on_deck_ids=set()
+    )
+    db.flush()
+
+    assert out == {"watched": 1, "in_progress": 0, "probed": 1}
+    st = (
+        db.execute(
+            select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id)
+        )
+        .scalars()
+        .one()
+    )
+    assert st.status == "watched"
+    assert st.position_seconds == 0
+
+
+def test_movie_probe_skips_ondeck_and_episodes(db):
+    """The probe must NOT touch (a) a movie Plex still lists on the deck — it's legitimately in
+    progress — nor (b) an in-progress EPISODE (episodes ride onDeck/history + rollup). Only stale
+    in-progress movies are probed, and metadata() is called for those alone."""
+    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
+    db.add(lib)
+    db.flush()
+    on_deck_movie = PlexItem(rating_key="mDeck", library_id=lib.id, kind="movie", title="On deck")
+    episode = PlexItem(rating_key="e1", library_id=lib.id, kind="episode", title="An episode")
+    db.add_all([on_deck_movie, episode])
+    user = User(email="probe2@example.com")
+    db.add(user)
+    db.flush()
+    now = datetime.now(timezone.utc)
+    for iid in (on_deck_movie.id, episode.id):
+        db.add(
+            PlexState(
+                user_id=user.id,
+                item_id=iid,
+                status="new",
+                position_seconds=300,
+                progress_updated_at=now,
+                synced_to_plex=True,
+            )
+        )
+    db.flush()
+
+    # Plex would report both watched — but neither should be probed at all, so the fake returning
+    # "watched" must have NO effect.
+    plex = _FakePlex(
+        {
+            "mDeck": {"viewCount": 1, "lastViewedAt": int(now.timestamp())},
+            "e1": {"viewCount": 1, "lastViewedAt": int(now.timestamp())},
+        }
+    )
+    out = _reconcile_inprogress_movies(
+        db, plex, SimpleNamespace(user_id=user.id), on_deck_ids={on_deck_movie.id}
+    )
+    db.flush()
+
+    assert out == {"watched": 0, "in_progress": 0, "probed": 0}, "neither item is probed"
+    for iid in (on_deck_movie.id, episode.id):
+        st = (
+            db.execute(
+                select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == iid)
+            )
+            .scalars()
+            .one()
+        )
+        assert st.status == "new" and st.position_seconds == 300, "left untouched"
+
+
+def test_movie_probe_leaves_row_when_plex_has_no_signal(db):
+    """If Plex reports no watch signal (viewCount 0, no offset) for a probed movie, the incremental pass
+    must NOT clear the local in-progress row — un-watch is the daily reconcile's job, and a Siftlode-only
+    in-progress state must survive."""
+    item_id, user_id = _seed_item_and_user(db)
+    now = datetime.now(timezone.utc)
+    db.add(
+        PlexState(
+            user_id=user_id,
+            item_id=item_id,
+            status="new",
+            position_seconds=300,
+            progress_updated_at=now,
+            synced_to_plex=True,
+        )
+    )
+    db.flush()
+
+    plex = _FakePlex({"m1": {"viewCount": 0}})
+    out = _reconcile_inprogress_movies(
+        db, plex, SimpleNamespace(user_id=user_id), on_deck_ids=set()
+    )
+    db.flush()
+
+    assert out == {"watched": 0, "in_progress": 0, "probed": 1}
+    st = (
+        db.execute(
+            select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id)
+        )
+        .scalars()
+        .one()
+    )
+    assert st.status == "new" and st.position_seconds == 300, "in-progress row preserved"
+
+
 def test_should_push_resume_scopes_by_user_and_item():
     """The throttle window is keyed by the FULL (user, item) tuple: pushing for one key must not
     suppress a distinct user or a distinct item within the same window."""