feat(plex): Phase C — incremental + full Plex↔Siftlode watch reconcile

Completes the two-way watch-state sync with two scheduler jobs:

- plex_watch_sync (default 30m): pull recent Plex-side changes (watch history + on-deck, filtered
  to the owner account) into Siftlode under last-write-wins (_pull_apply + a _same_state ping-pong
  guard + skew tolerance), then re-push any still-unsynced local states.
- plex_watch_reconcile (default daily): full section rescan; uses synced_to_plex to settle what the
  incremental feed can't — notably propagating a Plex-side un-watch (clear a previously-mirrored row
  Plex no longer has) — while re-pushing never-synced local states and never touching hidden
  (Siftlode-only) rows. Union-preserving.

PlexClient gains accounts/watch_history/on_deck; _scan_plex_states is factored out and shared with
the one-time import. Owner accountID is resolved once and cached on the link. Both jobs are
registered in the scheduler (pause-skip, activity tracking, run-now, admin-tunable intervals) with
trilingual (HU/EN/DE) labels + descriptions. New config default plex_watch_reconcile_interval_min.

Verified live against the real Plex server: read feeds, last-write-wins, dirty re-push, the
incremental job end-to-end, and a full reconcile that cleared exactly the one un-watched item with
zero collateral across 17976 scanned.

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