refactor(e5): deferred review-tail (backend) — enrich off request path, reuse, tests

Address the deep-review's deferred backend items so E5 is review-complete:

- person-enrich OFF the request path (ABUSE finding): person_detail no longer makes
  the TMDB call inline (even bounded to 8s it sat on the serve thread, un-throttled,
  demo-reachable — it violated the module's "no live TMDB call at serve time"
  invariant). It now schedules a FastAPI BackgroundTask (people._enrich_person_bg,
  fresh session, best-effort) after the response; the first open serves the row as-is
  and the bio lands by the next open. The inline try/rollback recovery is gone with it.

- reuse: a shared tmdb.resolve_library_guids owns the collision-safe guid→library
  bucketing (movie vs tv ids are independent sequences) that resolve_recs and
  people._resolve_known_for each hand-duplicated; a new rollup.item_state_map dedups
  the "PlexState-by-item_id" fetch across the person + discover paths.

- test gaps: enrich_batch transient TmdbError leaves the watermark NULL to retry;
  guid_to_tmdb_id resolves a tvdb source + tv media_type via /find's tv_results;
  _continue_watching includes a NULL-status started movie and excludes a hidden one;
  /api/search/youtube returns 422 (not 5xx) when ingest fails (route-level, B-core-1).

Backend lane 403 → 413, ruff clean. No migration; serve contracts unchanged.

backend/app/routes/plex/discover.py

@@ -37,19 +37,8 @@ def _scope_libs(db: Session, scope: str) -> tuple[list[int], list[int]]:
     )
 
 
-def _movie_states(db: Session, uid: int, items: list[PlexItem]) -> dict[int, PlexState]:
-    """One batched fetch of the user's watch-state for a set of movies (avoids a per-card query)."""
-    ids = [it.id for it in items]
-    if not ids:
-        return {}
-    rows = db.execute(
-        select(PlexState).where(PlexState.user_id == uid, PlexState.item_id.in_(ids))
-    ).scalars().all()
-    return {r.item_id: r for r in rows}
-
-
 def _movie_cards(db: Session, uid: int, items: list[PlexItem]) -> list[dict]:
-    sts = _movie_states(db, uid, items)
+    sts = rollup.item_state_map(db, uid, [it.id for it in items])
     return [movie_card(it, sts.get(it.id)) for it in items]
 
 

backend/app/routes/plex/people.py

@@ -2,17 +2,16 @@
 
 import logging
 
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, BackgroundTasks, Depends
 from sqlalchemy import func, or_, select
 from sqlalchemy.orm import Session
 
 from app.auth import current_user
-from app.db import get_db
+from app.db import SessionLocal, get_db
 from app.models import (
     PlexItem,
     PlexPerson,
     PlexShow,
-    PlexState,
     User,
 )
 from app.plex import rollup
@@ -36,44 +35,13 @@ def _resolve_known_for(
     popularity order. Matched by the canonical guid `tmdb:<id>`; out-of-library credits are dropped."""
     if not known_for:
         return []
-    # Bucket by media_type: TMDB movie ids and tv ids collide numerically, so a movie credit must only
-    # match a library movie and a tv credit only a library show (mirrors tmdb.resolve_recs). Matching
-    # blind to media_type would render a tv credit as an unrelated movie whose tmdb id happens to match.
-    movie_guids = [
-        f"tmdb:{k['tmdb_id']}" for k in known_for if k.get("tmdb_id") and k.get("media_type") != "tv"
-    ]
-    tv_guids = [
-        f"tmdb:{k['tmdb_id']}" for k in known_for if k.get("tmdb_id") and k.get("media_type") == "tv"
-    ]
-    if not movie_guids and not tv_guids:
-        return []
-    movies: dict[str, PlexItem] = {}
-    if movie_lib_ids and movie_guids:
-        for m in db.execute(
-            select(PlexItem).where(
-                PlexItem.kind == "movie",
-                PlexItem.library_id.in_(movie_lib_ids),
-                PlexItem.guid.in_(movie_guids),
-            )
-        ).scalars():
-            movies.setdefault(m.guid, m)
-    shows: dict[str, PlexShow] = {}
-    if show_lib_ids and tv_guids:
-        for s in db.execute(
-            select(PlexShow).where(
-                PlexShow.library_id.in_(show_lib_ids), PlexShow.guid.in_(tv_guids)
-            )
-        ).scalars():
-            shows.setdefault(s.guid, s)
-    m_states = {
-        s.item_id: s
-        for s in db.execute(
-            select(PlexState).where(
-                PlexState.user_id == uid,
-                PlexState.item_id.in_([m.id for m in movies.values()] or [0]),
-            )
-        ).scalars()
-    }
+    # Match each credit to a library row by guid, bucketed by media_type (movie ids and tv ids collide
+    # numerically) and scoped to the enabled libraries — the shared helper owns that collision-safe
+    # logic (the detail-page recs use the same one).
+    movies, shows = plex_tmdb.resolve_library_guids(
+        db, known_for, movie_lib_ids=movie_lib_ids, show_lib_ids=show_lib_ids
+    )
+    m_states = rollup.item_state_map(db, uid, [m.id for m in movies.values()])
     show_statuses = rollup.state_map(db, uid, [s.id for s in shows.values()])
     out: list[dict] = []
     for k in known_for:
@@ -144,9 +112,28 @@ def people_cards(
 _PERSON_SECTION_MAX = 200
 
 
+def _enrich_person_bg(person_id: int) -> None:
+    """Fetch + persist one person's TMDB bio OFF the request path — a FastAPI background task that runs
+    AFTER the response is sent, so the outbound TMDB call never sits on the serve latency (the module's
+    "no live TMDB call at serve time" invariant). The request's session is already closed by then, so
+    open a fresh one. Best-effort: a TMDB failure just leaves `tmdb_enriched_at` NULL to retry on the
+    next open, and the bio lands from the row by then."""
+    db = SessionLocal()
+    try:
+        person = db.get(PlexPerson, person_id)
+        if person is not None:
+            plex_tmdb.enrich_person(db, person)
+    except Exception:  # noqa: BLE001 — background best-effort; never surfaces to a user
+        db.rollback()
+        log.warning("background person bio enrich failed for id=%s", person_id, exc_info=True)
+    finally:
+        db.close()
+
+
 @router.get("/person/{name}")
 def person_detail(
     name: str,
+    background_tasks: BackgroundTasks,
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
@@ -158,18 +145,13 @@ def person_detail(
 
     E5 S6: the first time a page is opened, the person's TMDB bio + "Known for" credits are lazily
     fetched (from the S2-linked `tmdb_person_id`) and persisted, then served from the row on every open.
+    The fetch runs OFF the request path (a background task); the first open serves whatever's in the row
+    (bio possibly still null) and the bio lands from the row by the next open.
     """
     person = db.execute(select(PlexPerson).where(PlexPerson.name == name)).scalars().first()
-    # Lazy one-time bio enrich — best-effort, never fail the page if TMDB is down / the id is stale.
+    # Lazy one-time bio enrich, scheduled AFTER the response so the TMDB call is never on the serve path.
     if person is not None and person.tmdb_person_id and person.tmdb_enriched_at is None:
-        try:
-            plex_tmdb.enrich_person(db, person)
-        except Exception:  # noqa: BLE001 — enrichment is optional; the filmography still renders
-            # A failed enrich_person commit leaves the session in a needs-rollback state; without this
-            # the following SELECTs would raise PendingRollbackError and 500 the page (SQLAlchemy 2.x
-            # deactivates the transaction on a commit-time flush failure and does NOT auto-reset it).
-            db.rollback()
-            log.warning("person bio enrich failed for %s", name, exc_info=True)
+        background_tasks.add_task(_enrich_person_bg, person.id)
     movie_lib_ids, show_lib_ids = _content_libs(db)
 
     movies: list[dict] = []
@@ -187,16 +169,7 @@ def person_detail(
             .order_by(PlexItem.year.desc().nullslast(), func.lower(PlexItem.title))
             .limit(_PERSON_SECTION_MAX)
         ).scalars().all()
-        item_ids = [it.id for it in items]
-        states = {
-            s.item_id: s
-            for s in db.execute(
-                select(PlexState).where(
-                    PlexState.user_id == user.id,
-                    PlexState.item_id.in_(item_ids or [0]),
-                )
-            ).scalars()
-        }
+        states = rollup.item_state_map(db, user.id, [it.id for it in items])
         movies = [movie_card(it, states.get(it.id)) for it in items]
 
     shows: list[dict] = []