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

@@ -70,9 +70,9 @@ The domain is four **units**, each owning its tables: Capture (`downloads/`), Re
 sync), Catalog (`catalog/`, the `sync/` ingests) and Plex (`plex/`). A unit's route does no database work: it reads through
 the unit's `queries`, writes through its `service` (which flushes) and commits. A unit's code reads another unit's tables
 only along a declared edge and writes them only through that unit's service; core reaches a unit's tables through its
-`queries`. **Fence-held, with no exemptions** (`backend/tests/test_unit_fences.py`) — as far as an AST sees: an ORM
-attribute write on another unit's loaded row is review's to catch. Conventions in
-[`backend/app/README.md`](backend/app/README.md).
+`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
@@ -99,7 +99,7 @@ CQRS, a repository/DI framework, a router library, a generic CRUD-router factory
 | Upstream failure status              | `422` with JSON detail                                                                                                                                                                                                                        | `502` (reserve 502 for raw byte-proxies)                                                                                                                                                    |
 | User-facing error                    | structured `detail` (`{code, reason}`)                                                                                                                                                                                                        | a bare English string under a bilingual UI                                                                                                                                                  |
 | Mutation response                    | `{"ok": true, …}` (`{"deleted": id}`), test-held                                                                                                                                                                                              | a bare `{"unsubscribed"}`/`{"cleared": n}`, no `ok`                                                                                                                                         |
-| Transaction commit                   | the handler commits; a unit's `service` and every helper `flush` (fence-held, every module)                                                                                                                                                   | a helper that commits — only what owns its session does, each named with its reason: the fence's `UNIT_COMMITS` / `CORE_COMMITS`, a route's `COMMITTING_HELPERS` (`test_route_conventions`) |
+| 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                                                                                                                                                                |

VERSION

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

backend/alembic/versions/0112_one_watch_later.py

@@ -7,7 +7,9 @@ Create Date: 2026-10-20
 Watch later was made on a user's first save by a read then an insert, so two first saves at once
 each made a list, and the reader then picked between them at random. A partial unique index holds
 "one per user" now. An instance that already has a second list keeps the oldest: the others' videos
-move into it (after its own, in their order; one it already holds is not added twice), then they go.
+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
@@ -41,6 +43,19 @@ def upgrade() -> None:
         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",

backend/app/README.md

@@ -62,10 +62,13 @@ The domain is four units, each owning its tables (`models.py` in its package):
 
 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: an inline owner filter in any form (an alias, `.in_()`,
-`filter_by`), a model built or a statement write (`insert`/`update`/`delete` under any name) across
-units, and a commit off the lists all fail it. An ORM attribute write on another unit's loaded row
-is beyond it — review's to catch.
+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
 

backend/app/notifications.py

@@ -49,7 +49,9 @@ 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. Flushes; the caller commits."""
+    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)

backend/app/plex/service.py

@@ -302,16 +302,29 @@ def mark_finished(db: Session, user_id: int, item_id: int, now: datetime) -> boo
     return False
 
 
-def _state_row(db: Session, user: User, item_id: int, st: PlexState | None) -> PlexState:
+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."""
+    `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.id, item_id=item_id)
+        .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)
@@ -342,7 +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."""
-    st = _state_row(db, user, item_id, st)
+    st = state_row(db, user, item_id, st)
     st.position_seconds = position
     st.progress_updated_at = now
     if dirty:
@@ -360,7 +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."""
-    st = _state_row(db, user, item_id, st)
+    st = state_row(db, user, item_id, st)
     st.status = status
     if status == "watched":
         st.watched_at = datetime.now(timezone.utc)
@@ -383,9 +396,8 @@ def bulk_state(
         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/reader/service.py

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

backend/tests/conftest.py

@@ -119,17 +119,17 @@ 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}
-        before = _write_xact(db)
 
         def flushed(session, ctx):
             state["dirty"] = True
@@ -145,10 +145,10 @@ def _request_dep(db, unfinished: list[str]):
             yield db
             staged = db.new or db.deleted or any(db.is_modified(o) for o in db.dirty)
             # Core DML (`db.execute(update(...))`) fires no flush and stages nothing; the database
-            # itself says whether a write is open — a transaction that wrote and was not committed
-            # holds an id it did not hold when the request began (see `release_read_connection`). A
-            # row lock (`FOR UPDATE`) takes one too: the handler ends what it locked.
-            core_dml = (after := _write_xact(db)) is not None and after != before
+            # 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:

backend/tests/test_entry_points.py

@@ -9,6 +9,7 @@ schema on its own.
 """
 
 import ast
+import json
 import subprocess
 import sys
 from pathlib import Path
@@ -22,13 +23,16 @@ 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, sys\n"
+    "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"
-    "print('\\n'.join(sorted(Base.metadata.tables)))\n"
+    f"print({_MARK!r} + json.dumps(sorted(Base.metadata.tables)))\n"
 )
 
 
@@ -44,7 +48,9 @@ def test_each_entry_point_registers_the_whole_schema_on_its_own():
             check=False,
         )
         assert r.returncode == 0, (module, r.stderr[-2000:])
-        assert r.stdout.split() == whole, module
+        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():

backend/tests/test_migration_0112.py

@@ -1,8 +1,8 @@
 """Migration 0112's data step, DB-lane: a user who already has two Watch later lists keeps the oldest,
-with the other's videos moved in after its own (one it already held is not added twice), and the
-one-per-user index then refuses a second list.
+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 lost on the way is invisible there.
+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
@@ -17,6 +17,25 @@ 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"))
@@ -28,15 +47,8 @@ def test_a_second_watch_later_is_folded_into_the_first(db, make_user):
     command.downgrade(cfg, "0111_audit_request_id")
     try:
         with engine.begin() as conn:
-            ids = conn.execute(
-                text(
-                    "INSERT INTO playlists (user_id, name, kind, position) VALUES"
-                    " (:u, 'Watch later', 'watch_later', 0), (:u, 'Watch later', 'watch_later', 1),"
-                    " (:o, 'Watch later', 'watch_later', 0) RETURNING id"
-                ),
-                {"u": user.id, "o": other.id},
-            ).scalars()
-            first, second, theirs = list(ids)
+            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"
@@ -44,6 +56,12 @@ def test_a_second_watch_later_is_folded_into_the_first(db, make_user):
                 ),
                 {"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")
 
@@ -58,6 +76,10 @@ def test_a_second_watch_later_is_folded_into_the_first(db, make_user):
             )
         ).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(

backend/tests/test_override_dep_convention.py

@@ -87,7 +87,8 @@ def test_a_request_that_raises_leaves_nothing_it_did_not_commit(db):
 def test_the_commit_check_sees_a_core_dml_write(db, make_user):
     """`db.execute(update(...))` flushes nothing and stages nothing, so the session's own bookkeeping
     never sees it: a handler whose only write is Core DML and that forgot its commit passed. The
-    database's transaction id is what tells. A committed one must still pass."""
+    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
@@ -96,7 +97,9 @@ def test_the_commit_check_sees_a_core_dml_write(db, make_user):
 
     user = make_user()
 
-    def drive(commit: bool) -> None:
+    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""}
         )
@@ -113,3 +116,8 @@ def test_the_commit_check_sees_a_core_dml_write(db, make_user):
     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_plex_service.py

@@ -436,3 +436,17 @@ def test_a_first_write_that_lost_the_race_updates_the_winners_row(db, make_user)
     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

@@ -326,3 +326,18 @@ def test_a_watch_later_save_that_lost_the_race_uses_the_winners_list(db, make_us
     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

@@ -2,12 +2,14 @@
 
 A route's helper only flushes; the handler commits. The `db` fixture fails a test whose request left
 a write uncommitted, so every test here drives one successful write through its route and then
-reads the result back — a handler that lost its commit fails on both counts. The services' own
+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, timezone
+from datetime import datetime, timedelta, timezone
 
 import pytest
 from sqlalchemy import select
@@ -17,7 +19,9 @@ 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
 
@@ -343,14 +347,37 @@ def test_a_season_marked_watched_through_the_route_stays_watched(db, make_user,
     assert _plex_marks(db, user) == [("e1", "watched"), ("e2", "watched")]
 
 
-def test_reading_one_notice_is_committed_with_its_trim(db, make_user, as_user):
-    """Marking a notice read and trimming the read ones past the cap are one write the handler
-    commits — the trim no longer commits on its own."""
-    user = make_user()
-    n = Notification(user_id=user.id, type="t", title="x")
-    db.add(n)
+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()
-    r = as_user(user).post(f"/api/me/notifications/{n.id}/read")
+    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}
-    db.expire_all()
-    assert db.get(Notification, n.id).read is True
+    assert _committed_notices(user) == (READ_SOFT_CAP, False)

backend/tests/test_unit_fences.py

@@ -636,6 +636,27 @@ def test_every_unit_package_exists():
             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

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