refactor(plex): dedup movie/show watch-state predicates into query.py (E3 S1)

The `show` scope (all|watched|in_progress|unwatched) filter and the status CASE
were hand-duplicated between unified_library (the grid) and facets.wsq — the
exact drift risk the review flagged (a hidden movie kept out of the grid must
not inflate facets). Extract them into four shared query.py helpers
(movie_status_case, show_status_case, apply_movie_state_filter,
apply_show_state_filter), model-agnostic (callers pass the joined state
expressions), and point all three call sites at them.

Pure behaviour-equivalence refactor: browse.py -67 lines, no query changes.
Green under the E3 S1 parity net + query_conversion/facets/field-scoped/lang
tests (348 DB-lane pass, ruff clean).

backend/app/plex/query.py

@@ -6,7 +6,7 @@ Pure query fragments — every function returns SQLAlchemy expressions or a para
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 
-from sqlalchemy import and_, func, or_, type_coerce
+from sqlalchemy import and_, case, func, or_, type_coerce
 from sqlalchemy.dialects.postgresql import JSONB
 
 TS_CONFIG = "public.unaccent_simple"
@@ -189,3 +189,62 @@ class LibraryFilters:
 
 def like_escape(s: str) -> str:
     return s.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+
+
+# --- Watch-state scope (shared by the /library grid AND the /facets aggregation) ---------------
+#
+# The `show` scope (all | watched | in_progress | unwatched) filters the browse result by per-user
+# watch state. The SAME predicate has to be applied in two places — `unified_library` (the grid) and
+# `facets.wsq` (so the facet counts reflect exactly the visible set) — and they MUST stay identical:
+# a hidden movie kept out of the grid must not inflate the facets either. These helpers are the one
+# home for that predicate so the two call sites can't drift (E3 S1; pinned by test_plex_facets_grid_
+# parity). Callers pass the already-joined state expressions, so this module stays model-agnostic:
+#   * movies — `st` is an aliased PlexState outer-joined on (item_id, user_id).
+#   * shows  — `tot`/`wat`/`inp` are the coalesced totals from `rollup.agg_subquery`.
+
+
+def movie_status_case(st):
+    """The per-movie status label (new | in_progress | watched) from its PlexState alias `st`."""
+    return case(
+        (st.status == "watched", "watched"),
+        (
+            and_(st.position_seconds > 0, or_(st.status.is_(None), st.status != "watched")),
+            "in_progress",
+        ),
+        else_="new",
+    )
+
+
+def show_status_case(tot, wat, inp):
+    """The per-show status label rolled up from its episodes' totals (total/watched/in-progress)."""
+    return case(
+        (and_(tot > 0, wat >= tot), "watched"),
+        (or_(wat > 0, inp > 0), "in_progress"),
+        else_="new",
+    )
+
+
+def apply_movie_state_filter(query, st, show: str):
+    """Restrict a movie query to the `show` scope via its PlexState alias `st`. The default (`all`)
+    excludes hidden movies — the exclusion that must match between grid and facets."""
+    if show == "watched":
+        return query.where(st.status == "watched")
+    if show == "in_progress":
+        return query.where(
+            st.position_seconds > 0, or_(st.status.is_(None), st.status != "watched")
+        )
+    if show == "unwatched":
+        return query.where(or_(st.status.is_(None), st.status.notin_(["watched", "hidden"])))
+    return query.where(or_(st.status.is_(None), st.status != "hidden"))
+
+
+def apply_show_state_filter(query, tot, wat, inp, show: str):
+    """Restrict a show query to the `show` scope via its rollup totals. `all` is a no-op (shows have
+    no hidden concept); the caller only joins the rollup for the three non-`all` scopes."""
+    if show == "watched":
+        return query.where(tot > 0, wat >= tot)
+    if show == "in_progress":
+        return query.where(or_(wat > 0, inp > 0), wat < tot)
+    if show == "unwatched":
+        return query.where(wat == 0, inp == 0)
+    return query