fix(api): mutations answer ok; route helpers flush and leave the commit to the handler

Two rows of the canonical-patterns table had drifted, found while closing the roadmap audit:

- 63 mutation responses carried no `ok` — `{"unsubscribed": true}`, `{"status": "ok"}`,
  `{"added": 3, …}` and so on. Each now leads with `"ok": true` (additive: every field a client
  reads is still there).
- Twelve route helpers committed on their own. Six were just the end of their handler's work and now
  flush: the playlist item add/remove and Watch later get-or-create, the Plex series/season bulk
  mark, and the invite decision + account activation — which an approval used to land in separate
  transactions. The other six commit on purpose (a token durable before its link is mailed, a
  single-use token burned as its own unit, the Google callback's tail, a background registration
  with its own session, the staged search ingest around YouTube calls), each listed with its reason.

`tests/test_route_conventions.py` holds both rules in code; CLAUDE.md's two rows point at it.

backend/app/routes/plex/admin.py

@@ -149,7 +149,7 @@ def watch_link_set(
     result: dict = {}
     if enabled and not link.initial_import_done:
         result = plex_watch.import_owner_watch_state(db, link)
-    return {**_link_dict(link), "import": result or None}
+    return {"ok": True, **_link_dict(link), "import": result or None}
 
 
 @router.post("/watch/import")
@@ -160,4 +160,4 @@ def watch_import_now(user: User = Depends(admin_user), db: Session = Depends(get
     if link is None or not link.sync_enabled:
         raise refused(409, "watch_sync_off")
     result = plex_watch.import_owner_watch_state(db, link)
-    return {**_link_dict(link), "import": result}
+    return {"ok": True, **_link_dict(link), "import": result}

backend/app/routes/plex/playback.py

@@ -385,7 +385,7 @@ def item_progress(
         # item — or a second one racing this one — would otherwise bump Plex's viewCount again.
         if flipped:
             _push_watch(background, db, user.id, it.id, rating_key, "watched")
-        return {"status": "watched", "position_seconds": 0}
+        return {"ok": True, "status": "watched", "position_seconds": 0}
 
     st = (
         db.execute(
@@ -402,7 +402,7 @@ def item_progress(
             st.position_seconds = 0
             st.progress_updated_at = now
         db.commit()
-        return {"position_seconds": 0}
+        return {"ok": True, "position_seconds": 0}
 
     if st is None:
         st = PlexState(user_id=user.id, item_id=it.id)
@@ -427,7 +427,7 @@ def item_progress(
             duration * 1000,
             intent=now,
         )
-    return {"position_seconds": position}
+    return {"ok": True, "position_seconds": position}
 
 
 class ItemStateIn(BaseModel):
@@ -473,7 +473,7 @@ def item_state(
             db.commit()
         if mirrored_on_plex:
             _push_watch(background, db, user.id, it.id, rating_key, "unwatched")
-        return {"status": "new"}
+        return {"ok": True, "status": "new"}
     if st is None:
         st = PlexState(user_id=user.id, item_id=it.id)
         db.add(st)
@@ -485,7 +485,7 @@ def item_state(
     db.commit()
     if status == "watched" and prev_status != "watched":
         _push_watch(background, db, user.id, it.id, rating_key, "watched")
-    return {"status": status}
+    return {"ok": True, "status": status}
 
 
 def _bulk_state(
@@ -527,7 +527,9 @@ def _bulk_state(
                 db.delete(st)
                 if prev != "new":
                     to_push.append((e.id, e.rating_key, "unwatched"))
-    db.commit()
+    # Flush, not commit: the handler commits, and it does so before the response — which is when
+    # the background push below runs, so the push always reads the committed marks.
+    db.flush()
     if pushable and to_push:
         # One coalesced background task for the whole batch (reuses a single session + Plex client),
         # not one task per episode — a 200-ep show mark is 1 task, not 200.
@@ -558,7 +560,8 @@ def show_state(
         .all()
     )
     changed = _bulk_state(background, db, user.id, eps, watched)
-    return {"changed": changed, "watched": watched}
+    db.commit()
+    return {"ok": True, "changed": changed, "watched": watched}
 
 
 @router.post("/season/{rating_key}/state")
@@ -580,7 +583,8 @@ def season_state(
         .all()
     )
     changed = _bulk_state(background, db, user.id, eps, watched)
-    return {"changed": changed, "watched": watched}
+    db.commit()
+    return {"ok": True, "changed": changed, "watched": watched}
 
 
 @router.post("/stream/{rating_key}/session")
@@ -605,6 +609,7 @@ def stream_session(
     # the client can switch audio without a restart — even for otherwise direct-playable files.
     if it.playable == "direct" and audio is None and not aoff and not multi:
         return {
+            "ok": True,
             "mode": "direct",
             "url": f"/api/plex/stream/{it.rating_key}/file",
             "start_s": 0.0,
@@ -620,6 +625,7 @@ def stream_session(
     # surface the active mode + encoder in the stats overlay (`tonemap` = the HW HDR→SDR path); both
     # segment streams are otherwise identical.
     return {
+        "ok": True,
         "mode": "hls",
         "start_s": s.media_start_s,
         # The session id is part of the path (not a query param) so the playlist's RELATIVE segment

backend/app/routes/plex/playlists.py

@@ -325,7 +325,7 @@ def playlist_add_bulk(
         )
         or 0
     )
-    return {"added": added, "item_count": total}
+    return {"ok": True, "added": added, "item_count": total}
 
 
 @router.post("/playlists/{pid}/items/remove-bulk")
@@ -356,7 +356,7 @@ def playlist_remove_bulk(
         )
         or 0
     )
-    return {"removed": removed, "item_count": total}
+    return {"ok": True, "removed": removed, "item_count": total}
 
 
 @router.put("/playlists/{pid}/order")