Merge: Plex watch-state sync Phase C (incremental + full reconcile, 2-way complete)

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

@@ -198,6 +198,41 @@ class PlexClient:
             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,7 +14,7 @@ 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
@@ -59,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
@@ -66,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():
-                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):
-                    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] = {
+            scanned_states, stats["scanned"] = _scan_plex_states(db, plex)
+            # 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 — it re-updates.
+            rows = []
+            for item_id, (status, pos, watched_at, prog_at) in scanned_states.items():
+                rows.append(
+                    {
                         "user_id": link.user_id,
                         "item_id": item_id,
                         "status": status,
@@ -110,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).
-            rows = list(rows_by_item.values())
+            # scanned) are never referenced, so they're preserved untouched (union).
             for i in range(0, len(rows), 1000):
                 stmt = pg_insert(PlexState).values(rows[i : i + 1000])
                 stmt = stmt.on_conflict_do_update(
@@ -195,3 +204,214 @@ def push_state_to_plex(
         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/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/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",