Merge: promote dev to prod

VERSION

@@ -1 +1 @@
-0.34.2
+0.35.0

backend/app/config.py

@@ -202,8 +202,13 @@ class Settings(BaseSettings):
     plex_client_id: str = "siftlode-server"
     # How often the catalog sync job mirrors Plex metadata, in minutes.
     plex_sync_interval_min: int = 360
-    # How often the incremental Plex→Siftlode watch-state reconcile runs, in minutes (Phase C).
+    # 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
+    # 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.
+    plex_watch_reconcile_interval_min: int = 1440
     # Max concurrent transcodes for the P3 fallback. Low by default — CPU-only transcode is
     # expensive; direct-serve (browser-compatible files) has no such limit.
     plex_max_transcodes: int = 1

backend/app/plex/client.py

@@ -166,6 +166,73 @@ 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)}",
+        )
+
+    # --- Watch-state read for incremental sync (P5 Phase C) ---
+    def accounts(self) -> list[dict]:
+        """Server accounts (the owner + any managed/home users): id, name. Used to resolve the
+        owner's accountID so the shared watch-history feed can be filtered to just their views."""
+        return self._get("/accounts").get("Account", []) or []
+
+    def watch_history(
+        self, min_viewed_at: int = 0, account_id: int | None = None, size: int = 500
+    ) -> list[dict]:
+        """The most recent watch-history rows (ratingKey / viewedAt / accountID / type), newest
+        first — works without Plex Pass. Filtered client-side to `account_id` and to rows at/after
+        `min_viewed_at` (epoch seconds): the efficient "what changed since T" feed. Sorted desc, so
+        we stop at the first row older than the cutoff."""
+        mc = self._get(
+            "/status/sessions/history/all",
+            params={
+                "sort": "viewedAt:desc",
+                "X-Plex-Container-Start": 0,
+                "X-Plex-Container-Size": size,
+            },
+        )
+        out = []
+        for r in mc.get("Metadata", []) or []:
+            if int(r.get("viewedAt") or 0) < min_viewed_at:
+                break
+            if account_id is not None and int(r.get("accountID") or -1) != account_id:
+                continue
+            out.append(r)
+        return out
+
+    def on_deck(self) -> list[dict]:
+        """The token account's Continue-Watching list (ratingKey / viewOffset / lastViewedAt /
+        type) — the incremental source for resume-position changes made on the Plex side."""
+        return self._get("/library/onDeck").get("Metadata", []) or []
+
     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

@@ -14,13 +14,14 @@ Phases B (Siftlode→Plex push via scrobble/timeline) and C (incremental Plex→
 history + onDeck, last-write-wins) build on this and land in later ships.
 """
 import logging
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 
 from sqlalchemy import func
 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 (
@@ -58,6 +59,39 @@ def _plex_watch_to_state(
     return None
 
 
+def _scan_plex_states(
+    db: Session, plex: PlexClient
+) -> tuple[dict[int, tuple[str, int, datetime | None, datetime | None]], int]:
+    """Scan the enabled movie/show sections and return ``{item_id: (status, pos, watched_at,
+    prog_at)}`` for every mirrored leaf Plex holds a watch signal for, plus the scanned-leaf count.
+    Shared by the one-time import and the full reconcile (both need Plex's whole current picture)."""
+    item_id_by_rk: dict[str, int] = {
+        rk: iid for rk, iid in db.query(PlexItem.rating_key, PlexItem.id)
+    }
+    wanted = _enabled_section_keys(db)
+    scanned = 0
+    out: dict[int, tuple[str, int, datetime | None, datetime | None]] = {}
+    for s in plex.sections():
+        stype = s.get("type")
+        if stype not in ("movie", "show"):
+            continue
+        key = str(s.get("key"))
+        if wanted is not None and key not in wanted:
+            continue
+        item_type = _TYPE_MOVIE if stype == "movie" else _TYPE_EPISODE
+        for meta in _paginate(plex, key, item_type):
+            scanned += 1
+            item_id = item_id_by_rk.get(str(meta.get("ratingKey") or ""))
+            if item_id is None:
+                continue
+            res = _plex_watch_to_state(
+                meta.get("viewCount"), meta.get("viewOffset"), meta.get("lastViewedAt")
+            )
+            if res is not None:
+                out[item_id] = res
+    return out, scanned
+
+
 def import_owner_watch_state(db: Session, link: PlexLink) -> dict:
     """One-time "Plex is master" import of the linked user's Plex watch state into their
     `plex_states`. Idempotent (safe to re-run — Plex simply re-wins on the intersection). Only
@@ -65,41 +99,17 @@ def import_owner_watch_state(db: Session, link: PlexLink) -> dict:
     if not sysconfig.get_bool(db, "plex_enabled"):
         return {"skipped": "disabled"}
 
-    # rating_key -> plex_items.id for every mirrored playable leaf (movie/episode).
-    item_id_by_rk: dict[str, int] = {
-        rk: iid for rk, iid in db.query(PlexItem.rating_key, PlexItem.id)
-    }
     stats = {"watched": 0, "in_progress": 0, "scanned": 0}
-    wanted = _enabled_section_keys(db)
-    # Accumulate one row per item (keyed by item_id → inherently de-duped) then bulk-UPSERT. An upsert
-    # is idempotent AND concurrency-safe: a repeated/overlapping enable (the toggle firing twice while
-    # the multi-second import runs) can't raise a duplicate-key error — the conflict just re-updates.
-    rows_by_item: dict[int, dict] = {}
-
     try:
         with PlexClient(db) as plex:
-            for s in plex.sections():
+            scanned_states, stats["scanned"] = _scan_plex_states(db, plex)
-                stype = s.get("type")
+            # One row per item (keyed by item_id → inherently de-duped) then bulk-UPSERT. An upsert is
-                if stype not in ("movie", "show"):
+            # idempotent AND concurrency-safe: a repeated/overlapping enable (the toggle firing twice
-                    continue
+            # while the multi-second import runs) can't raise a duplicate-key error — it re-updates.
-                key = str(s.get("key"))
+            rows = []
-                if wanted is not None and key not in wanted:
+            for item_id, (status, pos, watched_at, prog_at) in scanned_states.items():
-                    continue
+                rows.append(
-                item_type = _TYPE_MOVIE if stype == "movie" else _TYPE_EPISODE
+                    {
-                for meta in _paginate(plex, key, item_type):
-                    stats["scanned"] += 1
-                    item_id = item_id_by_rk.get(str(meta.get("ratingKey") or ""))
-                    if item_id is None:
-                        continue
-                    res = _plex_watch_to_state(
-                        meta.get("viewCount"),
-                        meta.get("viewOffset"),
-                        meta.get("lastViewedAt"),
-                    )
-                    if res is None:
-                        continue
-                    status, pos, watched_at, prog_at = res
-                    rows_by_item[item_id] = {
                         "user_id": link.user_id,
                         "item_id": item_id,
                         "status": status,
@@ -109,11 +119,11 @@ def import_owner_watch_state(db: Session, link: PlexLink) -> dict:
                         # Mirrors Plex now — so Phase B's push doesn't bounce it straight back.
                         "synced_to_plex": True,
                     }
-                    stats["watched" if status == "watched" else "in_progress"] += 1
+                )
+                stats["watched" if status == "watched" else "in_progress"] += 1
 
             # Bulk upsert in chunks. Plex wins on the intersection; Siftlode-only states (item_ids not
-            # in rows_by_item) are never referenced, so they're preserved untouched (union).
+            # scanned) are never referenced, so they're preserved untouched (union).
-            rows = list(rows_by_item.values())
             for i in range(0, len(rows), 1000):
                 stmt = pg_insert(PlexState).values(rows[i : i + 1000])
                 stmt = stmt.on_conflict_do_update(
@@ -140,3 +150,268 @@ 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()
+
+
+# --- 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
+    P5b friend links slot in later without touching this loop."""
+    return (
+        db.query(PlexLink)
+        .filter_by(uses_admin=True, sync_enabled=True, initial_import_done=True)
+        .all()
+    )
+
+
+def _resolve_account_id(db: Session, plex: PlexClient, link: PlexLink) -> int:
+    """The owner's Plex accountID, cached on the link. The server owner is accountID 1 on their own
+    server; we confirm + grab their username from /accounts, falling back to 1 if unavailable."""
+    if link.plex_account_id:
+        return link.plex_account_id
+    acct_id, username = 1, None
+    try:
+        for a in plex.accounts():
+            if int(a.get("id") or -1) == 1:
+                username = a.get("name")
+                break
+    except PlexError:
+        pass
+    link.plex_account_id = acct_id
+    if username:
+        link.plex_username = username
+    db.commit()
+    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)."""
+    if st is None:
+        return False
+    if status == "watched":
+        return st.status == "watched"
+    return st.status != "watched" and abs((st.position_seconds or 0) - pos) <= _PROGRESS_MIN_S
+
+
+def _pull_apply(
+    db: Session, user_id: int, item_id: int, status: str, pos: int, plex_ts: datetime | None
+) -> str | None:
+    """Bring one Plex-side positive signal into Siftlode under last-write-wins. Returns
+    "watched"/"in_progress" if it changed the row, else None. Never clears a row (un-watch is the
+    full reconcile's job) — this only pulls in watched/resume."""
+    st = db.query(PlexState).filter_by(user_id=user_id, item_id=item_id).first()
+    if _same_state(st, status, pos):
+        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
+    if st is None:
+        st = PlexState(user_id=user_id, item_id=item_id)
+        db.add(st)
+    if status == "watched":
+        st.status, st.position_seconds, st.watched_at, st.progress_updated_at = "watched", 0, plex_ts, plex_ts
+    else:
+        st.status, st.position_seconds, st.progress_updated_at = "new", pos, plex_ts
+    st.synced_to_plex = True
+    return "watched" if status == "watched" else "in_progress"
+
+
+def _repush_dirty(db: Session, plex: PlexClient, link: PlexLink) -> int:
+    """Belt-and-suspenders: push local states that never reached Plex (synced_to_plex=False) — e.g.
+    an immediate Phase B push that failed while Plex was down. `hidden` is excluded (Siftlode-only,
+    never goes to Plex). Returns the count re-pushed."""
+    rk_by_id = dict(db.query(PlexItem.id, PlexItem.rating_key))
+    dirty = (
+        db.query(PlexState)
+        .filter(
+            PlexState.user_id == link.user_id,
+            PlexState.synced_to_plex.is_(False),
+            PlexState.status != "hidden",
+        )
+        .all()
+    )
+    n = 0
+    for st in dirty:
+        rk = rk_by_id.get(st.item_id)
+        if rk is None:
+            continue
+        try:
+            if st.status == "watched":
+                plex.scrobble(rk)
+            elif st.position_seconds and st.position_seconds >= _PROGRESS_MIN_S:
+                dur = int((plex.metadata(rk) or {}).get("duration") or 0)
+                plex.set_timeline(rk, st.position_seconds * 1000, dur)
+            else:
+                plex.unscrobble(rk)
+            st.synced_to_plex = True
+            n += 1
+        except PlexError as e:
+            log.warning(
+                "Plex dirty re-push failed (user %s, item %s): %s", link.user_id, st.item_id, e
+            )
+    return n
+
+
+def run_plex_watch_sync(db: Session) -> dict:
+    """Scheduler job (default 30 min) — the cheap incremental pass. Pulls Plex-side changes since the
+    high-water mark (watch history + on-deck, filtered to the owner account) into Siftlode under
+    last-write-wins, then re-pushes any still-unsynced local states. Does NOT propagate un-watches
+    (the history feed only carries positive views) — that's `run_plex_watch_reconcile`'s job."""
+    if not sysconfig.get_bool(db, "plex_enabled"):
+        return {"skipped": "disabled"}
+    total = {"links": 0, "watched": 0, "in_progress": 0, "repushed": 0}
+    for link in _active_sync_links(db):
+        total["links"] += 1
+        try:
+            with PlexClient(db) as plex:
+                acct = _resolve_account_id(db, plex, link)
+                since = int(link.last_watch_sync_at.timestamp()) if link.last_watch_sync_at else 0
+                now = datetime.now(timezone.utc)
+                rk_to_id = {rk: iid for rk, iid in db.query(PlexItem.rating_key, PlexItem.id)}
+
+                for row in plex.watch_history(min_viewed_at=since, account_id=acct):
+                    iid = rk_to_id.get(str(row.get("ratingKey") or ""))
+                    if iid is None:
+                        continue
+                    ch = _pull_apply(db, link.user_id, iid, "watched", 0, _epoch(row.get("viewedAt")))
+                    if ch:
+                        total[ch] += 1
+
+                for row in plex.on_deck():
+                    iid = rk_to_id.get(str(row.get("ratingKey") or ""))
+                    if iid is None:
+                        continue
+                    off = int(row.get("viewOffset") or 0) // 1000
+                    if off < _PROGRESS_MIN_S:
+                        continue
+                    ch = _pull_apply(db, link.user_id, iid, "new", off, _epoch(row.get("lastViewedAt")))
+                    if ch:
+                        total[ch] += 1
+
+                total["repushed"] += _repush_dirty(db, plex, link)
+                link.last_watch_sync_at = now
+                db.commit()
+        except PlexNotConfigured as e:
+            return {"skipped": str(e)}
+        except PlexError as e:
+            db.rollback()
+            log.warning("Plex watch sync failed (user %s): %s", link.user_id, e)
+    log.info("Plex watch sync: %s", total)
+    return total
+
+
+def run_plex_watch_reconcile(db: Session) -> dict:
+    """Scheduler job (default daily) — the authoritative full reconcile. Rescans every section and,
+    per item, uses `synced_to_plex` to settle what the incremental feed can't:
+      - Plex has a state, Siftlode differs → last-write-wins (Plex usually the newer here).
+      - Siftlode has a state, Plex has NO record:
+          · synced_to_plex=True  → Plex un-watched it since our last sync → clear it (un-watch).
+          · synced_to_plex=False → a local change that never reached Plex → re-push it (via
+            `_repush_dirty`), never dropped.
+      - `hidden` rows are Siftlode-only and never cleared.
+    Union-preserving: a never-synced Siftlode-only state is pushed up, not lost."""
+    if not sysconfig.get_bool(db, "plex_enabled"):
+        return {"skipped": "disabled"}
+    total = {"links": 0, "applied": 0, "cleared": 0, "repushed": 0, "scanned": 0}
+    for link in _active_sync_links(db):
+        total["links"] += 1
+        try:
+            with PlexClient(db) as plex:
+                _resolve_account_id(db, plex, link)
+                plex_states, scanned = _scan_plex_states(db, plex)
+                total["scanned"] += scanned
+
+                for item_id, (status, pos, watched_at, prog_at) in plex_states.items():
+                    ch = _pull_apply(db, link.user_id, item_id, status, pos, watched_at or prog_at)
+                    if ch:
+                        total["applied"] += 1
+
+                # Siftlode rows Plex no longer has a record for: clear the ones that were mirrored
+                # (Plex un-watched them); leave hidden + dirty (dirty gets re-pushed just below).
+                for st in db.query(PlexState).filter_by(user_id=link.user_id).all():
+                    if st.item_id in plex_states or st.status == "hidden":
+                        continue
+                    if st.synced_to_plex:
+                        db.delete(st)
+                        total["cleared"] += 1
+
+                total["repushed"] += _repush_dirty(db, plex, link)
+                link.last_watch_sync_at = datetime.now(timezone.utc)
+                db.commit()
+        except PlexNotConfigured as e:
+            return {"skipped": str(e)}
+        except PlexError as e:
+            db.rollback()
+            log.warning("Plex watch reconcile failed (user %s): %s", link.user_id, e)
+    log.info("Plex watch reconcile: %s", total)
+    return total

backend/app/routes/plex.py

@@ -16,7 +16,7 @@ from pathlib import Path
 from urllib.parse import quote
 
 import httpx
-from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, Request, Response
 from fastapi.responses import FileResponse
 from sqlalchemy import and_, func, or_, text
 from sqlalchemy.orm import Session, aliased
@@ -1486,22 +1486,49 @@ def item_detail(
     }
 
 
+def _push_watch(
+    background: BackgroundTasks,
+    db: Session,
+    user_id: int,
+    item_id: int,
+    rating_key: str,
+    action: str,
+    time_ms: int = 0,
+    duration_ms: int = 0,
+) -> 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."""
+    if plex_watch.link_for_push(db, user_id) is None:
+        return
+    background.add_task(
+        plex_watch.push_state_to_plex, user_id, item_id, rating_key, action, time_ms, duration_ms
+    )
+
+
 @router.post("/item/{rating_key}/progress")
 def item_progress(
     rating_key: str,
     payload: dict,
+    background: BackgroundTasks,
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    """Checkpoint the resume position while playing (near the end → mark watched + clear it)."""
+    """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."""
     it = _item_or_404(db, rating_key)
     try:
         position = max(0, int(payload.get("position_seconds") or 0))
         duration = int(payload.get("duration_seconds") or 0)
     except (TypeError, ValueError):
         raise HTTPException(status_code=400, detail="position_seconds must be an integer")
+    final = bool(payload.get("final"))
     now = datetime.now(timezone.utc)
     st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first()
+    was_watched = st is not None and st.status == "watched"
 
     near_end = duration > 0 and position > duration - _FINISH_MARGIN_S
     if near_end:
@@ -1512,7 +1539,12 @@ def item_progress(
         st.watched_at = now
         st.position_seconds = 0
         st.progress_updated_at = now
+        st.synced_to_plex = False
         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).
+        if not was_watched:
+            _push_watch(background, db, user.id, it.id, rating_key, "watched")
         return {"status": "watched", "position_seconds": 0}
 
     if position < _PROGRESS_MIN_S:
@@ -1529,7 +1561,13 @@ def item_progress(
         db.add(st)
     st.position_seconds = position
     st.progress_updated_at = now
+    if final:
+        st.synced_to_plex = False
     db.commit()
+    if final:
+        _push_watch(
+            background, db, user.id, it.id, rating_key, "resume", position * 1000, duration * 1000
+        )
     return {"position_seconds": position}
 
 
@@ -1537,19 +1575,25 @@ def item_progress(
 def item_state(
     rating_key: str,
     payload: dict,
+    background: BackgroundTasks,
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    """Set the per-user watch status (new|watched|hidden)."""
+    """Set the per-user watch status (new|watched|hidden). watched→scrobble / →new→unscrobble push
+    to a linked Plex account immediately; `hidden` is a Siftlode-only concept with no Plex analogue,
+    so it never touches Plex."""
     it = _item_or_404(db, rating_key)
     status = payload.get("status")
     if status not in ("new", "watched", "hidden"):
         raise HTTPException(status_code=400, detail="Invalid status")
     st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first()
+    prev_status = st.status if st is not None else "new"
     if status == "new":
         if st is not None:
             db.delete(st)
             db.commit()
+        if prev_status != "new":  # something → unwatched: mirror the un-mark to Plex
+            _push_watch(background, db, user.id, it.id, rating_key, "unwatched")
         return {"status": "new"}
     if st is None:
         st = PlexState(user_id=user.id, item_id=it.id)
@@ -1558,7 +1602,10 @@ def item_state(
     if status == "watched":
         st.watched_at = datetime.now(timezone.utc)
         st.position_seconds = 0
+    st.synced_to_plex = False
     db.commit()
+    if status == "watched" and prev_status != "watched":
+        _push_watch(background, db, user.id, it.id, rating_key, "watched")
     return {"status": status}
 
 

backend/app/scheduler.py

@@ -15,6 +15,7 @@ from app.db import SessionLocal
 from app.downloads.gc import run_download_gc
 from app.models import SchedulerSetting
 from app.plex.sync import sync as run_plex_sync
+from app.plex.watch_sync import run_plex_watch_reconcile, run_plex_watch_sync
 from app.notifications import create_notification
 from app.state import is_sync_paused
 from app.sync.autotag import run_autotag_all
@@ -63,6 +64,8 @@ JOB_INTERVALS: dict[str, int] = {
     "explore_cleanup": settings.explore_cleanup_minutes,
     "download_gc": settings.download_gc_minutes,
     "plex_sync": settings.plex_sync_interval_min,
+    "plex_watch_sync": settings.plex_watch_sync_interval_min,
+    "plex_watch_reconcile": settings.plex_watch_reconcile_interval_min,
 }
 
 # Sane bounds for an admin-set interval (minutes).
@@ -292,6 +295,18 @@ def _plex_sync_job() -> None:
     _job("plex_sync", run_plex_sync)
 
 
+def _plex_watch_sync_job() -> None:
+    # Incremental Plex→Siftlode watch-state pull (history + on-deck) + re-push of unsynced local
+    # states. Cheap; a no-op when Plex is disabled or no owner link has sync enabled.
+    _job("plex_watch_sync", run_plex_watch_sync)
+
+
+def _plex_watch_reconcile_job() -> None:
+    # Full watch-state reconcile (whole-section rescan) — catches un-watches the incremental feed
+    # can't. Heavier, so it runs far less often (daily default).
+    _job("plex_watch_reconcile", run_plex_watch_reconcile)
+
+
 # job_id -> wrapper. The single source of truth for which jobs exist and how to run one,
 # shared by start_scheduler (recurring registration) and trigger_job (manual "run now").
 JOB_FUNCS: dict[str, Callable[[], None]] = {
@@ -307,6 +322,8 @@ JOB_FUNCS: dict[str, Callable[[], None]] = {
     "explore_cleanup": _explore_cleanup_job,
     "download_gc": _download_gc_job,
     "plex_sync": _plex_sync_job,
+    "plex_watch_sync": _plex_watch_sync_job,
+    "plex_watch_reconcile": _plex_watch_reconcile_job,
 }
 
 

frontend/src/components/PlexPlayer.tsx

@@ -452,23 +452,31 @@ export default function PlexPlayer({ itemId, onClose, queue }: Props) {
   const absRef = useRef(0);
   absRef.current = abs;
   useEffect(() => {
-    const flush = () => {
+    // `final` = a settled checkpoint (pause / page-hide / leaving the player) vs. the periodic tick.
+    // Only final checkpoints get mirrored to a linked Plex account (backend-side debounce), so Plex
+    // sees the resume position when the user actually stops — not a timeline call every 10 seconds.
+    const flush = (final = false) => {
       if (absRef.current > 0 && durationRef.current > 0)
-        api.plexProgress(id, Math.floor(absRef.current), durationRef.current).catch(() => {});
+        api.plexProgress(id, Math.floor(absRef.current), durationRef.current, final).catch(() => {});
     };
     const iv = window.setInterval(flush, 10000);
+    // Pausing is a settled position — push it to Plex now rather than waiting for unmount/unload.
+    const onPauseFlush = () => flush(true);
+    const video = videoRef.current;
+    video?.addEventListener("pause", onPauseFlush);
     // 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.
+    // keepalive beacon so the POST survives the unload (a final checkpoint — the user is leaving).
     const onHide = () => {
       if (absRef.current > 0 && durationRef.current > 0)
-        api.plexProgressBeacon(id, Math.floor(absRef.current), durationRef.current);
+        api.plexProgressBeacon(id, Math.floor(absRef.current), durationRef.current, true);
     };
     window.addEventListener("pagehide", onHide);
     return () => {
       window.clearInterval(iv);
       window.removeEventListener("pagehide", onHide);
-      flush();
+      video?.removeEventListener("pause", onPauseFlush);
+      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
       // watch — so the resume landed at the old spot (the progress GET could also race the POST).
@@ -558,6 +566,15 @@ export default function PlexPlayer({ itemId, onClose, queue }: Props) {
   const go = useCallback(
     (otherId: string | null | undefined) => {
       if (otherId) {
+        // Stop the outgoing media BEFORE switching. If the next item fails to load (e.g. no local
+        // file), loadSession returns without replacing the <video> source, leaving the old episode
+        // still playing — it would then fire 'ended' against the NEW id and mark the wrong item
+        // watched (and scrobble it to Plex). Pausing here severs that stale-tail race.
+        try {
+          videoRef.current?.pause();
+        } catch {
+          /* ignore */
+        }
         setAbs(0);
         setId(otherId);
       }
@@ -702,6 +719,9 @@ export default function PlexPlayer({ itemId, onClose, queue }: Props) {
     const video = videoRef.current;
     if (!video) return;
     const onEnded = () => {
+      // Only a genuine play-to-the-end counts. A next episode that failed to load never played
+      // (absRef 0) — its stray 'ended' must not mark anything watched or cascade another advance.
+      if (absRef.current <= 0) return;
       api.plexSetState(id, "watched").catch(() => {});
       if (nextId) go(nextId);
     };
@@ -837,10 +857,14 @@ export default function PlexPlayer({ itemId, onClose, queue }: Props) {
   const doSkip = useCallback(
     (m: PlexMarker) => {
       setSkipProgress(null);
-      if (m.type === "credits" && nextId) go(nextId);
+      if (m.type === "credits" && nextId) {
-      else seekTo(m.end_s);
+        // Binge auto-advance: the episode we're leaving was watched through the credits, so mark IT
+        // (the current id, captured here — not the reactive id, which go() is about to change).
+        api.plexSetState(id, "watched").catch(() => {});
+        go(nextId);
+      } else seekTo(m.end_s);
     },
-    [go, nextId, seekTo],
+    [id, go, nextId, seekTo],
   );
   const cancelAutoSkip = useCallback(() => {
     const m = activeMarkerRef.current;

frontend/src/i18n/locales/de/scheduler.json

@@ -69,7 +69,9 @@
     "maintenance": "Findet Videos, die nicht mehr abspielbar sind (gelöscht, auf privat gesetzt, abgebrochene Premieren), blendet sie aus dem Feed aus und entfernt sie nach einer Schonfrist; prüft die am längsten nicht geprüften Videos erneut. Fällt er aus, bleiben tote Videos im Katalog.",
     "explore_cleanup": "Entfernt Kanäle, die du erkundet, aber nie abonniert hast (und ihre Videos), nach einer Karenzzeit, damit Neugier-Stöbern den Katalog nicht überfüllt. Wenn es stoppt, bleiben verlassene erkundete Kanäle bestehen.",
     "download_gc": "Löscht heruntergeladene Dateien nach Ablauf der Aufbewahrungsfrist, gibt nicht mehr belegten Speicher frei und warnt vor dem Ablauf einer geteilten Datei. Fällt er aus, häufen sich alte Downloads und Speicherplatz wird nicht freigegeben.",
-    "plex_sync": "Spiegelt die aktivierten Plex-Bibliotheken in den App-Katalog — die Metadaten hinter Plex-Stöbern, Suche, Filtern und Info-Seiten (Genres, Besetzung, Wertungen, Artwork). Fällt er aus, erscheinen neu hinzugefügte oder geänderte Plex-Titel erst beim nächsten Sync."
+    "plex_sync": "Spiegelt die aktivierten Plex-Bibliotheken in den App-Katalog — die Metadaten hinter Plex-Stöbern, Suche, Filtern und Info-Seiten (Genres, Besetzung, Wertungen, Artwork). Fällt er aus, erscheinen neu hinzugefügte oder geänderte Plex-Titel erst beim nächsten Sync.",
+    "plex_watch_sync": "Holt kürzliche Plex-Wiedergabeänderungen (beendete Episoden, Fortsetzungspositionen) nach Siftlode und schiebt lokale Wiedergabeänderungen nach, die Plex nicht erreicht haben. Der günstige Zwei-Wege-Abgleich zwischen den vollen Reconciles. Fällt er aus, erscheinen in der Plex-App gesetzte Gesehen-/Fortsetzungsstände hier später.",
+    "plex_watch_reconcile": "Ein vollständiger Wiedergabe-Abgleich, der jede Bibliothek neu scannt — erfasst, was der inkrementelle Lauf nicht kann, vor allem eine auf Plex-Seite zurückgenommene Gesehen-Markierung. Schwerer, läuft daher etwa täglich. Fällt er aus, werden in Plex vorgenommene Un-Watches nicht zurückgespiegelt."
   },
   "jobs": {
     "rss_poll": "RSS-Abfrage (neue Uploads)",
@@ -83,7 +85,9 @@
     "demo_reset": "Demo-Konto-Reset",
     "explore_cleanup": "Bereinigung erkundeter Kanäle",
     "download_gc": "Download-Bereinigung",
-    "plex_sync": "Plex-Bibliothek-Sync"
+    "plex_sync": "Plex-Bibliothek-Sync",
+    "plex_watch_sync": "Plex-Wiedergabe-Sync (inkrementell)",
+    "plex_watch_reconcile": "Plex-Wiedergabe-Abgleich (vollständig)"
   },
   "queue": {
     "recentPending": "Zu synchronisierende Kanäle",

frontend/src/i18n/locales/en/scheduler.json

@@ -69,7 +69,9 @@
     "maintenance": "Finds videos that can no longer be played (deleted, made private, abandoned premieres), hides them from the feed and removes them after a grace period; re-checks the least-recently-checked videos. If it stops, dead videos linger in the catalogue.",
     "explore_cleanup": "Removes channels you explored but never subscribed to (and their videos) after a grace period, so curiosity browsing doesn't pile up in the catalogue. If it stops, abandoned explored channels linger.",
     "download_gc": "Deletes downloaded files past their retention window, reclaims space no one is holding anymore, and warns owners before a shared file expires. If it stops, old downloads pile up and disk space isn't freed.",
-    "plex_sync": "Mirrors the enabled Plex libraries into the app's catalogue — the metadata behind Plex browsing, search, filters and the info pages (genres, cast, ratings, artwork). If it stops, newly added or changed Plex titles don't appear until the next sync."
+    "plex_sync": "Mirrors the enabled Plex libraries into the app's catalogue — the metadata behind Plex browsing, search, filters and the info pages (genres, cast, ratings, artwork). If it stops, newly added or changed Plex titles don't appear until the next sync.",
+    "plex_watch_sync": "Pulls recent Plex watch changes (finished episodes, resume positions) into Siftlode and re-pushes any local watch changes that didn't reach Plex. The cheap two-way keep-in-sync between full reconciles. If it stops, watched/resume made in the Plex app takes longer to show here.",
+    "plex_watch_reconcile": "A full watch-state reconcile that rescans every library — catches what the incremental pass can't, notably an episode un-marked as watched on the Plex side. Heavier, so it runs about once a day. If it stops, un-watches done in Plex aren't mirrored back."
   },
   "jobs": {
     "rss_poll": "RSS poll (new uploads)",
@@ -83,7 +85,9 @@
     "demo_reset": "Demo account reset",
     "explore_cleanup": "Explored-channel cleanup",
     "download_gc": "Download cleanup",
-    "plex_sync": "Plex library sync"
+    "plex_sync": "Plex library sync",
+    "plex_watch_sync": "Plex watch sync (incremental)",
+    "plex_watch_reconcile": "Plex watch reconcile (full)"
   },
   "queue": {
     "recentPending": "Channels to sync",

frontend/src/i18n/locales/hu/scheduler.json

@@ -69,7 +69,9 @@
     "maintenance": "Megkeresi a már sehol le nem játszható videókat (törölt, priváttá tett, elhagyott premier), elrejti őket a hírfolyamból, és türelmi idő után eltávolítja; újraellenőrzi a legrégebben ellenőrzött videókat. Ha leáll, a halott videók a katalógusban maradnak.",
     "explore_cleanup": "A türelmi idő után eltávolítja a felfedezett, de nem követett csatornákat (és videóikat), hogy a kíváncsiság-böngészés ne halmozódjon a katalógusban. Ha leáll, az elhagyott felfedezett csatornák megmaradnak.",
     "download_gc": "Törli a megőrzési időn túli letöltéseket, felszabadítja a már senki által nem használt helyet, és a lejárat előtt figyelmezteti a tulajdonost. Ha leáll, a régi letöltések felhalmozódnak, és a tárhely nem szabadul fel.",
-    "plex_sync": "Tükrözi az engedélyezett Plex-könyvtárakat az app katalógusába — ez adja a Plex böngészés, keresés, szűrők és infó-oldalak metaadatait (műfaj, szereplők, pontszám, borítók). Ha leáll, az újonnan hozzáadott vagy módosított Plex-címek csak a következő sync után jelennek meg."
+    "plex_sync": "Tükrözi az engedélyezett Plex-könyvtárakat az app katalógusába — ez adja a Plex böngészés, keresés, szűrők és infó-oldalak metaadatait (műfaj, szereplők, pontszám, borítók). Ha leáll, az újonnan hozzáadott vagy módosított Plex-címek csak a következő sync után jelennek meg.",
+    "plex_watch_sync": "Áthozza a Plex-oldali friss megtekintés-változásokat (befejezett epizódok, folytatás-pozíciók) a Siftlode-ba, és felnyomja a Plexet el nem érő helyi változásokat. Ez az olcsó kétirányú szinkron a teljes reconcile-ok között. Ha leáll, a Plex-appban jelölt megtekintés/folytatás lassabban jelenik meg itt.",
+    "plex_watch_reconcile": "Teljes megtekintés-állapot reconcile, ami minden könyvtárat újraszken­nel — elkapja, amit az inkrementális menet nem, főleg ha egy epizódról a Plex oldalán levették a megtekintve jelet. Nehezebb, ezért kb. naponta fut. Ha leáll, a Plexben végzett un-watch nem tükröződik vissza."
   },
   "jobs": {
     "rss_poll": "RSS-lekérdezés (új feltöltések)",
@@ -83,7 +85,9 @@
     "demo_reset": "Demo fiók visszaállítása",
     "explore_cleanup": "Felfedezett csatornák takarítása",
     "download_gc": "Letöltések takarítása",
-    "plex_sync": "Plex-könyvtár szinkron"
+    "plex_sync": "Plex-könyvtár szinkron",
+    "plex_watch_sync": "Plex megtekintés-szinkron (inkrementális)",
+    "plex_watch_reconcile": "Plex megtekintés-reconcile (teljes)"
   },
   "queue": {
     "recentPending": "Szinkronra váró csatorna",

frontend/src/lib/api.ts

@@ -1244,16 +1244,29 @@ export const api = {
   // cue times onto the HLS session's zero-based clock so subtitles line up with speech after a seek.
   plexSubtitleUrl: (id: string, ord: number, offset = 0): string =>
     `/api/plex/subtitle/${encodeURIComponent(id)}/${ord}${offset ? `?offset=${offset}` : ""}`,
-  plexProgress: (id: string, position_seconds: number, duration_seconds: number): Promise<unknown> =>
+  // `final` marks a settled checkpoint (pause / page-hide / leaving the player) as opposed to the
+  // periodic 10s tick — the backend mirrors the resume position to a linked Plex account only on a
+  // final checkpoint, so a single watch doesn't spray Plex with timeline updates.
+  plexProgress: (
+    id: string,
+    position_seconds: number,
+    duration_seconds: number,
+    final = false,
+  ): Promise<unknown> =>
     req(`/api/plex/item/${encodeURIComponent(id)}/progress`, {
       method: "POST",
-      body: JSON.stringify({ position_seconds, duration_seconds }),
+      body: JSON.stringify({ position_seconds, duration_seconds, final }),
     }),
   // Fire-and-forget progress save that SURVIVES page unload (F5/close/navigate): a normal fetch is
   // cancelled on unload and React effect cleanup doesn't run on a full reload, so the last position
   // (esp. right after a seek) would be lost. `keepalive` lets the POST complete during unload; we
   // replicate req()'s credentials + account header (no retry — there's no page left to retry on).
-  plexProgressBeacon: (id: string, position_seconds: number, duration_seconds: number): void => {
+  plexProgressBeacon: (
+    id: string,
+    position_seconds: number,
+    duration_seconds: number,
+    final = false,
+  ): void => {
     const active = getActiveAccount();
     try {
       void fetch(`/api/plex/item/${encodeURIComponent(id)}/progress`, {
@@ -1264,7 +1277,7 @@ export const api = {
           "Content-Type": "application/json",
           ...(active != null ? { [ACTIVE_ACCOUNT_HEADER]: String(active) } : {}),
         },
-        body: JSON.stringify({ position_seconds, duration_seconds }),
+        body: JSON.stringify({ position_seconds, duration_seconds, final }),
       }).catch(() => {});
     } catch {
       /* ignore */

frontend/src/lib/releaseNotes.ts

@@ -14,6 +14,17 @@ export interface ReleaseEntry {
 }
 
 export const RELEASE_NOTES: ReleaseEntry[] = [
+  {
+    version: "0.35.0",
+    date: "2026-07-10",
+    summary: "Two-way Plex watch sync: watch in either app and both stay in step.",
+    features: [
+      "Two-way Plex watch sync is complete. Mark something watched — or leave it half-finished — in Siftlode and it now updates Plex too; and anything you watch (or un-mark) in the Plex app flows back to Siftlode automatically. Turn it on in Settings → Account. (Your private “hidden” marks stay in Siftlode and are never sent to Plex.)",
+    ],
+    fixes: [
+      "Auto-advancing to the next episode now marks the episode you actually finished — not the next one. Previously, if the next episode wasn't available locally, the wrong episode could be marked watched.",
+    ],
+  },
   {
     version: "0.34.2",
     date: "2026-07-09",