test(fence): owner filters in any form, statement writes under any name, core commits by list

The owner-filter check missed an aliased() model, an import alias, .in_() and
filter_by; four live inline filters on aliased VideoState go through owned_by now.
Cross-unit writes are seen under any import name, through sa.update and via
__table__. Core modules may commit only where CORE_COMMITS names the reason;
trim_read and audit.gc flush now and their callers commit. The pin target, the
debug-session header and a link's target fetch through owned_row. The docs
claim what the fence holds, and say what is beyond it.

backend/app/README.md

@@ -56,11 +56,16 @@ The domain is four units, each owning its tables (`models.py` in its package):
 - **Whose a row is, `userscope` says** — in every module, core included: `owned_by(Model.user_id,
   user)` for a filter, `owned_row(db, Model, id, user)` for a row by id (someone else's row is the
   same answer as a missing one). A unit's `queries.own_x` returns None and its route answers
-  `not_found`; a core route calls `owned_or_404`.
+  `not_found`; a core route calls `owned_or_404`. A rule `owned_row` cannot say — a job its owner
+  shared, a built-in profile nobody owns, a folder read under `FOR SHARE` — is written out in the
+  unit's own code, beside the reason.
 
 All of it is held by `tests/test_unit_fences.py`, with no exemptions: the unit map, each unit's
-packages and edges (`UNIT_PACKAGES`, `READS`) and the jobs that commit (`UNIT_COMMITS`) are listed
-there.
+packages and edges (`UNIT_PACKAGES`, `READS`) and what may commit (`UNIT_COMMITS`, `CORE_COMMITS`)
+are listed there. It reads the code's AST: an inline owner filter in any form (an alias, `.in_()`,
+`filter_by`), a model built or a statement write (`insert`/`update`/`delete` under any name) across
+units, and a commit off the lists all fail it. An ORM attribute write on another unit's loaded row
+is beyond it — review's to catch.
 
 ## Transaction ownership
 
@@ -73,10 +78,12 @@ mutation it records, so an audit entry can never exist without its change (or vi
 
 - The **operation owner** is the route handler, or a job. A unit's `service` flushes even when it
   makes a whole transition (`service.cancel`): the handler commits it, together with whatever else the
-  request changed. What owns its OWN session commits it: `quota` and `jobruns`' run rows, and a unit's
-  jobs (the worker's `jobdb` writes, the syncs and ingests that commit per page or batch so a later
-  failure loses nothing already stored) — the fence's `UNIT_COMMITS` names each. Everything else
-  flushes.
+  request changed. What owns its OWN session commits it: `quota` and `jobruns`' run rows, the
+  scheduler's jobs, a unit's jobs (the worker's `jobdb` writes, the syncs and ingests that commit per
+  page or batch so a later failure loses nothing already stored), and a few core steps that must land
+  whatever their caller does next (a token's last-used stamp, a refreshed YouTube grant) — the fence's
+  `UNIT_COMMITS` and `CORE_COMMITS` name each with its reason, `test_route_conventions` a route's own
+  committing helpers. Everything else flushes.
 - `SessionLocal` is `expire_on_commit=False`: after a raw `UPDATE` + commit, an ORM object you already
   loaded is **stale**. Read the fresh value from the row (RETURNING, or a re-`SELECT`), not the ORM
   attribute — see `service.decr_ref_count`.

backend/app/audit.py

@@ -110,10 +110,9 @@ def record(
 
 def gc(db: Session, retention_days: int) -> int:
     """Delete audit rows older than `retention_days`. Returns the number removed. A
-    retention_days <= 0 disables pruning (keep forever)."""
+    retention_days <= 0 disables pruning (keep forever). Flushes; the job commits."""
     if retention_days <= 0:
         return 0
     cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
     result = db.execute(delete(AuditLog).where(AuditLog.created_at < cutoff))
-    db.commit()
     return result.rowcount or 0

backend/app/debugsession.py

@@ -26,7 +26,7 @@ from uuid import uuid4
 from sqlalchemy import and_, delete, func, not_, select, update
 from sqlalchemy.orm import Session, defer
 
-from app.userscope import owned_by
+from app.userscope import owned_by, owned_row
 from app import diagnostics
 from app.models import DebugScreenshot, DebugSession, LogEvent, User
 
@@ -129,12 +129,8 @@ def from_header(db: Session, user: User, value: str | None) -> DebugSession | No
     """The session a request names, if it is the caller's own and still live; else None."""
     if not value or len(value) != _ID_LEN or not all(c in "0123456789abcdef" for c in value):
         return None
-    session = db.get(DebugSession, value)
-    if (
-        session is None
-        or session.user_id != user.id
-        or not is_live(session, datetime.now(timezone.utc))
-    ):
+    session = owned_row(db, DebugSession, value, user)
+    if session is None or not is_live(session, datetime.now(timezone.utc)):
         return None
     return session
 

backend/app/discover/pins.py

@@ -72,11 +72,11 @@ def resolve_ref(db: Session, user: User, kind: str, ref: str) -> tuple[str, obje
     # such pin target", which is the same answer this function gives for one that does not exist.
     except ValueError:
         return None
-    pl = db.get(Playlist, pid)
     # A playlist is per-user, unlike the other four (which are shared catalog rows whose PRIVACY
     # lives in the state tables). Someone else's list must read as "no such thing", not as a 403 —
     # the same reason `owned_or_404` 404s rather than 403s.
-    return (col, pl.id) if pl is not None and pl.user_id == user.id else None
+    pl = owned_row(db, Playlist, pid, user)
+    return (col, pl.id) if pl is not None else None
 
 
 def reorder(db: Session, user: User, ids: list[int]) -> None:

backend/app/discover/queries.py

@@ -93,7 +93,7 @@ def yt_in_progress(db: Session, user: User, limit: int = RAIL_LIMIT) -> list[tup
     q = (
         select(*feed_columns(user, st), st.progress_updated_at.label("progressed_at"))
         .join(Channel, Channel.id == Video.channel_id)
-        .join(st, and_(st.video_id == Video.id, st.user_id == user.id))
+        .join(st, and_(st.video_id == Video.id, owned_by(st.user_id, user)))
         .where(
             st.position_seconds > 0,
             st.status.notin_(("watched", "hidden")),
@@ -136,7 +136,7 @@ def yt_video_row(db: Session, user: User, video_id: str) -> dict | None:
     row = db.execute(
         select(*feed_columns(user, st))
         .join(Channel, Channel.id == Video.channel_id)
-        .outerjoin(st, and_(st.video_id == Video.id, st.user_id == user.id))
+        .outerjoin(st, and_(st.video_id == Video.id, owned_by(st.user_id, user)))
         .where(Video.id == video_id)
     ).first()
     return serialize(row) if row is not None else None
@@ -179,7 +179,7 @@ def playlists_summary(
         select(*feed_columns(user, st), PlaylistItem.playlist_id.label("_playlist"))
         .join(PlaylistItem, PlaylistItem.video_id == Video.id)
         .join(Channel, Channel.id == Video.channel_id)
-        .outerjoin(st, and_(st.video_id == Video.id, st.user_id == user.id))
+        .outerjoin(st, and_(st.video_id == Video.id, owned_by(st.user_id, user)))
         .where(PlaylistItem.playlist_id.in_(playlist_ids))
         .order_by(PlaylistItem.playlist_id, PlaylistItem.position.asc())
     ).all()

backend/app/downloads/queries.py

@@ -296,13 +296,11 @@ def own_link(db: Session, user: User, link_id: int) -> DownloadLink | None:
     if link is None:
         return None
     target = (
-        db.get(DownloadJob, link.job_id)
+        owned_row(db, DownloadJob, link.job_id, user)
         if link.job_id is not None
-        else db.get(FileFolder, link.folder_id)
+        else owned_row(db, FileFolder, link.folder_id, user)
     )
-    if target is None or target.user_id != user.id:
-        return None
-    return link
+    return link if target is not None else None
 
 
 def links_to(

backend/app/feedquery.py

@@ -212,7 +212,7 @@ def filtered_query(
     # exactly. 'unwatched'/'all' include row-less 'new' videos, so they keep the outer join.
     state_required = show in ("watched", "hidden", "in_progress")
     join_state = query.join if state_required else query.outerjoin
-    query = join_state(state, and_(state.video_id == Video.id, state.user_id == user.id))
+    query = join_state(state, and_(state.video_id == Video.id, owned_by(state.user_id, user)))
 
     # Hide videos the maintenance job has flagged as currently unplayable (deleted/private/
     # abandoned). They're either deleted after a grace period or unhidden if they recover;

backend/app/notifications.py

@@ -49,7 +49,7 @@ def create_notification(
 
 def trim_read(db: Session, user_id: int, cap: int = READ_SOFT_CAP) -> int:
     """Delete a user's oldest READ notifications beyond `cap`. Returns the number removed.
-    Idempotent and cheap; safe to call after marking things read."""
+    Idempotent and cheap; safe to call after marking things read. Flushes; the caller commits."""
     ids = (
         db.execute(
             select(Notification.id)
@@ -64,5 +64,4 @@ def trim_read(db: Session, user_id: int, cap: int = READ_SOFT_CAP) -> int:
         return 0
     # One bulk DELETE rather than a get+delete per row (the ids are already in hand).
     db.execute(delete(Notification).where(Notification.id.in_(ids)))
-    db.commit()
     return len(ids)

backend/app/routes/_common.py

@@ -89,8 +89,9 @@ def owned_or_404(
 ) -> M:
     """Fetch a row by primary key that `user` owns, or raise 404 (`lock`: see `get_or_404`). A row owned by someone
     else 404s (NOT 403) on purpose — a 403 would confirm the id exists to a stranger.
-    It replaced the near-identical per-route ownership helpers, and is called directly — a thin
-    `_own_x` wrapper around it is the ad-hoc shape CLAUDE.md's patterns table rules out."""
+    A core route calls it directly. A unit's route cannot (it may not name a unit's model as a value):
+    it asks its unit's `queries.own_x`, which goes through the same `owned_row`, and answers
+    `not_found` itself."""
     obj = owned_row(db, model, id_, user, owner_attr=owner_attr, lock=lock)
     if obj is None:
         raise not_found(what)

backend/app/routes/notifications.py

@@ -78,8 +78,8 @@ def mark_read(
 ) -> dict:
     n = owned_or_404(db, Notification, notification_id, user, what="notification")
     n.read = True
-    db.commit()
     trim_read(db, user.id)
+    db.commit()
     return {"ok": True}
 
 
@@ -90,8 +90,8 @@ def mark_all_read(user: User = Depends(current_user), db: Session = Depends(get_
         .where(owned_by(Notification.user_id, user), Notification.read.is_(False))
         .values(read=True)
     )
-    db.commit()
     trim_read(db, user.id)
+    db.commit()
     return {"ok": True}
 
 

backend/app/userscope.py

@@ -3,8 +3,9 @@
 **Ownership.** A user-owned table says whose a row is in one column (`user_id`, or `to_user_id` for
 something handed to them). Every read and write names that column through `owned_by` and fetches a
 row by id through `owned_row`, never with an inline `Model.user_id == …`: the rule "a row owned by
-someone else is answered like a missing one" then lives here, and a contract test
-(`test_unit_fences.py`) holds every module to it, core included.
+someone else is answered like a missing one" then lives here. A contract test
+(`test_unit_fences.py`) holds every module, core included, to the filter's form; a rule this module
+cannot say (own or shared, built-in or own) is written out where it applies, with its reason.
 
 **Recipients.** "An active, non-suspended account" — the download share-by-email and the recipient
 picker filter on it. Kept here so the two forms (SQL clauses for a query, a Python check for a loaded

backend/tests/test_unit_fences.py

@@ -408,7 +408,38 @@ UNIT_COMMITS = {
         "plex/watch_sync.py:run_plex_watch_reconcile",
     },
 }
+# Where core code may commit, each with why. Everything else in core flushes and leaves the commit
+# to its caller, as a unit's code does; a route's own helpers are held by test_route_conventions.
+CORE_COMMITS = {
+    # Quota spend and a background run's row are durable whatever the caller does next: each opens
+    # a session of its own.
+    "quota.py",
+    "jobruns.py",
+    # The scheduler's jobs own their session.
+    "scheduler.py",
+    # The instance's one settings row: created on first read, and the setup steps that finish it.
+    "state.py",
+    # A token's last-used stamp lands ahead of the route: a read-only route never commits, and a
+    # write route's rollback must not take it along. The expiry warning is a scheduled job.
+    "apitokens.py:_token_caller",
+    "apitokens.py:warn_expiring",
+    # An account's erasure is one operation shared by the self-service and the admin delete; the
+    # Google revoke and the confirmation run after it, in the background.
+    "auth.py:purge_user",
+    # The undelivered-email notice opens its own session (its callers are often background tasks).
+    "email.py:_alert_admins",
+    # One notice, committed unless its caller batches (`commit=False`).
+    "notifications.py:create_notification",
+    # A refreshed (or dead) YouTube grant must persist even when the call that needed it fails and
+    # its caller rolls back.
+    "youtube/client.py:_access_token",
+    "youtube/client.py:_drop_dead_grant",
+}
 SQL_WRITES = {"insert", "update", "delete", "pg_insert"}
+ALL_MODEL_NAMES = set(UNIT_MODEL_NAMES) | CORE_MODELS
+# The column methods that compare an owner column as a filter does (`Model.user_id.in_(ids)`).
+# `.is_(None)` is not one: it asks for the rows nobody owns (the system's own tags).
+OWNER_METHODS = {"in_", "not_in", "notin_", "__eq__", "__ne__"}
 
 
 def _package_unit(rel: str) -> str | None:
@@ -440,6 +471,64 @@ def _imported_models(tree: ast.AST) -> dict[str, str]:
     return out
 
 
+def _model_aliases(tree: ast.AST) -> dict[str, str]:
+    """Every other name a module binds to a model, by that name: an import under another name, and
+    `x = aliased(Model)` — an alias is the same table, and its owner column the same filter."""
+    out = {}
+    for node in ast.walk(tree):
+        if isinstance(node, ast.ImportFrom) and "models" in (node.module or ""):
+            for alias in node.names:
+                if alias.name in ALL_MODEL_NAMES:
+                    out[alias.asname or alias.name] = alias.name
+    for node in ast.walk(tree):
+        if (
+            isinstance(node, ast.Assign)
+            and isinstance(node.value, ast.Call)
+            and _called_name(node.value) == "aliased"
+            and node.value.args
+            and isinstance(node.value.args[0], ast.Name)
+        ):
+            source = node.value.args[0].id
+            model = out.get(source, source if source in ALL_MODEL_NAMES else None)
+            for target in node.targets:
+                if model and isinstance(target, ast.Name):
+                    out[target.id] = model
+    return out
+
+
+def _called_name(call: ast.Call) -> str | None:
+    func = call.func
+    return func.id if isinstance(func, ast.Name) else getattr(func, "attr", None)
+
+
+def _sql_write_names(tree: ast.AST) -> tuple[set[str], set[str]]:
+    """The names a module calls SQLAlchemy's statement writes by — `insert`/`update`/`delete` as
+    imported, under any alias — and the names it binds SQLAlchemy's modules to (`sa.update(...)`)."""
+    functions, modules = set(SQL_WRITES), set()
+    for node in ast.walk(tree):
+        if isinstance(node, ast.ImportFrom) and (node.module or "").split(".")[0] == "sqlalchemy":
+            for alias in node.names:
+                if alias.name in {"insert", "update", "delete"}:
+                    functions.add(alias.asname or alias.name)
+                elif alias.name in {"dialects", "postgresql", "sql"}:
+                    modules.add(alias.asname or alias.name)
+        elif isinstance(node, ast.Import):
+            for alias in node.names:
+                if alias.name.split(".")[0] == "sqlalchemy":
+                    modules.add(alias.asname or alias.name.split(".")[0])
+    return functions, modules
+
+
+def _written_table(call: ast.Call) -> str | None:
+    """The name of what a statement write's first argument names: `Model`, or `Model.__table__`."""
+    if not call.args:
+        return None
+    arg = call.args[0]
+    if isinstance(arg, ast.Attribute) and arg.attr == "__table__":
+        arg = arg.value
+    return arg.id if isinstance(arg, ast.Name) else None
+
+
 def _enclosing_functions(tree: ast.AST) -> dict[ast.AST, str]:
     out = {}
     for fn in ast.walk(tree):
@@ -461,49 +550,67 @@ def package_violations(rel: str, source: str) -> list[str]:
             continue
         if theirs not in READS.get(mine or "", set()):
             found.append(f"0: {mine or 'core'} imports {theirs}'s {name}")
+    write_functions, sql_modules = _sql_write_names(tree)
     for node in ast.walk(tree):
-        if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Name):
+        if not isinstance(node, ast.Call):
             continue
+        func = node.func
         written = None
-        if node.func.id in models:
-            written = models[node.func.id]
-        elif node.func.id in SQL_WRITES and node.args and isinstance(node.args[0], ast.Name):
-            written = models.get(node.args[0].id)
+        if isinstance(func, ast.Name) and func.id in models:
+            written = models[func.id]
+        elif (isinstance(func, ast.Name) and func.id in write_functions) or (
+            isinstance(func, ast.Attribute)
+            and func.attr in {"insert", "update", "delete"}
+            and isinstance(_root(func), ast.Name)
+            and _root(func).id in sql_modules
+        ):
+            written = models.get(_written_table(node) or "")
         if written is None:
             continue
         theirs = UNIT_MODEL_NAMES[written]
         if theirs != mine:
             found.append(f"{node.lineno}: {mine or 'core'} writes {theirs}'s {written}")
     found += _owner_filters(tree)
-    if mine is not None:
-        functions = _enclosing_functions(tree)
-        allowed = UNIT_COMMITS.get(mine, set())
-        for node in ast.walk(tree):
-            if (
-                isinstance(node, ast.Call)
-                and isinstance(node.func, ast.Attribute)
-                and node.func.attr == "commit"
-                and rel not in allowed
-                and f"{rel}:{functions.get(node)}" not in allowed
-            ):
-                found.append(f"{node.lineno}: commits in {functions.get(node)}")
+    functions = _enclosing_functions(tree)
+    allowed = UNIT_COMMITS.get(mine, set()) if mine is not None else CORE_COMMITS
+    for node in ast.walk(tree):
+        if (
+            isinstance(node, ast.Call)
+            and isinstance(node.func, ast.Attribute)
+            and node.func.attr == "commit"
+            and rel not in allowed
+            and f"{rel}:{functions.get(node)}" not in allowed
+        ):
+            found.append(f"{node.lineno}: commits in {functions.get(node)}")
     return sorted(found, key=lambda v: int(v.split(":")[0]))
 
 
 def _owner_filters(tree: ast.AST) -> list[str]:
     """A comparison on a table's owner column written inline (`Model.user_id == …`), not through
     `userscope.owned_by`."""
+    names = {name: name for name in ALL_MODEL_NAMES} | _model_aliases(tree)
+
+    def owner_column(node: ast.expr) -> bool:
+        return (
+            isinstance(node, ast.Attribute)
+            and isinstance(node.value, ast.Name)
+            and node.value.id in names
+            and node.attr in OWNER_COLUMNS
+        )
+
     found = []
     for node in ast.walk(tree):
         if isinstance(node, ast.Compare):
             for side in (node.left, *node.comparators):
-                if (
-                    isinstance(side, ast.Attribute)
-                    and isinstance(side.value, ast.Name)
-                    and side.value.id in set(UNIT_MODEL_NAMES) | CORE_MODELS
-                    and side.attr in OWNER_COLUMNS
-                ):
+                if owner_column(side):
                     found.append(f"{node.lineno}: filters {side.value.id}.{side.attr} inline")
+        elif isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
+            if node.func.attr in OWNER_METHODS and owner_column(node.func.value):
+                found.append(f"{node.lineno}: filters an owner column with {node.func.attr}")
+            elif node.func.attr == "filter_by" and any(
+                k.arg in OWNER_COLUMNS for k in node.keywords
+            ):
+                found.append(f"{node.lineno}: filters an owner column with filter_by")
     return found
 
 
@@ -524,7 +631,7 @@ def test_every_unit_package_exists():
     for prefixes in UNIT_PACKAGES.values():
         for p in prefixes:
             assert (APP / p).exists(), p
-    for commits in UNIT_COMMITS.values():
+    for commits in (*UNIT_COMMITS.values(), CORE_COMMITS):
         for entry in commits:
             assert (APP / entry.split(":")[0]).exists(), entry
 
@@ -544,6 +651,22 @@ PACKAGE_CASES = {
     ("downloads/x.py", "q = DownloadJob.user_id == 1"): 1,
     ("downloads/x.py", "q = owned_by(DownloadJob.user_id, 1)"): 0,
     ("quota.py", "q = QuotaEvent.user_id == 1"): 1,  # core too
+    ("downloads/x.py", "from app.downloads.models import DownloadJob as J\nq = J.user_id == 1"): 1,
+    ("downloads/x.py", "st = aliased(VideoState)\nq = st.user_id == 1"): 1,
+    ("downloads/x.py", "q = DownloadJob.user_id.in_([1])"): 1,
+    ("downloads/x.py", "q = select(DownloadJob).filter_by(user_id=1)"): 1,
+    ("downloads/x.py", "q = Tag.user_id.is_(None)"): 0,  # the rows nobody owns
+    (
+        "downloads/x.py",
+        "from sqlalchemy import update as up\nfrom app.models import Video\nup(Video)",
+    ): 1,
+    (
+        "downloads/x.py",
+        "import sqlalchemy as sa\nfrom app.models import Video\nsa.update(Video)",
+    ): 1,
+    ("downloads/x.py", "from app.models import Video\nupdate(Video.__table__)"): 1,
+    ("notifications.py", "def f(db):\n    db.commit()"): 1,  # core flushes too
+    ("state.py", "def f(db):\n    db.commit()"): 0,
     ("downloads/x.py", "def f(db):\n    db.commit()"): 1,
     ("downloads/gc.py", "def f(db):\n    db.commit()"): 0,
     ("downloads/upload.py", "def ingest_upload(db):\n    db.commit()"): 0,