refactor(quota): own the quota session; Plex per-section commit (R6 S2)

Transaction ownership (C-A5):
- quota.record_usage / log_action write in their OWN SessionLocal(), committed
  immediately, and no longer take/commit the caller's db. The spend stays DURABLE
  regardless of the caller (read-only YouTube paths never commit; a failed job rolls
  back) — under-counting the shared daily budget would risk overspending the real
  YouTube quota — AND quota stops flushing a job's partial state mid-run behind its
  own `except: db.rollback()`. units_used_today switches to a scalar SELECT so
  measured()'s before/after diff sees the separate session's commits (READ COMMITTED).
  Call sites updated (youtube/client.py ×3, routes/search.py ×1).
- plex sync commits PER SECTION (moved the end-of-loop commit inside), so a PlexError
  on a later library keeps the ones already mirrored (idempotent; next run reconciles).

The sync jobs did NOT rely on record_usage's incidental commit for persistence — their
per-item functions (apply_rss_feed, backfill_channel_recent, import_subscriptions,
sync_user_playlists, apply_channel_autotags) all commit internally, so the in-loop
rollback handlers already isolate a failed item. No loop restructuring needed.

DB-lane tests (test_quota_ownership, 4): spend survives a caller rollback while the
caller's uncommitted row does NOT leak; accumulation; measured-style mid-txn read.
Gate: ruff clean; DB lane 180 green.

backend/app/quota.py

@@ -9,10 +9,12 @@ import contextvars
 from datetime import datetime, timezone
 from zoneinfo import ZoneInfo
 
+from sqlalchemy import func, select
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.orm import Session
 
 from app import sysconfig
+from app.db import SessionLocal
 from app.models import ApiQuotaUsage, QuotaEvent
 
 _PACIFIC = ZoneInfo("America/Los_Angeles")
@@ -99,8 +101,13 @@ def pacific_day_start_utc() -> datetime:
 
 
 def units_used_today(db: Session) -> int:
-    row = db.get(ApiQuotaUsage, pacific_today())
-    return row.units_used if row else 0
+    # A scalar SELECT (not db.get) so it bypasses the caller session's identity map — quota writes
+    # commit in a SEPARATE session (see record_usage), and a cached ORM row would otherwise hide
+    # their effect from a caller that already loaded today's row (measured()'s before/after diff).
+    # Under READ COMMITTED a fresh SELECT sees the other session's committed writes.
+    return db.scalar(
+        select(ApiQuotaUsage.units_used).where(ApiQuotaUsage.day == pacific_today())
+    ) or 0
 
 
 def remaining_today(db: Session) -> int:
@@ -111,21 +118,24 @@ def can_spend(db: Session, units: int) -> bool:
     return remaining_today(db) >= units
 
 
-def log_action(db: Session, user_id: int, action: str) -> None:
+def log_action(user_id: int, action: str) -> None:
     """Record a zero-cost action event so per-user daily caps still count it.
 
     `record_usage` only logs when units are actually charged; a scrape-based search spends no
-    quota yet must still be rate-limited per user, so it logs its event here directly."""
-    db.add(QuotaEvent(user_id=user_id, action=action, units=0))
-    db.commit()
+    quota yet must still be rate-limited per user, so it logs its event here directly.
+
+    Writes in its OWN session (like record_usage) — quota accounting is durable independent of the
+    caller's transaction and must never commit the caller's session (that would flush a job's
+    partial state mid-run, defeating its own rollback). See [[quota-owns-its-session]]."""
+    with SessionLocal() as qs:
+        qs.add(QuotaEvent(user_id=user_id, action=action, units=0))
+        qs.commit()
 
 
 def actions_today(db: Session, user_id: int, action: str) -> int:
     """How many quota events of `action` this user has logged so far in the current Pacific
     day — for per-user, per-action daily caps (e.g. the live-search limit). Counts events,
     not units, so it only works for actions charged exactly once per user action."""
-    from sqlalchemy import func, select  # local import: keep the module's import head lean
-
     return (
         db.scalar(
             select(func.count())
@@ -140,8 +150,15 @@ def actions_today(db: Session, user_id: int, action: str) -> int:
     )
 
 
-def record_usage(db: Session, units: int) -> None:
-    """Atomically add `units` to today's counter (upsert) and log an attribution event."""
+def record_usage(units: int) -> None:
+    """Atomically add `units` to today's counter (upsert) and log an attribution event.
+
+    Uses its OWN session, committed immediately, so the spend is DURABLE regardless of what the
+    calling job does next: the units were really consumed at YouTube, so a caller that later rolls
+    back (or is read-only and never commits) must NOT lose the record — under-counting the shared
+    daily budget risks overspending the real API quota. Equally, quota must never commit the
+    CALLER's session (the old behaviour) — that persisted a job's partial state mid-run behind its
+    own `except: db.rollback()`. See [[quota-owns-its-session]]."""
     if units <= 0:
         return
     day = pacific_today()
@@ -153,7 +170,8 @@ def record_usage(db: Session, units: int) -> None:
             set_={"units_used": ApiQuotaUsage.units_used + units},
         )
     )
-    db.execute(stmt)
-    # Audit detail: who/what spent it (per-user attribution; NULL actor = system).
-    db.add(QuotaEvent(user_id=_actor_id.get(), action=_action.get(), units=units))
-    db.commit()
+    with SessionLocal() as qs:
+        qs.execute(stmt)
+        # Audit detail: who/what spent it (per-user attribution; NULL actor = system).
+        qs.add(QuotaEvent(user_id=_actor_id.get(), action=_action.get(), units=units))
+        qs.commit()