refactor(api): Pydantic request models for plex routes + status unification (R6 S1b)

All 14 plex payload:dict endpoints typed (watch-link, collections, native
playlists, item/show/season state & progress). Shared PlexItemKeysIn (bulk add/
remove/reorder) and WatchedIn (show/season). item_progress's manual int()+400
becomes a 422 on a non-number.

Status-code unification (ROADMAP S1):
- the 5 collection write-failure 502s → 422, matching the house "external write
  failed" convention (playlists' YouTube 422). The image/subtitle proxy-fetch 502s
  stay (genuine gateway fetches).
- the field-validation 422s ("title is required", "title and item_rating_key are
  required") → 400, matching every other module's required-field 400.

Ends S1b: no `payload: dict` remains in app/routes. Gate: ruff clean, pytest 164.

backend/app/routes/plex.py

@@ -21,6 +21,7 @@ from pathlib import Path
 import httpx
 from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, Request, Response
 from fastapi.responses import FileResponse
+from pydantic import BaseModel
 from sqlalchemy import Integer, String, and_, case, cast, func, literal, null, or_
 from sqlalchemy.orm import Session, aliased
 
@@ -156,9 +157,13 @@ def watch_link_status(
     return _link_dict(link)
 
 
+class WatchLinkIn(BaseModel):
+    enabled: bool = False
+
+
 @router.post("/watch/link")
 def watch_link_set(
-    payload: dict,
+    body: WatchLinkIn,
     user: User = Depends(admin_user),
     db: Session = Depends(get_db),
 ) -> dict:
@@ -167,7 +172,7 @@ def watch_link_set(
     "Plex is master" import immediately so the owner's existing Plex history shows up in Siftlode."""
     if not sysconfig.get_bool(db, "plex_enabled"):
         raise HTTPException(status_code=409, detail="The Plex module is disabled.")
-    enabled = bool(payload.get("enabled"))
+    enabled = body.enabled
     link = db.query(PlexLink).filter_by(user_id=user.id).first()
     if link is None:
         link = PlexLink(user_id=user.id, uses_admin=True, linked_at=datetime.now(timezone.utc))
@@ -797,14 +802,22 @@ def _editable_collection_or_403(db: Session, rating_key: str) -> PlexCollection:
     return col
 
 
+class CreateCollectionIn(BaseModel):
+    library: str = ""
+    title: str = ""
+    item_rating_key: str = ""
+
+
 @router.post("/collections")
-def create_collection(payload: dict, _: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict:
+def create_collection(
+    body: CreateCollectionIn, _: User = Depends(admin_user), db: Session = Depends(get_db)
+) -> dict:
     """Create a manual collection (seeded with one item — Plex needs at least one). Editable by default."""
-    library = str(payload.get("library") or "")
+    library = body.library
-    title = (payload.get("title") or "").strip()
+    title = body.title.strip()
-    seed = str(payload.get("item_rating_key") or "")
+    seed = body.item_rating_key
     if not title or not seed:
-        raise HTTPException(status_code=422, detail="title and item_rating_key are required")
+        raise HTTPException(status_code=400, detail="title and item_rating_key are required")
     lib = db.query(PlexLibrary).filter_by(plex_key=library).first()
     if lib is None:
         raise HTTPException(status_code=404, detail="Unknown library")
@@ -816,7 +829,7 @@ def create_collection(payload: dict, _: User = Depends(admin_user), db: Session
         db.commit()
     except (PlexError, PlexNotConfigured) as e:
         db.rollback()
-        raise HTTPException(status_code=502, detail=f"Plex write failed: {e}")
+        raise HTTPException(status_code=422, detail=f"Plex write failed: {e}")
     return _collection_card(col)
 
 
@@ -833,7 +846,7 @@ def collection_add_item(
         db.commit()
     except (PlexError, PlexNotConfigured) as e:
         db.rollback()
-        raise HTTPException(status_code=502, detail=f"Plex write failed: {e}")
+        raise HTTPException(status_code=422, detail=f"Plex write failed: {e}")
     return _collection_card(col)
 
 
@@ -850,18 +863,22 @@ def collection_remove_item(
         db.commit()
     except (PlexError, PlexNotConfigured) as e:
         db.rollback()
-        raise HTTPException(status_code=502, detail=f"Plex write failed: {e}")
+        raise HTTPException(status_code=422, detail=f"Plex write failed: {e}")
     return _collection_card(col)
 
 
+class CollectionRenameIn(BaseModel):
+    title: str = ""
+
+
 @router.patch("/collections/{rating_key}")
 def collection_rename(
-    rating_key: str, payload: dict, _: User = Depends(admin_user), db: Session = Depends(get_db)
+    rating_key: str, body: CollectionRenameIn, _: User = Depends(admin_user), db: Session = Depends(get_db)
 ) -> dict:
     col = _editable_collection_or_403(db, rating_key)
-    title = (payload.get("title") or "").strip()
+    title = body.title.strip()
     if not title:
-        raise HTTPException(status_code=422, detail="title is required")
+        raise HTTPException(status_code=400, detail="title is required")
     lib = db.get(PlexLibrary, col.library_id)
     try:
         with PlexClient(db) as plex:
@@ -870,7 +887,7 @@ def collection_rename(
         db.commit()
     except (PlexError, PlexNotConfigured) as e:
         db.rollback()
-        raise HTTPException(status_code=502, detail=f"Plex write failed: {e}")
+        raise HTTPException(status_code=422, detail=f"Plex write failed: {e}")
     return _collection_card(col)
 
 
@@ -887,20 +904,24 @@ def collection_delete(
         db.commit()
     except (PlexError, PlexNotConfigured) as e:
         db.rollback()
-        raise HTTPException(status_code=502, detail=f"Plex write failed: {e}")
+        raise HTTPException(status_code=422, detail=f"Plex write failed: {e}")
     return {"deleted": rating_key}
 
 
+class CollectionEditableIn(BaseModel):
+    editable: bool = False
+
+
 @router.post("/collections/{rating_key}/editable")
 def collection_set_editable(
-    rating_key: str, payload: dict, _: User = Depends(admin_user), db: Session = Depends(get_db)
+    rating_key: str, body: CollectionEditableIn, _: User = Depends(admin_user), db: Session = Depends(get_db)
 ) -> dict:
     """Mark an existing Plex collection as 'mine to edit' (or unmark). Smart / auto-list collections
     (IMDb/TMDb/…) can never be made editable — they're regenerated outside Plex."""
     col = db.query(PlexCollection).filter_by(rating_key=str(rating_key)).first()
     if col is None:
         raise HTTPException(status_code=404, detail="Unknown collection")
-    want = bool(payload.get("editable"))
+    want = body.editable
     if want and (col.smart or _collection_source(col) != "collection"):
         raise HTTPException(status_code=400, detail="Smart / auto-list collections can't be made editable")
     col.editable = want
@@ -975,15 +996,22 @@ def playlists(
     return resp
 
 
+class CreatePlexPlaylistIn(BaseModel):
+    title: str = ""
+    item_rating_key: str = ""
+
+
 @router.post("/playlists")
-def create_playlist(payload: dict, user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict:
+def create_playlist(
-    title = (payload.get("title") or "").strip()
+    body: CreatePlexPlaylistIn, user: User = Depends(current_user), db: Session = Depends(get_db)
+) -> dict:
+    title = body.title.strip()
     if not title:
-        raise HTTPException(status_code=422, detail="title is required")
+        raise HTTPException(status_code=400, detail="title is required")
     pl = PlexPlaylist(user_id=user.id, title=title)
     db.add(pl)
     db.flush()
-    seed = str(payload.get("item_rating_key") or "")
+    seed = body.item_rating_key
     count = 0
     thumb_rk = None
     if seed:
@@ -1022,12 +1050,18 @@ def playlist_detail(pid: int, user: User = Depends(current_user), db: Session =
     return {"id": pl.id, "title": pl.title, "items": cards}
 
 
+class RenamePlexPlaylistIn(BaseModel):
+    title: str = ""
+
+
 @router.patch("/playlists/{pid}")
-def rename_playlist(pid: int, payload: dict, user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict:
+def rename_playlist(
+    pid: int, body: RenamePlexPlaylistIn, user: User = Depends(current_user), db: Session = Depends(get_db)
+) -> dict:
     pl = _own_playlist_or_404(db, pid, user)
-    title = (payload.get("title") or "").strip()
+    title = body.title.strip()
     if not title:
-        raise HTTPException(status_code=422, detail="title is required")
+        raise HTTPException(status_code=400, detail="title is required")
     pl.title = title
     db.commit()
     count = db.query(func.count(PlexPlaylistItem.id)).filter_by(playlist_id=pl.id).scalar() or 0
@@ -1043,10 +1077,16 @@ def delete_playlist(pid: int, user: User = Depends(current_user), db: Session =
     return {"deleted": pid}
 
 
+class PlexItemRefIn(BaseModel):
+    item_rating_key: str = ""
+
+
 @router.post("/playlists/{pid}/items")
-def playlist_add_item(pid: int, payload: dict, user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict:
+def playlist_add_item(
+    pid: int, body: PlexItemRefIn, user: User = Depends(current_user), db: Session = Depends(get_db)
+) -> dict:
     pl = _own_playlist_or_404(db, pid, user)
-    it = db.query(PlexItem).filter_by(rating_key=str(payload.get("item_rating_key") or "")).first()
+    it = db.query(PlexItem).filter_by(rating_key=body.item_rating_key).first()
     if it is None:
         raise HTTPException(status_code=404, detail="Unknown item")
     exists = db.query(PlexPlaylistItem).filter_by(playlist_id=pid, item_id=it.id).first()
@@ -1069,6 +1109,10 @@ def playlist_remove_item(pid: int, item_rating_key: str, user: User = Depends(cu
     return {"ok": True}
 
 
+class PlexItemKeysIn(BaseModel):
+    item_rating_keys: list[str] = []
+
+
 def _items_by_rks(db: Session, rks: list) -> list[PlexItem]:
     """Resolve rating_keys to PlexItems, preserving the input order (drops unknown keys)."""
     order = [str(k) for k in rks if str(k)]
@@ -1086,11 +1130,13 @@ def _items_by_rks(db: Session, rks: list) -> list[PlexItem]:
 
 
 @router.post("/playlists/{pid}/items/bulk")
-def playlist_add_bulk(pid: int, payload: dict, user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict:
+def playlist_add_bulk(
+    pid: int, body: PlexItemKeysIn, user: User = Depends(current_user), db: Session = Depends(get_db)
+) -> dict:
     """Append many items at once (a whole season/show). Already-present items are skipped; the input
     order is preserved for the newly-added ones. Returns how many were added + the new total."""
     pl = _own_playlist_or_404(db, pid, user)
-    items = _items_by_rks(db, payload.get("item_rating_keys") or [])
+    items = _items_by_rks(db, body.item_rating_keys)
     existing = {row[0] for row in db.query(PlexPlaylistItem.item_id).filter_by(playlist_id=pid)}
     pos = int(db.query(func.coalesce(func.max(PlexPlaylistItem.position), -1)).filter_by(playlist_id=pid).scalar())
     added = 0
@@ -1109,11 +1155,13 @@ def playlist_add_bulk(pid: int, payload: dict, user: User = Depends(current_user
 
 
 @router.post("/playlists/{pid}/items/remove-bulk")
-def playlist_remove_bulk(pid: int, payload: dict, user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict:
+def playlist_remove_bulk(
+    pid: int, body: PlexItemKeysIn, user: User = Depends(current_user), db: Session = Depends(get_db)
+) -> dict:
     """Remove many items at once (a whole season/show, or a multi-select). Returns how many were
     removed + the new total."""
     pl = _own_playlist_or_404(db, pid, user)
-    ids = [it.id for it in _items_by_rks(db, payload.get("item_rating_keys") or [])]
+    ids = [it.id for it in _items_by_rks(db, body.item_rating_keys)]
     removed = 0
     if ids:
         removed = (
@@ -1129,10 +1177,12 @@ def playlist_remove_bulk(pid: int, payload: dict, user: User = Depends(current_u
 
 
 @router.put("/playlists/{pid}/order")
-def reorder_playlist(pid: int, payload: dict, user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict:
+def reorder_playlist(
+    pid: int, body: PlexItemKeysIn, user: User = Depends(current_user), db: Session = Depends(get_db)
+) -> dict:
     """Set the playlist order from an explicit list of item rating_keys (0-based positions)."""
     pl = _own_playlist_or_404(db, pid, user)
-    order = [str(k) for k in (payload.get("item_rating_keys") or [])]
+    order = body.item_rating_keys
     id_by_rk = {it.rating_key: it.id for it in db.query(PlexItem).filter(PlexItem.rating_key.in_(order or [""]))}
     for pos, rk in enumerate(order):
         iid = id_by_rk.get(rk)
@@ -1892,10 +1942,16 @@ def _push_watch(
     )
 
 
+class ItemProgressIn(BaseModel):
+    position_seconds: int = 0
+    duration_seconds: int = 0
+    final: bool = False
+
+
 @router.post("/item/{rating_key}/progress")
 def item_progress(
     rating_key: str,
-    payload: dict,
+    body: ItemProgressIn,
     background: BackgroundTasks,
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
@@ -1906,12 +1962,9 @@ def item_progress(
     debounces the Plex resume push: we mirror the settled position to Plex only on a final flush, so
     a single watch doesn't spray `/:/timeline` calls at the server every 10 seconds."""
     it = _item_or_404(db, rating_key)
-    try:
+    position = max(0, body.position_seconds)
-        position = max(0, int(payload.get("position_seconds") or 0))
+    duration = body.duration_seconds
-        duration = int(payload.get("duration_seconds") or 0)
+    final = body.final
-    except (TypeError, ValueError):
-        raise HTTPException(status_code=400, detail="position_seconds must be an integer")
-    final = bool(payload.get("final"))
     now = datetime.now(timezone.utc)
     st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first()
     was_watched = st is not None and st.status == "watched"
@@ -1957,10 +2010,14 @@ def item_progress(
     return {"position_seconds": position}
 
 
+class ItemStateIn(BaseModel):
+    status: str = ""
+
+
 @router.post("/item/{rating_key}/state")
 def item_state(
     rating_key: str,
-    payload: dict,
+    body: ItemStateIn,
     background: BackgroundTasks,
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
@@ -1969,7 +2026,7 @@ def item_state(
     to a linked Plex account immediately; `hidden` is a Siftlode-only concept with no Plex analogue,
     so it never touches Plex."""
     it = _item_or_404(db, rating_key)
-    status = payload.get("status")
+    status = body.status
     if status not in ("new", "watched", "hidden"):
         raise HTTPException(status_code=400, detail="Invalid status")
     st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first()
@@ -2035,10 +2092,15 @@ def _bulk_state(
     return len(to_push)
 
 
+# Shared by the show- and season-level "mark all episodes watched/unwatched" endpoints.
+class WatchedIn(BaseModel):
+    watched: bool = False
+
+
 @router.post("/show/{rating_key}/state")
 def show_state(
     rating_key: str,
-    payload: dict,
+    body: WatchedIn,
     background: BackgroundTasks,
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
@@ -2047,7 +2109,7 @@ def show_state(
     sh = db.query(PlexShow).filter_by(rating_key=str(rating_key)).first()
     if sh is None:
         raise HTTPException(status_code=404, detail="Unknown Plex show")
-    watched = bool(payload.get("watched"))
+    watched = body.watched
     eps = db.query(PlexItem).filter_by(show_id=sh.id, kind="episode").all()
     changed = _bulk_state(background, db, user.id, eps, watched)
     return {"changed": changed, "watched": watched}
@@ -2056,7 +2118,7 @@ def show_state(
 @router.post("/season/{rating_key}/state")
 def season_state(
     rating_key: str,
-    payload: dict,
+    body: WatchedIn,
     background: BackgroundTasks,
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
@@ -2065,7 +2127,7 @@ def season_state(
     se = db.query(PlexSeason).filter_by(rating_key=str(rating_key)).first()
     if se is None:
         raise HTTPException(status_code=404, detail="Unknown Plex season")
-    watched = bool(payload.get("watched"))
+    watched = body.watched
     eps = db.query(PlexItem).filter_by(season_id=se.id, kind="episode").all()
     changed = _bulk_state(background, db, user.id, eps, watched)
     return {"changed": changed, "watched": watched}