release: 0.108.1 (improvement/e42-backend-units)

The test's own writes are committed before a request, so a handler's Core DML can no longer hide
in a transaction the test had already opened. The entry-point probe answers on a marked line the
API's logging cannot mix into. A listed commit whose function no longer commits fails the fence.
CLAUDE.md and the backend README now name what an AST cannot see; the route-commit tests no longer
claim their read-back holds the commit.

CHANGELOG.md

@@ -7,6 +7,17 @@ edit the release notes and re-run the generator.
 
 Upgrading, pinning a version and rolling back: [docs/self-hosting.md](docs/self-hosting.md#upgrading).
 
+## 0.108.1 — 2026-10-20
+
+### New
+
+- Development: the backend is split into four units (downloads, reader, catalog, Plex), each owning its tables. A contract test holds what each one reads and writes, and that every query names whose rows it asks for.
+
+### Fixed
+
+- Watch later: saving the first two videos at once no longer makes two lists. An existing second list is merged into the first, its videos and its comfort-shelf pin with it.
+- Playback: the first resume point of a video or a Plex item no longer fails when it is saved together with the watched mark, a whole season marked at once, or a Plex sync.
+
 ## 0.108.0 — 2026-10-18
 
 ### New

CLAUDE.md

@@ -66,6 +66,13 @@ how it gets built. Concretely:
 `downloads/`, `reader/`, `discover/`, `catalog/`, `youtube/`, `issues/` (shared by routes + scheduler + worker) → flat core (`auth`, `apitokens`, `quota`, `jobruns`,
 `sysconfig`, `state`, `audit`, `notifications`, `userscope`, `ratelimit`, `bodylimit`, `progress`, `collation`,
 `security`, `logs`, `logsink`, `diagnostics`, `debugsession`) → `models/db/config`. **Dependency direction is one-way: core never imports routes.**
+The domain is four **units**, each owning its tables: Capture (`downloads/`), Reader (`reader/`, `discover/`, the playlist
+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`) — as far as an AST sees: what it
+cannot (an ORM write on another unit's loaded row, SQL in `text()`, a commit by `begin()`) is review's to catch, listed in
+[`backend/app/README.md`](backend/app/README.md) with the conventions.
 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
 components → the **platform layer `components/ui/`** (domain-free primitives: the popover engine and
@@ -82,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                      | `owned_or_404` / `get_or_404`                                                                                                                                                                     | ad-hoc `_own_playlist`/`_owned`/`_own_job`                                                                                                              |
-| 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; helpers `flush` (test-held list)                                                                                                                                             | helpers that commit (only `quota` and `jobruns` 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 a listed one (what owns its session, or a core step that must land whatever its caller does next), each 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

VERSION

@@ -1 +1 @@
-0.108.0
+0.108.1

backend/alembic/versions/0112_one_watch_later.py

@@ -0,0 +1,70 @@
+"""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), a Home pin
+of one of them moves to it (unless it is pinned itself; of several, the first on the shelf), then
+they go — a pin left on them goes with them, as the kept list already holds one.
+"""
+
+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"""
+        UPDATE comfort_pins c SET playlist_id = m.keep_id
+        FROM (
+            SELECT DISTINCT ON (e.keep_id) p.id AS pin_id, e.keep_id
+            FROM comfort_pins p
+            JOIN ({_EXTRA}) e ON e.id = p.playlist_id
+            WHERE NOT EXISTS (SELECT 1 FROM comfort_pins k WHERE k.playlist_id = e.keep_id)
+            ORDER BY e.keep_id, p.position, p.id
+        ) m
+        WHERE c.id = m.pin_id
+        """
+    )
+    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")

backend/app/README.md

@@ -1,9 +1,8 @@
 # Backend conventions
 
 The rules below are already followed across `backend/app` — they lived only in a few module
-docstrings (`userscope.py`, `routes/_common.py`) until this file collected them (E3 S3,
-B-core-20). When you add code, match these so the next reader doesn't have to guess "which pattern do
-I copy".
+docstrings (`userscope.py`, `routes/_common.py`) until this file collected them. When you add code,
+match these so the next reader doesn't have to guess "which pattern do I copy".
 
 ## Module layering
 
@@ -13,12 +12,13 @@ Three layers, by *who* runs the code — not by feature:
    response models, status-code mapping, serialization. It owns logic that runs *only* on a request.
    Business logic that a request happens to trigger does **not** live here if anything else runs it too.
 
-2. **`plex/`, `downloads/`, `sync/`, `issues/` — domain logic shared by more than one caller.** Anything the
-   **scheduler or the worker** runs (not just a request) lives in one of these domain packages, so the
-   route, the worker, and the scheduler share one definition. Examples from this codebase:
-   `downloads/service.py` owns enqueue + the job state transitions (routes call it); `plex/sync.py` is
-   run by both the admin route and the scheduler; `downloads/pipeline.py` is the worker's child pipeline;
-   `issues/bundle.py` is the one definition of a bug report's files, for the zip export and the tracker.
+2. **The units and the other domain packages — logic shared by more than one caller.** A unit's
+   database work (below), and anything the **scheduler or the worker** runs (not just a request), lives
+   in a domain package, so the route, the worker, and the scheduler share one definition. Examples from
+   this codebase: `downloads/service.py` owns enqueue + the job state transitions (routes call it);
+   `plex/sync.py` is run by both the admin route and the scheduler; `downloads/pipeline.py` is the
+   worker's child pipeline; `issues/bundle.py` is the one definition of a bug report's files, for the
+   zip export and the tracker.
 
 3. **Flat `app/*.py` core modules — cross-cutting primitives.** `auth` (with `apitokens`, a machine's
    folder-scoped way into the Files library), `sysconfig`, `quota`, `audit`, `jobruns` (every
@@ -34,19 +34,59 @@ Three layers, by *who* runs the code — not by feature:
 a layer — into the relevant domain package, or a flat core module if it's cross-cutting. Don't
 pre-abstract for a single caller; don't leave shared logic duplicated once there are two.
 
+## Units
+
+The domain is four units, each owning its tables (`models.py` in its package):
+
+| Unit | Package | Its routes |
+| --- | --- | --- |
+| Capture | `downloads/`, the worker | `downloads.py`, `files.py`, `public.py` |
+| Reader | `reader/`, `discover/`, `feedquery.py`, the playlist sync and tag suggestions in `sync/` | `feed.py`, `playlists.py`, `saved_views.py`, `discover.py`, `tags.py` |
+| Catalog | `catalog/`, the ingests in `sync/` | `channels.py`, `search.py`, `sync.py` |
+| Plex | `plex/` | `plex/` |
+
+- **A unit's route does no database work.** It validates, reads through the unit's `queries.py`,
+  writes through its `service.py`, serializes, and commits. It may name a unit's model as a type
+  (under `TYPE_CHECKING`), never use one as a value, and it passes a kind (`"movie"`), not a model, to
+  a query.
+- **A unit's code keeps to its edges.** It reads another unit's tables only along a declared edge
+  (Capture and Reader read Catalog's and Plex's: the feed joins `Video` with `VideoState` in one
+  query) and writes them only through the owner's service. Core code reaches a unit's tables through
+  that unit's `queries`.
+- **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`. 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 what may commit (`UNIT_COMMITS`, `CORE_COMMITS`)
+are listed there. It reads the code's AST, so it holds the forms written here: an owner column
+compared inline (a model's, an import alias's or an `aliased()` one's, by `==`, `!=`, `.in_()`, or a
+`filter_by(user_id=…)`), a model built or an `insert`/`update`/`delete` (under any import name, of a
+model or its `__table__`) across units, and a `.commit()` call off the lists all fail it. What an AST
+cannot tell is review's to catch: an ORM write on another unit's loaded row (an attribute, a
+`db.delete`), SQL in `text()`, a column or model reached another way (bound to a variable, through
+`__table__.c` or a module path), and a commit made by a `begin()` block or a `commit` handed in.
+
 ## Transaction ownership
 
-This follows CLAUDE.md's canonical rule verbatim — **"the handler/job-wrapper commits; helpers flush
-(only `quota` and `jobruns` own their session)"** — so read it that way, not as an exception to it. The commit belongs
-to the **operation owner**; anything called *inside* another transaction only stages its change
+This follows CLAUDE.md's canonical rule — **the handler commits; a unit's `service` and every helper
+flush** — so read it that way, not as an exception to it. The commit belongs to the **operation
+owner**; anything called *inside* another transaction only stages its change
 (`db.add` / `flush`) and lets the owner commit, so the whole operation is one transaction.
 `audit.record()` is the reference: it adds the audit row and the caller commits it together with the
 mutation it records, so an audit entry can never exist without its change (or vice-versa).
 
-- The **operation owner** is usually the route handler. A `service.*` operation the handler delegates a
-  whole transition to (e.g. `service.cancel`) *is* that owner and commits its one transition — it isn't
-  a mid-operation helper. A helper that owns its OWN short-lived session (the worker's `jobdb` writes,
-  `jobruns`' run rows — the same carve-out CLAUDE.md grants `quota`) commits that session. Everything else flushes.
+- 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, the
+  scheduler's jobs, a unit's jobs (the worker's `jobdb` writes, the syncs and ingests that commit per
+  page or batch so a later failure loses nothing already stored), and a few core steps that must land
+  whatever their caller does next (a token's last-used stamp, a refreshed YouTube grant) — the fence's
+  `UNIT_COMMITS` and `CORE_COMMITS` name each with its reason, `test_route_conventions` a route's own
+  committing helpers. Everything else flushes.
 - `SessionLocal` is `expire_on_commit=False`: after a raw `UPDATE` + commit, an ORM object you already
   loaded is **stale**. Read the fresh value from the row (RETURNING, or a re-`SELECT`), not the ORM
   attribute — see `service.decr_ref_count`.

backend/app/apitokens.py

@@ -26,6 +26,7 @@ from fastapi import Depends, Request
 from sqlalchemy import delete, or_, select, update
 from sqlalchemy.orm import Session
 
+from app.userscope import owned_by
 from app import audit, debugsession, logs
 from app.audit import AuditAction
 from app.auth import current_user
@@ -223,7 +224,9 @@ def revoke_all(db: Session, user_id: int) -> int:
     change, "log out everywhere", a revoked untrusted password — ends its tokens too: one minted
     from a stolen session must not outlive the owner's answer to that theft. Returns how many."""
     gone = (
-        db.execute(delete(ApiToken).where(ApiToken.user_id == user_id).returning(ApiToken.id))
+        db.execute(
+            delete(ApiToken).where(owned_by(ApiToken.user_id, user_id)).returning(ApiToken.id)
+        )
         .scalars()
         .all()
     )

backend/app/audit.py

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

backend/app/auth.py

@@ -4,6 +4,7 @@ from sqlalchemy import delete, func, select
 from sqlalchemy.orm import Session
 from starlette.requests import HTTPConnection
 
+from app.userscope import owned_by
 from app import debugsession
 from app import email as email_mod
 from app import logs
@@ -230,7 +231,7 @@ def purge_user(db: Session, user: User, background: BackgroundTasks) -> None:
     # request's access line, a queued row, a later background task would otherwise put rows back
     # under the id after this DELETE (logs.forget_user).
     logs.forget_user(user_id)
-    db.execute(delete(LogEvent).where(LogEvent.user_id == user_id))
+    db.execute(delete(LogEvent).where(owned_by(LogEvent.user_id, user_id)))
     # Every canonically-equal invite goes, not just the literal spelling: is_allowed matches on the
     # canonical form, so leaving a sibling row behind would keep granting access to a mailbox whose
     # account was just erased — and prod really does carry approved pairs for one mailbox.

backend/app/debugsession.py

@@ -26,6 +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, owned_row
 from app import diagnostics
 from app.models import DebugScreenshot, DebugSession, LogEvent, User
 
@@ -76,7 +77,9 @@ def open_session(db: Session, user: User) -> DebugSession | None:
 
 def _unclosed(db: Session, user: User) -> DebugSession | None:
     return db.scalar(
-        select(DebugSession).where(DebugSession.user_id == user.id, DebugSession.ended_at.is_(None))
+        select(DebugSession).where(
+            owned_by(DebugSession.user_id, user), DebugSession.ended_at.is_(None)
+        )
     )
 
 
@@ -126,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
 
@@ -354,7 +353,7 @@ def stored_screenshot_bytes(db: Session, user_id: int) -> int:
         db.scalar(
             select(func.coalesce(func.sum(DebugScreenshot.size), 0))
             .join(DebugSession, DebugSession.id == DebugScreenshot.session_id)
-            .where(DebugSession.user_id == user_id)
+            .where(owned_by(DebugSession.user_id, user_id))
         )
         or 0
     )

backend/app/discover/pins.py

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

backend/app/discover/queries.py

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

backend/app/downloads/folders.py

@@ -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:

backend/app/downloads/queries.py

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

backend/app/downloads/runner.py

@@ -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

backend/app/feedquery.py

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

backend/app/main.py

@@ -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

backend/app/notifications.py

@@ -9,6 +9,7 @@ table can't grow without bound for a heavy user.
 from sqlalchemy import delete, select
 from sqlalchemy.orm import Session
 
+from app.userscope import owned_by
 from app.models import Notification
 
 # Keep at most this many READ notifications per user; older read ones are pruned. Unread
@@ -48,11 +49,13 @@ 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 first, so a notice the
+    caller has just marked read counts; the caller commits."""
+    db.flush()
     ids = (
         db.execute(
             select(Notification.id)
-            .where(Notification.user_id == user_id, Notification.read.is_(True))
+            .where(owned_by(Notification.user_id, user_id), Notification.read.is_(True))
             .order_by(Notification.created_at.desc())
             .offset(cap)
         )
@@ -63,5 +66,4 @@ def trim_read(db: Session, user_id: int, cap: int = READ_SOFT_CAP) -> int:
         return 0
     # One bulk DELETE rather than a get+delete per row (the ids are already in hand).
     db.execute(delete(Notification).where(Notification.id.in_(ids)))
-    db.commit()
     return len(ids)

backend/app/plex/queries.py

@@ -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 ----------------------------------------------------------------------------------
 
 

backend/app/plex/rollup.py

@@ -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 ()`)."""

backend/app/plex/service.py

@@ -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,36 @@ def mark_finished(db: Session, user_id: int, item_id: int, now: datetime) -> boo
     return False
 
 
+def state_row(
+    db: Session,
+    user: User | int,
+    item_id: int,
+    st: PlexState | None = None,
+    *,
+    synced: bool = False,
+) -> 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. Every first row goes through
+    here: a checkpoint, a status, a season marked at once, a pull from Plex. `synced` is what a row
+    made now says about Plex; a row another writer made first keeps its own."""
+    if st is not None:
+        return st
+    db.execute(
+        pg_insert(PlexState)
+        .values(
+            user_id=user if isinstance(user, int) else user.id,
+            item_id=item_id,
+            synced_to_plex=synced,
+        )
+        .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 +355,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 +373,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,15 +390,14 @@ 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)
         prev = st.status if st is not None else "new"
         if watched:
-            if st is None:
-                st = PlexState(user_id=user_id, item_id=e.id)
-                db.add(st)
+            st = state_row(db, user_id, e.id, st)
+            prev = st.status  # a first checkpoint beside this may have made it meanwhile
             st.status = "watched"
             st.watched_at = now
             st.position_seconds = 0

backend/app/plex/watch_sync.py

@@ -32,6 +32,8 @@ from app.db import SessionLocal
 from app.userscope import owned_by
 from app.plex.models import PlexItem, PlexLibrary, PlexLink, PlexState
 from app.plex.client import PlexClient, PlexError, PlexNotConfigured
+from app.plex import queries
+from app.plex.service import state_row
 from app.plex.sync import (
     _TYPE_EPISODE,
     _TYPE_MOVIE,
@@ -252,15 +254,7 @@ def push_state_to_plex(
         # Flag the row synced — but an "unwatched" via status=new deletes the row, so there may be
         # nothing to flag; that's fine (unscrobble still happened). Skip the flag if the row was
         # re-edited since we captured `expect_ts` (leave it dirty for its own push / the reconcile).
-        st = (
-            db.execute(
-                select(PlexState).where(
-                    owned_by(PlexState.user_id, user_id), PlexState.item_id == item_id
-                )
-            )
-            .scalars()
-            .first()
-        )
+        st = queries.state(db, user_id, item_id)
         if st is not None and _row_matches(st, expect_ts):
             st.synced_to_plex = True
             db.commit()
@@ -394,15 +388,7 @@ def _pull_apply(
     """Bring one Plex-side positive signal into Siftlode under last-write-wins. Returns
     "watched"/"in_progress" if it changed the row, else None. Never clears a row (un-watch is the
     full reconcile's job) — this only pulls in watched/resume."""
-    st = (
-        db.execute(
-            select(PlexState).where(
-                owned_by(PlexState.user_id, user_id), PlexState.item_id == item_id
-            )
-        )
-        .scalars()
-        .first()
-    )
+    st = queries.state(db, user_id, item_id)
     if _same_state(st, status, pos):
         if st is not None and not st.synced_to_plex:
             st.synced_to_plex = True  # already matches Plex → just settle the flag, no churn
@@ -423,8 +409,9 @@ def _pull_apply(
         # "almost always already pushed" assumption above excludes.
         return None
     if st is None:
-        st = PlexState(user_id=user_id, item_id=item_id)
-        db.add(st)
+        st = state_row(db, user_id, item_id, synced=True)
+        if not st.synced_to_plex:  # a local change made it first, beside this pull: the guard above
+            return None
     if status == "watched":
         st.status, st.position_seconds, st.watched_at, st.progress_updated_at = (
             "watched",

backend/app/quota.py

@@ -16,6 +16,7 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.engine import CursorResult
 from sqlalchemy.orm import Session
 
+from app.userscope import owned_by
 from app import logs, sysconfig
 from app.db import SessionLocal
 from app.models import ApiQuotaUsage, QuotaEvent
@@ -173,7 +174,7 @@ def actions_today(db: Session, user_id: int, action: str) -> int:
             select(func.count())
             .select_from(QuotaEvent)
             .where(
-                QuotaEvent.user_id == user_id,
+                owned_by(QuotaEvent.user_id, user_id),
                 QuotaEvent.action == action,
                 QuotaEvent.created_at >= pacific_day_start_utc(),
             )

backend/app/reader/models.py

@@ -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)

backend/app/reader/service.py

@@ -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
 
@@ -239,24 +239,33 @@ def clear_state(db: Session, user: User, video_id: str) -> None:
 
 def set_progress(db: Session, user: User, video_id: str, position: int) -> None:
     """The player's resume point. 0 clears it: a row holding nothing else goes, one with a status
-    (watched, hidden) keeps it with the position zeroed. The status itself is never changed here."""
-    row = queries.video_state(db, user, video_id)
+    (watched, hidden) keeps it with the position zeroed. The status itself is never changed here.
+    A position is one upsert on (user, video): the first checkpoint and the end-of-playback mark
+    beside it both read "no row", and a plain add let the second one fail on `uq_user_video`."""
     now = datetime.now(timezone.utc)
-    if not position:
-        if row is None:
-            return
-        if row.status == "new":
-            db.delete(row)
-        else:
-            row.position_seconds = 0
-            row.progress_updated_at = now
-        db.flush()
+    if position:
+        db.execute(
+            pg_insert(VideoState)
+            .values(
+                user_id=user.id,
+                video_id=video_id,
+                position_seconds=position,
+                progress_updated_at=now,
+            )
+            .on_conflict_do_update(
+                constraint="uq_user_video",
+                set_={"position_seconds": position, "progress_updated_at": now, "updated_at": now},
+            )
+        )
         return
+    row = queries.video_state(db, user, video_id)
     if row is None:
-        row = VideoState(user_id=user.id, video_id=video_id)
-        db.add(row)
-    row.position_seconds = position
-    row.progress_updated_at = now
+        return
+    if row.status == "new":
+        db.delete(row)
+    else:
+        row.position_seconds = 0
+        row.progress_updated_at = now
     db.flush()
 
 
@@ -319,17 +328,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
 
 

backend/app/routes/_common.py

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

backend/app/routes/auth.py

@@ -16,6 +16,7 @@ from sqlalchemy import select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
+from app.userscope import owned_by
 from app import apitokens
 from app import email as email_mod
 from app import logs
@@ -1124,7 +1125,7 @@ def password_reset_confirm(body: TokenPasswordIn, db: Session = Depends(get_db))
     # Burn any other outstanding reset tokens for this user.
     for row in db.execute(
         select(AuthToken).where(
-            AuthToken.user_id == user.id,
+            owned_by(AuthToken.user_id, user),
             AuthToken.kind.in_(_PASSWORD_TOKEN_KINDS),
             AuthToken.used_at.is_(None),
         )

backend/app/routes/debug_sessions.py

@@ -18,6 +18,7 @@ from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
+from app.userscope import owned_by
 from app import audit, debugsession, logs, logsink
 from app.audit import AuditAction
 from app.auth import current_user
@@ -95,7 +96,7 @@ def stop_session(
 def list_sessions(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict:
     sessions = db.scalars(
         select(DebugSession)
-        .where(DebugSession.user_id == user.id)
+        .where(owned_by(DebugSession.user_id, user))
         .order_by(DebugSession.started_at.desc())
         .limit(LIST_MAX)
     ).all()
@@ -219,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:

backend/app/routes/files.py

@@ -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:

backend/app/routes/notifications.py

@@ -5,6 +5,7 @@ from fastapi import APIRouter, Depends, Query
 from sqlalchemy import func, select, update
 from sqlalchemy.orm import Session
 
+from app.userscope import owned_by
 from app.auth import current_user
 from app.db import get_db
 from app.models import Notification, User
@@ -36,7 +37,7 @@ def list_notifications(
     user: User = Depends(current_user),
     db: Session = Depends(get_db),
 ) -> dict:
-    where = [Notification.user_id == user.id]
+    where = [owned_by(Notification.user_id, user)]
     if not include_dismissed:
         where.append(Notification.dismissed.is_(False))
     total = db.scalar(select(func.count()).select_from(Notification).where(*where)) or 0
@@ -61,7 +62,7 @@ def unread_count(user: User = Depends(current_user), db: Session = Depends(get_d
         select(func.count())
         .select_from(Notification)
         .where(
-            Notification.user_id == user.id,
+            owned_by(Notification.user_id, user),
             Notification.read.is_(False),
             Notification.dismissed.is_(False),
         )
@@ -77,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}
 
 
@@ -86,11 +87,11 @@ def mark_read(
 def mark_all_read(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict:
     db.execute(
         update(Notification)
-        .where(Notification.user_id == user.id, Notification.read.is_(False))
+        .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}
 
 
@@ -113,7 +114,7 @@ def clear_all(user: User = Depends(current_user), db: Session = Depends(get_db))
     """Dismiss every (still-visible) notification for this user."""
     db.execute(
         update(Notification)
-        .where(Notification.user_id == user.id, Notification.dismissed.is_(False))
+        .where(owned_by(Notification.user_id, user), Notification.dismissed.is_(False))
         .values(dismissed=True, read=True)
     )
     db.commit()

backend/app/routes/plex/detail.py

@@ -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,

backend/app/routes/plex/playlists.py

@@ -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(

backend/app/routes/quota.py

@@ -10,6 +10,7 @@ from fastapi import APIRouter, Depends, Query
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
+from app.userscope import owned_by
 from app import quota
 from app.auth import admin_user, current_user
 from app.db import get_db
@@ -27,7 +28,7 @@ def _sum(db: Session, *conds) -> int:
 @router.get("/my-usage")
 def my_usage(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict:
     now = datetime.now(timezone.utc)
-    mine = QuotaEvent.user_id == user.id
+    mine = owned_by(QuotaEvent.user_id, user)
     by_action = {
         action: int(units)
         for action, units in db.execute(

backend/app/routes/tokens.py

@@ -16,6 +16,7 @@ from pydantic import BaseModel, Field
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
+from app.userscope import owned_by
 from app import apitokens, audit
 from app.apitokens import GRANT_COLUMNS, Right
 from app.audit import AuditAction
@@ -86,7 +87,7 @@ def _serialize(db: Session, token: ApiToken) -> dict:
 @router.get("")
 def list_tokens(user: User = Depends(current_user), db: Session = Depends(get_db)) -> list[dict]:
     tokens = (
-        db.execute(select(ApiToken).where(ApiToken.user_id == user.id).order_by(ApiToken.id))
+        db.execute(select(ApiToken).where(owned_by(ApiToken.user_id, user)).order_by(ApiToken.id))
         .scalars()
         .all()
     )
@@ -134,7 +135,7 @@ def create_token(
     if (epoch or 0) != (user.session_epoch or 0):
         raise refused(401, "not_authenticated")
     count = db.execute(
-        select(func.count(ApiToken.id)).where(ApiToken.user_id == user.id)
+        select(func.count(ApiToken.id)).where(owned_by(ApiToken.user_id, user))
     ).scalar_one()
     if count >= MAX_TOKENS:
         raise refused(409, "api_token_limit", limit=MAX_TOKENS)

backend/app/userscope.py

@@ -1,10 +1,11 @@
 """Whose a row is, and who an account is to the others — one place for both.
 
 **Ownership.** A user-owned table says whose a row is in one column (`user_id`, or `to_user_id` for
-something handed to them). A unit's reads and writes name that column through `owned_by` and fetch a
+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 a switched-on unit to it.
+someone else is answered like a missing one" then lives here. A contract test
+(`test_unit_fences.py`) holds every module, core included, to the filter's form; a rule this module
+cannot say (own or shared, built-in or own) is written out where it applies, with its reason.
 
 **Recipients.** "An active, non-suspended account" — the download share-by-email and the recipient
 picker filter on it. Kept here so the two forms (SQL clauses for a query, a Python check for a loaded

backend/app/worker.py

@@ -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

backend/tests/conftest.py

@@ -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.
 
@@ -114,15 +119,16 @@ def _request_dep(db, unfinished: list[str]):
     and the suite stays green over a handler that forgot to commit (deep review, debt burn). So a
     request that ends normally with a write it did not commit — flushed, or still only staged — is
     recorded, and the test fails on it at teardown. Only the ROOT transaction's end clears it: a
-    savepoint released inside the handler commits nothing. The test's own staged rows are flushed before the request starts — they are the
-    test's business, not the handler's. A request that raises is rolled back, as production's
+    savepoint released inside the handler commits nothing. What the test itself staged or wrote is
+    committed before the request starts — it is the test's business, not the handler's, and a
+    request then begins with no write open, so any write left at its end is the handler's. A request that raises is rolled back, as production's
     `close()` rolls it back: without that, what a failing handler flushed but never committed stayed
     readable to the rest of the test — the one state production cannot be in (an issue created
     on the tracker before a failed upload was kept only by an explicit commit, and nothing saw it
     missing)."""
 
     def dep(conn: HTTPConnection):
-        db.flush()
+        db.commit()
         state = {"dirty": False}
 
         def flushed(session, ctx):
@@ -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 — the request began with none, and a transaction
+            # that wrote and was not committed holds an id (see `release_read_connection`). A row
+            # lock (`FOR UPDATE`) takes one too: the handler ends what it locked.
+            core_dml = _write_xact(db) is not None
+            if state["dirty"] or staged or core_dml:
                 unfinished.append(f"{conn.scope.get('method', 'WS')} {conn.url.path}")
         except Exception:
             db.rollback()

backend/tests/test_autotag_orphans.py

@@ -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

backend/tests/test_entry_points.py

@@ -0,0 +1,69 @@
+"""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 json
+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")
+
+# The table list goes out on one marked line: the API's logging writes to the same stdout, and a
+# line it logs while importing is not the probe's answer.
+_MARK = "entry-point-tables:"
+_PROBE = (
+    "import importlib, json, 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"
+    f"print({_MARK!r} + json.dumps(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:])
+        answer = [ln for ln in r.stdout.splitlines() if ln.startswith(_MARK)]
+        assert len(answer) == 1, (module, r.stdout[-2000:])
+        assert json.loads(answer[0].removeprefix(_MARK)) == 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)

backend/tests/test_files_api.py

@@ -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)

backend/tests/test_migration_0112.py

@@ -0,0 +1,90 @@
+"""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 a Home
+pin of the other moved onto it, and the one-per-user index then refuses a second list.
+
+The schema half is checked by test_migrations.py; a video or a pin 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 _lists(conn, user_id: int, n: int) -> list[int]:
+    values = ", ".join(f"(:u, 'Watch later', 'watch_later', {i})" for i in range(n))
+    return list(
+        conn.execute(
+            text(
+                f"INSERT INTO playlists (user_id, name, kind, position) VALUES {values} RETURNING id"
+            ),
+            {"u": user_id},
+        ).scalars()
+    )
+
+
+def _pin(conn, user_id: int, playlist_id: int, position: int) -> None:
+    conn.execute(
+        text("INSERT INTO comfort_pins (user_id, playlist_id, position) VALUES (:u, :p, :pos)"),
+        {"u": user_id, "p": playlist_id, "pos": position},
+    )
+
+
+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:
+            first, second, third = _lists(conn, user.id, 3)
+            theirs, their_second = _lists(conn, other.id, 2)
+            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},
+            )
+            # The user pinned two of the copies, the kept one neither: the first on the shelf moves.
+            _pin(conn, user.id, third, 0)
+            _pin(conn, user.id, second, 1)
+            # The other pinned both: the kept one's pin stays, and is the only one.
+            _pin(conn, other.id, their_second, 0)
+            _pin(conn, other.id, theirs, 1)
+    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)]
+        pins = conn.execute(
+            text("SELECT user_id, playlist_id, position FROM comfort_pins ORDER BY user_id")
+        ).all()
+        assert [tuple(r) for r in pins] == [(user.id, first, 0), (other.id, theirs, 1)]
+    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},
+        )

backend/tests/test_override_dep_convention.py

@@ -82,3 +82,42 @@ 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 — also when the test itself had a write open before the
+    request, which the handler's write would otherwise share. 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, *, test_wrote: bool = False) -> None:
+        if test_wrote:  # the test's own write, open and uncommitted, as the request starts
+            db.execute(update(User).where(User.id == user.id).values(display_name="test"))
+        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
+    drive(commit=True, test_wrote=True)
+    assert db.info["unfinished"] == []
+    drive(commit=False, test_wrote=True)
+    assert db.info["unfinished"] == ["POST /probe"]
+    db.info["unfinished"].clear()

backend/tests/test_owner_scoping.py

@@ -0,0 +1,248 @@
+"""The core's owner filters, each held by a second account whose rows must not count, show or change.
+
+Every one of these reads or writes a user-owned table through `userscope.owned_by`; the fence holds
+the form (no inline `Model.user_id == …`), these hold that the filter is there at all. Each test gives
+a second account the same kind of rows, so a filter that went missing answers with theirs. DB lane.
+"""
+
+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
+
+
+def _notify(db, user, n=1, *, read=False):
+    for i in range(n):
+        db.add(Notification(user_id=user.id, type="t", title=f"n{i}", read=read))
+    db.commit()
+
+
+def _visible(db, user) -> int:
+    return db.scalar(
+        select(func.count())
+        .select_from(Notification)
+        .where(Notification.user_id == user.id, Notification.dismissed.is_(False))
+    )
+
+
+def test_the_notification_list_shows_only_mine(db, make_user, as_user):
+    me, other = make_user(), make_user()
+    _notify(db, me)
+    _notify(db, other, 2)
+    body = as_user(me).get("/api/me/notifications").json()
+    assert (body["total"], len(body["items"])) == (1, 1)
+
+
+def test_the_unread_count_counts_only_mine(db, make_user, as_user):
+    me, other = make_user(), make_user()
+    _notify(db, me)
+    _notify(db, other, 2)
+    assert as_user(me).get("/api/me/notifications/unread_count").json() == {"count": 1}
+
+
+def test_reading_all_reads_only_mine(db, make_user, as_user):
+    me, other = make_user(), make_user()
+    _notify(db, me)
+    _notify(db, other, 2)
+    assert as_user(me).post("/api/me/notifications/read_all").status_code == 200
+    assert as_user(other).get("/api/me/notifications/unread_count").json() == {"count": 2}
+
+
+def test_clearing_all_dismisses_only_mine(db, make_user, as_user):
+    me, other = make_user(), make_user()
+    _notify(db, me)
+    _notify(db, other, 2)
+    assert as_user(me).post("/api/me/notifications/clear").status_code == 200
+    db.expire_all()
+    assert (_visible(db, me), _visible(db, other)) == (0, 2)
+
+
+def test_trimming_read_notifications_trims_only_mine(db, make_user):
+    me, other = make_user(), make_user()
+    _notify(db, me, 3, read=True)
+    _notify(db, other, 3, read=True)
+    assert notifications.trim_read(db, me.id, cap=1) == 2
+    db.commit()
+    assert _visible(db, other) == 3
+
+
+def test_todays_actions_count_only_mine(db, make_user):
+    me, other = make_user(), make_user()
+    quota.log_action(me.id, quota.QuotaAction.VIDEOS_SEARCH)
+    quota.log_action(other.id, quota.QuotaAction.VIDEOS_SEARCH)
+    quota.log_action(other.id, quota.QuotaAction.VIDEOS_SEARCH)
+    assert quota.actions_today(db, me.id, quota.QuotaAction.VIDEOS_SEARCH) == 1
+
+
+def test_my_usage_sums_only_mine(db, make_user, as_user):
+    me, other = make_user(), make_user()
+    db.add_all(
+        [
+            QuotaEvent(user_id=me.id, action="videos.list", units=5),
+            QuotaEvent(user_id=other.id, action="videos.list", units=7),
+        ]
+    )
+    db.commit()
+    body = as_user(me).get("/api/quota/my-usage").json()
+    assert (body["today"], body["all_time"], body["by_action"]) == (5, 5, {"videos.list": 5})
+
+
+def test_the_token_ceiling_counts_only_mine(db, make_user, as_user, monkeypatch):
+    monkeypatch.setattr(tokens_routes, "MAX_TOKENS", 1)
+    me, other = make_user(), make_user()
+    first = as_user(other).post("/api/tokens", json={"name": "theirs", "preset": "hub"})
+    assert first.status_code == 200, first.text
+    r = as_user(me).post("/api/tokens", json={"name": "mine", "preset": "hub"})
+    assert r.status_code == 200, r.text
+
+
+def test_a_reset_burns_only_my_other_links(client, db, make_user):
+    me, other = make_user(), make_user()
+    mine = _issue_token(db, me, "reset", RESET_TTL)
+    theirs = _issue_token(db, other, "reset", RESET_TTL)
+    db.commit()
+    r = client.post(
+        "/auth/password-reset/confirm", json={"token": mine, "password": "a brand new passphrase"}
+    )
+    assert r.json() == {"ok": True}
+    assert client.post("/auth/password-reset/check", json={"token": theirs}).json() == {"ok": True}
+
+
+def test_screenshot_storage_counts_only_mine(db, make_user):
+    me, other = make_user(), make_user()
+    now = datetime.now(timezone.utc)
+    for i, (user, size) in enumerate([(me, 10), (other, 1000)]):
+        db.add(
+            DebugSession(
+                id=f"s{i}",
+                user_id=user.id,
+                level="DEBUG",
+                started_at=now,
+                until=now + timedelta(hours=1),
+            )
+        )
+        db.flush()
+        db.add(
+            DebugScreenshot(
+                session_id=f"s{i}", created_at=now, mime="image/png", size=size, data=b"x" * size
+            )
+        )
+    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]

backend/tests/test_plex_service.py

@@ -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,32 @@ 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")]
+
+
+def test_a_season_marked_beside_a_first_checkpoint_updates_its_row(db, make_user, monkeypatch):
+    """Marking a season reads every episode's state once; a first checkpoint that made one row since
+    must be updated, not collided with on `uq_plex_user_item`."""
+    user = make_user()
+    lib = _lib(db)
+    ep = _movie(db, lib, "e-race")
+    _state(db, user, ep, position_seconds=10)  # the checkpoint that won, committed
+    monkeypatch.setattr(rollup, "item_state_map", lambda *a, **k: {})  # read before it committed
+    service.bulk_state(db, user.id, [ep], watched=True)
+    db.commit()
+    rows = db.scalars(select(PlexState).where(PlexState.item_id == ep.id)).all()
+    assert [(r.status, r.position_seconds) for r in rows] == [("watched", 0)]

backend/tests/test_plex_watch_sync.py

@@ -14,6 +14,7 @@ from types import SimpleNamespace
 from sqlalchemy import select
 
 from app.models import User
+from app.plex import queries
 from app.plex.models import PlexItem, PlexLibrary, PlexState
 from app.plex.watch_sync import _pull_apply, _reconcile_inprogress_movies
 from app.routes.plex.playback import (
@@ -385,3 +386,30 @@ def test_should_push_resume_periodic_is_throttled():
     )
     # Once the window elapses, the next tick pushes again.
     assert _should_push_resume(*key, now + timedelta(seconds=_PUSH_THROTTLE_S), final=False) is True
+
+
+def test_a_pull_that_lost_the_race_keeps_the_local_change_it_met(db, monkeypatch):
+    """The pull read "no state", and a first local checkpoint made one before it wrote: that row is
+    still waiting for its push, so the pull leaves it (the dirty guard), and fails on nothing."""
+    item_id, user_id = _seed_item_and_user(db)
+    now = datetime.now(timezone.utc)
+    db.add(PlexState(user_id=user_id, item_id=item_id, position_seconds=300, synced_to_plex=False))
+    db.commit()
+    real = queries.state
+    stale = [True]
+    monkeypatch.setattr(
+        queries, "state", lambda d, u, i: None if stale and stale.pop() else real(d, u, i)
+    )
+    assert _pull_apply(db, user_id, item_id, "new", 120, plex_ts=now) is None
+    db.commit()
+    rows = db.scalars(select(PlexState).where(PlexState.item_id == item_id)).all()
+    assert [(r.position_seconds, r.synced_to_plex) for r in rows] == [(300, False)]
+
+
+def test_a_first_pull_makes_a_row_in_step_with_plex(db):
+    item_id, user_id = _seed_item_and_user(db)
+    now = datetime.now(timezone.utc)
+    assert _pull_apply(db, user_id, item_id, "new", 120, plex_ts=now) == "in_progress"
+    db.commit()
+    rows = db.scalars(select(PlexState).where(PlexState.item_id == item_id)).all()
+    assert [(r.position_seconds, r.synced_to_plex) for r in rows] == [(120, True)]

backend/tests/test_reader_service.py

@@ -308,3 +308,36 @@ 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]
+
+
+def test_a_first_resume_point_beside_the_watched_mark_lands_on_its_row(db, make_user, monkeypatch):
+    """The first checkpoint of a video and the end-of-playback mark both read "no row"; the
+    checkpoint that writes second must update the mark's row, not fail on `uq_user_video`."""
+    user = make_user()
+    _videos(db, "v1")
+    service.mark(db, user, "v1", "watched")  # the mark that won, committed
+    db.commit()
+    monkeypatch.setattr(queries, "video_state", lambda *a, **k: None)  # read before it committed
+    service.set_progress(db, user, "v1", 90)
+    db.commit()
+    db.expire_all()
+    rows = db.execute(select(VideoState)).scalars().all()
+    assert [(r.status, r.position_seconds) for r in rows] == [("watched", 90)]

backend/tests/test_route_commits.py

@@ -0,0 +1,383 @@
+"""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. The read-back holds that the handler called its unit; it runs in the
+request's own session, so it sees a write that was never committed — the commit is the fixture's
+to hold (a test that must see it committed reads from a session of its own). 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, timedelta, 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.db import SessionLocal
+from app.models import Notification
+from app.notifications import READ_SOFT_CAP
+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 _read_past_the_cap(db, user) -> list[Notification]:
+    """READ_SOFT_CAP read notices, oldest first, and one unread notice after them."""
+    base = datetime(2024, 1, 1, tzinfo=timezone.utc)
+    rows = [
+        Notification(
+            user_id=user.id,
+            type="t",
+            title=f"n{i}",
+            read=i < READ_SOFT_CAP,
+            created_at=base + timedelta(minutes=i),
+        )
+        for i in range(READ_SOFT_CAP + 1)
+    ]
+    db.add_all(rows)
+    db.commit()
+    return rows
+
+
+def _committed_notices(user) -> tuple[int, bool]:
+    """What another session sees: how many of the user's notices are left, and whether the oldest is."""
+    with SessionLocal() as s:
+        ids = s.scalars(select(Notification.title).where(Notification.user_id == user.id)).all()
+    return len(ids), "n0" in ids
+
+
+@pytest.mark.parametrize("path", ["{id}/read", "read_all"])
+def test_reading_a_notice_past_the_cap_commits_the_trim(db, make_user, as_user, path):
+    """Marking a notice read makes READ_SOFT_CAP + 1 read ones: the handler commits the read flag and
+    the trim of the oldest together, and the trim counts the notice just read."""
+    user = make_user()
+    rows = _read_past_the_cap(db, user)
+    r = as_user(user).post("/api/me/notifications/" + path.format(id=rows[-1].id))
+    assert r.json() == {"ok": True}
+    assert _committed_notices(user) == (READ_SOFT_CAP, False)

backend/tests/test_unit_fences.py

@@ -5,17 +5,14 @@ What it may not do is build SQL, call the session's data methods, change a model
 import a unit's tables: when a route did those, ownership, the commit and the conditional `WHERE`
 lived in a different place in each handler, and a fix in one never reached its siblings.
 
-The rule is switched on unit by unit. `ALLOWED` pins what each HTTP file still does today, and the
-pin is exact: a count that rises fails (new database work in a route), and one that falls fails
-until its entry is lowered (the fence only tightens, as `baseline.json` does for basedpyright). An
-empty `ALLOWED` is the end state — a file listed there is a debt, not an exemption.
+The rule has no exemptions: a single `select()` in a unit's route file fails it.
 
 A core route (identity, admin, ops) is not a unit's HTTP layer and may use the session; it may
 still not reach into a unit's tables. A route may NAME a unit's model as a type (under
 `TYPE_CHECKING`), never use one as a value.
 
-Below the HTTP layer, a fenced unit's own code keeps to its edges (`FENCED`, `READS`): see the
-second half of this file.
+Below the HTTP layer, a unit's own code keeps to its edges (`READS`): see the second half of this
+file. Whose a row is, every module says through `userscope` — a core one too.
 """
 
 import ast
@@ -107,10 +104,6 @@ UNIT_ROUTES = {
     "plex/": "plex",
 }
 
-# What each HTTP file still does, by file: the debt a unit's sprint pays off. Exact counts. A core
-# route's count is only the unit tables it imports; it is paid by the sprint of the unit it reaches.
-ALLOWED: dict[str, int] = {}
-
 # The session methods that read or write data. `commit` and `rollback` stay with the handler.
 SESSION_DATA = {
     "add",
@@ -257,19 +250,13 @@ def test_every_listed_route_exists():
     rels = [rel for rel, _ in _http_files()]
     for key in UNIT_ROUTES:
         assert any(rel == key or (key.endswith("/") and rel.startswith(key)) for rel in rels), key
-    assert set(ALLOWED) <= set(rels), set(ALLOWED) - set(rels)
 
 
-def test_no_http_file_does_database_work_beyond_its_pin():
-    drift = {}
-    for rel, src in _http_files():
-        found = violations(src, _unit_of(rel))
-        pinned = ALLOWED.get(rel, 0)
-        if len(found) != pinned:
-            drift[rel] = (pinned, len(found), found if len(found) > pinned else "lower its pin")
-    assert drift == {}, (
-        "an HTTP file's database work moved off its pin (pinned, now, what): move new work into"
-        " the unit's queries/service; when a count fell, lower ALLOWED to it"
+def test_no_http_file_does_database_work():
+    hits = {rel: found for rel, src in _http_files() if (found := violations(src, _unit_of(rel)))}
+    assert hits == {}, (
+        "a route does database work: reads go through the unit's queries, writes through its"
+        " service; a core route reaches a unit's tables through that unit's queries too"
     )
 
 
@@ -307,11 +294,11 @@ def test_the_detector_sees_each_kind_of_database_work():
 
 
 # --- inside a unit: its package --------------------------------------------------------------
-# A unit whose sprint has landed is fenced beyond its HTTP layer too. Its own code reads another
-# unit's tables only along a declared edge (the feed joins Video with VideoState in one query; a
-# boundary that split it would cost the speed the app is for), and writes another unit's tables
-# never — that goes through the owner's service. Whose a row is, it says through `userscope`, and
-# it commits only where a job owns its session.
+# Every unit is fenced beyond its HTTP layer too. Its own code reads another unit's tables only
+# along a declared edge (the feed joins Video with VideoState in one query; a boundary that split
+# it would cost the speed the app is for), and writes another unit's tables never — that goes
+# through the owner's service. It commits only where a job owns its session. Whose a row is, it
+# says through `userscope`, and so does core code: the owner filter is checked in every module.
 
 # Each unit's code outside the HTTP layer: its package, and the entry point that runs its jobs.
 UNIT_PACKAGES = {
@@ -341,9 +328,7 @@ UNIT_PACKAGES = {
     # pages' reads and writes.
     "plex": ("plex/",),
 }
-# The units fenced so far; the rest join with their sprint.
-FENCED = {"capture", "reader", "catalog", "plex"}
-# Whose tables a unit's own code may READ. Core code reads a fenced unit's tables through its queries.
+# Whose tables a unit's own code may READ. Core code reads a unit's tables through its queries.
 READS = {
     "capture": {"catalog", "plex"},
     "reader": {"catalog", "plex"},
@@ -352,7 +337,7 @@ READS = {
 }
 # The columns that say whose a row is.
 OWNER_COLUMNS = {"user_id", "to_user_id", "from_user_id", "created_by"}
-# Where a fenced unit's code may commit: the jobs that own their session (the worker and what it
+# Where a unit's code may commit: the jobs that own their session (the worker and what it
 # runs, the GC, the backfill — none of them inside a request), and the one write that must commit
 # where it stands: an upload's rows and the file it placed on disk land together or not at all, so
 # a failed commit can remove the file (`upload.ingest_upload`). Everything else flushes; the handler
@@ -423,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:
@@ -455,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):
@@ -465,64 +539,82 @@ def _enclosing_functions(tree: ast.AST) -> dict[ast.AST, str]:
 
 
 def package_violations(rel: str, source: str) -> list[str]:
-    """What a module outside the HTTP layer does against a fenced unit's boundary, as `line: what`."""
+    """What a module outside the HTTP layer does against a unit's boundary, as `line: what`."""
     tree = ast.parse(source)
     mine = _package_unit(rel)
     found = []
     models = _imported_models(tree)
     for name in models.values():
         theirs = UNIT_MODEL_NAMES[name]
-        if theirs == mine or (mine not in FENCED and theirs not in FENCED):
+        if theirs == mine:
             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 and (mine in FENCED or theirs in FENCED):
+        if theirs != mine:
             found.append(f"{node.lineno}: {mine or 'core'} writes {theirs}'s {written}")
-    if mine in FENCED:
-        found += _owner_filters(tree)
-        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)}")
+    found += _owner_filters(tree)
+    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
 
 
-def test_a_fenced_units_code_keeps_to_its_edges():
+def test_a_units_code_keeps_to_its_edges():
     hits = {rel: found for rel, src in _app_files() if (found := package_violations(rel, src))}
     assert hits == {}, (
         "read another unit's tables only along READS, write them only through their service, say"
@@ -530,12 +622,8 @@ def test_a_fenced_units_code_keeps_to_its_edges():
     )
 
 
-def test_a_fenced_units_routes_filter_owners_through_userscope():
-    hits = {
-        rel: found
-        for rel, src in _http_files()
-        if _unit_of(rel) in FENCED and (found := _owner_filters(ast.parse(src)))
-    }
+def test_every_route_filters_owners_through_userscope():
+    hits = {rel: found for rel, src in _http_files() if (found := _owner_filters(ast.parse(src)))}
     assert hits == {}
 
 
@@ -543,11 +631,32 @@ 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
 
 
+def test_every_listed_commit_is_still_there():
+    """An entry names a commit that exists: one left behind by a rename, or by a function that no
+    longer commits, would go on allowing a commit to whatever takes its name next."""
+    stale = []
+    for commits in (*UNIT_COMMITS.values(), CORE_COMMITS):
+        for entry in commits:
+            rel, _, name = entry.partition(":")
+            tree = ast.parse((APP / rel).read_text(encoding="utf-8"))
+            functions = _enclosing_functions(tree)
+            committing = {
+                functions.get(node)
+                for node in ast.walk(tree)
+                if isinstance(node, ast.Call)
+                and isinstance(node.func, ast.Attribute)
+                and node.func.attr == "commit"
+            }
+            if (name not in committing) if name else not committing:
+                stale.append(entry)
+    assert stale == []
+
+
 PACKAGE_CASES = {
     # (module path, source): violations
     ("downloads/x.py", "from app.models import Video"): 0,  # capture reads catalog
@@ -562,6 +671,23 @@ PACKAGE_CASES = {
     ("sync/videos.py", "from app.models import Playlist"): 1,  # no catalog→reader edge
     ("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,

docs/gates.md

@@ -113,7 +113,6 @@ project checks needs a row.
 | CVE gate | HIGH/CRITICAL with a fix available only | `--ignore-unfixed` + `--severity HIGH,CRITICAL`: a finding with no fix cannot be acted on by an upgrade; it shows in the scan output. A fixable one fails the publish, and the only way past it is `.trivyignore.yaml`. | 0 open (E32 S1) |
 | npm audit | `--omit=dev --audit-level=high` | What ships to the browser, at high or above. The dev toolchain's advisories were burned down by hand in E32 S1 (18 → 0). | 0 |
 | knip | `entry` — one file added | Not a narrowing: knip's Vite plugin reads entries from index.html only, so the shared link's page (`watch.html` → `src/watchMain.tsx`) is named here, or the lane reports that whole entry unused. It adds files to the lane, never takes any out. | 1 entry |
-| unit fence | the database work the HTTP layer still does, pinned | `backend/tests/test_unit_fences.py`: a unit's route file may not build SQL, call the session's data methods, set a model's attributes or import a unit's tables; a core route may not import a unit's tables. Switched on unit by unit: `ALLOWED` pins each file's count exactly, so a rise fails and a fall fails until the pin is lowered — the list may only shrink. End state: empty, and the pin mechanism goes. A unit in `FENCED` is held below its HTTP layer too: its code reads another unit's tables only along `READS`, writes them never, filters owners through `userscope`, and commits only where `UNIT_COMMITS` names a job's own session. | 826 in 23 files (E42 S0) → 667 in 19 (S1: Capture 0, fenced) → 525 in 14 (S2: Reader 0, fenced) → 354 in 9 (S3: Catalog 0, fenced; core routes 0) → 178 in 4 (S4a: Plex browse, detail, people, collections 0) → 0 (S4b: Plex 0, fenced — every unit fenced) |
 | React Compiler | the functions it skips, pinned | `components/compilerCoverage.contract.test.ts` — the list may only shrink. End state: only VirtualGrid (TanStack Virtual, an IncompatibleLibrary) remains. | 93 (E32 S1) → 29 (E32 S3) → 1, VirtualGrid (E32 S4) |
 
 ## Accepted findings

frontend/src/content/release-notes/en.json

@@ -1,4 +1,15 @@
 [
+  {
+    "version": "0.108.1",
+    "date": "2026-10-20",
+    "new": [
+      "Development: the backend is split into four units (downloads, reader, catalog, Plex), each owning its tables. A contract test holds what each one reads and writes, and that every query names whose rows it asks for."
+    ],
+    "fixed": [
+      "Watch later: saving the first two videos at once no longer makes two lists. An existing second list is merged into the first, its videos and its comfort-shelf pin with it.",
+      "Playback: the first resume point of a video or a Plex item no longer fails when it is saved together with the watched mark, a whole season marked at once, or a Plex sync."
+    ]
+  },
   {
     "version": "0.108.0",
     "date": "2026-10-18",

frontend/src/content/release-notes/hu.json

@@ -1,4 +1,15 @@
 [
+  {
+    "version": "0.108.1",
+    "date": "2026-10-20",
+    "new": [
+      "Fejlesztés: a backend négy egységre bomlik (letöltések, olvasó, katalógus, Plex), mindegyik a saját tábláival. Egy szerződésteszt őrzi, melyik mit olvas és ír, és hogy minden lekérdezés megnevezi, kinek a sorait kéri."
+    ],
+    "fixed": [
+      "Megnézem később: ha az első két videót egyszerre mented, már nem lesz két lista. Egy meglévő második lista beolvad az elsőbe, a videóival és a komfort polcos kitűzésével együtt.",
+      "Lejátszás: egy videó vagy Plex-elem első folytatási pontja már nem hiúsul meg, ha egyszerre ment a megnézett jelöléssel, egy egész évad jelölésével vagy egy Plex-szinkronnal."
+    ]
+  },
   {
     "version": "0.108.0",
     "date": "2026-10-18",

scripts/e2e_seed.py

@@ -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