fix(e1): address /code-review high findings on the E1 stack

Review round on the whole E1 stack (main..HEAD). 6 findings, all low/medium,
none blocking; fixed for a clean gate-green handoff.

- plex/sync (correctness): the un-enriched cast guard compared the stored cast
  against THIS listing's length, so a trimmed Plex cast on a not-yet-enriched
  row was never applied. Compare against the listing CAP (_LISTING_CAST_CAP=20)
  instead: >cap = enriched (keep, survives a watermark reset), <=cap =
  listing-sourced (track edits in both directions). +2 regression tests.
- main (security): add X-Frame-Options: DENY as the legacy-browser fallback for
  the CSP frame-ancestors clickjacking guard.
- downloads/service (altitude): finalize_if_running now uses a SQLAlchemy 2.0
  update() construct instead of a hand-built text() UPDATE string — same atomic
  WHERE status='running', idiomatic, no identifier interpolation, no manual
  StrEnum coercion. Drops the now-unused text import.
- auth/admin (reuse): extract the SEC-10 "drop untrusted preset password + bump
  epoch" invariant into one helper (revoke_preset_password_if_unverified) so the
  Google-callback and admin-activation paths can't drift.
- tests/migrations (robustness): guarantee the shared schema is restored to head
  in a finally, so a downgrade/re-upgrade failure can't cascade into later tests.
- worker (simplification): align _finish_from_cache's display_name fill with the
  _download path's guard (no behavior change).

Gate green: ruff clean, 334 pytest (0 skip), frontend unchanged.

backend/app/downloads/service.py

@@ -8,7 +8,7 @@ the worker live in `worker.py` / the routes; this module owns enqueue + asset bo
 """
 from datetime import datetime, timezone
 
-from sqlalchemy import func, select, text
+from sqlalchemy import func, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -234,19 +234,21 @@ def finalize_if_running(db: Session, job_id: int, **fields) -> bool:
 
     Runs in the CALLER's session and does NOT commit — the caller commits it together with its asset
     writes in one transaction. It flushes first so any pending asset write takes its row lock BEFORE
-    this takes the job's, matching the cancel route's asset-then-job order (no deadlock). `fields` keys
-    are internal column names (never user input). Returns True when the row was still running and got
+    this takes the job's, matching the cancel route's asset-then-job order (no deadlock). `fields` are
+    mapped-column kwargs (never user input). Returns True when the row was still running and got
     updated, False when a route had already moved it out of 'running'."""
     if not fields:
         return False
     db.flush()  # asset-then-job lock order: push pending asset writes before we lock the job row
-    # StrEnum members carry through psycopg as strings, but coerce explicitly so a bound status value
-    # is always the plain literal the column stores.
-    params = {k: (v.value if isinstance(v, JobStatus) else v) for k, v in fields.items()}
-    assignments = ", ".join(f"{col} = :{col}" for col in fields)
+    # A Core UPDATE with the guard as a WHERE — same atomic "only while still running" write as the
+    # rest of the codebase's 2.0 statements, no hand-built SQL string. synchronize_session=False: this
+    # is a terminal write, so we don't pay to sync the (about-to-be-discarded) ORM identity map. Bound
+    # StrEnum values bind as their plain str literal (StrEnum is a str subclass).
     result = db.execute(
-        text(f"UPDATE download_jobs SET {assignments} WHERE id = :_job_id AND status = :_running"),
-        {**params, "_job_id": job_id, "_running": JobStatus.RUNNING.value},
+        update(DownloadJob)
+        .where(DownloadJob.id == job_id, DownloadJob.status == JobStatus.RUNNING.value)
+        .values(**fields)
+        .execution_options(synchronize_session=False)
     )
     return result.rowcount > 0
 

backend/app/main.py

@@ -164,13 +164,16 @@ async def static_cache_headers(request, call_next):
 # nosniff stops MIME-confusion, the Referrer-Policy keeps the (private) app URL from leaking to
 # external sites the user follows, `frame-ancestors 'none'` blocks click-jacking by refusing to be
 # iframed anywhere (we never embed our own pages; we embed OTHERS, which this does not affect), and
-# Permissions-Policy turns off powerful features the app never uses. A full Content-Security-Policy
+# Permissions-Policy turns off powerful features the app never uses. `X-Frame-Options: DENY` duplicates
+# the frame-ancestors intent for legacy engines that never implemented CSP frame-ancestors (belt and
+# suspenders — modern browsers use the CSP directive and ignore the header). A full Content-Security-Policy
 # (script/style/connect sources) is deferred — the SPA's inline styles + the YouTube/Plex embeds make
 # it real work — and HSTS belongs on the TLS-terminating edge (the Caddy front door), not here.
 _SECURITY_HEADERS = {
     "X-Content-Type-Options": "nosniff",
     "Referrer-Policy": "strict-origin-when-cross-origin",
     "Content-Security-Policy": "frame-ancestors 'none'",
+    "X-Frame-Options": "DENY",
     "Permissions-Policy": "camera=(), microphone=(), geolocation=()",
 }
 

backend/app/plex/sync.py

@@ -294,6 +294,11 @@ def _paginate(plex: PlexClient, key: str, item_type: int):
             break
 
 
+# The section listing exposes at most the top ~20 cast (Plex's own cap); a stored cast longer than this
+# can only have come from the rich enrich, never the listing.
+_LISTING_CAST_CAP = 20
+
+
 def _apply_facet_fields(row, meta: dict) -> None:
     """Set the filterable/orderable metadata columns shared by plex_items + plex_shows (identical
     column names) from the cheap section listing: rating, content_rating, studio, release date,
@@ -309,14 +314,14 @@ def _apply_facet_fields(row, meta: dict) -> None:
     # person-page cast filter for names past the top ~20. The watermark gate below (`people_enriched_at
     # is None`) normally protects this, but it is NOT enough on its own: a migration that resets the
     # watermark re-opens the regression (0065 nulled it, which is exactly how the S5b fix got undone in
-    # prod). So guard STRUCTURALLY too, by LENGTH: take the fresh listing UNLESS it is shorter than what
-    # is stored. A stored cast longer than the ~20-cap listing can only be an enriched fuller list, so
-    # keep it (robust to any watermark reset); at or below the cap the stored value is itself
-    # listing-sourced, so replacing it with the current listing stays self-correcting when Plex edits
-    # the cast (a pure union never would).
+    # prod). So guard STRUCTURALLY too, by LENGTH against the listing cap: a stored cast LONGER than the
+    # cap can only be an enriched fuller list, so keep it (robust to any watermark reset). A stored cast
+    # at or below the cap is itself listing-sourced, so take the fresh listing — which tracks Plex cast
+    # EDITS in EITHER direction (comparing against this listing's length instead would silently keep a
+    # stale longer cast when Plex TRIMS one).
     if row.people_enriched_at is None:
-        listing_cast = _tags(meta, "Role", limit=20)
-        if len(row.cast_names or []) <= len(listing_cast or []):
+        listing_cast = _tags(meta, "Role", limit=_LISTING_CAST_CAP)
+        if len(row.cast_names or []) <= _LISTING_CAST_CAP:
             row.cast_names = listing_cast
     people = list(dict.fromkeys((row.directors or []) + (row.cast_names or [])))
     row.people_text = " ".join(people) or None

backend/app/routes/admin.py

@@ -19,12 +19,12 @@ from app.auth import (
 )
 from app.db import get_db
 from app.demo import reset_demo_state
-from app.routes.auth import bump_session_epoch
 from app.models import (
     DemoWhitelist,
     Invite,
     User,
 )
+from app.routes.auth import revoke_preset_password_if_unverified
 from app.utils import iso, valid_email
 
 router = APIRouter(prefix="/api/admin", tags=["admin"])
@@ -64,12 +64,9 @@ def _activate_user_by_email(db: Session, email: str) -> None:
     user = db.execute(select(User).where(User.email == email.lower())).scalar_one_or_none()
     if user is not None and not user.is_active:
         # SEC-10: same guard as the Google-adoption path — approving an account whose email was never
-        # verified must not activate a password set on it by whoever registered the address (a
-        # pre-emption). Drop the untrusted password + revoke any session it minted; the owner re-sets
-        # one via password-reset. (No-op when email_verified is True, e.g. legit no-SMTP registrations.)
-        if not user.email_verified:
-            user.password_hash = None
-            bump_session_epoch(user)
+        # verified must not hand a working login to a password set on it by whoever registered the
+        # address (a pre-emption). Shared single-source helper so the two paths can't drift.
+        revoke_preset_password_if_unverified(user)
         user.is_active = True
         db.commit()
 

backend/app/routes/auth.py

@@ -286,14 +286,9 @@ async def callback(
                 return RedirectResponse(url="/?access=denied")
             user = existing
             user.google_sub = sub
-            # SEC-10: a password set on this account BEFORE its email was ever verified is untrusted —
-            # whoever registered the address (to pre-empt the real owner, who is only now arriving via
-            # Google) could have chosen it. Drop it now that Google attests ownership, and bump the epoch
-            # to revoke any session that pre-set password may have minted. The owner can set a fresh
-            # password via password-reset later if they also want email+password sign-in.
-            if not user.email_verified:
-                user.password_hash = None
-                bump_session_epoch(user)
+            # SEC-10: drop any password pre-set on this account while its email was still unverified,
+            # now that Google attests ownership (bumps the epoch to revoke any session it minted).
+            revoke_preset_password_if_unverified(user)
             # Google verified the email and is_allowed granted access, so finish activating a
             # previously pending password registration (otherwise current_user would bounce it).
             user.email_verified = True
@@ -817,6 +812,18 @@ def bump_session_epoch(user: User) -> None:
     user.session_epoch = (user.session_epoch or 0) + 1
 
 
+def revoke_preset_password_if_unverified(user: User) -> None:
+    """SEC-10 (account pre-emption): a password set on an account whose email was NEVER verified is
+    untrusted — whoever registered the address (to pre-empt the real owner) could have chosen it. When
+    ownership is finally attested (Google adoption, or admin approval), drop that password and bump the
+    epoch to revoke any session it minted; the owner re-sets one via password-reset. No-op for an
+    already-verified account, so a legit no-SMTP registration is unaffected. The single source for this
+    rule — both the Google-callback and admin-activation paths call it so they can't drift."""
+    if not user.email_verified:
+        user.password_hash = None
+        bump_session_epoch(user)
+
+
 @router.post("/logout-others")
 def logout_others(
     request: Request, user: User = Depends(current_user), db: Session = Depends(get_db)

backend/app/worker.py

@@ -484,9 +484,9 @@ def _finish_from_cache(job_id: int, asset_id: int) -> None:
         asset = db.get(MediaAsset, asset_id)
         if job and asset:
             fields = {"status": JobStatus.DONE, "progress": 100, "phase": None}
-            if not job.display_name:
+            if not job.display_name and asset.title:
                 # display_name is VARCHAR(255); a title can be longer (e.g. a recipe in the title).
-                fields["display_name"] = asset.title[:255] if asset.title else None
+                fields["display_name"] = asset.title[:255]
             asset.last_access_at = datetime.now(timezone.utc)
             # Atomic finalize: only marks the job done while it is still running, so a concurrent
             # cancel/pause is never resurrected (see service.finalize_if_running).

backend/tests/test_migrations.py

@@ -59,5 +59,10 @@ def test_head_migration_downgrades_and_reupgrades(db):
     # it lands, not during a rollback. Leaves the schema back at head for the rest of the session.
     cfg = Config("alembic.ini")
     command.downgrade(cfg, "-1")
+    try:
         command.upgrade(cfg, "head")
         assert _structural_drift() == []
+    finally:
+        # Always leave the SHARED schema at head, whatever the upgrade/assert above did — otherwise a
+        # failure here strands the DB one revision behind and cascades into every later DB-lane test.
+        command.upgrade(cfg, "head")

backend/tests/test_plex_detail.py

@@ -128,6 +128,35 @@ def test_apply_facet_fields_preserves_enriched_cast_names():
     assert "Diane Keaton" in enriched.people_text and "Francis Ford Coppola" in enriched.people_text
 
 
+def test_apply_facet_fields_keeps_over_cap_cast_after_a_watermark_reset():
+    """The enriched-cast guard must survive a watermark RESET (people_enriched_at nulled, e.g. by a
+    migration like 0065). A stored cast LONGER than the listing cap can only be an enriched fuller list,
+    so it's kept even though the row now reads as un-enriched — comparing against the cap, not the
+    current listing's length, is what makes this robust (regression guard for the review fix)."""
+    thin = {"Role": [{"tag": f"Actor {i}"} for i in range(20)]}  # a full 20-name listing
+    over_cap = SimpleNamespace(
+        people_enriched_at=None,  # watermark was reset
+        directors=None,
+        cast_names=[f"Enriched {i}" for i in range(22)],  # 22 > the 20 cap => enriched
+    )
+    _apply_facet_fields(over_cap, thin)
+    assert over_cap.cast_names == [f"Enriched {i}" for i in range(22)]  # kept, not downgraded
+
+
+def test_apply_facet_fields_tracks_a_trimmed_listing_when_unenriched():
+    """A listing-sourced cast (at or below the cap, never enriched) must TRACK Plex cast edits in both
+    directions. When Plex trims the cast, the shorter fresh listing replaces the stored one — the old
+    length-vs-this-listing guard would have wrongly kept the stale longer cast."""
+    stored = SimpleNamespace(
+        people_enriched_at=None,
+        directors=None,
+        cast_names=["A", "B", "C", "D", "E"],  # 5, at/below the cap => listing-sourced
+    )
+    trimmed = {"Role": [{"tag": "A"}, {"tag": "B"}]}  # Plex removed 3 cast members
+    _apply_facet_fields(stored, trimmed)
+    assert stored.cast_names == ["A", "B"]
+
+
 def test_enrich_people_batch_persists_detail(db):
     lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
     db.add(lib)