test: the rewritten write handlers commit, and three more owner filters hold

Each handler that now calls a service and commits is driven through HTTP once. A second account's watch state, Watch later list and private download profile are never read or changed. The autotag job's orphan prune is seen committed.
fix: a first watch state and a first Watch later survive two requests at once
2026-10-20 00:45:22 +02:00 · 2026-10-20 00:45:22 +02:00 · 2026-10-20 00:45:22 +02:00 · 2026-10-20 00:45:21 +02:00 · 2026-10-20 00:45:20 +02:00 · 2026-10-20 00:45:20 +02:00
38 changed files with 1036 additions and 132 deletions
@@ -70,7 +70,8 @@ The domain is four **units**, each owning its tables: Capture (`downloads/`), Re
 sync), Catalog (`catalog/`, the `sync/` ingests) and Plex (`plex/`). A unit's route does no database work: it reads through
 the unit's `queries`, writes through its `service` (which flushes) and commits. A unit's code reads another unit's tables
 only along a declared edge and writes them only through that unit's service; core reaches a unit's tables through its
-`queries`. **Fence-held, with no exemptions** (`backend/tests/test_unit_fences.py`); conventions in
+`queries`. **Fence-held, with no exemptions** (`backend/tests/test_unit_fences.py`) — as far as an AST sees: an ORM
+attribute write on another unit's loaded row is review's to catch. Conventions in
 [`backend/app/README.md`](backend/app/README.md).
 Frontend: `main.tsx` provider tree → `App.tsx` shell (a shared `/watch` link boots from its own
 entry, `watch.html` → `watchMain.tsx`, held app-free by a contract test) → `moduleRegistry` route-chunks → module
@@ -88,36 +89,36 @@ CQRS, a repository/DI framework, a router library, a generic CRUD-router factory

 ## Canonical patterns — when you have a choice, pick THIS

-| Area                                 | Use                                                                                                                                                                                               | Not                                                                                                                                                     |
-| ------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| ORM                                  | SQLAlchemy 2.0 `select()`                                                                                                                                                                         | `db.query(...)`                                                                                                                                         |
-| Request body                         | a Pydantic model                                                                                                                                                                                  | `payload: dict`                                                                                                                                         |
-| Request SIZE                         | nothing — `bodylimit` caps every route (SEC-20); need more? add a ceiling to `ROUTE_MAX_BODY_BYTES`                                                                                               | `capped_body` on a route that does not read raw bytes itself                                                                                            |
-| Ownership check                      | `userscope.owned_by` (a filter) / `owned_row` (a row by id): a unit's `queries.own_x` returns None and its route answers `not_found`; a core route `owned_or_404`                                 | an inline `Model.user_id == …` (fence-held, in every module) or a `db.get` and a compare                                                                |
-| Admin gate import                    | `app.auth`                                                                                                                                                                                        | the `app.routes.admin` re-export                                                                                                                        |
-| Upstream failure status              | `422` with JSON detail                                                                                                                                                                            | `502` (reserve 502 for raw byte-proxies)                                                                                                                |
-| User-facing error                    | structured `detail` (`{code, reason}`)                                                                                                                                                            | a bare English string under a bilingual UI                                                                                                              |
-| Mutation response                    | `{"ok": true, …}` (`{"deleted": id}`), test-held                                                                                                                                                  | a bare `{"unsubscribed"}`/`{"cleared": n}`, no `ok`                                                                                                     |
-| Transaction commit                   | the handler commits; a unit's `service` and every helper `flush` (fence-held)                                                                                                                     | a helper that commits (only `quota`, `jobruns` and the jobs the fence's `UNIT_COMMITS` names own their session)                                         |
-| Admin knob storage                   | `sysconfig` (values)                                                                                                                                                                              | env / `app_state` / a table of its own (the job intervals had one)                                                                                      |
-| Job status word                      | the `downloads/status.py` StrEnum                                                                                                                                                                 | loose `"canceled"`/`"cancelled"` literals                                                                                                               |
-| Current time                         | inline `datetime.now(timezone.utc)`                                                                                                                                                               | new `now_utc`/`_now` helpers                                                                                                                            |
-| FE per-account state                 | `usePref`                                                                                                                                                                                         | raw `readAccount`/`writeAccount` or bespoke hooks                                                                                                       |
-| FE keyboard shortcut                 | a row in `lib/keymap.ts` + `dispatch()`                                                                                                                                                           | a private `keydown` listener with its own key literals                                                                                                  |
-| FE list/grid keys                    | `NavGrid`/`keyboardNav` + `navItemAttrs` + `data-nav-open`                                                                                                                                        | a private arrow-key / roving-tabindex handler                                                                                                           |
-| FE view push (history)               | `rememberNavPosition()` right before `history.pushState`                                                                                                                                          | a push that forgets: Back strands focus on `<body>`                                                                                                     |
-| FE static label / button-like link   | `Badge` (flat, no edge) · `LinkButton` / `IconLink`                                                                                                                                               | a glass pill that reads as a chip; an `<a>` hand-painted like a button                                                                                  |
-| FE loading/empty/error               | `QueryState` / `StateMessage`                                                                                                                                                                     | ad-hoc conditionals                                                                                                                                     |
-| FE failed write                      | `meta.failTitle`, else `notifyActionError(err, title)`; a YouTube write action (subscribe, sync, …): `notifyYouTubeActionError` — a 403 offers Connect                                            | a bare `catch {}` or `notify()` with no error attached                                                                                                  |
-| FE floating or full-screen layer     | `ui/Overlay` (portals to `<body>`) or the popover engine (`AnchoredPopover` portals itself); stacking by the z-index scale (`z-menu` … `z-popover`, docs/DESIGN.md "Layers")                      | one rendered in place inside a virtualized row or a masked/transformed scroller — clipped or painted over; a magic `z-[n]`                              |
-| FE player on a native media element  | `lib/mediaSession` (`useMediaSession`) + `lib/mediaKeys` + `lib/resume` (`useResume`), the element held as STATE                                                                                  | a private `navigator.mediaSession` / `keydown` / position-saving effect per player                                                                      |
-| FE diagnostics                       | `lib/log.ts` (`log.info("event.code", msg, attrs)`); a filter change → `lib/filterLog.ts`, a player → `lib/mediaLog.ts`                                                                           | `console.*` (lint-enforced); a value only a higher level may carry, logged unguarded                                                                    |
-| FE state that MIRRORS a prop/query   | derive it, or hold a draft (`draft ?? server`)                                                                                                                                                    | a copy an effect re-syncs — that paints the stale value first                                                                                           |
-| FE state reset on a key change       | adjust it DURING render against a remembered key                                                                                                                                                  | `useEffect(() => setX(init), [key])`                                                                                                                    |
-| FE effect that must not re-subscribe | `useEffectEvent` (React 19.2+); a handler or timer: `useLatest` / `useStableCallback`                                                                                                             | a render-phase `ref.current = x` — the React Compiler skips the component                                                                               |
-| FE hook suppression                  | an inline `eslint-disable-next-line` NAMING the invariant                                                                                                                                         | `eslint-suppressions.json` (deliberately absent — E27 S2)                                                                                               |
-| FE popover / scroll-fade ref         | destructure it: `const { setReference } = pop.refs`, then `ref={setReference}`                                                                                                                    | `ref={pop.refs.setReference}` / `ref={fade.ref}` — the compiler then treats the whole object as a ref and skips                                         |
-| FE async handler with cleanup        | `try { … } catch { … }`, then the cleanup after it (steps that `return` early go in their own function); a conditional, loop or `?.` the `try` needs goes before it or into a module-level helper | `try/finally`, or a `?:` / `??` / loop / `?.` inside `try` — the React Compiler skips the whole component (and `react-hooks/todo` misses most of these) |
+| Area                                 | Use                                                                                                                                                                                                                                           | Not                                                                                                                                                                                         |
+| ------------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| ORM                                  | SQLAlchemy 2.0 `select()`                                                                                                                                                                                                                     | `db.query(...)`                                                                                                                                                                             |
+| Request body                         | a Pydantic model                                                                                                                                                                                                                              | `payload: dict`                                                                                                                                                                             |
+| Request SIZE                         | nothing — `bodylimit` caps every route (SEC-20); need more? add a ceiling to `ROUTE_MAX_BODY_BYTES`                                                                                                                                           | `capped_body` on a route that does not read raw bytes itself                                                                                                                                |
+| Ownership check                      | `userscope.owned_by` (a filter) / `owned_row` (a row by id): a unit's `queries.own_x` returns None and its route answers `not_found`; a core route `owned_or_404`. A composite rule (own or shared, built-in or own) says so in its `queries` | an inline owner filter — `Model.user_id == …`, an alias's, `.in_()`, `filter_by(user_id=…)` (fence-held, every module) — or a `db.get` and a compare where `owned_row` fits                 |
+| Admin gate import                    | `app.auth`                                                                                                                                                                                                                                    | the `app.routes.admin` re-export                                                                                                                                                            |
+| Upstream failure status              | `422` with JSON detail                                                                                                                                                                                                                        | `502` (reserve 502 for raw byte-proxies)                                                                                                                                                    |
+| User-facing error                    | structured `detail` (`{code, reason}`)                                                                                                                                                                                                        | a bare English string under a bilingual UI                                                                                                                                                  |
+| Mutation response                    | `{"ok": true, …}` (`{"deleted": id}`), test-held                                                                                                                                                                                              | a bare `{"unsubscribed"}`/`{"cleared": n}`, no `ok`                                                                                                                                         |
+| Transaction commit                   | the handler commits; a unit's `service` and every helper `flush` (fence-held, every module)                                                                                                                                                   | a helper that commits — only what owns its session does, each named with its reason: the fence's `UNIT_COMMITS` / `CORE_COMMITS`, a route's `COMMITTING_HELPERS` (`test_route_conventions`) |
+| Admin knob storage                   | `sysconfig` (values)                                                                                                                                                                                                                          | env / `app_state` / a table of its own (the job intervals had one)                                                                                                                          |
+| Job status word                      | the `downloads/status.py` StrEnum                                                                                                                                                                                                             | loose `"canceled"`/`"cancelled"` literals                                                                                                                                                   |
+| Current time                         | inline `datetime.now(timezone.utc)`                                                                                                                                                                                                           | new `now_utc`/`_now` helpers                                                                                                                                                                |
+| FE per-account state                 | `usePref`                                                                                                                                                                                                                                     | raw `readAccount`/`writeAccount` or bespoke hooks                                                                                                                                           |
+| FE keyboard shortcut                 | a row in `lib/keymap.ts` + `dispatch()`                                                                                                                                                                                                       | a private `keydown` listener with its own key literals                                                                                                                                      |
+| FE list/grid keys                    | `NavGrid`/`keyboardNav` + `navItemAttrs` + `data-nav-open`                                                                                                                                                                                    | a private arrow-key / roving-tabindex handler                                                                                                                                               |
+| FE view push (history)               | `rememberNavPosition()` right before `history.pushState`                                                                                                                                                                                      | a push that forgets: Back strands focus on `<body>`                                                                                                                                         |
+| FE static label / button-like link   | `Badge` (flat, no edge) · `LinkButton` / `IconLink`                                                                                                                                                                                           | a glass pill that reads as a chip; an `<a>` hand-painted like a button                                                                                                                      |
+| FE loading/empty/error               | `QueryState` / `StateMessage`                                                                                                                                                                                                                 | ad-hoc conditionals                                                                                                                                                                         |
+| FE failed write                      | `meta.failTitle`, else `notifyActionError(err, title)`; a YouTube write action (subscribe, sync, …): `notifyYouTubeActionError` — a 403 offers Connect                                                                                        | a bare `catch {}` or `notify()` with no error attached                                                                                                                                      |
+| FE floating or full-screen layer     | `ui/Overlay` (portals to `<body>`) or the popover engine (`AnchoredPopover` portals itself); stacking by the z-index scale (`z-menu` … `z-popover`, docs/DESIGN.md "Layers")                                                                  | one rendered in place inside a virtualized row or a masked/transformed scroller — clipped or painted over; a magic `z-[n]`                                                                  |
+| FE player on a native media element  | `lib/mediaSession` (`useMediaSession`) + `lib/mediaKeys` + `lib/resume` (`useResume`), the element held as STATE                                                                                                                              | a private `navigator.mediaSession` / `keydown` / position-saving effect per player                                                                                                          |
+| FE diagnostics                       | `lib/log.ts` (`log.info("event.code", msg, attrs)`); a filter change → `lib/filterLog.ts`, a player → `lib/mediaLog.ts`                                                                                                                       | `console.*` (lint-enforced); a value only a higher level may carry, logged unguarded                                                                                                        |
+| FE state that MIRRORS a prop/query   | derive it, or hold a draft (`draft ?? server`)                                                                                                                                                                                                | a copy an effect re-syncs — that paints the stale value first                                                                                                                               |
+| FE state reset on a key change       | adjust it DURING render against a remembered key                                                                                                                                                                                              | `useEffect(() => setX(init), [key])`                                                                                                                                                        |
+| FE effect that must not re-subscribe | `useEffectEvent` (React 19.2+); a handler or timer: `useLatest` / `useStableCallback`                                                                                                                                                         | a render-phase `ref.current = x` — the React Compiler skips the component                                                                                                                   |
+| FE hook suppression                  | an inline `eslint-disable-next-line` NAMING the invariant                                                                                                                                                                                     | `eslint-suppressions.json` (deliberately absent — E27 S2)                                                                                                                                   |
+| FE popover / scroll-fade ref         | destructure it: `const { setReference } = pop.refs`, then `ref={setReference}`                                                                                                                                                                | `ref={pop.refs.setReference}` / `ref={fade.ref}` — the compiler then treats the whole object as a ref and skips                                                                             |
+| FE async handler with cleanup        | `try { … } catch { … }`, then the cleanup after it (steps that `return` early go in their own function); a conditional, loop or `?.` the `try` needs goes before it or into a module-level helper                                             | `try/finally`, or a `?:` / `??` / loop / `?.` inside `try` — the React Compiler skips the whole component (and `react-hooks/todo` misses most of these)                                     |

 Not repeated here, because a lint rule refuses the wrong one and its message names the right one:
 buttons (`ui/Button`, `IconButton`, `Pressable`, `Chip`, `DragHandle`), text fields and native
@@ -0,0 +1,55 @@
+"""One Watch later list per user, held by the database
+
+Revision ID: 0112_one_watch_later
+Revises: 0111_audit_request_id
+Create Date: 2026-10-20
+
+Watch later was made on a user's first save by a read then an insert, so two first saves at once
+each made a list, and the reader then picked between them at random. A partial unique index holds
+"one per user" now. An instance that already has a second list keeps the oldest: the others' videos
+move into it (after its own, in their order; one it already holds is not added twice), then they go.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0112_one_watch_later"
+down_revision: str | None = "0111_audit_request_id"
+branch_labels: str | None = None
+depends_on: str | None = None
+
+_EXTRA = """
+    SELECT p.id, k.keep_id
+    FROM playlists p
+    JOIN (
+        SELECT user_id, min(id) AS keep_id FROM playlists WHERE kind = 'watch_later' GROUP BY user_id
+    ) k ON k.user_id = p.user_id
+    WHERE p.kind = 'watch_later' AND p.id <> k.keep_id
+"""
+
+
+def upgrade() -> None:
+    op.execute(
+        f"""
+        INSERT INTO playlist_items (playlist_id, video_id, position, added_at)
+        SELECT e.keep_id, i.video_id,
+               (SELECT coalesce(max(position), -1) FROM playlist_items WHERE playlist_id = e.keep_id)
+               + row_number() OVER (PARTITION BY e.keep_id ORDER BY i.playlist_id, i.position, i.id),
+               i.added_at
+        FROM playlist_items i
+        JOIN ({_EXTRA}) e ON e.id = i.playlist_id
+        ON CONFLICT ON CONSTRAINT uq_playlist_video DO NOTHING
+        """
+    )
+    op.execute(f"DELETE FROM playlists WHERE id IN (SELECT id FROM ({_EXTRA}) e)")
+    op.create_index(
+        "uq_playlist_watch_later",
+        "playlists",
+        ["user_id"],
+        unique=True,
+        postgresql_where=sa.text("kind = 'watch_later'"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("uq_playlist_watch_later", table_name="playlists")
@@ -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`.
@@ -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
@@ -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

@@ -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:
@@ -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()
@@ -328,9 +328,14 @@ def create_own(db: Session, user_id: int, name: str) -> FileFolder:
    return folder


+def renamable(folder: FileFolder) -> bool:
+    """Only a folder of the user's own making takes a name: a source folder's is its origin's."""
+    return folder.kind == FolderKind.OWN
+
+
 def rename(db: Session, folder: FileFolder, name: str) -> None:
    """Rename one of the user's own folders (a source folder's name is its origin's)."""
-    if folder.kind != FolderKind.OWN:
+    if not renamable(folder):
        raise FolderError("file_folder_not_own")
    folder.name = name
    try:
@@ -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(
@@ -16,6 +16,7 @@ import shutil
 import subprocess
 import sys

+import app.tables  # noqa: F401 — a fresh interpreter: every table, before the first query
 from app import logs
 from app import logsink
 from app.config import settings
@@ -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;
@@ -28,6 +28,7 @@ log = logs.get_logger(__name__)
 from app import bodylimit, collation, logsink, state, sysconfig  # noqa: E402 — logging is configured first
 from app.config import settings  # noqa: E402 — logging is configured first
 from app.db import SessionLocal, engine  # noqa: E402 — logging is configured first
+import app.tables  # noqa: E402 — every table, before the first query
 from app.downloads import service as download_service  # noqa: E402 — logging is configured first
 from app.downloads import storage  # noqa: E402 — logging is configured first
 from app.plex import stream as plex_stream  # noqa: E402 — logging is configured first
@@ -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)
@@ -108,20 +108,6 @@ def state(db: Session, user: User | int, item_id: int) -> PlexState | None:
    )


-def states(db: Session, user: User | int, item_ids: list[int]) -> dict[int, PlexState]:
-    """The user's state on each of these items that has one, by item id."""
-    if not item_ids:
-        return {}
-    return {
-        s.item_id: s
-        for s in db.execute(
-            select(PlexState).where(
-                owned_by(PlexState.user_id, user), PlexState.item_id.in_(item_ids)
-            )
-        ).scalars()
-    }
-
-
 # --- a show ----------------------------------------------------------------------------------


@@ -8,6 +8,7 @@ from datetime import datetime, timezone
 from sqlalchemy import and_, case, func, or_, select
 from sqlalchemy.orm import Session, aliased

+from app.models import User
 from app.userscope import owned_by
 from app.plex.models import PlexItem, PlexState
 from app.plex.cards import ShowState
@@ -106,7 +107,7 @@ def state_map(db: Session, user_id: int, show_ids: list[int]) -> dict[int, ShowS
    return {r.show_id: show_state(r.total, r.watched, r.inprog) for r in rows}


-def item_state_map(db: Session, user_id: int, item_ids: list[int]) -> dict[int, PlexState]:
+def item_state_map(db: Session, user_id: User | int, item_ids: list[int]) -> dict[int, PlexState]:
    """A user's PlexState rows for a set of leaf items (movies or episodes), keyed by item_id — the
    leaf analogue of `state_map` (which rolls a show's episodes up to a `ShowState`). One batched
    fetch, so the caller serializes cards without a per-card query. Empty in → empty dict (no `IN ()`)."""
@@ -19,7 +19,7 @@ from app.plex import people as plex_people
 from app.plex import tmdb as plex_tmdb
 from app.plex.lookups import by_rating_key
 from app.models import User
-from app.plex import queries
+from app.plex import queries, rollup
 from app.plex.models import (
    PlexCollection,
    PlexItem,
@@ -302,6 +302,23 @@ def mark_finished(db: Session, user_id: int, item_id: int, now: datetime) -> boo
    return False


+def _state_row(db: Session, user: User, item_id: int, st: PlexState | None) -> PlexState:
+    """The user's state row on this item, made if the read before found none. Made by the insert
+    itself (ON CONFLICT DO NOTHING) and then read: the first checkpoint of an item and the `final`
+    flush beside it routinely both read "none", and a plain add let the second one fail on
+    `uq_plex_user_item` — a 500, and the push it was carrying lost."""
+    if st is not None:
+        return st
+    db.execute(
+        pg_insert(PlexState)
+        .values(user_id=user.id, item_id=item_id)
+        .on_conflict_do_nothing(constraint="uq_plex_user_item")
+    )
+    made = queries.state(db, user, item_id)
+    assert made is not None  # inserted now, or committed by the request that won
+    return made
+
+
 def clear_progress(db: Session, st: PlexState | None, now: datetime) -> None:
    """A checkpoint too early to keep: a state that held nothing else goes, any other forgets its
    position."""
@@ -325,9 +342,7 @@ def checkpoint(
 ) -> PlexState:
    """Record a resume position. `dirty` marks it not yet mirrored to Plex — set before a push, so
    the reconcile re-pushes it if the push fails and the push can settle it once Plex confirms."""
-    if st is None:
-        st = PlexState(user_id=user.id, item_id=item_id)
-        db.add(st)
+    st = _state_row(db, user, item_id, st)
    st.position_seconds = position
    st.progress_updated_at = now
    if dirty:
@@ -345,9 +360,7 @@ def set_status(
    db: Session, user: User, item_id: int, st: PlexState | None, status: str
 ) -> PlexState:
    """Set a watched or hidden status; watched also clears the resume point."""
-    if st is None:
-        st = PlexState(user_id=user.id, item_id=item_id)
-        db.add(st)
+    st = _state_row(db, user, item_id, st)
    st.status = status
    if status == "watched":
        st.watched_at = datetime.now(timezone.utc)
@@ -364,7 +377,7 @@ def bulk_state(
    (item id, rating key, action) pushes a linked Plex account needs. Unwatching keeps the user's
    private "hidden" marks."""
    now = datetime.now(timezone.utc)
-    existing = queries.states(db, user_id, [e.id for e in eps])
+    existing = rollup.item_state_map(db, user_id, [e.id for e in eps])
    to_push: list[tuple[int, str, str]] = []
    for e in eps:
        st = existing.get(e.id)
@@ -11,12 +11,14 @@ from sqlalchemy import (
    DateTime,
    Float,
    ForeignKey,
+    Index,
    Integer,
    JSON,
    String,
    Text,
    UniqueConstraint,
    func,
+    text,
 )
 from sqlalchemy.orm import Mapped, mapped_column, relationship

@@ -93,6 +95,15 @@ class Playlist(Base, TimestampMixin, UpdatedAtMixin):
    sync phases; the foundation phase only uses local playlists.)"""

    __tablename__ = "playlists"
+    __table_args__ = (
+        # One Watch later per user, held by the database: two first saves at once each made one.
+        Index(
+            "uq_playlist_watch_later",
+            "user_id",
+            unique=True,
+            postgresql_where=text("kind = 'watch_later'"),
+        ),
+    )

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
@@ -9,7 +9,7 @@ unique key, a row lock taken before a presence check. The comfort shelf's writes

 from datetime import datetime, timezone

-from sqlalchemy import exists, func, select, update
+from sqlalchemy import exists, func, select, text, update
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.orm import Session

@@ -319,17 +319,25 @@ def create_playlist(db: Session, user: User, name: str) -> Playlist:


 def watch_later(db: Session, user: User) -> Playlist:
-    """The user's built-in Watch later playlist, created on first use."""
+    """The user's built-in Watch later playlist, created on first use. Created by the insert itself
+    (ON CONFLICT DO NOTHING on the one-per-user index) and then read: the first two saves at once
+    both read "none", and each used to make a list of its own."""
    pl = queries.watch_later(db, user)
    if pl is None:
-        pl = Playlist(
-            user_id=user.id,
-            name="Watch later",
-            kind="watch_later",
-            position=_next_playlist_position(db, user.id),
+        db.execute(
+            pg_insert(Playlist)
+            .values(
+                user_id=user.id,
+                name="Watch later",
+                kind="watch_later",
+                position=_next_playlist_position(db, user.id),
+            )
+            .on_conflict_do_nothing(
+                index_elements=["user_id"], index_where=text("kind = 'watch_later'")
+            )
        )
-        db.add(pl)
-        db.flush()  # the id, for the item the caller adds next
+        pl = queries.watch_later(db, user)
+        assert pl is not None  # inserted now, or committed by the request that won
    return pl


@@ -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)
@@ -220,6 +220,7 @@ def send_issue(sid: str, user: User = Depends(current_user), db: Session = Depen
        raise refused(409, "issue_sending")
    new = session.issue_url is None
    if not new and not debugsession.unfinished(session):
+        db.commit()  # nothing to send: the lock ends here, not when the request's session closes
        return {"ok": True, "session": _view(db, session)}
    key = str(user.id)
    if new:
@@ -375,6 +375,9 @@ def update_folder(
    folder = own_folder(db, user, folder_id)
    sent = body.model_fields_set
    if "name" in sent:
+        # Which folder may be renamed is answered before what the name is, as it always was.
+        if not folders.renamable(folder):
+            raise refused(409, "file_folder_not_own")
        try:
            folders.rename(db, folder, _folder_name(body.name or ""))
        except folders.FolderError as e:
@@ -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}


@@ -100,7 +100,7 @@ def show_detail(
        raise not_found("show")
    seasons = queries.seasons(db, sh)
    eps = queries.episodes(db, sh)
-    states = queries.states(db, user, [e.id for e in eps])
+    states = rollup.item_state_map(db, user, [e.id for e in eps])
    by_season: dict[int | None, list] = {}
    all_cards: list[dict] = []  # every episode, in season/episode order → show-level rollup
    for e in eps:
@@ -247,7 +247,7 @@ def _collection_strips(db: Session, it: "PlexItem", user_id: int, lang: str) ->
    for col in ordered:
        rk = col.rating_key
        members = queries.library_movies_in(db, it.library_id, rk, lang, limit=60)
-        states = queries.states(db, user_id, [m.id for m in members])
+        states = rollup.item_state_map(db, user_id, [m.id for m in members])
        out.append(
            {
                "id": rk,
@@ -9,7 +9,7 @@ from sqlalchemy.orm import Session
 from app.auth import current_user
 from app.db import get_db
 from app.models import User
-from app.plex import queries, service
+from app.plex import queries, rollup, service
 from app.plex.cards import display_title, leaf_card, norm_lang
 from app.refusal import refused
 from app.routes._common import not_found
@@ -99,7 +99,7 @@ def playlist_detail(
    lang = norm_lang(lang)
    pl = _own_playlist(db, user, pid)
    items = queries.playlist_items(db, pl.id)
-    states = queries.states(db, user, [it.id for it in items])
+    states = rollup.item_state_map(db, user, [it.id for it in items])
    shows = queries.shows_by_ids(db, {it.show_id for it in items if it.show_id})
    cards = [
        leaf_card(
@@ -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
@@ -31,6 +31,7 @@ from functools import partial
 from sqlalchemy import text
 from sqlalchemy.orm import Session

+import app.tables  # noqa: F401 — every table, before the first query
 from app import logs
 from app import logsink
 from app import sysconfig
@@ -105,6 +105,11 @@ def pytest_configure(config):
        )


+def _write_xact(db) -> str | None:
+    """The id of the open transaction if it has written anything, else None (a read never takes one)."""
+    return db.scalar(text("SELECT pg_current_xact_id_if_assigned()::text"))
+
+
 def _request_dep(db, unfinished: list[str]):
    """`get_db` for a test request: the test's own session, plus the check a shared session hides.

@@ -124,6 +129,7 @@ def _request_dep(db, unfinished: list[str]):
    def dep(conn: HTTPConnection):
        db.flush()
        state = {"dirty": False}
+        before = _write_xact(db)

        def flushed(session, ctx):
            state["dirty"] = True
@@ -138,7 +144,12 @@ def _request_dep(db, unfinished: list[str]):
        try:
            yield db
            staged = db.new or db.deleted or any(db.is_modified(o) for o in db.dirty)
-            if state["dirty"] or staged:
+            # Core DML (`db.execute(update(...))`) fires no flush and stages nothing; the database
+            # itself says whether a write is open — a transaction that wrote and was not committed
+            # holds an id it did not hold when the request began (see `release_read_connection`). A
+            # row lock (`FOR UPDATE`) takes one too: the handler ends what it locked.
+            core_dml = (after := _write_xact(db)) is not None and after != before
+            if state["dirty"] or staged or core_dml:
                unfinished.append(f"{conn.scope.get('method', 'WS')} {conn.url.path}")
        except Exception:
            db.rollback()
@@ -8,6 +8,8 @@ drops an unknown id, so the view silently showed the whole feed under its old na
 from sqlalchemy import select

 from app.catalog.models import Channel
+from app.db import SessionLocal
+from app.sync.autotag import run_autotag_all
 from app.reader.models import ChannelTag, SavedView, Tag
 from app.reader import service as reader_service

@@ -66,3 +68,15 @@ def test_a_view_kept_tag_list_larger_than_a_bind_parameter_limit_still_cleans_up
    db.commit()
    assert reader_service.prune_orphan_system_tags(db) == 1
    db.commit()
+
+
+def test_the_job_commits_its_cleanup(db):
+    """The prune only flushes; the job ends its own transaction. Read from a session of its own,
+    the orphan is gone only once the job committed — the job's session would see its own delete
+    either way."""
+    orphan = Tag(user_id=None, name="Cooking", category="topic")
+    db.add(orphan)
+    db.commit()
+    assert run_autotag_all(db, only_missing=True)["orphan_tags_removed"] == 1
+    with SessionLocal() as fresh:
+        assert fresh.execute(select(Tag).where(Tag.name == "Cooking")).first() is None
@@ -0,0 +1,63 @@
+"""Every process the app runs as sees every table before its first query.
+
+The units declare their tables in their own `models` modules, and a core model names a unit's by
+string (`User.subscriptions` → `Subscription`). A fresh interpreter that never imported the unit's
+module fails to configure its mappers on its first query: the download child once did exactly that,
+and every job failed, while this suite stayed green — it imports `app.tables` in-process, which hides
+the gap. So each entry point is imported here in a clean interpreter, and must register the whole
+schema on its own.
+"""
+
+import ast
+import subprocess
+import sys
+from pathlib import Path
+
+import app.tables  # noqa: F401
+from app.db import Base
+
+BACKEND = Path(__file__).parent.parent
+APP = BACKEND / "app"
+
+# Each `python -m` module the deployment starts, and the API's app module.
+ENTRY_POINTS = ("app.main", "app.worker", "app.downloads.runner")
+
+_PROBE = (
+    "import importlib, sys\n"
+    "importlib.import_module(sys.argv[1])\n"
+    "from sqlalchemy.orm import configure_mappers\n"
+    "from app.db import Base\n"
+    "configure_mappers()\n"
+    "print('\\n'.join(sorted(Base.metadata.tables)))\n"
+)
+
+
+def test_each_entry_point_registers_the_whole_schema_on_its_own():
+    whole = sorted(Base.metadata.tables)
+    for module in ENTRY_POINTS:
+        r = subprocess.run(
+            [sys.executable, "-c", _PROBE, module],
+            cwd=BACKEND,
+            capture_output=True,
+            text=True,
+            timeout=120,
+            check=False,
+        )
+        assert r.returncode == 0, (module, r.stderr[-2000:])
+        assert r.stdout.split() == whole, module
+
+
+def test_every_module_that_runs_as_a_process_is_listed():
+    mains = set()
+    for path in APP.rglob("*.py"):
+        tree = ast.parse(path.read_text(encoding="utf-8"))
+        for node in tree.body:
+            if (
+                isinstance(node, ast.If)
+                and isinstance(node.test, ast.Compare)
+                and isinstance(node.test.left, ast.Name)
+                and node.test.left.id == "__name__"
+            ):
+                rel = path.relative_to(BACKEND).with_suffix("").as_posix()
+                mains.add(rel.replace("/", "."))
+    assert mains <= set(ENTRY_POINTS), mains - set(ENTRY_POINTS)
@@ -124,6 +124,9 @@ def test_only_an_own_folder_is_renamed_and_only_by_its_user(db, make_user, as_us
    assert clash.json()["detail"]["reason"] == "file_folder_name_taken"
    no = c.patch(f"/api/files/folders/{src.id}", json={"name": "YouTube 2"})
    assert no.status_code == 409 and no.json()["detail"]["reason"] == "file_folder_not_own"
+    # Which folder is answered before what the name is: a blank one is still "not yours to name".
+    blank = c.patch(f"/api/files/folders/{src.id}", json={"name": " "})
+    assert blank.status_code == 409 and blank.json()["detail"]["reason"] == "file_folder_not_own"
    stranger = as_user(make_user()).patch(f"/api/files/folders/{own.id}", json={"name": "X"})
    assert stranger.status_code == 404
    db.refresh(own)
@@ -0,0 +1,68 @@
+"""Migration 0112's data step, DB-lane: a user who already has two Watch later lists keeps the oldest,
+with the other's videos moved in after its own (one it already held is not added twice), and the
+one-per-user index then refuses a second list.
+
+The schema half is checked by test_migrations.py; a video lost on the way is invisible there.
+"""
+
+from datetime import datetime, timezone
+
+import pytest
+from alembic import command
+from alembic.config import Config
+from sqlalchemy import text
+from sqlalchemy.exc import IntegrityError
+
+from app.catalog.models import Channel, Video
+from app.db import engine
+
+
+def test_a_second_watch_later_is_folded_into_the_first(db, make_user):
+    user, other = make_user(), make_user()
+    db.add(Channel(id="UCm", title="M"))
+    now = datetime.now(timezone.utc)
+    for v in ("v1", "v2", "v3"):
+        db.add(Video(id=v, channel_id="UCm", published_at=now, duration_seconds=60))
+    db.commit()
+    cfg = Config("alembic.ini")
+    command.downgrade(cfg, "0111_audit_request_id")
+    try:
+        with engine.begin() as conn:
+            ids = conn.execute(
+                text(
+                    "INSERT INTO playlists (user_id, name, kind, position) VALUES"
+                    " (:u, 'Watch later', 'watch_later', 0), (:u, 'Watch later', 'watch_later', 1),"
+                    " (:o, 'Watch later', 'watch_later', 0) RETURNING id"
+                ),
+                {"u": user.id, "o": other.id},
+            ).scalars()
+            first, second, theirs = list(ids)
+            conn.execute(
+                text(
+                    "INSERT INTO playlist_items (playlist_id, video_id, position) VALUES"
+                    " (:a, 'v1', 0), (:b, 'v2', 0), (:b, 'v1', 1), (:c, 'v3', 0)"
+                ),
+                {"a": first, "b": second, "c": theirs},
+            )
+    finally:
+        command.upgrade(cfg, "head")
+
+    with engine.connect() as conn:
+        lists = conn.execute(
+            text("SELECT id, user_id FROM playlists WHERE kind = 'watch_later' ORDER BY id")
+        ).all()
+        assert [tuple(r) for r in lists] == [(first, user.id), (theirs, other.id)]
+        items = conn.execute(
+            text(
+                "SELECT playlist_id, video_id, position FROM playlist_items ORDER BY playlist_id, position"
+            )
+        ).all()
+        assert [tuple(r) for r in items] == [(first, "v1", 0), (first, "v2", 1), (theirs, "v3", 0)]
+    with pytest.raises(IntegrityError), engine.begin() as conn:
+        conn.execute(
+            text(
+                "INSERT INTO playlists (user_id, name, kind, position)"
+                " VALUES (:u, 'Watch later', 'watch_later', 2)"
+            ),
+            {"u": user.id},
+        )
@@ -82,3 +82,34 @@ def test_a_request_that_raises_leaves_nothing_it_did_not_commit(db):
        request.throw(RuntimeError("the handler failed"))
    assert db.scalar(select(User).where(User.email == "rolled-back-probe@example.com")) is None
    assert db.info["unfinished"] == []  # a failing request is rolled back, not reported
+
+
+def test_the_commit_check_sees_a_core_dml_write(db, make_user):
+    """`db.execute(update(...))` flushes nothing and stages nothing, so the session's own bookkeeping
+    never sees it: a handler whose only write is Core DML and that forgot its commit passed. The
+    database's transaction id is what tells. A committed one must still pass."""
+    import pytest
+    from sqlalchemy import update
+    from starlette.requests import HTTPConnection
+
+    from app.models import User
+
+    user = make_user()
+
+    def drive(commit: bool) -> None:
+        conn = HTTPConnection(
+            {"type": "http", "method": "POST", "path": "/probe", "headers": [], "query_string": b""}
+        )
+        request = db.info["request"](conn)
+        session = next(request)
+        session.execute(update(User).where(User.id == user.id).values(display_name="dml"))
+        if commit:
+            session.commit()
+        with pytest.raises(StopIteration):
+            next(request)
+
+    drive(commit=True)
+    assert db.info["unfinished"] == []
+    drive(commit=False)
+    assert db.info["unfinished"] == ["POST /probe"]
+    db.info["unfinished"].clear()  # recorded as it should be — not this test's failure
@@ -10,7 +10,10 @@ from datetime import datetime, timedelta, timezone
 from sqlalchemy import func, select

 from app import debugsession, notifications, quota
+from app.catalog.models import Channel, Video
+from app.downloads.models import DownloadJob, DownloadProfile
 from app.models import DebugScreenshot, DebugSession, Notification, QuotaEvent
+from app.reader.models import Playlist, PlaylistItem, VideoState
 from app.routes import tokens as tokens_routes
 from app.routes.auth import RESET_TTL, _issue_token

@@ -133,3 +136,113 @@ def test_screenshot_storage_counts_only_mine(db, make_user):
        )
    db.commit()
    assert debugsession.stored_screenshot_bytes(db, me.id) == 10
+
+
+# --- the units' owner filters, through their routes --------------------------------------------
+# The same rule one layer up: a second account acting on the SAME video, list or profile id must
+# reach only its own row. Each test holds the first account's row unchanged afterwards.
+
+
+def _videos(db, *ids):
+    db.add(Channel(id="UCo", title="O"))
+    db.flush()
+    for vid in ids:
+        db.add(
+            Video(
+                id=vid,
+                channel_id="UCo",
+                published_at=datetime.now(timezone.utc),
+                duration_seconds=3600,
+            )
+        )
+    db.commit()
+
+
+def _video_state(db, user, vid="v1"):
+    db.expire_all()
+    row = db.execute(
+        select(VideoState).where(VideoState.user_id == user.id, VideoState.video_id == vid)
+    ).scalar_one_or_none()
+    return None if row is None else (row.status, row.position_seconds)
+
+
+def test_unmarking_a_video_leaves_another_users_state(db, make_user, as_user):
+    me, other = make_user(), make_user()
+    _videos(db, "v1")
+    db.add(VideoState(user_id=other.id, video_id="v1", status="watched"))
+    db.commit()
+    r = as_user(me).post("/api/videos/v1/state", json={"status": "new"})
+    assert r.status_code == 200, r.text
+    assert _video_state(db, other) == ("watched", 0)
+
+
+def test_clearing_a_video_leaves_another_users_state(db, make_user, as_user):
+    me, other = make_user(), make_user()
+    _videos(db, "v1")
+    db.add(VideoState(user_id=other.id, video_id="v1", status="hidden", position_seconds=90))
+    db.commit()
+    assert as_user(me).delete("/api/videos/v1/state").status_code == 200
+    assert _video_state(db, other) == ("hidden", 90)
+
+
+def test_a_resume_point_is_set_and_cleared_on_my_own_row_only(db, make_user, as_user):
+    me, other = make_user(), make_user()
+    _videos(db, "v1")
+    db.add(VideoState(user_id=other.id, video_id="v1", status="hidden", position_seconds=90))
+    db.commit()
+    client = as_user(me)
+    body = {"position_seconds": 600, "duration_seconds": 3600}
+    assert client.post("/api/videos/v1/progress", json=body).status_code == 200
+    assert (_video_state(db, me), _video_state(db, other)) == (("new", 600), ("hidden", 90))
+    body = {"position_seconds": 0, "duration_seconds": 3600}
+    assert client.post("/api/videos/v1/progress", json=body).status_code == 200
+    assert (_video_state(db, me), _video_state(db, other)) == (None, ("hidden", 90))
+
+
+def _watch_later_ids(db, user):
+    db.expire_all()
+    return sorted(
+        db.execute(
+            select(PlaylistItem.video_id)
+            .join(Playlist, Playlist.id == PlaylistItem.playlist_id)
+            .where(Playlist.user_id == user.id, Playlist.kind == "watch_later")
+        ).scalars()
+    )
+
+
+def test_watch_later_adds_and_removes_use_my_own_list(db, make_user, as_user):
+    me, other = make_user(), make_user()
+    _videos(db, "v1", "v2")
+    theirs = as_user(other).post("/api/playlists/watch-later", json={"video_id": "v1"}).json()
+    client = as_user(me)
+    mine = client.post("/api/playlists/watch-later", json={"video_id": "v2"}).json()
+    assert mine["playlist_id"] != theirs["playlist_id"]
+    # Their saved video is not on my list, so removing it from mine leaves theirs alone.
+    assert client.delete("/api/playlists/watch-later/v1").status_code == 200
+    assert (_watch_later_ids(db, me), _watch_later_ids(db, other)) == (["v2"], ["v1"])
+
+
+def test_another_users_private_profile_is_refused_like_a_missing_one(db, make_user, as_user):
+    me, other = make_user(), make_user()
+    theirs = DownloadProfile(user_id=other.id, name="Theirs", spec={"format": "best"})
+    builtin = DownloadProfile(user_id=None, name="Best", spec={"format": "best"}, is_builtin=True)
+    db.add_all([theirs, builtin])
+    db.commit()
+    client = as_user(me)
+
+    def enqueue(profile_id):
+        return client.post(
+            "/api/downloads", json={"source": "dQw4w9WgXcQ", "profile_id": profile_id}
+        )
+
+    missing = enqueue(theirs.id + builtin.id + 1)
+    refused = enqueue(theirs.id)
+    assert (refused.status_code, refused.json()) == (404, missing.json())
+    assert refused.json()["detail"] == {"code": "not_found", "reason": "profile"}
+    db.expire_all()
+    assert db.execute(select(DownloadJob)).first() is None
+    # A built-in is everyone's to use.
+    r = enqueue(builtin.id)
+    assert r.status_code == 200, r.text
+    db.expire_all()
+    assert db.execute(select(DownloadJob.profile_id)).scalars().all() == [builtin.id]
@@ -12,7 +12,7 @@ from datetime import datetime, timezone

 from sqlalchemy import select

-from app.plex import library, queries, service
+from app.plex import library, queries, rollup, service
 from app.plex.models import PlexCollection, PlexItem, PlexLibrary, PlexShow, PlexState
 from app.plex.query import LibraryFilters

@@ -71,9 +71,9 @@ def test_a_users_watch_state_is_theirs_alone(db, make_user):
    _state(db, owner, it, status="watched")
    assert queries.state(db, owner, it.id).status == "watched"
    assert queries.state(db, stranger, it.id) is None
-    assert queries.states(db, stranger, [it.id]) == {}
-    assert set(queries.states(db, owner, [it.id])) == {it.id}
-    assert queries.states(db, owner, []) == {}
+    assert rollup.item_state_map(db, stranger, [it.id]) == {}
+    assert set(rollup.item_state_map(db, owner, [it.id])) == {it.id}
+    assert rollup.item_state_map(db, owner, []) == {}


 def test_the_grid_shows_the_callers_state_and_hides_only_what_they_hid(db, make_user):
@@ -421,3 +421,18 @@ def test_only_the_owners_own_link_pushes_their_watch_state(db, make_user):
    db.commit()
    assert watch_sync.link_for_push(db, owner.id).id == link.id
    assert watch_sync.link_for_push(db, other.id) is None
+
+
+def test_a_first_write_that_lost_the_race_updates_the_winners_row(db, make_user):
+    """The first checkpoint of an item and the `final` flush beside it both read "no state"; the one
+    that writes second must land on the row the first made, not fail on `uq_plex_user_item`."""
+    user = make_user()
+    it = _movie(db, _lib(db), "m-race")
+    _state(db, user, it, position_seconds=10)  # the request that won, committed
+    st = service.checkpoint(db, user, it.id, None, 42, NOW, dirty=True)  # read "none" before it
+    assert (st.position_seconds, st.synced_to_plex) == (42, False)
+    st = service.set_status(db, user, it.id, None, "hidden")
+    assert st.status == "hidden"
+    db.commit()
+    rows = db.scalars(select(PlexState).where(PlexState.item_id == it.id)).all()
+    assert [(r.position_seconds, r.status) for r in rows] == [(42, "hidden")]
@@ -308,3 +308,21 @@ def test_a_pin_is_one_per_thing_and_only_its_owners(db, make_user):
    pins.remove(db, pins.own_pin(db, user, pin.id))
    db.commit()
    assert db.execute(select(ComfortPin)).first() is None
+
+
+def test_a_watch_later_save_that_lost_the_race_uses_the_winners_list(db, make_user, monkeypatch):
+    """The first two saves at once both read "no Watch later"; the second must land in the list the
+    first made, not make one of its own."""
+    user = make_user()
+    won = service.watch_later(db, user)
+    db.commit()
+    real = queries.watch_later
+    stale = [True]  # this request's first read happened before the winner committed
+
+    def read(d, u):
+        return None if stale and stale.pop() else real(d, u)
+
+    monkeypatch.setattr(queries, "watch_later", read)
+    assert service.watch_later(db, user).id == won.id
+    db.commit()
+    assert [p.id for p in queries.playlists_for(db, user) if p.kind == "watch_later"] == [won.id]
@@ -0,0 +1,356 @@
+"""Writes driven through HTTP, each read back: the handler called its unit and committed.
+
+A route's helper only flushes; the handler commits. The `db` fixture fails a test whose request left
+a write uncommitted, so every test here drives one successful write through its route and then
+reads the result back — a handler that lost its commit fails on both counts. The services' own
+tests hold what each write does; these hold that the route finishes it. Nothing here reaches
+YouTube or Plex: the YouTube client is a stub and no user has a Plex link. Needs the DB.
+"""
+
+from datetime import datetime, timezone
+
+import pytest
+from sqlalchemy import select
+
+from app import quota
+from app.auth import require_write_scope
+from app.catalog.models import Channel, ExploredChannel, SearchFind, Subscription, Video
+from app.downloads.models import DownloadJob, DownloadProfile, DownloadShare
+from app.plex.models import PlexItem, PlexLibrary, PlexSeason, PlexShow, PlexState
+from app.models import Notification
+from app.reader.models import SavedView, Tag, VideoState
+from app.routes import channels
+
+NOW = datetime.now(timezone.utc)
+
+
+def _video(db, vid="v1", **kw):
+    if db.get(Channel, "UCr") is None:
+        db.add(Channel(id="UCr", title="R"))
+        db.flush()
+    db.add(Video(id=vid, channel_id="UCr", published_at=NOW, duration_seconds=60, **kw))
+    db.commit()
+
+
+# --- tags ------------------------------------------------------------------------------------
+
+
+def test_a_tag_made_through_the_route_is_listed_back(db, make_user, as_user):
+    user = make_user()
+    client = as_user(user)
+    r = client.post("/api/tags", json={"name": "Science"})
+    assert r.status_code == 200, r.text
+    assert [t["name"] for t in client.get("/api/tags").json()] == ["Science"]
+
+
+def test_a_tag_renamed_through_the_route_is_listed_back(db, make_user, as_user):
+    user = make_user()
+    tag = Tag(user_id=user.id, name="Old")
+    db.add(tag)
+    db.commit()
+    client = as_user(user)
+    assert client.patch(f"/api/tags/{tag.id}", json={"name": "New"}).status_code == 200
+    assert [t["name"] for t in client.get("/api/tags").json()] == ["New"]
+
+
+def test_a_tag_deleted_through_the_route_is_gone(db, make_user, as_user):
+    user = make_user()
+    tag = Tag(user_id=user.id, name="Gone")
+    db.add(tag)
+    db.commit()
+    client = as_user(user)
+    assert client.delete(f"/api/tags/{tag.id}").json() == {"deleted": tag.id}
+    assert client.get("/api/tags").json() == []
+
+
+# --- watch state ------------------------------------------------------------------------------
+
+
+def _state(db, user, vid="v1"):
+    db.expire_all()
+    return db.execute(
+        select(VideoState).where(VideoState.user_id == user.id, VideoState.video_id == vid)
+    ).scalar_one_or_none()
+
+
+def test_a_video_marked_watched_through_the_route_stays_watched(db, make_user, as_user):
+    user = make_user()
+    _video(db)
+    r = as_user(user).post("/api/videos/v1/state", json={"status": "watched"})
+    assert r.status_code == 200, r.text
+    assert _state(db, user).status == "watched"
+
+
+def test_a_video_unmarked_through_the_route_stays_unmarked(db, make_user, as_user):
+    # "new" is its own branch of the handler, with its own commit.
+    user = make_user()
+    _video(db)
+    db.add(VideoState(user_id=user.id, video_id="v1", status="watched"))
+    db.commit()
+    r = as_user(user).post("/api/videos/v1/state", json={"status": "new"})
+    assert r.status_code == 200, r.text
+    assert _state(db, user) is None
+
+
+def test_a_video_state_cleared_through_the_route_is_gone(db, make_user, as_user):
+    user = make_user()
+    _video(db)
+    db.add(VideoState(user_id=user.id, video_id="v1", status="hidden", position_seconds=90))
+    db.commit()
+    assert as_user(user).delete("/api/videos/v1/state").status_code == 200
+    assert _state(db, user) is None
+
+
+# --- saved views ------------------------------------------------------------------------------
+
+
+def test_views_reordered_through_the_route_list_back_in_that_order(db, make_user, as_user):
+    user = make_user()
+    a = SavedView(user_id=user.id, source="feed", name="A", filters={}, position=0)
+    b = SavedView(user_id=user.id, source="feed", name="B", filters={}, position=1)
+    db.add_all([a, b])
+    db.commit()
+    client = as_user(user)
+    assert client.post("/api/saved-views/reorder", json={"ids": [b.id, a.id]}).status_code == 200
+    listed = client.get("/api/saved-views").json()
+    assert [(v["name"], v["position"]) for v in listed] == [("B", 0), ("A", 1)]
+
+
+# --- download profiles and shares -------------------------------------------------------------
+
+_SPEC = {"format": "best"}
+
+
+def _profile(db, user, name="Mine"):
+    p = DownloadProfile(user_id=user.id, name=name, spec=_SPEC, is_builtin=False, sort_order=101)
+    db.add(p)
+    db.commit()
+    return p
+
+
+def _profile_names(client):
+    return [p["name"] for p in client.get("/api/downloads/profiles").json()]
+
+
+def test_a_profile_made_through_the_route_is_listed_back(db, make_user, as_user):
+    user = make_user()
+    client = as_user(user)
+    r = client.post("/api/downloads/profiles", json={"name": "Phone", "spec": _SPEC})
+    assert r.status_code == 200, r.text
+    assert _profile_names(client) == ["Phone"]
+
+
+def test_a_profile_renamed_through_the_route_is_listed_back(db, make_user, as_user):
+    user = make_user()
+    p = _profile(db, user)
+    client = as_user(user)
+    r = client.patch(f"/api/downloads/profiles/{p.id}", json={"name": "Renamed"})
+    assert r.status_code == 200, r.text
+    assert _profile_names(client) == ["Renamed"]
+
+
+def test_a_profile_deleted_through_the_route_is_gone(db, make_user, as_user):
+    user = make_user()
+    p = _profile(db, user)
+    client = as_user(user)
+    assert client.delete(f"/api/downloads/profiles/{p.id}").json() == {"deleted": p.id}
+    assert _profile_names(client) == []
+
+
+def test_a_share_taken_back_through_the_route_is_gone(db, make_user, as_user):
+    owner, friend = make_user(), make_user()
+    job = DownloadJob(
+        user_id=owner.id, source_kind="youtube", source_ref="dQw4w9WgXcQ", profile_snapshot={}
+    )
+    db.add(job)
+    db.flush()
+    db.add(DownloadShare(job_id=job.id, from_user_id=owner.id, to_user_id=friend.id))
+    db.commit()
+    r = as_user(owner).delete(f"/api/downloads/{job.id}/share/{friend.email}")
+    assert r.json() == {"ok": True}
+    db.expire_all()
+    assert db.execute(select(DownloadShare)).first() is None
+
+
+# --- channels ---------------------------------------------------------------------------------
+
+
+class _YouTube:
+    """The YouTube client, answering every write the channel routes make."""
+
+    def __init__(self, db, user):
+        pass
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        return False
+
+    def insert_subscription(self, channel_id):
+        return "yt-sub-1"
+
+    def delete_subscription(self, sub_id):
+        return None
+
+    def get_channels(self, ids):
+        return []
+
+
+@pytest.fixture
+def youtube(monkeypatch):
+    monkeypatch.setattr(channels, "YouTubeClient", _YouTube)
+    monkeypatch.setattr(quota, "backfill_reserve_reached", lambda: False)
+
+
+def _channel(db, cid="UCc", **kw):
+    ch = Channel(id=cid, title="C", details_synced_at=NOW, **kw)
+    db.add(ch)
+    db.commit()
+    return ch
+
+
+def _subscription(db, user, cid="UCc"):
+    db.expire_all()
+    return db.execute(
+        select(Subscription).where(Subscription.user_id == user.id, Subscription.channel_id == cid)
+    ).scalar_one_or_none()
+
+
+def test_a_subscribe_through_the_route_is_recorded(db, make_user, as_user, override_dep, youtube):
+    user = make_user()
+    _channel(db)
+    client = as_user(user)
+    override_dep(require_write_scope, lambda: user)
+    assert client.post("/api/channels/UCc/subscribe").json() == {"ok": True, "subscribed": "UCc"}
+    assert _subscription(db, user).yt_subscription_id == "yt-sub-1"
+
+
+def test_an_explore_through_the_route_is_recorded(db, make_user, as_user, youtube, monkeypatch):
+    user = make_user()
+    _channel(db)
+    # The ingest pages YouTube; what the route itself records is the exploration.
+    monkeypatch.setattr(channels, "details_stale", lambda ch, now: False)
+    monkeypatch.setattr(
+        channels,
+        "explore_ingest_page",
+        lambda db, yt, ch, token: {"ingested": 0, "next_page_token": None},
+    )
+    r = as_user(user).post("/api/channels/UCc/explore")
+    assert r.status_code == 200, r.text
+    db.expire_all()
+    assert db.execute(select(ExploredChannel.user_id)).scalars().all() == [user.id]
+    assert db.get(Channel, "UCc").from_explore is True
+
+
+def test_an_unsubscribe_through_the_route_drops_the_subscription(
+    db, make_user, as_user, override_dep, youtube
+):
+    user = make_user()
+    _channel(db)
+    db.add(Subscription(user_id=user.id, channel_id="UCc", yt_subscription_id="yt-sub-1"))
+    db.commit()
+    client = as_user(user)
+    override_dep(require_write_scope, lambda: user)
+    r = client.delete("/api/channels/UCc/subscription")
+    assert r.json() == {"ok": True, "unsubscribed": "UCc"}
+    assert _subscription(db, user) is None
+
+
+def test_a_backfill_reset_through_the_route_is_kept(db, make_user, as_user, monkeypatch):
+    admin, reader = make_user(role="admin"), make_user()
+    _channel(db, backfill_done=True, backfill_cursor="page-9", recent_synced_at=NOW)
+    db.add(Subscription(user_id=reader.id, channel_id="UCc", deep_requested=False))
+    db.commit()
+    # The recent pull afterwards reaches YouTube; the reset is the route's own write.
+    monkeypatch.setattr(channels, "run_recent_backfill", lambda *a, **kw: None)
+    r = as_user(admin).post("/api/channels/UCc/reset-backfill")
+    assert r.status_code == 200, r.text
+    db.expire_all()
+    ch = db.get(Channel, "UCc")
+    assert (ch.backfill_done, ch.backfill_cursor, ch.recent_synced_at) == (False, None, None)
+    assert _subscription(db, reader).deep_requested is True
+
+
+# --- live search ------------------------------------------------------------------------------
+
+
+def test_search_results_cleared_through_the_route_are_gone(db, make_user, as_user):
+    user = make_user()
+    _video(db, "s1", via_search=True)
+    db.add(SearchFind(user_id=user.id, video_id="s1"))
+    db.commit()
+    r = as_user(user).post("/api/search/clear", json={"video_ids": ["s1"]})
+    assert r.json() == {"ok": True, "removed": 1}
+    db.expire_all()
+    assert db.get(Video, "s1") is None
+    assert db.execute(select(SearchFind)).first() is None
+
+
+# --- Plex show and season marks ---------------------------------------------------------------
+
+
+def _show(db):
+    lib = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True)
+    db.add(lib)
+    db.flush()
+    sh = PlexShow(rating_key="sh1", library_id=lib.id, title="Show")
+    db.add(sh)
+    db.flush()
+    se = PlexSeason(rating_key="se1", show_id=sh.id, season_number=1)
+    db.add(se)
+    db.flush()
+    for n in (1, 2):
+        db.add(
+            PlexItem(
+                rating_key=f"e{n}",
+                library_id=lib.id,
+                kind="episode",
+                title=f"E{n}",
+                show_id=sh.id,
+                season_id=se.id,
+                season_number=1,
+                episode_number=n,
+            )
+        )
+    db.commit()
+
+
+def _plex_marks(db, user):
+    db.expire_all()
+    return sorted(
+        db.execute(
+            select(PlexItem.rating_key, PlexState.status)
+            .join(PlexState, PlexState.item_id == PlexItem.id)
+            .where(PlexState.user_id == user.id)
+        ).all()
+    )
+
+
+def test_a_show_marked_watched_through_the_route_stays_watched(db, make_user, as_user):
+    user = make_user()
+    _show(db)
+    r = as_user(user).post("/api/plex/show/sh1/state", json={"watched": True})
+    assert r.json() == {"ok": True, "changed": 2, "watched": True}
+    assert _plex_marks(db, user) == [("e1", "watched"), ("e2", "watched")]
+
+
+def test_a_season_marked_watched_through_the_route_stays_watched(db, make_user, as_user):
+    user = make_user()
+    _show(db)
+    r = as_user(user).post("/api/plex/season/se1/state", json={"watched": True})
+    assert r.json() == {"ok": True, "changed": 2, "watched": True}
+    assert _plex_marks(db, user) == [("e1", "watched"), ("e2", "watched")]
+
+
+def test_reading_one_notice_is_committed_with_its_trim(db, make_user, as_user):
+    """Marking a notice read and trimming the read ones past the cap are one write the handler
+    commits — the trim no longer commits on its own."""
+    user = make_user()
+    n = Notification(user_id=user.id, type="t", title="x")
+    db.add(n)
+    db.commit()
+    r = as_user(user).post(f"/api/me/notifications/{n.id}/read")
+    assert r.json() == {"ok": True}
+    db.expire_all()
+    assert db.get(Notification, n.id).read is True
@@ -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,
@@ -24,6 +24,7 @@ sys.path.insert(0, "/app")
 from sqlalchemy import delete, func, select

 from app.db import SessionLocal
+import app.tables  # noqa: F401 — every table, before the first query
 from app.catalog.models import Channel, Subscription, Video
 from app.models import AppState, User
 from app.reader.models import Playlist, PlaylistItem, VideoState
Author	SHA1	Message	Date
peter	b04b2cbd62	test: the rewritten write handlers commit, and three more owner filters hold Each handler that now calls a service and commits is driven through HTTP once. A second account's watch state, Watch later list and private download profile are never read or changed. The autotag job's orphan prune is seen committed.	2026-10-20 00:45:22 +02:00
peter	3a5212cd8a	fix: a first watch state and a first Watch later survive two requests at once The first checkpoint of an item and the final flush beside it both read no state and both inserted; the loser failed on uq_plex_user_item. Two first saves each made a Watch later list. Both are created by an ON CONFLICT DO NOTHING insert and then read; a partial unique index holds one Watch later per user, and the migration folds an existing second list into the first. Plex keeps one per-item state map (rollup.item_state_map).	2026-10-20 00:45:22 +02:00
peter	b376373af5	fix(files): a folder's kind is answered before its new name, as before A blank name on a source folder answered 400 name_required; it answers 409 not_own again.	2026-10-20 00:45:22 +02:00
peter	8206384c91	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.	2026-10-20 00:45:21 +02:00
peter	e27656ee7e	test: the request commit check sees Core DML and row locks A db.execute(update/delete) fires no flush and stages nothing, so a handler whose only write was Core DML could forget its commit and stay green. The check now asks Postgres whether the request opened a writing transaction. It found one real case: a repeated issue send returned with the session row still locked.	2026-10-20 00:45:20 +02:00
peter	d73cdcee97	fix(capture): every process registers every table before its first query The models now live in their units' packages, and a core model names a unit's by string (User.subscriptions -> Subscription). The download child imports neither catalog nor app.tables before db.get, so mapper configuration failed and every download, clip, edit and upload job died with child_exit. Each entry point now imports app.tables, and a test imports each one in a clean interpreter.	2026-10-20 00:45:20 +02:00