feat(plex): Plex-authoritative watch-sync — resume write-back + faster pull (S2)

Make Plex the source of truth for watch state:

- Write-back during playback: item_progress now pushes the resume position to
  Plex on the periodic checkpoint too (not only the final flush), throttled to
  one push per item every ~10s (in-memory, single-process rate limiter). The FE
  now also fires an immediate (debounced) push on seek.
- Faster incremental pull: 30 → 15 min (plex_watch_sync_interval_min).
- Plex-authoritative pull: _pull_apply resolves a genuine divergence to Plex's
  state instead of last-write-wins; the _same_state + synced_to_plex short-circuit
  still prevents a freshly-pushed row from bouncing. Removed the now-unused
  _LWW_TOLERANCE_S / _sift_ts.

Tests: 5 new in test_plex_watch_sync.py (Plex-wins resume/watched, no-bounce,
throttle final vs periodic).

backend/app/config.py

@@ -232,7 +232,9 @@ class Settings(BaseSettings):
     plex_sync_interval_min: int = 360
     # How often the incremental Plex→Siftlode watch-state reconcile runs, in minutes (Phase C):
     # a cheap history + on-deck pull plus a re-push of any states that failed their immediate push.
-    plex_watch_sync_interval_min: int = 30
+    # 15 min (S2): with Plex now the source of truth (prompt write-back on the ~10s checkpoint +
+    # Plex-authoritative pull), a faster incremental pull keeps Siftlode close to Plex-side changes.
+    plex_watch_sync_interval_min: int = 15
     # How often the FULL watch-state reconcile runs, in minutes (Phase C). A whole-section rescan
     # that catches what the incremental history feed can't — notably an un-watch done on the Plex
     # side — so it runs far less often (daily) than the incremental pass.

backend/app/plex/watch_sync.py

@@ -11,11 +11,14 @@ already pages) and upsert them into the owner's `plex_states`. Plex wins on over
 exist only in Siftlode are left untouched (union, Plex-authoritative on the intersection).
 
 Phases B (Siftlode→Plex push via scrobble/timeline) and C (incremental Plex→Siftlode via the watch
-history + onDeck, last-write-wins) build on this and land in later ships.
+history + onDeck) build on this and land in later ships. As of S2 Plex is the source of truth: the
+player pushes the resume position back to Plex on the periodic checkpoint (not just the final flush),
+and the incremental pull resolves a genuine divergence to Plex's state (see `_pull_apply`) rather than
+running a last-write-wins comparison.
 """
 
 import logging
-from datetime import datetime, timedelta, timezone
+from datetime import datetime, timezone
 
 from sqlalchemy import func, select, update
 from sqlalchemy.dialects.postgresql import insert as pg_insert
@@ -293,12 +296,6 @@ def push_bulk_state_to_plex(user_id: int, changes: list[tuple[int, str, str]]) -
 
 # --- Phase C: incremental + full Plex ↔ Siftlode reconcile (scheduler jobs) ------------------------
 
-# Clock-skew / round-trip slack when comparing a Plex timestamp to a Siftlode one. A state we just
-# pushed (Plex stamps viewedAt a beat after our watched_at) must not read as "newer on Plex" and get
-# pulled back with a different shape. Genuine Plex-side edits are minutes from our writes — far
-# outside this — so the window only absorbs skew, never real changes.
-_LWW_TOLERANCE_S = 90
-
 
 def _active_sync_links(db: Session) -> list[PlexLink]:
     """Every link with two-way sync actually running (owner MVP = one row). Iterated uniformly so
@@ -338,11 +335,6 @@ def _resolve_account_id(db: Session, plex: PlexClient, link: PlexLink) -> int:
     return acct_id
 
 
-def _sift_ts(st: PlexState) -> datetime | None:
-    """The freshest Siftlode-side timestamp on a row, for last-write-wins."""
-    return max([t for t in (st.watched_at, st.progress_updated_at) if t], default=None)
-
-
 def _same_state(st: PlexState | None, status: str, pos: int) -> bool:
     """Whether a row already matches a target (status, pos) — the core ping-pong guard: when nothing
     would change we skip the write entirely (no row churn, no timestamp bump, no re-pull)."""
@@ -370,10 +362,13 @@ def _pull_apply(
         if st is not None and not st.synced_to_plex:
             st.synced_to_plex = True  # already matches Plex → just settle the flag, no churn
         return None
-    # Conflict: keep Siftlode when it's newer than Plex (plus skew slack).
-    sts = _sift_ts(st) if st else None
-    if sts and plex_ts and plex_ts <= sts + timedelta(seconds=_LWW_TOLERANCE_S):
-        return None
+    # Plex-authoritative (S2): on a genuine divergence for an item Plex knows about, Plex wins — no
+    # last-write-wins tug-of-war. This is safe because prompt write-back (playback.py pushes the resume
+    # position to Plex on the ~10s checkpoint, not just on the final flush) keeps Plex current, so a
+    # local edit that's "newer" than Plex has almost always already been pushed and matched above by
+    # `_same_state`; a row that reaches here really is out of step with the source of truth. The
+    # `_same_state` + `synced_to_plex` short-circuit above still absorbs the freshly-pushed case (Plex
+    # echoes the same shape a beat later) so a just-pushed row never bounces.
     if st is None:
         st = PlexState(user_id=user_id, item_id=item_id)
         db.add(st)

backend/app/routes/plex/playback.py

@@ -3,6 +3,7 @@
 import hashlib
 import os
 import tempfile
+import threading
 from datetime import datetime, timezone
 
 import httpx
@@ -155,6 +156,29 @@ _PROGRESS_MIN_S = 5
 
 _FINISH_MARGIN_S = 30
 
+# S2 write-back throttle: the resume position is now pushed to Plex on the periodic (non-final)
+# checkpoint too — not only on the final pause/unmount flush — so Plex (the source of truth) stays
+# current in near-real-time. To keep a single watch from spraying `/:/timeline` calls, a push is
+# rate-limited to at most one per item per user every _PUSH_THROTTLE_S seconds; a `final` checkpoint
+# (pause / seek / stop) always pushes immediately and resets the window. This is an in-memory, single-
+# process rate-limiter (the deployment is one FastAPI process): after a restart the map is empty, so at
+# most one extra push happens on the first checkpoint — harmless. It is NOT durable domain state.
+_PUSH_THROTTLE_S = 10
+_last_push_at: dict[tuple[int, int], datetime] = {}
+_push_throttle_lock = threading.Lock()
+
+
+def _should_push_resume(user_id: int, item_id: int, now: datetime, final: bool) -> bool:
+    """Whether to mirror this resume checkpoint to Plex now. `final` always pushes; a periodic tick
+    pushes only when ≥_PUSH_THROTTLE_S has elapsed since the last push for this (user, item). Records
+    `now` as the last-push time whenever it returns True so the window is measured from the real push."""
+    with _push_throttle_lock:
+        last = _last_push_at.get((user_id, item_id))
+        if final or last is None or (now - last).total_seconds() >= _PUSH_THROTTLE_S:
+            _last_push_at[(user_id, item_id)] = now
+            return True
+        return False
+
 
 @router.get("/person-image")
 def person_image(u: str, request: Request) -> Response:
@@ -232,9 +256,10 @@ def item_progress(
 ) -> dict:
     """Checkpoint the resume position while playing (near the end → mark watched + clear it).
 
-    `final` (sent on the pause/pagehide/unmount flush, not the periodic 10s checkpoint) is what
-    debounces the Plex resume push: we mirror the settled position to Plex only on a final flush, so
-    a single watch doesn't spray `/:/timeline` calls at the server every 10 seconds."""
+    S2 write-back: the resume position is mirrored to Plex (the source of truth) on the periodic
+    checkpoint too, not only on the final flush — throttled to ≥_PUSH_THROTTLE_S per item (see
+    `_should_push_resume`). `final` (sent on pause / seek / pagehide / unmount) pushes immediately and
+    resets the throttle window, so Plex sees a settled position the instant the user pauses or seeks."""
     it = item_or_404(db, rating_key)
     position = max(0, int(body.position_seconds))
     duration = int(body.duration_seconds)
@@ -280,10 +305,13 @@ def item_progress(
         db.add(st)
     st.position_seconds = position
     st.progress_updated_at = now
-    if final:
+    push_now = _should_push_resume(user.id, it.id, now, final)
+    if push_now:
+        # Mark dirty before the (background) push so the reconcile knows to re-push if it fails, and so
+        # `push_state_to_plex` can settle synced_to_plex once Plex confirms.
         st.synced_to_plex = False
     db.commit()
-    if final:
+    if push_now:
         _push_watch(
             background,
             db,

backend/tests/test_plex_watch_sync.py

@@ -0,0 +1,148 @@
+"""S2 watch-sync semantics: Plex is the source of truth.
+
+Two independent pieces:
+- `_pull_apply` (DB lane): on a genuine divergence for an item Plex knows about, Plex now WINS —
+  the old "keep Siftlode when it's newer within a tolerance" tug-of-war is gone. The anti-bounce
+  short-circuit (`_same_state` + `synced_to_plex`) must still keep a freshly-pushed row from bouncing.
+- `_should_push_resume` (pure): the in-memory write-back throttle — a `final` checkpoint always pushes,
+  a periodic one pushes at most once per _PUSH_THROTTLE_S seconds per (user, item).
+"""
+
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import select
+
+from app.models import PlexItem, PlexLibrary, PlexState, User
+from app.plex.watch_sync import _pull_apply
+from app.routes.plex.playback import _PUSH_THROTTLE_S, _should_push_resume
+
+
+def _seed_item_and_user(db):
+    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
+    db.add(lib)
+    db.flush()
+    item = PlexItem(rating_key="m1", library_id=lib.id, kind="movie", title="A Movie")
+    db.add(item)
+    user = User(email="watch@example.com")
+    db.add(user)
+    db.flush()
+    return item.id, user.id
+
+
+def test_pull_plex_wins_over_newer_local_resume(db):
+    """The core S2 change: a locally-in-progress item whose local edit is NEWER than Plex's still
+    resolves to PLEX's position on the incremental pull — Plex is authoritative, no LWW."""
+    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,  # local touched just now — "newer" than Plex
+            synced_to_plex=False,
+        )
+    )
+    db.flush()
+
+    # Plex reports a DIFFERENT (older) resume position for the same item.
+    changed = _pull_apply(db, user_id, item_id, "new", 120, plex_ts=now - timedelta(seconds=60))
+    db.flush()
+
+    assert changed == "in_progress"
+    st = (
+        db.execute(
+            select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id)
+        )
+        .scalars()
+        .one()
+    )
+    assert st.position_seconds == 120, "Plex's position must win, even though local was newer"
+    assert st.synced_to_plex is True
+
+
+def test_pull_plex_watched_wins_over_local_in_progress(db):
+    """Plex marking the item watched propagates even over a fresh local in-progress row."""
+    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=False,
+        )
+    )
+    db.flush()
+
+    changed = _pull_apply(db, user_id, item_id, "watched", 0, plex_ts=now - timedelta(seconds=60))
+    db.flush()
+
+    assert changed == "watched"
+    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_pull_no_bounce_on_matching_row(db):
+    """Anti-bounce short-circuit: when the row already matches what Plex reports (a just-pushed
+    resume), the pull makes NO change and only settles the synced flag — no churn, no re-pull."""
+    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=100,
+            progress_updated_at=now,
+            synced_to_plex=True,  # freshly pushed
+        )
+    )
+    db.flush()
+
+    # Plex echoes the same position a beat later (well within the _PROGRESS_MIN_S fuzz).
+    changed = _pull_apply(db, user_id, item_id, "new", 102, plex_ts=now + timedelta(seconds=2))
+    db.flush()
+
+    assert changed is None, "a matching row must not bounce"
+    st = (
+        db.execute(
+            select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id)
+        )
+        .scalars()
+        .one()
+    )
+    assert st.position_seconds == 100, "no churn to the stored position"
+
+
+def test_should_push_resume_final_always_pushes():
+    """A final checkpoint (pause / seek / stop) always pushes, bypassing the throttle."""
+    now = datetime.now(timezone.utc)
+    key = (9001, 9001)
+    assert _should_push_resume(*key, now, final=True) is True
+    # A second final a fraction of a second later still pushes.
+    assert _should_push_resume(*key, now + timedelta(seconds=1), final=True) is True
+
+
+def test_should_push_resume_periodic_is_throttled():
+    """Periodic (non-final) checkpoints push at most once per _PUSH_THROTTLE_S per (user, item)."""
+    now = datetime.now(timezone.utc)
+    key = (9002, 9002)
+    assert _should_push_resume(*key, now, final=False) is True  # first tick pushes
+    # A tick within the window is suppressed.
+    assert (
+        _should_push_resume(*key, now + timedelta(seconds=_PUSH_THROTTLE_S - 1), final=False)
+        is False
+    )
+    # Once the window elapses, the next tick pushes again.
+    assert _should_push_resume(*key, now + timedelta(seconds=_PUSH_THROTTLE_S), final=False) is True

frontend/src/lib/plexPlayerHooks.ts

@@ -102,6 +102,16 @@ export function usePlexProgress(opts: {
     const onPauseFlush = () => flush(true);
     const video = videoRef.current;
     video?.addEventListener("pause", onPauseFlush);
+    // A seek is a settled position too (S2: Plex is the source of truth, so it must see the new spot
+    // right away, not up to 10s later). Debounced so scrubbing — many `seeked` events in a row — pushes
+    // just ONCE, once the user lands. `final=true` bypasses the backend's ≥10s push-throttle, which is
+    // exactly why the debounce lives here: it bounds a scrub to one immediate Plex timeline call.
+    let seekTimer: number | undefined;
+    const onSeekFlush = () => {
+      window.clearTimeout(seekTimer);
+      seekTimer = window.setTimeout(() => flush(true), 800);
+    };
+    video?.addEventListener("seeked", onSeekFlush);
     // Save on page unload too (F5/close/navigate): React effect cleanup does NOT run on a full reload,
     // so without this the resume point lags up to 10s — e.g. a seek right before F5 was lost. Uses a
     // keepalive beacon so the POST survives the unload (a final checkpoint — the user is leaving).
@@ -112,8 +122,10 @@ export function usePlexProgress(opts: {
     window.addEventListener("pagehide", onHide);
     return () => {
       window.clearInterval(iv);
+      window.clearTimeout(seekTimer);
       window.removeEventListener("pagehide", onHide);
       video?.removeEventListener("pause", onPauseFlush);
+      video?.removeEventListener("seeked", onSeekFlush);
       flush(true);
       // Back (unmount) persists the position server-side via flush(), but reopening the same item in
       // the SAME session read react-query's CACHED detail — whose position_seconds predated this