fix(plex,channels): deep-review fixes — durable watch-sync + no shared-link clobber

Deep-review of the quickfix batch surfaced two data/preference-loss bugs plus
smaller items; all fixed with tests.

- watch_sync _pull_apply: don't let the Plex-authoritative pull overwrite a DIRTY
  local row (a change whose immediate push failed) — that stranded the value
  forever since _repush_dirty only retries dirty rows. Leave it dirty for re-push.
- PlexProvider / ChannelsProvider: a partial shared link opened in a fresh tab
  persisted mount-time DEFAULTS over the account's stored scope/sort/filters
  (base read before the account was pinned). Persist/keep only URL-present dims;
  re-hydrate the omitted ones from storage.
- playback resume write-back: guard against out-of-order delivery (a superseded
  older push now drops) and evict the throttle map so it can't grow unbounded.
- PlexPlayer: announce mid-playback buffering to assistive tech (role=status +
  sr-only label); new plex.player.buffering (HU+EN).
- columnSort: hoist the shared key:dir sort codec used by both URL-state modules.
- Tests: dirty-guard + flag-settle + per-key throttle scoping + superseded-push
  (backend); hold-repeat/suppress-click/unmount + partial-link no-clobber +
  refetch-doesn't-clobber-undo (frontend). Backend pytest 475, FE vitest 519.

backend/app/routes/plex/playback.py

@@ -4,7 +4,7 @@ import hashlib
 import os
 import tempfile
 import threading
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 
 import httpx
 from fastapi import (
@@ -171,15 +171,33 @@ _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."""
+    `now` as the last-push time whenever it returns True so the window is measured from the real push
+    (and so `_resume_push_superseded` can drop an older, still-pending push — see there)."""
     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
+            # Opportunistic eviction so the map can't grow unbounded over a long-lived process (one
+            # entry per item ever checkpointed). Anything older than a few windows is dead — a push
+            # completes in seconds, so a stale entry protects nothing.
+            if len(_last_push_at) > 2048:
+                cutoff = now - timedelta(seconds=_PUSH_THROTTLE_S * 6)
+                for k in [k for k, t in _last_push_at.items() if t < cutoff]:
+                    del _last_push_at[k]
             return True
         return False
 
 
+def _resume_push_superseded(user_id: int, item_id: int, intent: datetime) -> bool:
+    """True if a newer resume push for this (user, item) was scheduled after `intent`. Called at the
+    HTTP boundary of a background resume push: because periodic and seek/pause pushes are independent
+    background tasks with no ordering guard, an older still-pending periodic push could otherwise land
+    at Plex AFTER a newer backward-seek push and — with the Plex-authoritative pull — durably revive the
+    stale position. Skipping a superseded push keeps the last-scheduled position the one Plex settles on."""
+    with _push_throttle_lock:
+        return _last_push_at.get((user_id, item_id), intent) > intent
+
+
 @router.get("/person-image")
 def person_image(u: str, request: Request) -> Response:
     """Proxy a cast/crew photo from Plex's public metadata CDN (host-whitelisted; no token needed).
@@ -202,6 +220,26 @@ def person_image(u: str, request: Request) -> Response:
     return imgcache.response(r.content, ct)
 
 
+def _push_resume_guarded(
+    user_id: int,
+    item_id: int,
+    rating_key: str,
+    time_ms: int,
+    duration_ms: int,
+    expect_ts: datetime | None,
+    intent: datetime,
+) -> None:
+    """Background entrypoint for a resume push that drops itself if a newer resume push for the same
+    (user, item) was scheduled after it (see `_resume_push_superseded`), then delegates to the real
+    push. This narrows the periodic-vs-seek write-ordering race to the (rare) case where the older push
+    has already left the process before the newer one is scheduled."""
+    if _resume_push_superseded(user_id, item_id, intent):
+        return
+    plex_watch.push_state_to_plex(
+        user_id, item_id, rating_key, "resume", time_ms, duration_ms, expect_ts
+    )
+
+
 def _push_watch(
     background: BackgroundTasks,
     db: Session,
@@ -211,10 +249,14 @@ def _push_watch(
     action: str,
     time_ms: int = 0,
     duration_ms: int = 0,
+    intent: datetime | None = None,
 ) -> None:
     """Schedule a best-effort Siftlode→Plex watch push, but only for a user whose owner link has
     push enabled — non-owner users never touch Plex. Guarding here (with the request session) avoids
-    spinning up a throwaway background session on every state change for the common no-link case."""
+    spinning up a throwaway background session on every state change for the common no-link case.
+
+    A `resume` push with an `intent` timestamp is routed through `_push_resume_guarded` so a superseded
+    (older) push can't land out of order; `watched`/`unwatched` scrobbles go straight through."""
     if plex_watch.link_for_push(db, user_id) is None:
         return
     # Capture the row's freshest timestamp NOW so the background push can tell if a newer local edit
@@ -226,6 +268,12 @@ def _push_watch(
         .scalars()
         .first()
     )
+    marker = plex_watch._state_marker(st)
+    if action == "resume" and intent is not None:
+        background.add_task(
+            _push_resume_guarded, user_id, item_id, rating_key, time_ms, duration_ms, marker, intent
+        )
+        return
     background.add_task(
         plex_watch.push_state_to_plex,
         user_id,
@@ -234,7 +282,7 @@ def _push_watch(
         action,
         time_ms,
         duration_ms,
-        plex_watch._state_marker(st),
+        marker,
     )
 
 
@@ -284,6 +332,10 @@ def item_progress(
         st.position_seconds = 0
         st.progress_updated_at = now
         st.synced_to_plex = False
+        # Terminal state for resume write-back: drop the throttle/ordering entry so the map doesn't
+        # retain a dead key for every finished item.
+        with _push_throttle_lock:
+            _last_push_at.pop((user.id, it.id), None)
         db.commit()
         # Immediate on the watched transition; skip a repeat scrobble if it was already watched
         # (a later checkpoint of the same finished item would otherwise bump Plex's viewCount again).
@@ -321,6 +373,7 @@ def item_progress(
             "resume",
             position * 1000,
             duration * 1000,
+            intent=now,
         )
     return {"position_seconds": position}