release: 0.75.0 (improvement/quickfix-batch-2)

- Button: `loading` always disables (|| not ??) — stops a double-submit
  when a call-site passes an explicit boolean `disabled` (VideoEditor).
- playlists: track un-ingestable-but-live YouTube ids in the sync baseline
  so a linked playlist containing a private/deleted video converges instead
  of re-classifying it as a pending pull forever; re-attempt ingestion each
  sync so a restored (private->public) video heals, and prune ids that
  leave YouTube.
- SettingsPanel: download-auth section renders an honest error/loading
  state (StateMessage + retry) instead of a bare empty section on a failed
  credentials fetch; HU+EN loadError string.
- tests: reorder-order, partial-failure baseline hold, un-ingestable
  converge/heal/prune, Button loading-disables, download-auth error/retry.

.git-blame-ignore-revs

@@ -4,3 +4,6 @@
 #
 # Prettier one-time repo-wide format pass (R2 S1b, 2026-07-21)
 51c36c2071728d2cd7ae580b7bfdf21f6867a347
+
+# ruff format one-time backend + scripts pass (line-length 100, 2026-08-23)
+b1832afd180866b922521148603db447020c57a4

.gitea/workflows/check.yml

@@ -59,6 +59,8 @@ jobs:
       - run: pip install --no-cache-dir -r backend/requirements.lock -r backend/requirements-dev.txt
       - name: ruff
         run: python -m ruff check --config backend/ruff.toml backend scripts
+      - name: ruff format --check
+        run: python -m ruff format --check --config backend/ruff.toml backend scripts
       - name: pytest
         run: python -m pytest
         working-directory: backend

CLAUDE.md

@@ -12,12 +12,12 @@ provider.
 
 ## The gate — never merge or ship red
 
-The local gate is **ruff + typecheck (3 tsc projects) + eslint + prettier --check + knip + vitest +
+The local gate is **ruff (`check` + `format --check`) + typecheck (3 tsc projects) + eslint + prettier
-pytest**. It runs on every `publish`, which **refuses to build on failure — there is deliberately no
+--check + knip + vitest + pytest**. It runs on every `publish`, which **refuses to build on failure — there is deliberately no
 `--skip` flag.** Run it and get it green **before** any merge, release, or handing work to review.
 
 - Full gate (agents with the private ops dispatcher): `siftlode check`.
-- Directly, without the dispatcher: `cd frontend && npm run typecheck && npm run lint && npm run knip && npm run test`, plus backend `ruff check backend` and `pytest` (the DB-lane needs Postgres).
+- Directly, without the dispatcher: `cd frontend && npm run typecheck && npm run lint && npm run knip && npm run test`, plus backend `ruff check backend` + `ruff format --check backend` and `pytest` (the DB-lane needs Postgres).
 - New DB-backed test? It must actually run, not silently skip — assume `SIFTLODE_TEST_REQUIRE_DB=1`.
 
 ## Workflow
@@ -26,12 +26,33 @@ pytest**. It runs on every `publish`, which **refuses to build on failure — th
   keep commits logical and Conventional-Commit-styled.
 - **No AI attribution in commits.** No `Co-Authored-By` trailer, no "generated by" lines. (This file
   being AI-written is the intended, visible exception; commits are not.)
-- **Release is a deliberate, user-owned step** (`siftlode release` from the feature branch: merge to
+- **Release: user-owned in timing, agent-run in execution.** The user decides *when* to ship; on their
-  `main` → publish → deploy → verify `/api/version`). Do not self-ship; bring work to a green,
+  go-ahead *you* run `siftlode release` from the feature branch (merge to `main` → publish → deploy →
-  committed state and stop for the user's go-ahead.
+  verify `/api/version`). Never self-ship — bring work to a green, committed state and stop for the
+  go-ahead first, then run it yourself.
 - **Public repo.** Keep ops/topology/secrets out — those live in a separate private repo. Nothing here
   should assume or leak them.
 
+## Collaboration — how to work with the human
+
+The human is the product owner: they bring ideas and requirements. You are the technical lead — you own
+how it gets built. Concretely:
+
+- **Plan before you build.** For anything non-trivial, present a short plan (sprint-level for larger
+  work) and get a go-ahead before writing code.
+- **Offer choices, mark the optimal — and explain the trade-off.** When several solutions are viable,
+  list them and mark the recommended one. This is the *one* place to spend extra words: give the *why*
+  behind each option and the weight of the decision. These decision points are where the product owner
+  engages with the build and learns from it, so make them understandable — a reasoned menu, not a bare one.
+- **Mockups for design/UX decisions.** For any visual or interaction choice, show mockup options to
+  choose from before building.
+- **Challenge unsound requests.** If a request is technically ill-advised and you can back it with real
+  arguments, push back or propose an alternative — don't comply blindly.
+- **Verify UI in the browser before calling it done.** tsc/unit/DOM tests miss visual and interaction
+  bugs; do a real visual check of any UI change.
+- **Be terse everywhere else.** Outside those decision points, lead with the answer or decision. During
+  implementation and debugging especially: no restating the same point reworded, no filler.
+
 ## Architecture shape (don't fight it)
 
 `routes/*` (HTTP only: validation, auth-dependency, serialization) → domain packages `sync/`, `plex/`,
@@ -79,5 +100,6 @@ CQRS, a repository/DI framework, a router library, a generic CRUD-router factory
 
 ---
 
-*A fuller written-conventions doc (`docs/development.md`) is planned; once it lands, this file slims to a
+*Run-it-locally and the full gate commands live in [`docs/development.md`](docs/development.md); backend
-pointer plus the gate/workflow/no-attribution rules. Until then, the table above is the source of truth.*
+code conventions in [`backend/app/README.md`](backend/app/README.md). This file stays the source of
+truth for the canonical-patterns table and the gate/workflow/collaboration rules above.*

VERSION

@@ -1 +1 @@
-0.71.0
+0.75.0

backend/alembic/versions/0001_initial.py

@@ -4,6 +4,7 @@ Revision ID: 0001_initial
 Revises:
 Create Date: 2026-06-11
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0002_ingest_core.py

@@ -4,6 +4,7 @@ Revision ID: 0002_ingest_core
 Revises: 0001_initial
 Create Date: 2026-06-11
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0003_tags.py

@@ -4,6 +4,7 @@ Revision ID: 0003_tags
 Revises: 0002_ingest_core
 Create Date: 2026-06-11
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0004_feed_state_prefs.py

@@ -4,6 +4,7 @@ Revision ID: 0004_feed_state_prefs
 Revises: 0003_tags
 Create Date: 2026-06-11
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0005_shorts_probed.py

@@ -4,6 +4,7 @@ Revision ID: 0005_shorts_probed
 Revises: 0004_feed_state_prefs
 Create Date: 2026-06-11
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0006_app_state.py

@@ -4,6 +4,7 @@ Revision ID: 0006_app_state
 Revises: 0005_shorts_probed
 Create Date: 2026-06-11
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0007_deep_requested.py

@@ -4,6 +4,7 @@ Revision ID: 0007_deep_requested
 Revises: 0006_app_state
 Create Date: 2026-06-11
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0008_invites.py

@@ -4,6 +4,7 @@ Revision ID: 0008_invites
 Revises: 0007_deep_requested
 Create Date: 2026-06-12
 """
+
 from typing import Sequence, Union
 
 from alembic import op
@@ -22,9 +23,7 @@ def upgrade() -> None:
         "invites",
         sa.Column("id", sa.Integer(), primary_key=True),
         sa.Column("email", sa.String(length=320), nullable=False),
-        sa.Column(
+        sa.Column("status", sa.String(length=16), nullable=False, server_default="pending"),
-            "status", sa.String(length=16), nullable=False, server_default="pending"
-        ),
         sa.Column("note", sa.Text(), nullable=True),
         sa.Column(
             "requested_at",

backend/alembic/versions/0009_quota_events.py

@@ -4,6 +4,7 @@ Revision ID: 0009_quota_events
 Revises: 0008_invites
 Create Date: 2026-06-12
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0010_watch_progress.py

@@ -4,6 +4,7 @@ Revision ID: 0010_watch_progress
 Revises: 0009_quota_events
 Create Date: 2026-06-14
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0011_playlists.py

@@ -4,6 +4,7 @@ Revision ID: 0011_playlists
 Revises: 0010_watch_progress
 Create Date: 2026-06-15
 """
+
 from typing import Sequence, Union
 
 from alembic import op
@@ -26,16 +27,10 @@ def upgrade() -> None:
             nullable=False,
         ),
         sa.Column("name", sa.String(length=255), nullable=False),
-        sa.Column(
+        sa.Column("kind", sa.String(length=16), nullable=False, server_default="user"),
-            "kind", sa.String(length=16), nullable=False, server_default="user"
+        sa.Column("source", sa.String(length=16), nullable=False, server_default="local"),
-        ),
-        sa.Column(
-            "source", sa.String(length=16), nullable=False, server_default="local"
-        ),
         sa.Column("yt_playlist_id", sa.String(length=64), nullable=True),
-        sa.Column(
+        sa.Column("dirty", sa.Boolean(), nullable=False, server_default="false"),
-            "dirty", sa.Boolean(), nullable=False, server_default="false"
-        ),
         sa.Column("position", sa.Integer(), nullable=False, server_default="0"),
         sa.Column(
             "created_at",

backend/alembic/versions/0012_watch_later.py

@@ -7,6 +7,7 @@ Create Date: 2026-06-15
 Migrates the old per-video status='saved' into membership of a built-in 'watch_later'
 playlist (one per user who had saved videos), then demotes those states to 'new'.
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0013_playlist_fingerprint.py

@@ -11,6 +11,7 @@ a YouTube read. Backfills the baseline for already-clean linked playlists (the c
 state == the synced state for them); dirty ones are left NULL so they stay dirty until the
 next real sync establishes a baseline.
 """
+
 import hashlib
 from typing import Sequence, Union
 
@@ -32,24 +33,18 @@ def _fingerprint(name: str, ids: list[str]) -> str:
 
 
 def upgrade() -> None:
-    op.add_column(
+    op.add_column("playlists", sa.Column("synced_fingerprint", sa.String(), nullable=True))
-        "playlists", sa.Column("synced_fingerprint", sa.String(), nullable=True)
-    )
     bind = op.get_bind()
     # Backfill clean, YouTube-linked playlists: their current state IS the synced state.
     rows = bind.execute(
-        sa.text(
+        sa.text("SELECT id, name FROM playlists WHERE yt_playlist_id IS NOT NULL AND dirty = false")
-            "SELECT id, name FROM playlists "
-            "WHERE yt_playlist_id IS NOT NULL AND dirty = false"
-        )
     ).fetchall()
     for pid, name in rows:
         ids = [
             r[0]
             for r in bind.execute(
                 sa.text(
-                    "SELECT video_id FROM playlist_items "
+                    "SELECT video_id FROM playlist_items WHERE playlist_id = :pid ORDER BY position"
-                    "WHERE playlist_id = :pid ORDER BY position"
                 ),
                 {"pid": pid},
             ).fetchall()

backend/alembic/versions/0014_demo_account.py

@@ -11,6 +11,7 @@ Adds the shared demo-account plumbing:
 The demo user row itself is created lazily on first demo login (works across all three DBs
 without a data migration), so this only adds the schema.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -47,9 +48,7 @@ def upgrade() -> None:
             nullable=False,
         ),
     )
-    op.create_index(
+    op.create_index("ix_demo_whitelist_email", "demo_whitelist", ["email"], unique=True)
-        "ix_demo_whitelist_email", "demo_whitelist", ["email"], unique=True
-    )
 
 
 def downgrade() -> None:

backend/alembic/versions/0015_scheduler_settings.py

@@ -7,6 +7,7 @@ Create Date: 2026-06-16
 Adds scheduler_settings: a per-job run-interval override (minutes), editable from the admin
 Scheduler dashboard. Absent row = use the env/config default, so this is purely additive.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0016_video_maintenance.py

@@ -10,6 +10,7 @@ the validation lifecycle (last_checked_at for rolling re-validation; unavailable
 unavailable_reason to flag a video as currently unplayable — hidden from the feed
 immediately, hard-deleted after a grace period). Purely additive; all columns nullable.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -26,7 +27,9 @@ def upgrade() -> None:
     op.add_column("videos", sa.Column("privacy_status", sa.String(length=16), nullable=True))
     op.add_column("videos", sa.Column("upload_status", sa.String(length=16), nullable=True))
     op.add_column("videos", sa.Column("last_checked_at", sa.DateTime(timezone=True), nullable=True))
-    op.add_column("videos", sa.Column("unavailable_since", sa.DateTime(timezone=True), nullable=True))
+    op.add_column(
+        "videos", sa.Column("unavailable_since", sa.DateTime(timezone=True), nullable=True)
+    )
     op.add_column("videos", sa.Column("unavailable_reason", sa.String(length=24), nullable=True))
     op.create_index("ix_videos_last_checked_at", "videos", ["last_checked_at"])
     op.create_index("ix_videos_unavailable_since", "videos", ["unavailable_since"])

backend/alembic/versions/0017_notifications.py

@@ -9,6 +9,7 @@ JSON payload column is named `data` (SQLAlchemy's declarative Base reserves `met
 First producer is the maintenance job's batched "videos removed" notice. Coexists with the
 existing client-side transient bell.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -36,18 +37,14 @@ def upgrade() -> None:
         sa.Column("data", sa.JSON(), nullable=True),
         sa.Column("read", sa.Boolean(), nullable=False, server_default="false"),
         sa.Column("dismissed", sa.Boolean(), nullable=False, server_default="false"),
-        sa.Column(
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
-            "created_at", sa.DateTime(timezone=True), server_default=sa.func.now()
-        ),
     )
     op.create_index("ix_notifications_user_id", "notifications", ["user_id"])
     op.create_index("ix_notifications_type", "notifications", ["type"])
     op.create_index("ix_notifications_read", "notifications", ["read"])
     op.create_index("ix_notifications_created_at", "notifications", ["created_at"])
     # Fast unread-badge count + inbox listing per user.
-    op.create_index(
+    op.create_index("ix_notifications_user_read", "notifications", ["user_id", "read"])
-        "ix_notifications_user_read", "notifications", ["user_id", "read"]
-    )
 
 
 def downgrade() -> None:

backend/alembic/versions/0018_maintenance_batch_setting.py

@@ -8,6 +8,7 @@ Adds app_state.maintenance_revalidate_batch: an admin-tunable override (from the
 dashboard) for how many videos the maintenance job re-validates per run. NULL = use the
 env/config default. Purely additive.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0019_user_yt_channel_id.py

@@ -9,6 +9,7 @@ channels.list?mine=true and cached. Used to exclude the user's own channel from
 playlist-discovery list (you don't subscribe to yourself). NULL = not resolved yet.
 Purely additive.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0020_rename_quota_actions.py

@@ -10,6 +10,7 @@ stored historical values to the canonical scheme defined in app.quota.QuotaActio
 fully reversible. (Scheduler job ids and progress phase labels are a separate namespace and
 are NOT touched.)
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0021_system_config.py

@@ -8,6 +8,7 @@ Adds the system_config table: admin overrides for operational config that otherw
 from env/config defaults (see app.sysconfig for the registry of known keys). Secret values
 are stored Fernet-encrypted (encrypted flag). Purely additive; absent rows = use defaults.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0022_password_auth.py

@@ -9,6 +9,7 @@ google_sub nullable (password-only accounts have no Google sub until they link),
 single-use auth_tokens table for email verification + password reset. Existing rows are all
 Google/demo accounts → backfilled email_verified=true (is_active defaults true).
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0023_user_suspension.py

@@ -8,6 +8,7 @@ Adds users.is_suspended — an admin-controlled access block, distinct from is_a
 approval lifecycle). A suspended account can't sign in (password or Google) and any live
 session is rejected; the user is told why and given the operator's contact.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0024_google_email_verified.py

@@ -9,6 +9,7 @@ email_verified=false (the callback didn't set it). Backfill those so the flag re
 matters because password-login gates on email_verified, so a Google user who later adds a password
 must count as verified. Demo/password-only rows (google_sub IS NULL) are left untouched.
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0025_install_wizard.py

@@ -8,6 +8,7 @@ Adds app_state.configured + setup_token_hash for the first-run install wizard. E
 (any users present) are marked configured so they never see the wizard; a genuinely fresh DB (no
 users) starts unconfigured and boots into setup mode.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0026_messages.py

@@ -8,6 +8,7 @@ Adds the `messages` table for direct user-to-user messaging (notification module
 A conversation is the set of rows between two users; both FKs cascade so a GDPR account
 erasure removes that user's messages.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0027_message_e2ee.py

@@ -12,6 +12,7 @@ hold each user's public key + passphrase-wrapped private key (the server can nev
 Any pre-existing rows were plaintext with no keys, so they're dropped (the feature is unshipped;
 prod has none).
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0028_live_search_provenance.py

@@ -11,6 +11,7 @@ the Library by default:
   - channels.from_search   — this channel exists only because search surfaced one of its videos
 Both default false (every existing row is organic), indexed for the feed's exclude filter.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -25,16 +26,12 @@ depends_on: Union[str, Sequence[str], None] = None
 def upgrade() -> None:
     op.add_column(
         "videos",
-        sa.Column(
+        sa.Column("via_search", sa.Boolean(), nullable=False, server_default="false"),
-            "via_search", sa.Boolean(), nullable=False, server_default="false"
-        ),
     )
     op.create_index("ix_videos_via_search", "videos", ["via_search"])
     op.add_column(
         "channels",
-        sa.Column(
+        sa.Column("from_search", sa.Boolean(), nullable=False, server_default="false"),
-            "from_search", sa.Boolean(), nullable=False, server_default="false"
-        ),
     )
     op.create_index("ix_channels_from_search", "channels", ["from_search"])
 

backend/alembic/versions/0029_unaccent_search.py

@@ -10,6 +10,7 @@ query wraps both sides in unaccent(); this migration just makes the function ava
 `unaccent` ships with the official postgres image (postgresql-contrib). The DB role is a
 superuser in our deployments, so CREATE EXTENSION succeeds; IF NOT EXISTS keeps it idempotent.
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0030_search_finds.py

@@ -9,6 +9,7 @@ live YouTube search. This lets "videos I searched for" appear in that user's Min
 Mine Source filter), independent of the global `videos.via_search` flag (which marks a video as
 search-discovered by anyone, for the shared Library's Source filter).
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0031_title_fts.py

@@ -15,6 +15,7 @@ search stays accent-insensitive ("tiesto" matches "Tiësto"). A GIN expression i
 query uses the exact same expression so the planner picks the index. `unaccent` is already
 enabled (migration 0029). The config-based 2-arg to_tsvector is IMMUTABLE, so it's index-safe.
 """
+
 from typing import Sequence, Union
 
 from alembic import op
@@ -26,9 +27,7 @@ depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
-    op.execute(
+    op.execute("CREATE TEXT SEARCH CONFIGURATION public.unaccent_simple (COPY = pg_catalog.simple)")
-        "CREATE TEXT SEARCH CONFIGURATION public.unaccent_simple (COPY = pg_catalog.simple)"
-    )
     op.execute(
         """
         ALTER TEXT SEARCH CONFIGURATION public.unaccent_simple

backend/alembic/versions/0032_search_document.py

@@ -21,6 +21,7 @@ title-only expression index from 0031 is dropped (superseded). Existing rows get
 description indexed immediately; keywords/search_terms fill in as videos are (re-)enriched and
 searched. Uses the `unaccent_simple` config from 0031 (accent-insensitive).
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -48,9 +49,7 @@ def upgrade() -> None:
         f"ALTER TABLE videos ADD COLUMN search_vector tsvector "
         f"GENERATED ALWAYS AS ({_SEARCH_VECTOR_EXPR}) STORED"
     )
-    op.execute(
+    op.execute("CREATE INDEX ix_videos_search_vector ON videos USING gin (search_vector)")
-        "CREATE INDEX ix_videos_search_vector ON videos USING gin (search_vector)"
-    )
 
 
 def downgrade() -> None:

backend/alembic/versions/0033_channel_explore.py

@@ -17,6 +17,7 @@ un-subscribed channels):
     access to its ephemeral videos and drives the grace-period purge of un-kept channels.
 All new flags default false / NULL, so every existing row stays organic.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -30,9 +31,7 @@ depends_on: Union[str, Sequence[str], None] = None
 
 def upgrade() -> None:
     op.add_column("channels", sa.Column("total_view_count", sa.BigInteger(), nullable=True))
-    op.add_column(
+    op.add_column("channels", sa.Column("published_at", sa.DateTime(timezone=True), nullable=True))
-        "channels", sa.Column("published_at", sa.DateTime(timezone=True), nullable=True)
-    )
     op.add_column("channels", sa.Column("banner_url", sa.String(length=1024), nullable=True))
     op.add_column("channels", sa.Column("external_links", sa.JSON(), nullable=True))
     op.add_column(

backend/alembic/versions/0034_blocked_channels.py

@@ -8,6 +8,7 @@ Adds the `blocked_channels` table: a per-user blocklist. Videos from a blocked c
 never shown in that user's live YouTube search (and not ingested), and are hidden from their
 feed / explore views. Per-user, so one user's block doesn't affect anyone else.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0035_saved_views.py

@@ -8,6 +8,7 @@ Adds the `saved_views` table: per-user named snapshots of the feed's filter/sort
 row stores the serialized FeedFilters (JSON), a sidebar `position` for drag-reorder, and an
 `is_default` flag (at most one per user, enforced by the API) for the view applied on load.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -26,9 +27,7 @@ def upgrade() -> None:
         sa.Column("user_id", sa.Integer(), nullable=False),
         sa.Column("name", sa.String(length=80), nullable=False),
         sa.Column("filters", sa.JSON(), nullable=False),
-        sa.Column(
+        sa.Column("position", sa.Integer(), server_default=sa.text("0"), nullable=False),
-            "position", sa.Integer(), server_default=sa.text("0"), nullable=False
-        ),
         sa.Column(
             "is_default",
             sa.Boolean(),

backend/alembic/versions/0036_download_center.py

@@ -10,6 +10,7 @@ FOR UPDATE SKIP LOCKED, live progress written back for polling); `download_profi
 reusable format presets (builtins seeded in the next migration); `download_shares` grant a file
 to another user; `download_quotas` hold admin-set per-user limits (default via sysconfig).
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -47,9 +48,16 @@ def upgrade() -> None:
         sa.Column("ref_count", sa.Integer(), server_default="0", nullable=False),
         sa.Column("last_access_at", sa.DateTime(timezone=True), nullable=True),
         sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
         sa.PrimaryKeyConstraint("id"),
-        sa.UniqueConstraint("source_kind", "source_ref", "format_sig", name="uq_media_asset_cache_key"),
+        sa.UniqueConstraint(
+            "source_kind", "source_ref", "format_sig", name="uq_media_asset_cache_key"
+        ),
     )
     op.create_index("ix_media_assets_status", "media_assets", ["status"])
 
@@ -61,7 +69,12 @@ def upgrade() -> None:
         sa.Column("spec", sa.JSON(), nullable=False),
         sa.Column("is_builtin", sa.Boolean(), server_default="false", nullable=False),
         sa.Column("sort_order", sa.Integer(), server_default="0", nullable=False),
-        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
         sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="CASCADE"),
         sa.PrimaryKeyConstraint("id"),
     )
@@ -84,8 +97,18 @@ def upgrade() -> None:
         sa.Column("phase", sa.String(length=32), nullable=True),
         sa.Column("error", sa.Text(), nullable=True),
         sa.Column("queue_pos", sa.Integer(), server_default="0", nullable=False),
-        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column(
-        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
         sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="CASCADE"),
         sa.ForeignKeyConstraint(["asset_id"], ["media_assets.id"], ondelete="SET NULL"),
         sa.ForeignKeyConstraint(["profile_id"], ["download_profiles.id"], ondelete="SET NULL"),
@@ -102,7 +125,12 @@ def upgrade() -> None:
         sa.Column("job_id", sa.Integer(), nullable=False),
         sa.Column("from_user_id", sa.Integer(), nullable=False),
         sa.Column("to_user_id", sa.Integer(), nullable=False),
-        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
         sa.ForeignKeyConstraint(["job_id"], ["download_jobs.id"], ondelete="CASCADE"),
         sa.ForeignKeyConstraint(["from_user_id"], ["users.id"], ondelete="CASCADE"),
         sa.ForeignKeyConstraint(["to_user_id"], ["users.id"], ondelete="CASCADE"),
@@ -120,7 +148,12 @@ def upgrade() -> None:
         sa.Column("max_concurrent", sa.Integer(), nullable=False),
         sa.Column("max_jobs", sa.Integer(), nullable=False),
         sa.Column("unlimited", sa.Boolean(), server_default="false", nullable=False),
-        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
         sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="CASCADE"),
         sa.PrimaryKeyConstraint("user_id"),
     )

backend/alembic/versions/0037_download_builtin_profiles.py

@@ -10,6 +10,7 @@ definition consumed by app.downloads.formats: mode ("av" audio+video / "v" video
 the embed_*/sponsorblock postprocessor flags. Names are English (the UI default); the frontend
 localizes builtin labels by a stable key derived from the name.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0038_asset_gc_notified.py

@@ -7,6 +7,7 @@ Create Date: 2026-07-03
 The download GC warns each holder once, a grace window before an asset's TTL expiry; this flag
 records that so subsequent GC runs don't re-notify until the asset is deleted.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0039_video_title_normalize.py

@@ -9,6 +9,7 @@ display-friendly form (emoji stripped, ALL-CAPS de-shouted, trailing SEO hashtag
 see app.titles). Reversible: original_title preserves the source, and title can be re-derived.
 The generated search_vector column regenerates automatically as each title is updated.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -45,5 +46,7 @@ def upgrade() -> None:
 
 def downgrade() -> None:
     conn = op.get_bind()
-    conn.execute(sa.text("UPDATE videos SET title = original_title WHERE original_title IS NOT NULL"))
+    conn.execute(
+        sa.text("UPDATE videos SET title = original_title WHERE original_title IS NOT NULL")
+    )
     op.drop_column("videos", "original_title")

backend/alembic/versions/0040_reset_builtin_profiles.py

@@ -11,6 +11,7 @@ rows so both fresh installs (0037 then this) and existing DBs converge on the sa
 jobs are unaffected (they snapshot their spec; profile_id just SET NULLs if it pointed at a
 replaced row).
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0041_download_edit.py

@@ -10,6 +10,7 @@ still hangs off a `media_assets` row (with `source_kind='edit'` and a per-user c
 existing file-serve / ref-count / GC / quota machinery is reused unchanged. `source_job_id` /
 `source_asset_id` point at the parent download; `edit_spec` is the trim/crop recipe.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -24,16 +25,10 @@ depends_on: Union[str, Sequence[str], None] = None
 def upgrade() -> None:
     op.add_column(
         "download_jobs",
-        sa.Column(
+        sa.Column("job_kind", sa.String(length=16), server_default="download", nullable=False),
-            "job_kind", sa.String(length=16), server_default="download", nullable=False
-        ),
-    )
-    op.add_column(
-        "download_jobs", sa.Column("source_job_id", sa.Integer(), nullable=True)
-    )
-    op.add_column(
-        "download_jobs", sa.Column("source_asset_id", sa.Integer(), nullable=True)
     )
+    op.add_column("download_jobs", sa.Column("source_job_id", sa.Integer(), nullable=True))
+    op.add_column("download_jobs", sa.Column("source_asset_id", sa.Integer(), nullable=True))
     op.add_column("download_jobs", sa.Column("edit_spec", sa.JSON(), nullable=True))
     op.create_foreign_key(
         "fk_download_jobs_source_job",
@@ -54,12 +49,8 @@ def upgrade() -> None:
 
 
 def downgrade() -> None:
-    op.drop_constraint(
+    op.drop_constraint("fk_download_jobs_source_asset", "download_jobs", type_="foreignkey")
-        "fk_download_jobs_source_asset", "download_jobs", type_="foreignkey"
+    op.drop_constraint("fk_download_jobs_source_job", "download_jobs", type_="foreignkey")
-    )
-    op.drop_constraint(
-        "fk_download_jobs_source_job", "download_jobs", type_="foreignkey"
-    )
     op.drop_column("download_jobs", "edit_spec")
     op.drop_column("download_jobs", "source_asset_id")
     op.drop_column("download_jobs", "source_job_id")

backend/alembic/versions/0042_download_links.py

@@ -9,6 +9,7 @@ can watch the file on the public `/watch/{token}` player page — no account nee
 per-link controls: an expiry, an "allow download" toggle, and an (argon2-hashed) password. Revoke
 = delete the row. Distinct from `download_shares` (an ACL grant to another *registered* user).
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -31,7 +32,12 @@ def upgrade() -> None:
         sa.Column("password_hash", sa.Text(), nullable=True),
         sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True),
         sa.Column("view_count", sa.Integer(), server_default="0", nullable=False),
-        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
         sa.ForeignKeyConstraint(["job_id"], ["download_jobs.id"], ondelete="CASCADE"),
         sa.ForeignKeyConstraint(["created_by"], ["users.id"], ondelete="CASCADE"),
         sa.PrimaryKeyConstraint("id"),

backend/alembic/versions/0043_asset_source_url.py

@@ -8,6 +8,7 @@ Stores yt-dlp's canonical `webpage_url` for a download's source, so the Download
 clean "downloaded from" reference (copyable + openable). Nullable — filled by the worker on
 extraction; existing/queued rows fall back to a URL derived from source_kind+source_ref.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0044_plex_catalog.py

@@ -18,6 +18,7 @@ Plex server (playback is from the local physical file; Plex is metadata only). T
 The generated `search_vector` columns + their GIN indexes are added via raw SQL (same approach
 as migration 0032), reusing the `unaccent_simple` text-search config from migration 0031.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -52,8 +53,12 @@ def upgrade() -> None:
         sa.Column("kind", sa.String(length=16), nullable=False),
         sa.Column("enabled", sa.Boolean(), server_default="true", nullable=False),
         sa.Column("synced_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column(
-        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
     )
     op.create_index("ix_plex_libraries_plex_key", "plex_libraries", ["plex_key"], unique=True)
 
@@ -61,7 +66,12 @@ def upgrade() -> None:
         "plex_shows",
         sa.Column("id", sa.Integer(), primary_key=True),
         sa.Column("rating_key", sa.String(length=32), nullable=False),
-        sa.Column("library_id", sa.Integer(), sa.ForeignKey("plex_libraries.id", ondelete="CASCADE"), nullable=False),
+        sa.Column(
+            "library_id",
+            sa.Integer(),
+            sa.ForeignKey("plex_libraries.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
         sa.Column("title", sa.String(length=512), nullable=False),
         sa.Column("summary", sa.Text(), nullable=True),
         sa.Column("year", sa.Integer(), nullable=True),
@@ -69,8 +79,12 @@ def upgrade() -> None:
         sa.Column("art_key", sa.String(length=512), nullable=True),
         sa.Column("child_count", sa.Integer(), nullable=True),
         sa.Column("added_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column(
-        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
     )
     op.create_index("ix_plex_shows_rating_key", "plex_shows", ["rating_key"], unique=True)
     op.create_index("ix_plex_shows_library_id", "plex_shows", ["library_id"])
@@ -81,12 +95,21 @@ def upgrade() -> None:
         "plex_seasons",
         sa.Column("id", sa.Integer(), primary_key=True),
         sa.Column("rating_key", sa.String(length=32), nullable=False),
-        sa.Column("show_id", sa.Integer(), sa.ForeignKey("plex_shows.id", ondelete="CASCADE"), nullable=False),
+        sa.Column(
+            "show_id",
+            sa.Integer(),
+            sa.ForeignKey("plex_shows.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
         sa.Column("title", sa.String(length=255), nullable=True),
         sa.Column("season_number", sa.Integer(), nullable=True),
         sa.Column("thumb_key", sa.String(length=512), nullable=True),
-        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column(
-        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
     )
     op.create_index("ix_plex_seasons_rating_key", "plex_seasons", ["rating_key"], unique=True)
     op.create_index("ix_plex_seasons_show_id", "plex_seasons", ["show_id"])
@@ -96,10 +119,25 @@ def upgrade() -> None:
         "plex_items",
         sa.Column("id", sa.Integer(), primary_key=True),
         sa.Column("rating_key", sa.String(length=32), nullable=False),
-        sa.Column("library_id", sa.Integer(), sa.ForeignKey("plex_libraries.id", ondelete="CASCADE"), nullable=False),
+        sa.Column(
+            "library_id",
+            sa.Integer(),
+            sa.ForeignKey("plex_libraries.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
         sa.Column("kind", sa.String(length=16), nullable=False),
-        sa.Column("show_id", sa.Integer(), sa.ForeignKey("plex_shows.id", ondelete="CASCADE"), nullable=True),
+        sa.Column(
-        sa.Column("season_id", sa.Integer(), sa.ForeignKey("plex_seasons.id", ondelete="CASCADE"), nullable=True),
+            "show_id",
+            sa.Integer(),
+            sa.ForeignKey("plex_shows.id", ondelete="CASCADE"),
+            nullable=True,
+        ),
+        sa.Column(
+            "season_id",
+            sa.Integer(),
+            sa.ForeignKey("plex_seasons.id", ondelete="CASCADE"),
+            nullable=True,
+        ),
         sa.Column("season_number", sa.Integer(), nullable=True),
         sa.Column("episode_number", sa.Integer(), nullable=True),
         sa.Column("title", sa.String(length=512), nullable=False),
@@ -119,8 +157,12 @@ def upgrade() -> None:
         sa.Column("probed_at", sa.DateTime(timezone=True), nullable=True),
         sa.Column("markers", sa.JSON(), nullable=True),
         sa.Column("added_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column(
-        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
     )
     op.create_index("ix_plex_items_rating_key", "plex_items", ["rating_key"], unique=True)
     op.create_index("ix_plex_items_library_id", "plex_items", ["library_id"])
@@ -137,14 +179,23 @@ def upgrade() -> None:
     op.create_table(
         "plex_states",
         sa.Column("id", sa.Integer(), primary_key=True),
-        sa.Column("user_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
+        sa.Column(
-        sa.Column("item_id", sa.Integer(), sa.ForeignKey("plex_items.id", ondelete="CASCADE"), nullable=False),
+            "user_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False
+        ),
+        sa.Column(
+            "item_id",
+            sa.Integer(),
+            sa.ForeignKey("plex_items.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
         sa.Column("status", sa.String(length=12), server_default="new", nullable=False),
         sa.Column("position_seconds", sa.Integer(), server_default="0", nullable=False),
         sa.Column("watched_at", sa.DateTime(timezone=True), nullable=True),
         sa.Column("progress_updated_at", sa.DateTime(timezone=True), nullable=True),
         sa.Column("synced_to_plex", sa.Boolean(), server_default="false", nullable=False),
-        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
         sa.UniqueConstraint("user_id", "item_id", name="uq_plex_user_item"),
     )
     op.create_index("ix_plex_states_user_id", "plex_states", ["user_id"])

backend/alembic/versions/0045_plex_filter_meta.py

@@ -9,6 +9,7 @@ Plex section listing (no per-item API calls): rating (audienceRating ~ IMDb), co
 release date, and GIN-indexed genre / director / cast arrays. These also double as the content
 signals a future watch-habit recommender would use.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -33,9 +34,13 @@ def upgrade() -> None:
     op.create_index("ix_plex_items_rating", "plex_items", ["rating"])
     op.create_index("ix_plex_items_content_rating", "plex_items", ["content_rating"])
     op.create_index("ix_plex_items_studio", "plex_items", ["studio"])
-    op.create_index("ix_plex_items_originally_available_at", "plex_items", ["originally_available_at"])
+    op.create_index(
+        "ix_plex_items_originally_available_at", "plex_items", ["originally_available_at"]
+    )
     op.create_index("ix_plex_items_genres_gin", "plex_items", ["genres"], postgresql_using="gin")
-    op.create_index("ix_plex_items_directors_gin", "plex_items", ["directors"], postgresql_using="gin")
+    op.create_index(
+        "ix_plex_items_directors_gin", "plex_items", ["directors"], postgresql_using="gin"
+    )
     op.create_index("ix_plex_items_cast_gin", "plex_items", ["cast_names"], postgresql_using="gin")
 
 

backend/alembic/versions/0046_plex_people_search.py

@@ -10,6 +10,7 @@ finds titles by an actor/director name and ranks those above films that merely m
 their synopsis. A generated column's expression can't be altered in place, so we drop + recreate it
 (and its GIN index). `people_text` is populated on the next Plex sync.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0047_plex_collections.py

@@ -10,6 +10,7 @@ A `plex_collections` table holds each collection's card metadata; membership is
 combinable filter and a title's "collection strip" is a pure local lookup. `editable` marks the
 (curated) collections we may later write back to Plex; everything is read-only by default.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -27,7 +28,12 @@ def upgrade() -> None:
         "plex_collections",
         sa.Column("id", sa.Integer(), primary_key=True),
         sa.Column("rating_key", sa.String(length=32), nullable=False),
-        sa.Column("library_id", sa.Integer(), sa.ForeignKey("plex_libraries.id", ondelete="CASCADE"), nullable=False),
+        sa.Column(
+            "library_id",
+            sa.Integer(),
+            sa.ForeignKey("plex_libraries.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
         sa.Column("title", sa.String(length=512), nullable=False),
         sa.Column("summary", sa.Text(), nullable=True),
         sa.Column("thumb_key", sa.String(length=512), nullable=True),
@@ -35,16 +41,32 @@ def upgrade() -> None:
         sa.Column("smart", sa.Boolean(), nullable=False, server_default="false"),
         sa.Column("editable", sa.Boolean(), nullable=False, server_default="false"),
         sa.Column("synced_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column(
-        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+    )
+    op.create_index(
+        "ix_plex_collections_rating_key", "plex_collections", ["rating_key"], unique=True
     )
-    op.create_index("ix_plex_collections_rating_key", "plex_collections", ["rating_key"], unique=True)
     op.create_index("ix_plex_collections_library_id", "plex_collections", ["library_id"])
 
     op.add_column("plex_items", sa.Column("collection_keys", postgresql.JSONB(), nullable=True))
-    op.create_index("ix_plex_items_collections_gin", "plex_items", ["collection_keys"], postgresql_using="gin")
+    op.create_index(
+        "ix_plex_items_collections_gin", "plex_items", ["collection_keys"], postgresql_using="gin"
+    )
     op.add_column("plex_shows", sa.Column("collection_keys", postgresql.JSONB(), nullable=True))
-    op.create_index("ix_plex_shows_collections_gin", "plex_shows", ["collection_keys"], postgresql_using="gin")
+    op.create_index(
+        "ix_plex_shows_collections_gin", "plex_shows", ["collection_keys"], postgresql_using="gin"
+    )
 
 
 def downgrade() -> None:

backend/alembic/versions/0048_plex_playlists.py

@@ -9,6 +9,7 @@ Plex account (like plex_states) — the "your own lists" counterpart to the shar
 collections. `plex_playlists` is owned by a user; `plex_playlist_items` holds the ordered entries
 (`position`). `plex_rating_key` is reserved for an optional later Plex-direction sync.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -24,11 +25,23 @@ def upgrade() -> None:
     op.create_table(
         "plex_playlists",
         sa.Column("id", sa.Integer(), primary_key=True),
-        sa.Column("user_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
+        sa.Column(
+            "user_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False
+        ),
         sa.Column("title", sa.String(length=512), nullable=False),
         sa.Column("plex_rating_key", sa.String(length=32), nullable=True),
-        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column(
-        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
     )
     op.create_index("ix_plex_playlists_user_id", "plex_playlists", ["user_id"])
 
@@ -36,12 +49,16 @@ def upgrade() -> None:
         "plex_playlist_items",
         sa.Column("id", sa.Integer(), primary_key=True),
         sa.Column(
-            "playlist_id", sa.Integer(),
+            "playlist_id",
-            sa.ForeignKey("plex_playlists.id", ondelete="CASCADE"), nullable=False,
+            sa.Integer(),
+            sa.ForeignKey("plex_playlists.id", ondelete="CASCADE"),
+            nullable=False,
         ),
         sa.Column(
-            "item_id", sa.Integer(),
+            "item_id",
-            sa.ForeignKey("plex_items.id", ondelete="CASCADE"), nullable=False,
+            sa.Integer(),
+            sa.ForeignKey("plex_items.id", ondelete="CASCADE"),
+            nullable=False,
         ),
         sa.Column("position", sa.Integer(), nullable=False, server_default="0"),
         sa.UniqueConstraint("playlist_id", "item_id", name="uq_plex_playlist_item"),

backend/alembic/versions/0049_download_display_meta.py

@@ -13,6 +13,7 @@ and captures yt-dlp's canonical channel/uploader URL so an auto-filled channel b
   overrides (like display_name — display metadata only, never touches the on-disk file).
 - download_jobs.extra_links: optional list of extra reference URLs shown on the card + watch page.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -26,7 +27,9 @@ depends_on: Union[str, Sequence[str], None] = None
 
 def upgrade() -> None:
     op.add_column("media_assets", sa.Column("uploader_url", sa.Text(), nullable=True))
-    op.add_column("download_jobs", sa.Column("display_uploader", sa.String(length=255), nullable=True))
+    op.add_column(
+        "download_jobs", sa.Column("display_uploader", sa.String(length=255), nullable=True)
+    )
     op.add_column("download_jobs", sa.Column("display_uploader_url", sa.Text(), nullable=True))
     op.add_column("download_jobs", sa.Column("extra_links", sa.JSON(), nullable=True))
 

backend/alembic/versions/0050_asset_poster.py

@@ -9,6 +9,7 @@ download shows a blank image box. The worker now extracts a representative frame
 saves it as a poster sidecar; `poster_path` (relative to download_root) records it so the card /
 watch page / link preview can fall back to it. Nullable — only set when there was no thumbnail.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0051_plex_link.py

@@ -5,6 +5,7 @@ state with a Plex account. The owner MVP uses the server admin token (uses_admin
 token); a personal token column is reserved for later friend-account linking (PIN OAuth). See the
 PlexLink model docstring.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0052_plex_show_meta.py

@@ -11,6 +11,7 @@ blob folded into the generated `search_vector` (weight B). This lets the TV grid
 sorted like the movie grid. A generated column's expression can't be altered in place, so the
 search_vector is dropped + recreated. All new columns are populated on the next Plex sync.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -57,9 +58,13 @@ def upgrade() -> None:
     op.create_index("ix_plex_shows_rating", "plex_shows", ["rating"])
     op.create_index("ix_plex_shows_content_rating", "plex_shows", ["content_rating"])
     op.create_index("ix_plex_shows_studio", "plex_shows", ["studio"])
-    op.create_index("ix_plex_shows_originally_available_at", "plex_shows", ["originally_available_at"])
+    op.create_index(
+        "ix_plex_shows_originally_available_at", "plex_shows", ["originally_available_at"]
+    )
     op.create_index("ix_plex_shows_genres_gin", "plex_shows", ["genres"], postgresql_using="gin")
-    op.create_index("ix_plex_shows_directors_gin", "plex_shows", ["directors"], postgresql_using="gin")
+    op.create_index(
+        "ix_plex_shows_directors_gin", "plex_shows", ["directors"], postgresql_using="gin"
+    )
     op.create_index("ix_plex_shows_cast_gin", "plex_shows", ["cast_names"], postgresql_using="gin")
     # NB: ix_plex_shows_collections_gin already exists (created with collections in 0047).
 

backend/alembic/versions/0053_user_session_epoch.py

@@ -5,6 +5,7 @@ current_user rejects any cookie whose epoch is stale. Bumped on password reset,
 or "log out everywhere" — so a stolen/copied client-side cookie dies on those events instead of
 staying valid until it expires. Existing rows default to 0 (matching a fresh cookie's absent/0 epoch).
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0054_audit_log.py

@@ -4,6 +4,7 @@ Revision ID: 0054_audit_log
 Revises: 0053_user_session_epoch
 Create Date: 2026-07-12
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0055_channel_keywords.py

@@ -4,6 +4,7 @@ Revision ID: 0055_channel_keywords
 Revises: 0054_audit_log
 Create Date: 2026-07-12
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0056_plex_people.py

@@ -4,6 +4,7 @@ Revision ID: 0056_plex_people
 Revises: 0055_channel_keywords
 Create Date: 2026-07-19
 """
+
 from typing import Sequence, Union
 
 from alembic import op
@@ -23,8 +24,12 @@ def upgrade() -> None:
         # A representative proxied headshot URL (or NULL when none is available). Captured
         # opportunistically from an item's rich metadata (info-page view + the rich re-sync).
         sa.Column("photo_thumb", sa.Text(), nullable=True),
-        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column(
-        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
     )
     op.create_index("ix_plex_people_name", "plex_people", ["name"], unique=True)
 

backend/alembic/versions/0057_plex_ppl_enriched.py

@@ -4,6 +4,7 @@ Revision ID: 0057_plex_ppl_enriched
 Revises: 0056_plex_people
 Create Date: 2026-07-19
 """
+
 from typing import Sequence, Union
 
 from alembic import op
@@ -17,7 +18,9 @@ depends_on: Union[str, Sequence[str], None] = None
 
 def upgrade() -> None:
     for tbl in ("plex_items", "plex_shows"):
-        op.add_column(tbl, sa.Column("people_enriched_at", sa.DateTime(timezone=True), nullable=True))
+        op.add_column(
+            tbl, sa.Column("people_enriched_at", sa.DateTime(timezone=True), nullable=True)
+        )
 
 
 def downgrade() -> None:

backend/alembic/versions/0058_dl_default_720p.py

@@ -10,6 +10,7 @@ Best still one click away. Re-seeds the builtin rows so fresh installs and exist
 (same mechanism as 0040). Existing jobs are unaffected (they snapshot their spec; a profile_id that
 pointed at a replaced row just SET NULLs).
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0059_dl_link_pw_version.py

@@ -10,6 +10,7 @@ did NOT invalidate outstanding grants — they lived out their 6h TTL. Fold a pe
 into the signed grant and bump it on every password change, so a rotation revokes old grants at once.
 Small table (one row per share link), so this is a quick add-column with a default.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0060_plex_jsonb_null_normalize.py

@@ -9,6 +9,7 @@ This one-time data fixup rewrites those scalars to SQL NULL so the columns are u
 forward-only: a SQL NULL is indistinguishable from the old JSON `null` to every reader (the guards
 treat both as "not an array"), so there is nothing meaningful to downgrade. No schema change.
 """
+
 from typing import Sequence, Union
 
 from alembic import op
@@ -25,9 +26,7 @@ _COLS = ("genres", "directors", "cast_names", "collection_keys")
 def upgrade() -> None:
     for table in _TABLES:
         for col in _COLS:
-            op.execute(
+            op.execute(f"UPDATE {table} SET {col} = NULL WHERE jsonb_typeof({col}) = 'null'")
-                f"UPDATE {table} SET {col} = NULL WHERE jsonb_typeof({col}) = 'null'"
-            )
 
 
 def downgrade() -> None:

backend/alembic/versions/0061_dl_job_error_code.py

@@ -7,6 +7,7 @@ small category code (`unavailable` / `login_required` / `geo_blocked` / `format_
 a retry makes sense, keeping the raw text as a technical-detail disclosure. Nullable; only set on the
 error path. Forward-only — an existing errored row simply has NULL until its next failure/retry.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0062_drop_redundant_idx.py

@@ -17,6 +17,7 @@ Redundant index -> the composite that already covers it:
   ix_plex_states_user_id              <- uq_plex_user_item (user_id, item_id)
   ix_plex_playlist_items_playlist_id  <- uq_plex_playlist_item (playlist_id, item_id)
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0063_plex_library_overlay.py

@@ -5,6 +5,7 @@ OVERLAY library is the Hungarian-metadata twin of another section: it is never b
 its titles' HU fields are written onto the matched EN rows (paired by `guid`, see 0064). Cheap:
 metadata-only, no table rewrite.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0064_plex_hu_overlay_cols.py

@@ -9,6 +9,7 @@ Run BEFORE the first HU overlay sync: while the *_hu columns are all NULL the ge
 is trivial. ADD only (no drop/rebuild of the EN search_vector). NOTE: adding a STORED generated
 column + its GIN takes an ACCESS EXCLUSIVE lock and rewrites the table — brief, but flag it on prod.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0065_plex_detail_fields.py

@@ -11,6 +11,7 @@ so the data step below resets `people_enriched_at` to NULL on movie/show rows: t
 each once (60/run, self-throttled) and stamps the new columns. ADD-only, all nullable → no table
 rewrite, no generated column, no ACCESS EXCLUSIVE lock (unlike 0064).
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0066_notif_created_at_notnull.py

@@ -6,6 +6,7 @@ schema allowed NULL — a model/schema drift the new migration drift-test (E1 S2
 server_default has filled every row since day one, so no NULLs exist; enforcing NOT NULL just makes the
 schema match the contract the model already promises. Pure DDL, reversible.
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0067_feed_sort_indexes.py

@@ -22,6 +22,7 @@ CONCURRENTLY (in alembic's autocommit_block, since it can't run inside a transac
 these without blocking reads/writes; downtime ~0. IF (NOT) EXISTS keeps the drift test's
 downgrade->re-upgrade round-trip idempotent.
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0068_published_at_not_null.py

@@ -13,6 +13,7 @@ Note: `SET NOT NULL` validates with a brief ACCESS EXCLUSIVE lock + one table sc
 CONCURRENTLY). On the prod row count (~235k) that is sub-second; still, it is a short write-blocking lock
 rather than the ~0 of 0067's CONCURRENTLY index build.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa
@@ -27,9 +28,7 @@ depends_on: Union[str, Sequence[str], None] = None
 def upgrade() -> None:
     # Backfill any stray NULLs first so SET NOT NULL can't fail (0 in prod, but be safe).
     op.execute("UPDATE videos SET published_at = now() WHERE published_at IS NULL")
-    op.alter_column(
+    op.alter_column("videos", "published_at", nullable=False, server_default=sa.text("now()"))
-        "videos", "published_at", nullable=False, server_default=sa.text("now()")
-    )
 
 
 def downgrade() -> None:

backend/alembic/versions/0069_job_index_hygiene.py

@@ -18,6 +18,7 @@ biggest table, same lookups. live_status is left alone (a separate finding).
 All CONCURRENTLY (autocommit_block) — downtime ~0; IF (NOT) EXISTS keeps the drift test's
 downgrade->re-upgrade idempotent.
 """
+
 from typing import Sequence, Union
 
 from alembic import op

backend/alembic/versions/0070_fold_maintenance_batch.py

@@ -5,6 +5,7 @@ hand-rolled override-or-default reader in `app.state`, duplicating the sysconfig
 now a sysconfig key (`system_config` store), so every admin knob lives in one place. This migration
 carries any existing override into the sysconfig store, then drops the dead column. Reversible.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0071_plex_tmdb_fields.py

@@ -8,6 +8,7 @@ Plex-sourced `imdb_rating`/`rating`, which are left untouched. Plus `plex_people
 foundation for S6's person bios). ADD-only, all nullable → no table rewrite, no ACCESS EXCLUSIVE lock.
 No watermark reset needed: `tmdb_enriched_at` starts NULL, so every eligible row is picked up once.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0072_plex_person_bio.py

@@ -7,6 +7,7 @@ first time a person page is opened (from /person/{tmdb_person_id}), then SERVED
 `tmdb_person_id` link itself already exists (0071). ADD-only, all nullable → no table rewrite, no
 ACCESS EXCLUSIVE lock. `tmdb_enriched_at` starts NULL, so every person is enriched once on first view.
 """
+
 from typing import Sequence, Union
 
 import sqlalchemy as sa

backend/alembic/versions/0073_saved_view_source.py

@@ -0,0 +1,40 @@
+"""saved_views.source (ask 4: per-module saved views)
+
+Saved "smart views" were feed-only; the Plex module now has them too. Add a `source` discriminator
+(default "feed" so every existing row stays a feed view) and widen the per-user unique name constraint
+to (user_id, source, name) so a feed view and a Plex view can share a name. `source` is NOT NULL with a
+server_default, so the ADD is a metadata-only rewrite-free change; the constraint swap is quick on a
+small per-user table.
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0073_saved_view_source"
+down_revision: Union[str, None] = "0072_plex_person_bio"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "saved_views",
+        sa.Column("source", sa.String(length=16), nullable=False, server_default="feed"),
+    )
+    op.drop_constraint("uq_user_saved_view", "saved_views", type_="unique")
+    op.create_unique_constraint(
+        "uq_user_source_saved_view", "saved_views", ["user_id", "source", "name"]
+    )
+
+
+def downgrade() -> None:
+    # LOSSY / effectively forward-only: upgrade lets a user hold a feed view and a plex view of the same
+    # name, so re-creating the narrow (user_id, name) constraint can raise IntegrityError on such a pair.
+    # De-dup by dropping the non-feed rows first (they can't exist in the pre-0073 schema anyway) so the
+    # narrow constraint re-applies cleanly for the drift test's downgrade→re-upgrade cycle.
+    op.execute("DELETE FROM saved_views WHERE source <> 'feed'")
+    op.drop_constraint("uq_user_source_saved_view", "saved_views", type_="unique")
+    op.create_unique_constraint("uq_user_saved_view", "saved_views", ["user_id", "name"])
+    op.drop_column("saved_views", "source")

backend/alembic/versions/0074_download_credentials.py

@@ -0,0 +1,46 @@
+"""download_credentials (S6: per-user, per-host yt-dlp cookie/auth store)
+
+Each user stores their OWN login cookies (Netscape cookies.txt) for an auth-walled download host
+(Facebook, Instagram, …), Fernet-encrypted at rest. A download of a yt-dlp-supported, auth-walled
+link on that host is retried with the acting user's cookie. Unique per (user_id, host_key) so a
+user holds at most one credential per host. Empty for every existing install; forward-only in
+practice (a downgrade just drops the table + its secrets).
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0074_download_credentials"
+down_revision: Union[str, None] = "0073_saved_view_source"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "download_credentials",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column(
+            "user_id",
+            sa.Integer(),
+            sa.ForeignKey("users.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("host_key", sa.String(length=64), nullable=False),
+        sa.Column("cookies_enc", sa.Text(), nullable=False),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+        sa.UniqueConstraint("user_id", "host_key", name="uq_download_credential"),
+    )
+    op.create_index("ix_download_credentials_user_id", "download_credentials", ["user_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_download_credentials_user_id", table_name="download_credentials")
+    op.drop_table("download_credentials")

backend/alembic/versions/0075_playlist_sync_baseline.py

@@ -0,0 +1,26 @@
+"""playlist synced_video_ids (S7: 3-way merge baseline for two-way YouTube sync)
+
+The one-directional "local wins" push is replaced by a 3-way merge that classifies each item
+against the last-synced state. That needs the raw ordered id set as a baseline, not just the
+`synced_fingerprint` hash. Add a nullable `synced_video_ids` (JSON list) column; NULL on every
+existing row means "no baseline yet" — the first merge then unions losslessly and records one.
+Forward-only in practice (a downgrade just drops the column).
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0075_playlist_sync_baseline"
+down_revision: Union[str, None] = "0074_download_credentials"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column("playlists", sa.Column("synced_video_ids", sa.Text(), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("playlists", "synced_video_ids")

backend/app/audit.py

@@ -9,6 +9,7 @@ runs are always logged. Secrets are never stored (config changes log the key nam
 mutation it describes, so the trail can't drift from reality (and a rolled-back mutation drops
 its audit row too).
 """
+
 from datetime import datetime, timedelta, timezone
 
 from sqlalchemy import delete
@@ -43,6 +44,9 @@ class AuditAction:
     SYNC_RESUME = "sync_resume"
     # Scheduler job outcome / run-summary (actor_id set = a manual "run now"; NULL = automatic)
     JOB_RUN = "job_run"
+    # Per-user download cookie/auth credential set or removed (the value is NEVER recorded)
+    DOWNLOAD_CREDENTIAL_SET = "download_credential_set"
+    DOWNLOAD_CREDENTIAL_DELETE = "download_credential_delete"
     # The log itself was cleared (a single row survives the clear as tamper-evidence)
     AUDIT_CLEAR = "audit_clear"
 

backend/app/auth.py

@@ -41,8 +41,7 @@ log = logging.getLogger("siftlode.auth")
 def google_enabled(db: Session) -> bool:
     """Whether Sign in with Google is available (both client id + secret resolve to non-empty)."""
     return bool(
-        sysconfig.get_str(db, "google_client_id")
+        sysconfig.get_str(db, "google_client_id") and sysconfig.get_str(db, "google_client_secret")
-        and sysconfig.get_str(db, "google_client_secret")
     )
 
 
@@ -125,7 +124,9 @@ def purge_user(db: Session, user: User, background: BackgroundTasks) -> None:
     # ciphertext, so a token row with no refresh token silently failed to revoke). decrypt_optional so a
     # TOKEN_ENCRYPTION_KEY mismatch can't 500 the purge — a best-effort revoke degrades to "skip". (B-core-2)
     google_token = (
-        (decrypt_optional(tok.refresh_token_enc) or decrypt_optional(tok.access_token)) if tok else None
+        (decrypt_optional(tok.refresh_token_enc) or decrypt_optional(tok.access_token))
+        if tok
+        else None
     )
     # Raw delete so Postgres applies ON DELETE CASCADE / SET NULL on every dependent table.
     db.execute(delete(User).where(User.id == user_id))
@@ -197,9 +198,7 @@ def current_user(request: Request, db: Session = Depends(get_db)) -> User:
     return user
 
 
-def optional_current_user(
+def optional_current_user(request: Request, db: Session = Depends(get_db)) -> User | None:
-    request: Request, db: Session = Depends(get_db)
-) -> User | None:
     """Like `current_user` but returns None instead of raising 401 when there's no valid session.
     Used by the bootstrap /api/me probe so the logged-out landing doesn't log a failed request to
     the browser console (a non-2xx fetch is a console error no matter how the app handles it)."""
@@ -215,9 +214,7 @@ def require_human(user: User = Depends(current_user)) -> User:
     no token, so has_read_scope/has_write_scope are False); this makes the boundary explicit
     where there's no scope check to lean on."""
     if user.is_demo:
-        raise HTTPException(
+        raise HTTPException(status_code=403, detail="Not available in the demo account.")
-            status_code=403, detail="Not available in the demo account."
-        )
     return user
 
 

backend/app/config.py

@@ -169,6 +169,7 @@ class Settings(BaseSettings):
     @property
     def player_client_fallback_list(self) -> list[str]:
         return self._clients(self.download_player_clients_fallback)
+
     # How many downloads the worker runs concurrently (claimed via FOR UPDATE SKIP LOCKED).
     download_worker_concurrency: int = 2
     # How often (empty pending queue) the worker polls for a new job, in seconds.
@@ -185,6 +186,21 @@ class Settings(BaseSettings):
     download_supervisor_poll_seconds: int = 3
     # Grace between SIGTERM and SIGKILL when stopping a child (matches run_ffmpeg's escalation).
     download_kill_grace_seconds: int = 5
+    # --- worker-child resource de-prioritization ---
+    # The download/edit child (yt-dlp + ffmpeg) shares ONE LAN box with the API, Postgres, and Plex
+    # playback. Without a priority handicap a CPU re-encode pegs every core for ~30s and starves web
+    # requests + playback until it finishes. The child lowers its OWN priority at startup — inherited
+    # by every ffmpeg/yt-dlp descendant — so any foreground work always wins. Each knob is best-effort
+    # and independently disable-able with 0.
+    download_child_nice: int = 10  # CPU nice increment applied to the child; 0 = don't renice
+    download_child_ionice_class: int = 3  # best-effort IO class (3 = idle); 0 = leave IO untouched
+    # Cap the browser-safe re-encode's encoder threads so a transcode can't claim every core. 0 =
+    # reserve one core for foreground work (max(1, cpu_count - 1)); >0 pins exactly that many.
+    download_ffmpeg_threads: int = 0
+    # Throttle the retention GC's whole-tree orphan file sweep so its rglob can't monopolize disk IO
+    # in the API process: sleep `pause_ms` after every `batch` files touched. pause_ms=0 disables.
+    download_gc_sweep_pause_ms: int = 20
+    download_gc_sweep_batch: int = 400
     # Total cache size cap across ALL users (bytes). When >0 and exceeded, the GC evicts the
     # least-recently-accessed ready assets until under it. 0 = unlimited (disk is the limit).
     download_total_max_bytes: int = 0
@@ -231,7 +247,9 @@ class Settings(BaseSettings):
     plex_sync_interval_min: int = 360
     # How often the incremental Plex→Siftlode watch-state reconcile runs, in minutes (Phase C):
     # a cheap history + on-deck pull plus a re-push of any states that failed their immediate push.
-    plex_watch_sync_interval_min: int = 30
+    # 15 min (S2): with Plex now the source of truth (prompt write-back on the ~10s checkpoint +
+    # Plex-authoritative pull), a faster incremental pull keeps Siftlode close to Plex-side changes.
+    plex_watch_sync_interval_min: int = 15
     # How often the FULL watch-state reconcile runs, in minutes (Phase C). A whole-section rescan
     # that catches what the incremental history feed can't — notably an un-watch done on the Plex
     # side — so it runs far less often (daily) than the incremental pass.

backend/app/demo.py

@@ -4,6 +4,7 @@ Shared by the admin's manual reset endpoint (routes/admin.py) and the scheduled
 (scheduler.py). Lives in the service layer, not routes, so the scheduler can call it without an
 `app.routes -> app.scheduler` import cycle (the reason its import used to be lazy).
 """
+
 from sqlalchemy import delete, func, select
 from sqlalchemy.orm import Session
 
@@ -49,13 +50,17 @@ def _seed_demo_subscriptions(db: Session, demo: User, n: int = 14) -> int:
     Channel manager isn't empty — gives visitors something to explore and matches the landing-page
     screenshot. Idempotent: clears the demo's subscriptions first, then re-adds the top N."""
     db.execute(delete(Subscription).where(Subscription.user_id == demo.id))
-    top = db.execute(
+    top = (
-        select(Video.channel_id)
+        db.execute(
-        .where(Video.channel_id.is_not(None))
+            select(Video.channel_id)
-        .group_by(Video.channel_id)
+            .where(Video.channel_id.is_not(None))
-        .order_by(func.count().desc())
+            .group_by(Video.channel_id)
-        .limit(n)
+            .order_by(func.count().desc())
-    ).scalars().all()
+            .limit(n)
+        )
+        .scalars()
+        .all()
+    )
     for channel_id in top:
         db.add(Subscription(user_id=demo.id, channel_id=channel_id))
     return len(top)

backend/app/downloads/adapters.py

@@ -10,6 +10,7 @@ Adapters are stateless singletons; get one with `resolve_adapter(source_kind, so
 Resolution works both before extraction (from source_kind / URL host) and after (from the yt-dlp
 `extractor_key` in the info dict), so a non-YouTube "url" job still lands on the right adapter.
 """
+
 from urllib.parse import urlparse
 
 from app.config import settings
@@ -140,8 +141,10 @@ def resolve_adapter(
     def host_is(*domains: str) -> bool:
         return any(_host_is(h, *domains) for h in hosts)
 
-    if source_kind == "youtube" or "youtube" in extractor or host_is(
+    if (
-        "youtube.com", "youtu.be", "youtube-nocookie.com"
+        source_kind == "youtube"
+        or "youtube" in extractor
+        or host_is("youtube.com", "youtu.be", "youtube-nocookie.com")
     ):
         return _YOUTUBE
     if "facebook" in extractor or host_is("facebook.com", "fb.watch"):

backend/app/downloads/backfill.py

@@ -10,6 +10,7 @@ URL may have expired — those just get a locally-generated poster instead).
 Run once per environment:
     docker exec siftlode-worker-1 python -c "from app.downloads import backfill; print(backfill.run())"
 """
+
 import logging
 
 import yt_dlp

backend/app/downloads/credentials.py

@@ -0,0 +1,258 @@
+"""Per-user, per-host download cookies — the single source of truth for "does the acting user's
+cookie apply to this download, and how does that keep the result private".
+
+Two call sites depend on ONE consistent answer (house rule: an invariant lives in shared code, not
+in prose):
+  * `enqueue` salts the MediaAsset cache signature with the user id WHEN a cookie applies, so the
+    authed download becomes a PRIVATE asset that is never cache-served to another user.
+  * the worker attaches the cookie file to yt-dlp ONLY when a cookie still applies AND the asset it
+    is downloading into is that private, salted identity — so a cookie can never poison a shared
+    asset (the load-bearing cross-user privacy fix).
+
+Both go through `cookie_salt` / `authed_format_sig` / `host_key_for_url` here, so they can't drift.
+
+Security controls enforced here:
+  * **host scope (SEC-12)** — `host_key_for_url` matches the download URL's netloc against a curated
+    registry via the netloc-scoped `_host_is`, so a cookie is only ever offered to the exact site it
+    belongs to; an arbitrary `?url=` (or a look-alike host) resolves to no host_key and gets no
+    cookie.
+  * **secret handling** — cookies are Fernet-encrypted at rest; the blob is only ever decrypted
+    in-worker into a 0600 file that is unlinked in a `finally`, and is never returned by a route,
+    logged, or written to a job's error/snapshot.
+"""
+
+import logging
+import os
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.downloads import formats
+from app.downloads.adapters import _host_is
+from app.models import DownloadCredential
+from app.security import decrypt_optional, encrypt
+
+log = logging.getLogger("siftlode.downloads.credentials")
+
+
+class HostSpec:
+    """A supported cookie host: a short canonical `key` (stored on the credential + folded into the
+    cache salt), a human `label` for the UI, and the netloc `domains` it matches."""
+
+    __slots__ = ("key", "label", "domains")
+
+    def __init__(self, key: str, label: str, domains: tuple[str, ...]):
+        self.key = key
+        self.label = label
+        self.domains = domains
+
+
+# Curated registry (user's call: seed the common auth-walled hosts now). One `key` can cover several
+# domains of the same site. Add a host by adding a row — the FE dropdown, host-scope guard, and salt
+# all derive from this, so there is nothing else to wire up.
+SUPPORTED_HOSTS: tuple[HostSpec, ...] = (
+    HostSpec("facebook", "Facebook", ("facebook.com", "fb.watch")),
+    HostSpec("instagram", "Instagram", ("instagram.com",)),
+    HostSpec("twitter", "X (Twitter)", ("twitter.com", "x.com")),
+    HostSpec("tiktok", "TikTok", ("tiktok.com",)),
+    HostSpec("reddit", "Reddit", ("reddit.com",)),
+)
+
+_BY_KEY: dict[str, HostSpec] = {h.key: h for h in SUPPORTED_HOSTS}
+
+
+def is_supported_host(host_key: str) -> bool:
+    return host_key in _BY_KEY
+
+
+def host_label(host_key: str) -> str:
+    spec = _BY_KEY.get(host_key)
+    return spec.label if spec else host_key
+
+
+def host_key_for_url(url: str | None) -> str | None:
+    """The supported host_key whose domains own `url`'s netloc, or None. Netloc-scoped (via
+    `_host_is`), so a look-alike host ('myfacebook.com', 'facebook.com.evil') never matches. When
+    several specs could match, the most-specific (longest matched domain) wins so the result is
+    deterministic across the enqueue and worker call sites."""
+    if not url:
+        return None
+    from urllib.parse import urlparse
+
+    host = urlparse(url).netloc.lower()
+    if not host:
+        return None
+    best_key: str | None = None
+    best_len = -1
+    for spec in SUPPORTED_HOSTS:
+        for dom in spec.domains:
+            if _host_is(host, dom) and len(dom) > best_len:
+                best_key, best_len = spec.key, len(dom)
+    return best_key
+
+
+def cookie_salt(user_id: int, host_key: str) -> str:
+    """The cache-signature salt for a cookie-authed download. Keyed by user id + host so the authed
+    asset is private to that user (never dedup-shared) and stable across that user's own re-adds."""
+    return f"cookie:{user_id}:{host_key}"
+
+
+def authed_format_sig(spec: dict, user_id: int, host_key: str) -> str:
+    """The salted `format_sig` an authed download for (user, host) resolves to — the private asset
+    identity. Both enqueue (to create it) and the worker (to recognise it) compute it HERE."""
+    return formats.format_sig(spec, salt=cookie_salt(user_id, host_key))
+
+
+def salt_for_enqueue(db: Session, user_id: int, url: str | None) -> str | None:
+    """At enqueue: the cache salt to apply if this user has a cookie for `url`'s host, else None (a
+    normal shared download). Returns the salt string so the caller passes it straight to
+    `format_sig(spec, salt=...)`."""
+    host_key = host_key_for_url(url)
+    if host_key is None:
+        return None
+    if get_credential(db, user_id, host_key) is None:
+        return None
+    return cookie_salt(user_id, host_key)
+
+
+# --- storage (CRUD) ---------------------------------------------------------------------------
+
+
+def get_credential(db: Session, user_id: int, host_key: str) -> DownloadCredential | None:
+    return db.execute(
+        select(DownloadCredential).where(
+            DownloadCredential.user_id == user_id, DownloadCredential.host_key == host_key
+        )
+    ).scalar_one_or_none()
+
+
+class InvalidCookies(ValueError):
+    """The pasted cookies aren't a usable Netscape cookies.txt — surfaced as a 400 by the route."""
+
+
+def validate_cookies(raw: str) -> str:
+    """Sanity-check a pasted cookies.txt (Netscape format) and return the cleaned text. yt-dlp reads
+    the Netscape format: comment/`#` lines plus TAB-separated cookie lines (domain, flag, path,
+    secure, expiry, name, value). We don't fully parse — just reject obvious non-cookie paste (empty,
+    or no tab-separated data line) so a mistaken paste fails HERE with a clear message rather than
+    silently producing a cookie file yt-dlp ignores."""
+    text = (raw or "").strip()
+    if not text:
+        raise InvalidCookies("empty")
+    has_data_line = any(
+        line.strip() and not line.lstrip().startswith("#") and "\t" in line
+        for line in text.splitlines()
+    )
+    if not has_data_line:
+        raise InvalidCookies("format")
+    # Netscape cookie files must end with a newline for some parsers; normalise.
+    return text if text.endswith("\n") else text + "\n"
+
+
+def upsert_credential(db: Session, user_id: int, host_key: str, raw: str) -> DownloadCredential:
+    """Create or replace this user's cookie blob for `host_key`. Caller has validated host_key is
+    supported and `raw` via `validate_cookies`. Does NOT commit (the route commits with the audit
+    row so the trail can't drift). The blob is encrypted here; the plaintext never leaves this call."""
+    cred = get_credential(db, user_id, host_key)
+    enc = encrypt(raw)
+    if cred is None:
+        cred = DownloadCredential(user_id=user_id, host_key=host_key, cookies_enc=enc)
+        db.add(cred)
+    else:
+        cred.cookies_enc = enc
+    return cred
+
+
+def delete_credential(db: Session, user_id: int, host_key: str) -> bool:
+    """Delete this user's cookie for `host_key`. Returns whether a row existed. Does NOT commit."""
+    cred = get_credential(db, user_id, host_key)
+    if cred is None:
+        return False
+    db.delete(cred)
+    return True
+
+
+def status_list(db: Session, user_id: int) -> list[dict]:
+    """The supported hosts with this user's configured/updated state — the write-only Settings view.
+    Never includes the blob."""
+    have = {
+        c.host_key: c
+        for c in db.execute(
+            select(DownloadCredential).where(DownloadCredential.user_id == user_id)
+        ).scalars()
+    }
+    out = []
+    for spec in SUPPORTED_HOSTS:
+        cred = have.get(spec.key)
+        out.append(
+            {
+                "host_key": spec.key,
+                "label": spec.label,
+                "configured": cred is not None,
+                "updated_at": cred.updated_at.isoformat() if cred is not None else None,
+            }
+        )
+    return out
+
+
+# --- worker: resolve + write the cookie file --------------------------------------------------
+
+
+def resolve_cookiefile(
+    db: Session,
+    user_id: int,
+    url: str | None,
+    asset_format_sig: str,
+    spec: dict,
+    staging: Path,
+) -> Path | None:
+    """In the worker child: the cookies.txt to hand yt-dlp for this download, or None.
+
+    The double gate that prevents cross-user leakage:
+      1. a credential must still exist for `url`'s host (host-scoped via `host_key_for_url`), AND
+      2. the asset being downloaded into must BE this user's private, salted identity
+         (`asset_format_sig == authed_format_sig(...)`).
+
+    If (2) fails the asset is the SHARED (unsalted) one — attaching the cookie would poison the
+    global cache with a private-group video, so we attach nothing. An unreadable blob (rotated key →
+    InvalidToken) also yields None; the download then fails honestly on the login wall (S3's
+    `login_required`)."""
+    host_key = host_key_for_url(url)
+    if host_key is None:
+        return None
+    cred = get_credential(db, user_id, host_key)
+    if cred is None:
+        return None
+    if asset_format_sig != authed_format_sig(spec, user_id, host_key):
+        return None  # shared asset — never attach a cookie to a non-private identity
+    raw = decrypt_optional(cred.cookies_enc)
+    if not raw:
+        log.warning(
+            "download cookie for user %s host %s could not be decrypted (rotated key?) — "
+            "proceeding without it",
+            user_id,
+            host_key,
+        )
+        return None
+    return _write_cookiefile(raw, staging)
+
+
+def _write_cookiefile(raw: str, staging: Path) -> Path:
+    """Write the cookie blob to a private 0600 file under `staging/.auth/` and return its path.
+
+    Deliberately NOT in the staging root: the download retry loop unlinks every file it finds there,
+    and `_produced_file` picks the largest file there as the output — either would corrupt a
+    cookies.txt in the root. A `.auth` SUBDIR survives both (they skip directories) and is still
+    removed by the parent worker's staging cleanup on a hard kill. The caller unlinks it in a
+    `finally` on the normal/exception paths."""
+    auth_dir = staging / ".auth"
+    auth_dir.mkdir(mode=0o700, exist_ok=True)
+    path = auth_dir / "cookies.txt"
+    # Open with 0600 from the start (don't create world-readable then chmod).
+    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+    try:
+        os.write(fd, raw.encode("utf-8"))
+    finally:
+        os.close(fd)
+    return path

backend/app/downloads/edit.py

@@ -20,10 +20,12 @@ Also builds the editor **filmstrip** — a single tiled sprite of the source vid
 scrub track. Cached on the source asset's `storyboard_path` (reserved in phase 1) and reusable
 for a player hover-preview later.
 """
+
 import hashlib
 import json
 import logging
 import math
+import os
 import queue
 import shutil
 import subprocess
@@ -31,15 +33,28 @@ import threading
 import time
 from pathlib import Path
 
+from app.config import settings
+
 log = logging.getLogger("siftlode.edit")
 
 
+def encode_thread_cap() -> int:
+    """How many encoder threads a browser-safe re-encode may use. `download_ffmpeg_threads` when the
+    admin pinned one (>0), else reserve a core for foreground work (max(1, cpu_count - 1)). Shared by
+    the ffmpeg `-threads` flag and the child's `OMP_NUM_THREADS` so both honour the same budget."""
+    n = settings.download_ffmpeg_threads
+    if n > 0:
+        return n
+    return max(1, (os.cpu_count() or 2) - 1)
+
+
 class EditAborted(Exception):
     """Raised by run_ffmpeg when the job was paused/canceled mid-encode."""
 
 
 # --- edit spec ------------------------------------------------------------------------------
 
+
 def normalize_edit_spec(spec: dict | None) -> dict:
     """Canonicalize a raw edit spec: clamp/round trim, coerce crop to ints, drop empties.
 
@@ -140,6 +155,7 @@ def clip_duration(spec: dict, source_duration: int | None) -> int | None:
 
 # --- ffmpeg: trim / crop --------------------------------------------------------------------
 
+
 def build_edit_cmd(src: Path, dest: Path, spec: dict, out_ext: str) -> list[str]:
     """ffmpeg command for a trim/crop edit. `-progress pipe:1` streams machine-readable progress.
 
@@ -153,8 +169,15 @@ def build_edit_cmd(src: Path, dest: Path, spec: dict, out_ext: str) -> list[str]
     reencode = needs_reencode(spec)
 
     cmd = [
-        "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-nostdin",
+        "ffmpeg",
-        "-progress", "pipe:1", "-nostats",
+        "-y",
+        "-hide_banner",
+        "-loglevel",
+        "error",
+        "-nostdin",
+        "-progress",
+        "pipe:1",
+        "-nostats",
     ]
     if start:
         cmd += ["-ss", f"{start:.3f}"]
@@ -170,8 +193,16 @@ def build_edit_cmd(src: Path, dest: Path, spec: dict, out_ext: str) -> list[str]
         if crop:
             cmd += ["-vf", f"crop={crop['w']}:{crop['h']}:{crop['x']}:{crop['y']}"]
         cmd += [
-            "-c:v", "libx264", "-preset", "veryfast", "-crf", "20",
+            "-c:v",
-            "-c:a", "aac", "-b:a", "192k",
+            "libx264",
+            "-preset",
+            "veryfast",
+            "-crf",
+            "20",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "192k",
         ]
     else:
         cmd += ["-c", "copy", "-avoid_negative_ts", "make_zero"]
@@ -204,8 +235,15 @@ def build_concat_plan(src: Path, dest: Path, spec: dict, out_ext: str, staging:
     crop = spec.get("crop")
     prep: dict[Path, str] = {}
     head = [
-        "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-nostdin",
+        "ffmpeg",
-        "-progress", "pipe:1", "-nostats",
+        "-y",
+        "-hide_banner",
+        "-loglevel",
+        "error",
+        "-nostdin",
+        "-progress",
+        "pipe:1",
+        "-nostats",
     ]
 
     if needs_reencode(spec):
@@ -218,9 +256,24 @@ def build_concat_plan(src: Path, dest: Path, spec: dict, out_ext: str, staging:
             labels.append(f"[v{i}][a{i}]")
         parts.append(f"{''.join(labels)}concat=n={len(segs)}:v=1:a=1[v][a]")
         cmd = head + [
-            "-i", str(src), "-filter_complex", ";".join(parts),
+            "-i",
-            "-map", "[v]", "-map", "[a]",
+            str(src),
-            "-c:v", "libx264", "-preset", "veryfast", "-crf", "20", "-c:a", "aac", "-b:a", "192k",
+            "-filter_complex",
+            ";".join(parts),
+            "-map",
+            "[v]",
+            "-map",
+            "[a]",
+            "-c:v",
+            "libx264",
+            "-preset",
+            "veryfast",
+            "-crf",
+            "20",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "192k",
         ]
     else:
         listp = staging / "concat.txt"
@@ -314,7 +367,9 @@ def _stop(proc: subprocess.Popen) -> None:
     escalate_kill(proc, proc.terminate, proc.kill, grace_s=5, label="ffmpeg")
 
 
-def run_ffmpeg(cmd, total_s, on_progress, should_cancel, stall_timeout_s=_FFMPEG_STALL_TIMEOUT_S) -> None:
+def run_ffmpeg(
+    cmd, total_s, on_progress, should_cancel, stall_timeout_s=_FFMPEG_STALL_TIMEOUT_S
+) -> None:
     """Run an ffmpeg edit, reporting 0–100 progress and honouring cooperative cancel.
 
     Both pipes are drained by threads (see `_drain_tail`) and the main loop polls on a 1s tick, so
@@ -411,9 +466,20 @@ def probe_codecs(path: Path) -> tuple[str | None, str | None]:
     stream or if ffprobe fails (a failed probe is treated as 'leave it alone' upstream)."""
     try:
         out = subprocess.run(
-            ["ffprobe", "-v", "error", "-show_entries", "stream=codec_type,codec_name",
+            [
-             "-of", "json", str(path)],
+                "ffprobe",
-            capture_output=True, text=True, timeout=30, check=True,
+                "-v",
+                "error",
+                "-show_entries",
+                "stream=codec_type,codec_name",
+                "-of",
+                "json",
+                str(path),
+            ],
+            capture_output=True,
+            text=True,
+            timeout=30,
+            check=True,
         ).stdout
     except (subprocess.TimeoutExpired, subprocess.CalledProcessError, OSError):
         return None, None
@@ -425,16 +491,29 @@ def probe_duration(path: Path) -> int:
     re-encode when the yt-dlp info dict carried no `duration` (some URL/live extractions)."""
     try:
         out = subprocess.run(
-            ["ffprobe", "-v", "error", "-show_entries", "format=duration",
+            [
-             "-of", "default=nw=1:nk=1", str(path)],
+                "ffprobe",
-            capture_output=True, text=True, timeout=30, check=True,
+                "-v",
+                "error",
+                "-show_entries",
+                "format=duration",
+                "-of",
+                "default=nw=1:nk=1",
+                str(path),
+            ],
+            capture_output=True,
+            text=True,
+            timeout=30,
+            check=True,
         ).stdout.strip()
         return int(float(out)) if out else 0
     except (subprocess.TimeoutExpired, subprocess.CalledProcessError, OSError, ValueError):
         return 0
 
 
-def browser_transcode_flags(vcodec: str | None, acodec: str | None, is_mp4: bool) -> tuple[bool, bool, bool]:
+def browser_transcode_flags(
+    vcodec: str | None, acodec: str | None, is_mp4: bool
+) -> tuple[bool, bool, bool]:
     """Decide what (if anything) makes a file universally browser-playable. Returns
     (needs_change, reencode_video, reencode_audio). needs_change=False => already H.264 + AAC/MP3
     in mp4. A None codec (no such stream, or a failed probe) is never re-encoded."""
@@ -443,16 +522,34 @@ def browser_transcode_flags(vcodec: str | None, acodec: str | None, is_mp4: bool
     return (rv or ra or not is_mp4), rv, ra
 
 
-def build_browser_safe_cmd(src: Path, dest: Path, reencode_video: bool, reencode_audio: bool) -> list[str]:
+def build_browser_safe_cmd(
+    src: Path, dest: Path, reencode_video: bool, reencode_audio: bool, threads: int | None = None
+) -> list[str]:
     """ffmpeg command that lands `src` as a browser-safe mp4 (faststart): re-encode only the
     stream(s) that need it (libx264 / aac), stream-copy the rest, always into mp4. `-progress
-    pipe:1` streams machine-readable progress for run_ffmpeg."""
+    pipe:1` streams machine-readable progress for run_ffmpeg. `threads` caps the encoder so a
+    re-encode can't claim every core on the shared box (see the worker-child de-prioritization)."""
     cmd = [
-        "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-nostdin",
+        "ffmpeg",
-        "-progress", "pipe:1", "-nostats", "-i", str(src),
+        "-y",
+        "-hide_banner",
+        "-loglevel",
+        "error",
+        "-nostdin",
+        "-progress",
+        "pipe:1",
+        "-nostats",
+        "-i",
+        str(src),
     ]
-    cmd += ["-c:v", "libx264", "-preset", "veryfast", "-crf", "20"] if reencode_video else ["-c:v", "copy"]
+    cmd += (
+        ["-c:v", "libx264", "-preset", "veryfast", "-crf", "20"]
+        if reencode_video
+        else ["-c:v", "copy"]
+    )
     cmd += ["-c:a", "aac", "-b:a", "192k"] if reencode_audio else ["-c:a", "copy"]
+    if threads and threads > 0:
+        cmd += ["-threads", str(threads)]
     cmd += ["-movflags", "+faststart", str(dest)]
     return cmd
 
@@ -486,25 +583,63 @@ def build_storyboard_cmd(src: Path, dest: Path, duration_s: int | None, geom: di
     fps = geom["count"] / dur
     vf = f"fps={fps:.6f},scale={_SB_TILE_W}:-2,tile={geom['cols']}x{geom['rows']}"
     return [
-        "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-nostdin",
+        "ffmpeg",
-        "-i", str(src), "-frames:v", "1", "-q:v", "5", "-vf", vf, str(dest),
+        "-y",
+        "-hide_banner",
+        "-loglevel",
+        "error",
+        "-nostdin",
+        "-i",
+        str(src),
+        "-frames:v",
+        "1",
+        "-q:v",
+        "5",
+        "-vf",
+        vf,
+        str(dest),
     ]
 
 
 def build_poster_cmd(src: Path, dest: Path, at_s: float) -> list[str]:
     """Grab one representative frame as a poster JPEG. `-ss` before `-i` seeks fast."""
     return [
-        "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-nostdin",
+        "ffmpeg",
-        "-ss", f"{max(0.0, at_s):.3f}", "-i", str(src),
+        "-y",
-        "-frames:v", "1", "-q:v", "3", "-vf", "scale=640:-2", str(dest),
+        "-hide_banner",
+        "-loglevel",
+        "error",
+        "-nostdin",
+        "-ss",
+        f"{max(0.0, at_s):.3f}",
+        "-i",
+        str(src),
+        "-frames:v",
+        "1",
+        "-q:v",
+        "3",
+        "-vf",
+        "scale=640:-2",
+        str(dest),
     ]
 
 
 def build_thumbnail_jpg_cmd(src: Path, dest: Path) -> list[str]:
     """Decode a downloaded thumbnail (webp/png/…) to a single baseline JPEG frame."""
     return [
-        "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-nostdin",
+        "ffmpeg",
-        "-i", str(src), "-frames:v", "1", "-update", "1", str(dest),
+        "-y",
+        "-hide_banner",
+        "-loglevel",
+        "error",
+        "-nostdin",
+        "-i",
+        str(src),
+        "-frames:v",
+        "1",
+        "-update",
+        "1",
+        str(dest),
     ]
 
 
@@ -550,7 +685,10 @@ def thumbnail_to_jpg(src: Path | None, dest: Path, timeout_s: int = 30) -> Path
             return None
     err = _run_ffmpeg_best_effort(build_thumbnail_jpg_cmd(src, dest), timeout_s)
     if err is not None:
-        log.info("thumbnail->jpg conversion failed (%s); poster falls back to a cut frame", str(err)[:120])
+        log.info(
+            "thumbnail->jpg conversion failed (%s); poster falls back to a cut frame",
+            str(err)[:120],
+        )
         return None
     return dest if dest.exists() else None
 

backend/app/downloads/errors.py

@@ -24,6 +24,16 @@ _RULES: tuple[tuple[str, str], ...] = (
     # Some extractors (e.g. Reddit) gate content behind a logged-in account: "Account
     # authentication is required." — an auth wall we deliberately don't cross, not a transient error.
     ("authentication is required", "login_required"),
+    # Closed-group / private-post auth walls (Facebook groups, Instagram private, …). yt-dlp phrases
+    # these many ways; the common thread is "you must be signed in" or its own suggestion to supply
+    # cookies. Treating the cookie suggestion as an auth signal is what makes a login-walled link read
+    # honestly (and stay non-retryable) when the acting user has no usable cookie for that host.
+    ("log in to", "login_required"),
+    ("login required", "login_required"),
+    ("you must be logged in", "login_required"),
+    ("only available for registered users", "login_required"),
+    ("use --cookies", "login_required"),
+    ("--cookies-from-browser", "login_required"),
     # Geo-restricted.
     ("in your country", "geo_blocked"),
     ("available in your location", "geo_blocked"),
@@ -42,6 +52,10 @@ _RULES: tuple[tuple[str, str], ...] = (
     ("account associated with this video has been terminated", "unavailable"),
     ("this video is not available", "unavailable"),
     ("this video has been removed", "unavailable"),
+    # yt-dlp doesn't support this site at all ("Unsupported URL: …") — distinct from an auth wall on a
+    # supported site. Non-retryable: no cookie or retry can make an unsupported host downloadable.
+    ("unsupported url", "unsupported_host"),
+    ("no video could be found in this url", "unsupported_host"),
     # DRM.
     ("drm protected", "drm"),
     ("drm-protected", "drm"),

backend/app/downloads/formats.py

@@ -13,6 +13,7 @@ spec shape (all keys optional; defaults applied by `normalize`):
     vcodec         "h264" | "vp9" | "av1" | null (preference, best-effort)
     embed_subs     bool     embed_chapters bool     embed_thumbnail bool     sponsorblock bool
 """
+
 import hashlib
 import json
 
@@ -97,12 +98,17 @@ def build_ydl_opts(
     outtmpl: str,
     progress_hook,
     extra: dict | None = None,
+    cookiefile: str | None = None,
 ) -> dict:
     """Assemble yt-dlp options for a single download to `outtmpl` (a full path template).
 
     `extra` is source-specific options contributed by the SourceAdapter (e.g. YouTube's
     `extractor_args` with the player-client set + bgutil PO-token base_url); it is merged onto the
-    generic options below. Everything here is source-agnostic."""
+    generic options below. Everything here is source-agnostic.
+
+    `cookiefile` is the acting user's per-host cookies.txt path (S6), attached ONLY by the worker
+    after the host-scope + private-asset gate in `downloads.credentials`. It is set AFTER the `extra`
+    merge so a source adapter can never clobber it."""
     s = normalize(spec)
     h = s["max_height"]
     opts: dict = {
@@ -131,6 +137,11 @@ def build_ydl_opts(
     if extra:
         opts.update(extra)
 
+    # The acting user's per-host cookie (S6). AFTER the adapter merge on purpose: an adapter must not
+    # be able to override the authenticated session the worker resolved for this specific user.
+    if cookiefile:
+        opts["cookiefile"] = cookiefile
+
     if s["mode"] == "a":
         opts["format"] = "bestaudio/best"
         opts["postprocessors"].append(
@@ -215,8 +226,6 @@ def build_ydl_opts(
     # jpg/png thumbnail directly and converts anything else to PNG (not the crashing mjpeg2jpeg-to-jpg
     # path), so it never needed the convertor and does not reintroduce the crash.
     if s["embed_thumbnail"]:
-        opts["postprocessors"].append(
+        opts["postprocessors"].append({"key": "EmbedThumbnail", "already_have_thumbnail": True})
-            {"key": "EmbedThumbnail", "already_have_thumbnail": True}
-        )
 
     return opts

backend/app/downloads/gc.py

@@ -15,6 +15,7 @@ Five passes each run:
 
 Runs in the API process (which mounts DOWNLOAD_ROOT). Pure disk/DB work — no YouTube quota.
 """
+
 import logging
 import time
 from datetime import datetime, timedelta, timezone
@@ -40,9 +41,7 @@ _ORPHAN_MIN_AGE_S = 3600  # only sweep files older than this, so an in-flight do
 def _holders(db: Session, asset_id: int) -> list[int]:
     return list(
         db.execute(
-            select(DownloadJob.user_id)
+            select(DownloadJob.user_id).where(DownloadJob.asset_id == asset_id).distinct()
-            .where(DownloadJob.asset_id == asset_id)
-            .distinct()
         ).scalars()
     )
 
@@ -87,11 +86,25 @@ def _sweep_orphan_files(db: Session, download_root: str) -> int:
 
     cutoff = time.time() - _ORPHAN_MIN_AGE_S
     removed = 0
+    # Throttle the whole-tree walk: this runs in the API process, and an unpaced rglob+stat over a
+    # large download root monopolizes disk IO long enough to make web requests + playback stutter.
+    # Yield the IO channel after every `batch` files touched so foreground work interleaves.
+    pause_s = max(0, settings.download_gc_sweep_pause_ms) / 1000.0
+    batch = max(1, settings.download_gc_sweep_batch)
+    touched = 0
+
+    def _throttle() -> None:
+        nonlocal touched
+        touched += 1
+        if pause_s and touched % batch == 0:
+            time.sleep(pause_s)
+
     for entry in root.iterdir():
         if entry.name.startswith("."):  # .edits / .plex-* / .staging / .storyboards + any dotfile
             continue
         files = [entry] if entry.is_file() else [p for p in entry.rglob("*") if p.is_file()]
         for f in files:
+            _throttle()
             try:
                 if f.resolve() in keep or f.stat().st_mtime > cutoff:
                     continue
@@ -101,7 +114,9 @@ def _sweep_orphan_files(db: Session, download_root: str) -> int:
                 pass
         if entry.is_dir():  # prune emptied dirs (deepest first), then the channel dir itself
             for d in sorted(
-                (p for p in entry.rglob("*") if p.is_dir()), key=lambda p: len(p.parts), reverse=True
+                (p for p in entry.rglob("*") if p.is_dir()),
+                key=lambda p: len(p.parts),
+                reverse=True,
             ):
                 try:
                     d.rmdir()
@@ -121,15 +136,19 @@ def run_download_gc(db: Session) -> dict:
 
     # 1. Pre-expiry warnings.
     warn_before = now + timedelta(days=grace_days)
-    soon = db.execute(
+    soon = (
-        select(MediaAsset).where(
+        db.execute(
-            MediaAsset.status == "ready",
+            select(MediaAsset).where(
-            MediaAsset.gc_notified.is_(False),
+                MediaAsset.status == "ready",
-            MediaAsset.expires_at.is_not(None),
+                MediaAsset.gc_notified.is_(False),
-            MediaAsset.expires_at > now,
+                MediaAsset.expires_at.is_not(None),
-            MediaAsset.expires_at <= warn_before,
+                MediaAsset.expires_at > now,
+                MediaAsset.expires_at <= warn_before,
+            )
         )
-    ).scalars().all()
+        .scalars()
+        .all()
+    )
     for asset in soon:
         _notify_holders(db, asset, "download_expiring")
         asset.gc_notified = True
@@ -138,13 +157,17 @@ def run_download_gc(db: Session) -> dict:
         db.commit()
 
     # 2. Expiry deletion (ready assets past TTL, and any errored leftovers with a TTL).
-    dead = db.execute(
+    dead = (
-        select(MediaAsset).where(
+        db.execute(
-            MediaAsset.status.in_(("ready", "error")),
+            select(MediaAsset).where(
-            MediaAsset.expires_at.is_not(None),
+                MediaAsset.status.in_(("ready", "error")),
-            MediaAsset.expires_at <= now,
+                MediaAsset.expires_at.is_not(None),
+                MediaAsset.expires_at <= now,
+            )
         )
-    ).scalars().all()
+        .scalars()
+        .all()
+    )
     for asset in dead:
         _delete_asset(db, asset, "download_expired")
         expired += 1
@@ -154,17 +177,24 @@ def run_download_gc(db: Session) -> dict:
     # 3. LRU eviction over the total cache cap.
     cap = sysconfig.get_int(db, "download_total_max_bytes")
     if cap > 0:
-        total = db.execute(
+        total = (
-            select(func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where(
+            db.execute(
-                MediaAsset.status == "ready"
+                select(func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where(
-            )
+                    MediaAsset.status == "ready"
-        ).scalar() or 0
+                )
+            ).scalar()
+            or 0
+        )
         if total > cap:
-            candidates = db.execute(
+            candidates = (
-                select(MediaAsset)
+                db.execute(
-                .where(MediaAsset.status == "ready")
+                    select(MediaAsset)
-                .order_by(MediaAsset.last_access_at.asc().nulls_first())
+                    .where(MediaAsset.status == "ready")
-            ).scalars().all()
+                    .order_by(MediaAsset.last_access_at.asc().nulls_first())
+                )
+                .scalars()
+                .all()
+            )
             for asset in candidates:
                 if total <= cap:
                     break
@@ -176,13 +206,17 @@ def run_download_gc(db: Session) -> dict:
     # 4. Reap orphaned errored rows (no file, no holder). get_or_create_asset resets a still-
     #    referenced errored row to pending on re-enqueue, so these only accumulate once every
     #    holder is gone; they carry no expires_at, so passes 1-3 never touch them.
-    orphans = db.execute(
+    orphans = (
-        select(MediaAsset).where(
+        db.execute(
-            MediaAsset.status == "error",
+            select(MediaAsset).where(
-            MediaAsset.ref_count == 0,
+                MediaAsset.status == "error",
-            MediaAsset.rel_path.is_(None),
+                MediaAsset.ref_count == 0,
+                MediaAsset.rel_path.is_(None),
+            )
         )
-    ).scalars().all()
+        .scalars()
+        .all()
+    )
     for asset in orphans:
         db.delete(asset)  # no file to unlink; FK SET NULL clears any stray job.asset_id
         reaped += 1
@@ -194,7 +228,11 @@ def run_download_gc(db: Session) -> dict:
     swept = _sweep_orphan_files(db, settings.download_root)
 
     result = {
-        "warned": warned, "expired": expired, "evicted": evicted, "reaped": reaped, "swept": swept
+        "warned": warned,
+        "expired": expired,
+        "evicted": evicted,
+        "reaped": reaped,
+        "swept": swept,
     }
     log.info("download_gc %s", result)
     return result

backend/app/downloads/jobdb.py

@@ -1,4 +1,5 @@
 """Short-lived-session DB helpers + staging paths shared by the worker parent and the child pipeline."""
+
 from pathlib import Path
 
 from sqlalchemy import text
@@ -10,6 +11,7 @@ from app.models import DownloadJob, MediaAsset
 
 # --- small DB helpers (short-lived sessions; the download itself holds no transaction) -------
 
+
 def _set_job(job_id: int, **fields) -> None:
     with SessionLocal() as db:
         job = db.get(DownloadJob, job_id)

backend/app/downloads/links.py

@@ -6,6 +6,7 @@ require a short-lived signed **grant** on the file request — because a `<video
 header, the password is exchanged once at `/unlock` for an HMAC grant appended to the file URL,
 so the raw password never rides in the media URL (or server logs).
 """
+
 import hashlib
 import hmac
 import secrets
@@ -27,10 +28,9 @@ def is_expired(link: DownloadLink) -> bool:
 
 # --- signed grants (password-protected file access) ----------------------------------------
 
+
 def _sign(msg: str) -> str:
-    return hmac.new(
+    return hmac.new(settings.secret_key.encode(), msg.encode(), hashlib.sha256).hexdigest()[:32]
-        settings.secret_key.encode(), msg.encode(), hashlib.sha256
-    ).hexdigest()[:32]
 
 
 def make_grant(token: str, password_version: int, now: float | None = None) -> str:
@@ -58,6 +58,7 @@ def check_grant(token: str, grant: str | None, password_version: int) -> bool:
 
 # --- serialization -------------------------------------------------------------------------
 
+
 def owner_view(link: DownloadLink, base_url: str = "") -> dict:
     """What the file owner sees when managing their links."""
     return {
@@ -72,7 +73,9 @@ def owner_view(link: DownloadLink, base_url: str = "") -> dict:
     }
 
 
-def public_meta(link: DownloadLink, job, asset, file_url: str, poster_url: str | None = None) -> dict:
+def public_meta(
+    link: DownloadLink, job, asset, file_url: str, poster_url: str | None = None
+) -> dict:
     """Metadata for the public watch page (only ever returned once access is authorized).
 
     Resolves the owner's per-download display overrides (custom title/channel) over the asset's
@@ -80,6 +83,10 @@ def public_meta(link: DownloadLink, job, asset, file_url: str, poster_url: str |
     from app.downloads import service
 
     source_url = service.reference_url(job, asset)
+    # Audio-only assets (mp3 rips, extracted audio) carry a codec but no video stream. WebKit (all
+    # iOS browsers) refuses to play them inside a <video> element, so the watch page must render a
+    # native <audio> instead — flag it here from the recorded codecs (a real video always has vcodec).
+    is_audio = asset.vcodec is None and asset.acodec is not None
     return {
         "title": (job and job.display_name) or asset.title,
         "uploader": (job and job.display_uploader) or asset.uploader,
@@ -89,6 +96,7 @@ def public_meta(link: DownloadLink, job, asset, file_url: str, poster_url: str |
         "duration_s": asset.duration_s,
         "width": asset.width,
         "height": asset.height,
+        "audio": is_audio,
         "allow_download": link.allow_download,
         "file_url": file_url,
         "poster_url": poster_url,

backend/app/downloads/og.py

@@ -6,6 +6,7 @@ This injects per-video OG/Twitter tags into the served HTML so a shared link unf
 video's title, channel and thumbnail. Real browsers ignore the extra tags and hydrate the SPA as
 usual. Password-protected / expired / invalid links fall back to the generic card (no metadata leak).
 """
+
 import html
 import re
 from pathlib import Path
@@ -71,9 +72,7 @@ def _jpeg_size(path: Path) -> tuple[int, int] | None:
 
 def _watch_tags(token: str, db: Session) -> str | None:
     """Build the OG/Twitter meta block for a watch token, or None to keep the generic card."""
-    link = db.execute(
+    link = db.execute(select(DownloadLink).where(DownloadLink.token == token)).scalar_one_or_none()
-        select(DownloadLink).where(DownloadLink.token == token)
-    ).scalar_one_or_none()
     if link is None or linksmod.is_expired(link):
         return None
     url = f"{settings.app_base}/watch/{token}"

backend/app/downloads/pipeline.py

@@ -1,4 +1,5 @@
 """Child-side download/edit pipeline — the yt-dlp download + ffmpeg edit that runs in the spawned subprocess."""
+
 import logging
 import re
 import shutil
@@ -7,13 +8,14 @@ from datetime import date, datetime, timedelta, timezone
 from pathlib import Path
 
 import yt_dlp
+from sqlalchemy import select
 
 from app import sysconfig
 from app.config import settings
 from app.db import SessionLocal
 from app.downloads import edit as editmod
 from app.downloads import errors as dlerrors
-from app.downloads import adapters, formats, service, storage
+from app.downloads import adapters, credentials, formats, service, storage
 from app.downloads.jobdb import (
     _edit_staging_dir,
     _job_status,
@@ -57,9 +59,7 @@ _YTDLP_TEMP_RE = re.compile(
 )
 
 
-def _notify_download_done(
+def _notify_download_done(db, user_id: int | None, title: str | None, job_id, *, kind: str) -> None:
-    db, user_id: int | None, title: str | None, job_id, *, kind: str
-) -> None:
     """Post a best-effort 'ready' notice to the owner's inbox — downloads/edits used to finish
     silently while scheduler jobs and playlist-push already notified (R11 feedback symmetry). A
     notification failure must never mask the completed job, so it is isolated + rolled back (mirrors
@@ -113,7 +113,7 @@ def _fill_asset_meta_early(asset_id: int, info: dict) -> None:
 
 def _make_progress_hook(job_id: int, asset_id: int):
     state = {"meta_done": False}
-    cancelled = _cancel_poll(job_id)          # same policy as the ffmpeg path
+    cancelled = _cancel_poll(job_id)  # same policy as the ffmpeg path
     allow_write = _rate_limiter(_DB_WRITE_INTERVAL_S)
 
     def hook(d: dict) -> None:
@@ -247,7 +247,9 @@ def _make_pp_hook(job_id: int):
 
     def hook(d: dict) -> None:
         if d.get("status") in ("started", "processing"):
-            _set_job(job_id, phase=_pp_phase(d.get("postprocessor") or ""), speed_bps=None, eta_s=None)
+            _set_job(
+                job_id, phase=_pp_phase(d.get("postprocessor") or ""), speed_bps=None, eta_s=None
+            )
 
     return hook
 
@@ -269,7 +271,9 @@ def _produced_file(info: dict, staging: Path) -> Path:
     back to the largest file left in staging that is neither a thumbnail nor one of yt-dlp's own
     working files (see `_YTDLP_TEMP_RE`)."""
     req = (info.get("requested_downloads") or [{}])[0]
-    fp = req.get("filepath") or req.get("_filename") or info.get("filepath") or info.get("_filename")
+    fp = (
+        req.get("filepath") or req.get("_filename") or info.get("filepath") or info.get("_filename")
+    )
     if fp:
         produced = Path(str(fp))
         if produced.is_file():
@@ -301,17 +305,23 @@ def _media_info(info: dict) -> dict:
     return {**info, **chosen}
 
 
-def _extract_with_fallback(job_id, asset_id, url, spec, staging, adapter):
+def _extract_with_fallback(job_id, asset_id, url, spec, staging, adapter, cookiefile=None):
     """Download via yt-dlp using the source `adapter`'s client sets + options. For YouTube the adapter
     supplies the primary player-client set then a POT-backed fallback (retried on a bot/DRM/format
-    error); other sources make a single attempt. Returns the info dict of the successful attempt."""
+    error); other sources make a single attempt. Returns the info dict of the successful attempt.
+
+    `cookiefile` (S6) is the acting user's per-host cookies.txt, already gated by
+    `credentials.resolve_cookiefile`; passed to every attempt so the fallback stays authenticated."""
     client_sets = adapter.client_sets()
     outtmpl = str(staging / "%(id)s.%(ext)s")
     last_exc: Exception | None = None
     for i, clients in enumerate(client_sets):
         opts = formats.build_ydl_opts(
-            spec, outtmpl, _make_progress_hook(job_id, asset_id),
+            spec,
+            outtmpl,
+            _make_progress_hook(job_id, asset_id),
             extra=adapter.ydl_opts_extra(clients),
+            cookiefile=cookiefile,
         )
         opts["postprocessor_hooks"] = [_make_pp_hook(job_id)]
         _touch_job(job_id)  # stall-watchdog heartbeat: extraction writes no progress, so a slow /
@@ -324,7 +334,12 @@ def _extract_with_fallback(job_id, asset_id, url, spec, staging, adapter):
         except yt_dlp.utils.DownloadError as exc:
             last_exc = exc
             if i < len(client_sets) - 1 and adapter.retriable(str(exc)):
-                log.info("job %s: clients %s failed (%s); retrying with fallback", job_id, clients, str(exc)[:70])
+                log.info(
+                    "job %s: clients %s failed (%s); retrying with fallback",
+                    job_id,
+                    clients,
+                    str(exc)[:70],
+                )
                 for p in list(staging.iterdir()):  # drop partials before the retry
                     try:
                         p.unlink()
@@ -363,13 +378,17 @@ def _ensure_browser_playable(job_id: int, produced: Path, spec: dict, info: dict
 
     dest = produced.with_name(produced.stem + ".browser.mp4")
     action = "re-encode" if (reenc_v or reenc_a) else "remux"
-    log.info("job %s: %s (v=%s a=%s mp4=%s) -> browser-safe mp4", job_id, action, vcodec, acodec, is_mp4)
+    log.info(
+        "job %s: %s (v=%s a=%s mp4=%s) -> browser-safe mp4", job_id, action, vcodec, acodec, is_mp4
+    )
     _set_job(job_id, phase="processing", progress=0, speed_bps=None, eta_s=None)
     # Prefer the info duration; probe the file when the extractor didn't provide one, so a long
     # re-encode still shows real progress instead of a frozen 0%.
     total_s = int(info["duration"]) if info.get("duration") else editmod.probe_duration(produced)
     editmod.run_ffmpeg(
-        editmod.build_browser_safe_cmd(produced, dest, reenc_v, reenc_a),
+        editmod.build_browser_safe_cmd(
+            produced, dest, reenc_v, reenc_a, threads=editmod.encode_thread_cap()
+        ),
         total_s=total_s,
         on_progress=_ffmpeg_progress_writer(job_id, "processing"),
         should_cancel=_cancel_poll(job_id),
@@ -383,8 +402,11 @@ def _ensure_browser_playable(job_id: int, produced: Path, spec: dict, info: dict
     return dest
 
 
-def _download(job_id: int, asset_id: int, source_kind: str, source_ref: str, spec: dict) -> None:
+def _download(
+    job_id: int, asset_id: int, source_kind: str, source_ref: str, spec: dict, user_id: int
+) -> None:
     staging = _staging_dir(asset_id)
+    cookiefile: Path | None = None
     try:
         if staging.exists():
             shutil.rmtree(staging, ignore_errors=True)
@@ -393,7 +415,27 @@ def _download(job_id: int, asset_id: int, source_kind: str, source_ref: str, spe
         _set_job(job_id, phase="downloading", progress=0)
         adapter = adapters.resolve_adapter(source_kind, source_ref)
         url = service.source_url(source_kind, source_ref)
-        info = _media_info(_extract_with_fallback(job_id, asset_id, url, spec, staging, adapter))
+        # S6: attach the acting user's per-host cookie ONLY when it's host-scoped to this URL AND
+        # this asset is that user's private, salted identity (the double gate lives in credentials).
+        with SessionLocal() as cdb:
+            asset_sig = cdb.execute(
+                select(MediaAsset.format_sig).where(MediaAsset.id == asset_id)
+            ).scalar_one_or_none()
+            if asset_sig is not None:
+                cookiefile = credentials.resolve_cookiefile(
+                    cdb, user_id, url, asset_sig, spec, staging
+                )
+        info = _media_info(
+            _extract_with_fallback(
+                job_id,
+                asset_id,
+                url,
+                spec,
+                staging,
+                adapter,
+                cookiefile=str(cookiefile) if cookiefile else None,
+            )
+        )
         # Re-resolve now that yt-dlp's info reveals the true source (a bare/redirected url only
         # identifies post-extraction) so the .nfo uniqueid type + uploader_url agree on one source.
         adapter = adapters.resolve_adapter(source_kind, source_ref, info)
@@ -436,8 +478,12 @@ def _download(job_id: int, asset_id: int, source_kind: str, source_ref: str, spe
                 asset.width = int(info["width"]) if info.get("width") else None
                 asset.height = int(info["height"]) if info.get("height") else None
                 asset.container = ext
-                asset.vcodec = info.get("vcodec") if info.get("vcodec") not in (None, "none") else None
+                asset.vcodec = (
-                asset.acodec = info.get("acodec") if info.get("acodec") not in (None, "none") else None
+                    info.get("vcodec") if info.get("vcodec") not in (None, "none") else None
+                )
+                asset.acodec = (
+                    info.get("acodec") if info.get("acodec") not in (None, "none") else None
+                )
                 asset.title = meta.title
                 asset.uploader = meta.uploader
                 asset.uploader_url = adapter.uploader_url(info)
@@ -456,8 +502,13 @@ def _download(job_id: int, asset_id: int, source_kind: str, source_ref: str, spe
             job = db.get(DownloadJob, job_id)
             # Capture the owner before commit expires the row (avoids a refresh SELECT just to notify).
             owner_id = job.user_id if job else None
-            fields = {"status": JobStatus.DONE, "progress": 100, "phase": None, "speed_bps": None,
+            fields = {
-                      "eta_s": None}
+                "status": JobStatus.DONE,
+                "progress": 100,
+                "phase": None,
+                "speed_bps": None,
+                "eta_s": None,
+            }
             if job and not job.display_name and meta.title:
                 # display_name is VARCHAR(255) — clip a long title (see _fill_asset_meta_early).
                 fields["display_name"] = meta.title[:255]
@@ -484,12 +535,20 @@ def _download(job_id: int, asset_id: int, source_kind: str, source_ref: str, spe
             # Atomic fail: only while still running, so a route-set cancel/pause is never overwritten
             # with `error` (error also holds the asset → would re-open the double-release class).
             service.finalize_if_running(
-                db, job_id, status=JobStatus.ERROR, error=msg,
+                db,
-                error_code=dlerrors.classify(msg), phase=None,
+                job_id,
+                status=JobStatus.ERROR,
+                error=msg,
+                error_code=dlerrors.classify(msg),
+                phase=None,
             )
             db.commit()
         log.warning("job %s failed: %s", job_id, msg)
     finally:
+        # Unlink the decrypted cookie file explicitly (the rmtree below removes the whole staging
+        # tree too, but never leave a plaintext session cookie on disk a moment longer than needed).
+        if cookiefile is not None:
+            cookiefile.unlink(missing_ok=True)
         shutil.rmtree(staging, ignore_errors=True)
 
 
@@ -501,8 +560,12 @@ def _fail_edit(job_id: int, asset_id: int, msg: str) -> None:
             asset.error = msg
         # Atomic fail: only while still running (see the download error path).
         service.finalize_if_running(
-            db, job_id, status=JobStatus.ERROR, error=msg,
+            db,
-            error_code=dlerrors.classify(msg), phase=None,
+            job_id,
+            status=JobStatus.ERROR,
+            error=msg,
+            error_code=dlerrors.classify(msg),
+            phase=None,
         )
         db.commit()
     log.warning("edit job %s failed: %s", job_id, msg)
@@ -517,7 +580,9 @@ def _process_edit(job_id: int, asset_id: int) -> None:
         user_id = job.user_id if job else None
         src_ready = src is not None and src.status == "ready" and bool(src.rel_path)
         src_rel = src.rel_path if src else None
-        src_w, src_h, src_dur = (src.width, src.height, src.duration_s) if src else (None, None, None)
+        src_w, src_h, src_dur = (
+            (src.width, src.height, src.duration_s) if src else (None, None, None)
+        )
 
     if not src_ready or not src_rel:
         _fail_edit(job_id, asset_id, "The source download is no longer available.")
@@ -541,7 +606,9 @@ def _process_edit(job_id: int, asset_id: int) -> None:
 
         _set_job(job_id, phase="editing", progress=0, speed_bps=None, eta_s=None)
         if edit_spec.get("segments"):
-            cmd, prep = editmod.build_concat_plan(src_path, dest_staging, edit_spec, out_ext, staging)
+            cmd, prep = editmod.build_concat_plan(
+                src_path, dest_staging, edit_spec, out_ext, staging
+            )
             for p, content in prep.items():
                 p.write_text(content, encoding="utf-8")
         else:

backend/app/downloads/quota.py

@@ -8,6 +8,7 @@ Limits come from a per-user DownloadQuota row if the admin set one, else the sys
 Enforcement split: max_jobs + max_bytes are checked at enqueue (they bound holdings); per-user
 max_concurrent is enforced by the worker at claim time (it bounds simultaneous downloads).
 """
+
 from dataclasses import dataclass
 
 from sqlalchemy import func, select

backend/app/downloads/runner.py

@@ -10,17 +10,50 @@ Contract: the asset is already claimed ('downloading') by the parent. This proce
 writes the terminal job/asset state itself (done / error, or a cooperative pause via the in-flight
 progress hook), then exits 0. An unexpected crash exits non-zero and the parent fails the job.
 """
+
 import logging
+import os
+import shutil
+import subprocess
 import sys
 
+from app.config import settings
 from app.db import SessionLocal
+from app.downloads import edit as editmod
 from app.models import DownloadJob
 
 log = logging.getLogger("siftlode.worker.runner")
 
 
+def _deprioritize() -> None:
+    """Lower THIS child's CPU + IO priority so its yt-dlp/ffmpeg work — every descendant inherits the
+    handicap — can never starve the co-located API, Postgres, or Plex playback. Run once at startup,
+    before any heavy work. Every step is best-effort: a failed renice must never fail a download."""
+    nice_inc = settings.download_child_nice
+    if nice_inc > 0:
+        try:
+            os.nice(nice_inc)  # relative increment; the child starts at the worker's nice (0)
+        except OSError as exc:
+            log.info("could not renice download child: %s", str(exc)[:120])
+    cls = settings.download_child_ionice_class
+    if cls > 0 and shutil.which("ionice"):  # util-linux; absent on some minimal images
+        try:
+            subprocess.run(
+                ["ionice", "-c", str(cls), "-p", str(os.getpid())],
+                capture_output=True,
+                timeout=5,
+                check=False,
+            )
+        except (OSError, subprocess.SubprocessError) as exc:
+            log.info("could not ionice download child: %s", str(exc)[:120])
+    # Cap OpenMP-parallel ffmpeg filters to the same budget as the encoder threads, so a filter
+    # graph can't fan out across every core behind ffmpeg's own `-threads` cap.
+    os.environ.setdefault("OMP_NUM_THREADS", str(editmod.encode_thread_cap()))
+
+
 def main() -> None:
     logging.basicConfig(level=logging.INFO)
+    _deprioritize()
     if len(sys.argv) != 2:
         print("usage: python -m app.downloads.runner <job_id>", file=sys.stderr)
         raise SystemExit(2)
@@ -34,6 +67,7 @@ def main() -> None:
         job_kind = job.job_kind
         spec = dict(job.profile_snapshot or {})
         source_kind, source_ref = job.source_kind, job.source_ref
+        user_id = job.user_id
     if asset_id is None:
         return
 
@@ -44,7 +78,7 @@ def main() -> None:
     if job_kind == "edit":
         _process_edit(job_id, asset_id)
     else:
-        _download(job_id, asset_id, source_kind, source_ref, spec)
+        _download(job_id, asset_id, source_kind, source_ref, spec, user_id)
 
 
 if __name__ == "__main__":

backend/app/downloads/service.py

@@ -6,6 +6,7 @@ cache hit (asset already ready) — marks the job done instantly with no re-down
 handles everything else. State transitions used by the routes (pause/resume/cancel/delete) and
 the worker live in `worker.py` / the routes; this module owns enqueue + asset bookkeeping.
 """
+
 from datetime import datetime, timezone
 
 from sqlalchemy import func, select, update
@@ -14,6 +15,7 @@ from sqlalchemy.orm import Session
 
 from app.config import settings
 from app.downloads import adapters
+from app.downloads import credentials
 from app.downloads import edit as editmod
 from app.downloads import formats, quota, storage
 from app.downloads.status import HOLDING, JobStatus
@@ -154,7 +156,12 @@ def enqueue(
     Raises quota.QuotaExceeded if the user is at their job-count or footprint cap."""
     quota.check_enqueue(db, user_id)
     spec = formats.normalize(spec)
-    sig = formats.format_sig(spec)
+    # If this user has a stored cookie for the download's host, salt the cache signature with their
+    # id so the cookie-authed result is a PRIVATE asset — never dedup-shared with, or cache-served
+    # to, another user (the load-bearing privacy fix; the worker attaches the cookie to exactly this
+    # salted identity). A normal download (no matching cookie) gets salt=None and shares as before.
+    salt = credentials.salt_for_enqueue(db, user_id, source_url(source_kind, source_ref))
+    sig = formats.format_sig(spec, salt=salt)
     asset = get_or_create_asset(db, source_kind, source_ref, sig)
     # Show a real title + thumbnail the moment it's queued (feed downloads are catalog videos).
     populate_from_catalog(db, asset)
@@ -166,7 +173,8 @@ def enqueue(
         source_ref=source_ref,
         profile_id=profile_id,
         profile_snapshot=spec,
-        display_name=display_name or (asset.title[:255] if asset.status == "ready" and asset.title else None),
+        display_name=display_name
+        or (asset.title[:255] if asset.status == "ready" and asset.title else None),
         status="queued",
         queue_pos=next_queue_pos(db),
     )

backend/app/downloads/storage.py

@@ -10,6 +10,7 @@ Plus a Kodi/Plex-readable `<basename>.nfo` (episodedetails) and `<basename>.jpg`
 a Plex library using a metadata agent picks up rich info. Channel = "show", video = episode,
 date-based — the community-standard mapping for YouTube-in-Plex.
 """
+
 import re
 import shutil
 import unicodedata
@@ -36,9 +37,27 @@ _DROP_CATEGORIES = {"So", "Sk", "Cc", "Cf", "Cs", "Co", "Cn"}
 # that accented names show in non-UTF-8 tools.
 _FOLD_MAP = str.maketrans(
     {
-        "ß": "ss", "ẞ": "SS", "ø": "o", "Ø": "O", "ł": "l", "Ł": "L", "đ": "d", "Đ": "D",
+        "ß": "ss",
-        "ð": "d", "Ð": "D", "þ": "th", "Þ": "Th", "æ": "ae", "Æ": "AE", "œ": "oe", "Œ": "OE",
+        "ẞ": "SS",
-        "ı": "i", "İ": "I", "·": " ", "–": "-", "—": "-",
+        "ø": "o",
+        "Ø": "O",
+        "ł": "l",
+        "Ł": "L",
+        "đ": "d",
+        "Đ": "D",
+        "ð": "d",
+        "Ð": "D",
+        "þ": "th",
+        "Þ": "Th",
+        "æ": "ae",
+        "Æ": "AE",
+        "œ": "oe",
+        "Œ": "OE",
+        "ı": "i",
+        "İ": "I",
+        "·": " ",
+        "–": "-",
+        "—": "-",
     }
 )
 
@@ -174,7 +193,7 @@ def _nfo_xml(meta: MediaMeta) -> str:
         f"  <premiered>{aired}</premiered>",
         f"  <year>{year}</year>",
         f"  <plot>{escape(meta.description or '')}</plot>",
-        f"  <uniqueid type=\"{escape(meta.uniqueid_type)}\" default=\"true\">{escape(meta.video_id)}</uniqueid>",
+        f'  <uniqueid type="{escape(meta.uniqueid_type)}" default="true">{escape(meta.video_id)}</uniqueid>',
     ]
     if meta.duration_s:
         parts.append(f"  <runtime>{round(meta.duration_s / 60)}</runtime>")
@@ -190,9 +209,7 @@ def _sidecar(media: Path, ext: str) -> Path:
     return media.with_suffix(ext)
 
 
-def write_sidecars(
+def write_sidecars(download_root: str, rel: str, meta: MediaMeta, thumb_src: Path | None) -> bool:
-    download_root: str, rel: str, meta: MediaMeta, thumb_src: Path | None
-) -> bool:
     """Write `<basename>.nfo` + `<basename>.jpg` next to the media file. Returns True if written."""
     media = abs_path(download_root, rel)
     try:

backend/app/email.py

@@ -4,6 +4,7 @@ Gmail SMTP + App Password by default. Deliberately fail-soft: if SMTP isn't conf
 or a send errors, we log and return False so the caller can fall back to in-app
 notifications — email is never load-bearing for the app's core flow.
 """
+
 import logging
 import smtplib
 import ssl
@@ -77,8 +78,7 @@ def send_access_approved(email: str) -> bool:
         "Open Siftlode and sign in, and your YouTube subscriptions feed will be ready:\n\n"
         f"{settings.app_base}\n\n"
         "If you weren't expecting this, you can ignore the message.\n\n"
-        "— Siftlode"
+        "— Siftlode" + (f"\n\nQuestions? Just reply to this email ({admin})." if admin else "")
-        + (f"\n\nQuestions? Just reply to this email ({admin})." if admin else "")
     )
     return _send([email], "You're in — your Siftlode access is approved", body, reply_to=admin)
 
@@ -90,8 +90,7 @@ def send_access_denied(email: str) -> bool:
         "Thanks for your interest in Siftlode. After review, your access request wasn't approved "
         "this time.\n\n"
         "If you think this was a mistake, you're welcome to reply and let us know.\n\n"
-        "— Siftlode"
+        "— Siftlode" + (f"\n\nQuestions? Just reply to this email ({admin})." if admin else "")
-        + (f"\n\nQuestions? Just reply to this email ({admin})." if admin else "")
     )
     return _send([email], "About your Siftlode access request", body, reply_to=admin)
 
@@ -113,9 +112,7 @@ def send_account_unsuspended(to: str, operator: str | None) -> bool:
     body = (
         "Hi,\n\n"
         "Your Siftlode account has been reinstated — the suspension was lifted and you can sign "
-        "in again."
+        "in again." + contact + "\n\n— Siftlode"
-        + contact
-        + "\n\n— Siftlode"
     )
     return _send([to], "Your Siftlode account has been reinstated", body, reply_to=operator)
 
@@ -130,9 +127,7 @@ def send_account_deleted(to: str, operator: str | None) -> bool:
         "Hi,\n\n"
         "Your Siftlode account and all associated data — subscriptions, tags, watch history, "
         "playlists and settings — have been permanently deleted. Nothing is retained.\n\n"
-        "Thanks for having used Siftlode."
+        "Thanks for having used Siftlode." + contact + "\n\n— Siftlode"
-        + contact
-        + "\n\n— Siftlode"
     )
     return _send([to], "Your Siftlode account has been deleted", body, reply_to=operator)
 
@@ -147,9 +142,7 @@ def send_account_suspended(to: str, operator: str | None) -> bool:
     body = (
         "Hi,\n\n"
         "A sign-in to your Siftlode account was just attempted, but the account is currently "
-        "suspended, so access was denied.\n\n"
+        "suspended, so access was denied.\n\n" + contact + "— Siftlode"
-        + contact
-        + "— Siftlode"
     )
     return _send([to], "Your Siftlode account is suspended", body, reply_to=operator)
 

backend/app/feedquery.py

@@ -100,6 +100,11 @@ def serialize(row) -> dict:
     }
 
 
+def _tag_visible(user: User):
+    """A ChannelTag row this user may see: a shared system tag (user_id NULL) or their own."""
+    return or_(ChannelTag.user_id.is_(None), ChannelTag.user_id == user.id)
+
+
 def tags_by_category(db: Session, tag_ids: list[int]) -> dict[str, list[int]]:
     """Group tag ids by their category (language/topic/other). The single source for the category split
     shared by the feed's tag WHERE (filtered_query) and the facet/count summary (feed_summary), so the
@@ -118,6 +123,7 @@ def filtered_query(
     *,
     tags: list[int],
     tag_modes: dict[str, str],
+    untagged: bool = False,
     channel_ids: list[str],
     q: str | None,
     min_duration: int | None,
@@ -144,9 +150,7 @@ def filtered_query(
     state = aliased(VideoState)
     status_expr = func.coalesce(state.status, "new")
 
-    query = select(*feed_columns(user, state)).join(
+    query = select(*feed_columns(user, state)).join(Channel, Channel.id == Video.channel_id)
-        Channel, Channel.id == Video.channel_id
-    )
 
     if scope == "all":
         # Whole shared catalog; subscription is optional (only for priority sort).
@@ -181,9 +185,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(
+    query = join_state(state, and_(state.video_id == Video.id, state.user_id == user.id))
-        state, and_(state.video_id == Video.id, state.user_id == user.id)
-    )
 
     # 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;
@@ -197,9 +199,7 @@ def filtered_query(
     # asking for scope=all + channel_ids (the channel is in this accessible set).
     accessible_explore = (
         select(Subscription.channel_id).where(Subscription.user_id == user.id)
-    ).union(
+    ).union(select(ExploredChannel.channel_id).where(ExploredChannel.user_id == user.id))
-        select(ExploredChannel.channel_id).where(ExploredChannel.user_id == user.id)
-    )
     query = query.where(
         or_(
             Video.via_explore.is_(False),
@@ -244,13 +244,9 @@ def filtered_query(
         query = query.where(Video.duration_seconds <= max_duration)
     if max_age_days is not None:
         cutoff = datetime.now(timezone.utc).timestamp() - max_age_days * 86400
-        query = query.where(
+        query = query.where(Video.published_at >= datetime.fromtimestamp(cutoff, tz=timezone.utc))
-            Video.published_at >= datetime.fromtimestamp(cutoff, tz=timezone.utc)
-        )
     if published_after is not None:
-        start = datetime.combine(
+        start = datetime.combine(published_after, datetime.min.time(), tzinfo=timezone.utc)
-            published_after, datetime.min.time(), tzinfo=timezone.utc
-        )
         query = query.where(Video.published_at >= start)
     if published_before is not None:
         end = datetime.combine(
@@ -304,7 +300,7 @@ def filtered_query(
         # selections but still honours the other categories' filters).
         if exclude_tag_category:
             by_category.pop(exclude_tag_category, None)
-        visible = or_(ChannelTag.user_id.is_(None), ChannelTag.user_id == user.id)
+        visible = _tag_visible(user)
         for category, ids in by_category.items():
             if tag_modes.get(category, "or") == "and" and len(ids) > 1:
                 sub = (
@@ -314,11 +310,21 @@ def filtered_query(
                     .having(func.count(func.distinct(ChannelTag.tag_id)) == len(set(ids)))
                 )
             else:
-                sub = select(ChannelTag.channel_id).where(
+                sub = select(ChannelTag.channel_id).where(ChannelTag.tag_id.in_(ids), visible)
-                    ChannelTag.tag_id.in_(ids), visible
-                )
             query = query.where(Video.channel_id.in_(sub))
 
+    # "No tag" filter (item B): keep only channels the USER has not personally tagged in the channel
+    # manager — i.e. channels with no ChannelTag row owned by this user (user_id == me, source "user").
+    # System/auto tags (user_id NULL) don't count: this is "channels I haven't tagged yet". It's a plain
+    # channel-level narrowing (like channel_ids/date), NOT a tag-category selection, so feed_summary
+    # leaves it applied in the base pass and needs no mirror in passes().
+    if untagged:
+        query = query.where(
+            Video.channel_id.notin_(
+                select(ChannelTag.channel_id).where(ChannelTag.user_id == user.id)
+            )
+        )
+
     # Content type: Normal / Shorts / Live·Upcoming as a union of enabled types.
     # Explicit Watched/Saved/Hidden views show every type so nothing goes missing.
     explicit_view = show in ("watched", "hidden")
@@ -367,8 +373,9 @@ def feed_summary(db: Session, user: User, **params) -> dict:
     within, a category in "and" mode requires ALL its selected tags)."""
     tag_modes = params.get("tag_modes", {})
 
-    # ONE pass: per-channel video counts for the base filter, tags stripped so every candidate channel
+    # ONE pass: per-channel video counts for the base filter, the positive tag filter stripped so every
-    # is present to facet over.
+    # candidate channel is present to facet over. The "No tag" (untagged) filter is a plain channel-level
+    # narrowing, so it STAYS applied here (not stripped) and the facets naturally drill down within it.
     base, _rank = filtered_query(db, user, **{**params, "tags": []})
     ch_col = base.with_only_columns(Video.channel_id).subquery()
     channel_counts: dict[str, int] = dict(
@@ -398,12 +405,15 @@ def feed_summary(db: Session, user: User, **params) -> dict:
 
     def passes(ch_tags: set[int], exclude_cat: str | None) -> bool:
         # The Python mirror of filtered_query's SQL tag block (AND across categories, OR within,
-        # topic-AND requires all selected) — keep in lockstep; see the ⚠️ note there.
+        # topic-AND requires all selected) — keep in lockstep; see the ⚠️ note there. (The "No tag"
+        # filter is applied in the base query above, so it needs no mirror here.)
         for cat, sel in selected_by_cat.items():
             if cat == exclude_cat:
                 continue
             if tag_modes.get(cat, "or") == "and" and len(sel) > 1:
-                if not sel <= ch_tags:  # conjunctive: channel must carry ALL this category's selected tags
+                if (
+                    not sel <= ch_tags
+                ):  # conjunctive: channel must carry ALL this category's selected tags
                     return False
             elif not (sel & ch_tags):  # disjunctive: channel must carry ANY selected tag
                 return False
@@ -427,6 +437,30 @@ def feed_summary(db: Session, user: User, **params) -> dict:
     return {"count": count, "facets": facets}
 
 
+def feed_years(db: Session, user: User, **params) -> list[dict]:
+    """Per-year video counts for the current filter context (item 8 / E8·S2): powers the year control's
+    quick chips + the "more…" per-year-counts dialog. Every OTHER filter (tags, channels, content type,
+    search, watch state, scope) still applies, but the publish-date narrowing itself is DROPPED —
+    published_after/before and max_age_days — so the year list is the full set of years available and
+    switching years stays a live drill-down (the same principle as the tag facets dropping their own
+    category). One grouped query, newest year first."""
+    base, _rank = filtered_query(
+        db,
+        user,
+        **{**params, "published_after": None, "published_before": None, "max_age_days": None},
+    )
+    pub = base.with_only_columns(Video.published_at.label("pub")).subquery()
+    # Bucket by UTC year to stay in lockstep with the year filter's boundaries (published_after/before
+    # are pinned to UTC in filtered_query). published_at is timestamptz, so a bare extract('year', …)
+    # would use the DB session timezone and, on a non-UTC session, bucket a video near a year boundary
+    # into a different year than the filter includes — the count and the drill-down feed would disagree.
+    year = func.extract("year", func.timezone("UTC", pub.c.pub))
+    rows = db.execute(
+        select(year.label("y"), func.count()).group_by(year).order_by(year.desc())
+    ).all()
+    return [{"year": int(y), "count": n} for y, n in rows if y is not None]
+
+
 # --- Keyset (cursor) pagination ---------------------------------------------
 #
 # Each sort is described as an ordered list of key columns (expression + direction
@@ -443,19 +477,52 @@ def sort_keys(sort: str, seed: int, rank_expr=None) -> list[dict]:
         "oldest": [{**pub, "asc": True}],
         "views": [{"expr": Video.view_count, "asc": False, "nullable": True, "kind": "num"}],
         "views_asc": [{"expr": Video.view_count, "asc": True, "nullable": True, "kind": "num"}],
-        "duration_desc": [{"expr": Video.duration_seconds, "asc": False, "nullable": True, "kind": "num"}],
+        "duration_desc": [
-        "duration_asc": [{"expr": Video.duration_seconds, "asc": True, "nullable": True, "kind": "num"}],
+            {"expr": Video.duration_seconds, "asc": False, "nullable": True, "kind": "num"}
+        ],
+        "duration_asc": [
+            {"expr": Video.duration_seconds, "asc": True, "nullable": True, "kind": "num"}
+        ],
         "title": [{"expr": func.lower(Video.title), "asc": True, "nullable": True, "kind": "str"}],
-        "title_desc": [{"expr": func.lower(Video.title), "asc": False, "nullable": True, "kind": "str"}],
+        "title_desc": [
-        "subscribers": [{"expr": Channel.subscriber_count, "asc": False, "nullable": True, "kind": "num"}],
+            {"expr": func.lower(Video.title), "asc": False, "nullable": True, "kind": "str"}
-        "subscribers_asc": [{"expr": Channel.subscriber_count, "asc": True, "nullable": True, "kind": "num"}],
+        ],
+        "subscribers": [
+            {"expr": Channel.subscriber_count, "asc": False, "nullable": True, "kind": "num"}
+        ],
+        "subscribers_asc": [
+            {"expr": Channel.subscriber_count, "asc": True, "nullable": True, "kind": "num"}
+        ],
         # Your per-channel priority (set in the channel manager), newest first within a tier.
         # coalesce keeps it null-safe in "all" scope where unsubscribed channels have no row.
-        "priority": [{"expr": func.coalesce(Subscription.priority, 0), "asc": False, "nullable": False, "kind": "num"}, pub],
+        "priority": [
-        "priority_asc": [{"expr": func.coalesce(Subscription.priority, 0), "asc": True, "nullable": False, "kind": "num"}, pub],
+            {
+                "expr": func.coalesce(Subscription.priority, 0),
+                "asc": False,
+                "nullable": False,
+                "kind": "num",
+            },
+            pub,
+        ],
+        "priority_asc": [
+            {
+                "expr": func.coalesce(Subscription.priority, 0),
+                "asc": True,
+                "nullable": False,
+                "kind": "num",
+            },
+            pub,
+        ],
         # Deterministic per-seed shuffle: the same seed yields the same total order, so
         # the cursor stays valid across pages of one shuffled run.
-        "shuffle": [{"expr": func.md5(func.concat(Video.id, str(seed))), "asc": True, "nullable": False, "kind": "str"}],
+        "shuffle": [
+            {
+                "expr": func.md5(func.concat(Video.id, str(seed))),
+                "asc": True,
+                "nullable": False,
+                "kind": "str",
+            }
+        ],
     }
     # Relevance (search-result ranking) only exists when there's a query to rank against;
     # without `rank_expr` it falls through to newest. Ties break by newest, then id.
@@ -490,6 +557,7 @@ def keyset_predicate(keys: list[dict], cursor_id: str):
     `newest` pages went 51.8k -> ~0.5k buffers). A NULLABLE key can't take that bound (the
     NULL group sorts after every value and `col <= v` would drop it), so it keeps the pure
     NULL-aware OR form."""
+
     def eq(k):
         return k["expr"].is_(None) if k["value"] is None else k["expr"] == k["value"]