feat(plex): Phase B — Siftlode→Plex watch-state push

Mirror watched/unwatched/resume from Siftlode to a linked (owner) Plex account.

- PlexClient: scrobble / unscrobble / set_timeline (/:/scrobble, /:/unscrobble, /:/timeline).
- watch_sync: link_for_push() gates on owner + sync_enabled + initial_import_done; best-effort
  push_state_to_plex() runs as a background task with its own session, never raises (Plex being
  down must not break the user's action), and flips synced_to_plex on success so Phase C's pull
  won't bounce it back.
- item_state / item_progress schedule the push: watched/unwatched immediately, resume only on a
  "final" checkpoint (pause / pagehide / unmount) — not every 10s tick — so one watch doesn't spray
  /:/timeline at the server. `hidden` is Siftlode-only and never touches Plex.
- Frontend: plexProgress / plexProgressBeacon carry a `final` flag; the periodic checkpoint is
  non-final, pause/pagehide/leaving the player are final.

Verified live against the real Plex server (scrobble/unscrobble/timeline round-trip + restore;
link gating; synced_to_plex flip).

backend/app/plex/client.py

@@ -166,6 +166,38 @@ class PlexClient:
         items = mc.get("Metadata", []) or []
         return items[0] if items else None
 
+    # --- Watch-state write-back (P5 Phase B — Siftlode→Plex push; acts on the token's account) ---
+    _LIBRARY_IDENTIFIER = "com.plexapp.plugins.library"
+
+    def scrobble(self, rating_key: str) -> None:
+        """Mark the item *watched* on the token's Plex account (viewCount++, resume cleared)."""
+        self._write(
+            "GET",
+            f"/:/scrobble?identifier={self._LIBRARY_IDENTIFIER}"
+            f"&key={quote(str(rating_key), safe='')}",
+        )
+
+    def unscrobble(self, rating_key: str) -> None:
+        """Mark the item *unwatched* on the token's Plex account (clears viewCount)."""
+        self._write(
+            "GET",
+            f"/:/unscrobble?identifier={self._LIBRARY_IDENTIFIER}"
+            f"&key={quote(str(rating_key), safe='')}",
+        )
+
+    def set_timeline(
+        self, rating_key: str, time_ms: int, duration_ms: int, state: str = "stopped"
+    ) -> None:
+        """Report a resume position (viewOffset) for the item on the token's account. `state` is a
+        Plex timeline state (playing|paused|stopped); we push the settled position as ``stopped``."""
+        key = quote(f"/library/metadata/{rating_key}", safe="")
+        self._write(
+            "GET",
+            f"/:/timeline?identifier={self._LIBRARY_IDENTIFIER}"
+            f"&ratingKey={quote(str(rating_key), safe='')}&key={key}"
+            f"&state={state}&time={int(time_ms)}&duration={int(duration_ms)}",
+        )
+
     def raw_get(self, path: str) -> bytes:
         """Raw bytes of an arbitrary Plex resource path (e.g. an external subtitle stream `key` like
         ``/library/streams/553184``). Keeps the admin token server-side."""

backend/app/plex/watch_sync.py

@@ -21,6 +21,7 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.orm import Session
 
 from app import sysconfig
+from app.db import SessionLocal
 from app.models import PlexItem, PlexLink, PlexState
 from app.plex.client import PlexClient, PlexError, PlexNotConfigured
 from app.plex.sync import (
@@ -140,3 +141,57 @@ def import_owner_watch_state(db: Session, link: PlexLink) -> dict:
 
     log.info("Plex watch import (user %s): %s", link.user_id, stats)
     return stats
+
+
+# --- Phase B: Siftlode → Plex push -----------------------------------------------------------------
+
+
+def link_for_push(db: Session, user_id: int) -> PlexLink | None:
+    """The active owner watch-push link for this user, or None when push is off / not the owner.
+    Push is gated on `initial_import_done` so we never push a half-known local state over Plex's
+    authoritative one before the one-time "Plex is master" import has reconciled them."""
+    link = db.query(PlexLink).filter_by(user_id=user_id).first()
+    if link and link.uses_admin and link.sync_enabled and link.initial_import_done:
+        return link
+    return None
+
+
+def push_state_to_plex(
+    user_id: int,
+    item_id: int,
+    rating_key: str,
+    action: str,
+    time_ms: int = 0,
+    duration_ms: int = 0,
+) -> None:
+    """Best-effort Siftlode→Plex push, run in a FastAPI BackgroundTask (its OWN DB session — the
+    request's session is already closed by the time this runs). Never raises: Plex being down or slow
+    must not break the user's action. On success it flags `synced_to_plex` so Phase C's incremental
+    pull won't bounce the change straight back; on failure the row stays dirty (synced_to_plex=False)
+    for a later retry/reconcile.
+
+    `action` is one of ``watched`` (scrobble), ``unwatched`` (unscrobble), ``resume`` (timeline)."""
+    with SessionLocal() as db:
+        if link_for_push(db, user_id) is None:
+            return
+        try:
+            with PlexClient(db) as plex:
+                if action == "watched":
+                    plex.scrobble(rating_key)
+                elif action == "unwatched":
+                    plex.unscrobble(rating_key)
+                elif action == "resume":
+                    plex.set_timeline(rating_key, time_ms, duration_ms)
+                else:
+                    return
+        except PlexError as e:
+            log.warning(
+                "Plex watch push failed (user %s, item %s, %s): %s", user_id, item_id, action, e
+            )
+            return
+        # Flag the row synced — but an "unwatched" via status=new deletes the row, so there may be
+        # nothing to flag; that's fine (unscrobble still happened).
+        st = db.query(PlexState).filter_by(user_id=user_id, item_id=item_id).first()
+        if st is not None:
+            st.synced_to_plex = True
+            db.commit()