diff --git a/backend/alembic/versions/0083_comfort_pins.py b/backend/alembic/versions/0083_comfort_pins.py
new file mode 100644
index 00000000..e4774457
--- /dev/null
+++ b/backend/alembic/versions/0083_comfort_pins.py
@@ -0,0 +1,78 @@
+"""comfort_pins — the shelf you fall back to when you cannot decide
+
+Revision ID: 0083_comfort_pins
+Revises: 0082_invite_lang
+Create Date: 2026-09-06
+
+E24's one new table. The shelf holds five different things — a video, a YouTube channel, a Plex show,
+a Plex movie, a playlist — because comfort media is usually "put on X", a series or a channel, and
+almost never one particular video.
+
+Three shapes were possible and two are wrong here:
+
+- a boolean on `video_states` / `plex_states` cannot ORDER a shelf, and has nowhere to put a channel
+  or a show (neither has a per-user state row at all);
+- a polymorphic `(kind, ref)` string pair orders and holds anything, but drops the foreign key — and a
+  Plex item removed from the library, or a channel the user unsubscribes and prunes, would then leave
+  a dead tile on the one surface whose entire job is to be reliable when nothing else is.
+
+So: one nullable CASCADE-ing FK per kind, plus `num_nonnulls(...) = 1`. The union invariant lives in
+the schema rather than in a comment at every insert site, which is the same rule that the
+`cancelled`/`canceled` bug taught this codebase.
+
+NULLs are DISTINCT in a Postgres unique index, so the five per-kind `(user_id, <ref>)` uniques each
+stop a double-pin without colliding with the other four kinds' NULL rows.
+
+No backfill: an empty shelf is the honest state for every existing account. The surface renders its
+own empty state and invites the first pin.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0083_comfort_pins"
+down_revision: str | None = "0082_invite_lang"
+branch_labels: str | None = None
+depends_on: str | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "comfort_pins",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column(
+            "user_id",
+            sa.Integer(),
+            sa.ForeignKey("users.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("video_id", sa.String(length=16), sa.ForeignKey("videos.id", ondelete="CASCADE")),
+        sa.Column(
+            "channel_id", sa.String(length=32), sa.ForeignKey("channels.id", ondelete="CASCADE")
+        ),
+        sa.Column("plex_item_id", sa.Integer(), sa.ForeignKey("plex_items.id", ondelete="CASCADE")),
+        sa.Column("plex_show_id", sa.Integer(), sa.ForeignKey("plex_shows.id", ondelete="CASCADE")),
+        sa.Column("playlist_id", sa.Integer(), sa.ForeignKey("playlists.id", ondelete="CASCADE")),
+        sa.Column("position", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+        sa.CheckConstraint(
+            "num_nonnulls(video_id, channel_id, plex_item_id, plex_show_id, playlist_id) = 1",
+            name="ck_comfort_pin_one_ref",
+        ),
+        sa.UniqueConstraint("user_id", "video_id", name="uq_comfort_pin_video"),
+        sa.UniqueConstraint("user_id", "channel_id", name="uq_comfort_pin_channel"),
+        sa.UniqueConstraint("user_id", "plex_item_id", name="uq_comfort_pin_plex_item"),
+        sa.UniqueConstraint("user_id", "plex_show_id", name="uq_comfort_pin_plex_show"),
+        sa.UniqueConstraint("user_id", "playlist_id", name="uq_comfort_pin_playlist"),
+    )
+    op.create_index("ix_comfort_pins_user_id", "comfort_pins", ["user_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_comfort_pins_user_id", table_name="comfort_pins")
+    op.drop_table("comfort_pins")
diff --git a/backend/app/discover/__init__.py b/backend/app/discover/__init__.py
new file mode 100644
index 00000000..e6f779b7
--- /dev/null
+++ b/backend/app/discover/__init__.py
@@ -0,0 +1,10 @@
+"""E24 — the read-model behind the Home surface ("Mit nézzek most?").
+
+The anti-paralysis answer layer: one hero suggestion, a cross-source resume rail, and the user's own
+comfort shelf, all served from our own Postgres with no upstream call at serve time (the same rule
+the Plex Discover surface follows).
+
+Layering, mirroring `app/plex/`: `budget` (the time bands) and `cards` (the unified row shape) are
+pure; `queries` holds the SQL; `rails`, `pins` and `hero` compose. Nothing here imports from
+`app.routes` — `routes/discover.py` is a thin HTTP shell over these.
+"""
diff --git a/backend/app/discover/budget.py b/backend/app/discover/budget.py
new file mode 100644
index 00000000..23e2f558
--- /dev/null
+++ b/backend/app/discover/budget.py
@@ -0,0 +1,62 @@
+"""The time budget — E24's replacement for "what mood are you in?".
+
+`random-gondolatok` 15/D asked for a mood questionnaire before the recommendations. The same file's
+later BA-interview note (#23) says the mood half is not what makes this work: the friction is
+deciding at all. A mood→genre mapping would also be invented data — we store no mood signal and never
+will — whereas "how much time do you have" is a question the library can actually answer, from a
+column we already have.
+
+So the surface asks about TIME, and the answer is deterministic: four bands, no model, no latency.
+The band also picks the source for free — ten minutes is a YouTube video, an evening is a movie —
+which is exactly the cross-source behaviour the Home surface exists to provide.
+
+The bands overlap on purpose. "I have 30 minutes" should not be answered with a two-minute Short just
+because it fits; a lower bound is as much a part of the promise as the upper one.
+"""
+
+from typing import Literal
+
+Budget = Literal["short", "medium", "long", "any"]
+
+DEFAULT_BUDGET: Budget = "any"
+
+# budget -> (min_seconds, max_seconds); None = unbounded on that side.
+BANDS: dict[str, tuple[int | None, int | None]] = {
+    "short": (None, 12 * 60),  # "10 perc" — a bit of slack over the round number the chip shows
+    "medium": (8 * 60, 40 * 60),  # "30 perc"
+    "long": (40 * 60, None),  # "Egy este"
+    "any": (None, None),  # "Mindegy"
+}
+
+
+def band(budget: str) -> tuple[int | None, int | None]:
+    """The (min, max) seconds for a budget. Unknown budgets fall back to the unbounded band rather
+    than raising: the route validates the vocabulary with a Literal, so anything reaching here with a
+    bad value is our own bug, and an unfiltered shelf is a better failure than a 500 on the landing
+    page."""
+    return BANDS.get(budget, BANDS["any"])
+
+
+def remaining_seconds(duration_seconds: int | None, position_seconds: int) -> int | None:
+    """What is actually LEFT to watch. The question is "how much time do you have", so a 90-minute
+    film with 8 minutes to go belongs in the ten-minute band, not the evening one. Unknown duration
+    stays unknown — see `fits`."""
+    if duration_seconds is None:
+        return None
+    return max(0, duration_seconds - max(0, position_seconds))
+
+
+def fits(seconds: int | None, budget: str) -> bool:
+    """Does a runtime fit the budget?
+
+    An unknown duration fits only the unbounded band. We cannot promise an item finishes inside ten
+    minutes when we do not know how long it is, and quietly including it would make the one control
+    on this page dishonest — the failure mode the house rules call an un-honest state."""
+    lo, hi = band(budget)
+    if seconds is None:
+        return lo is None and hi is None
+    if lo is not None and seconds < lo:
+        return False
+    if hi is not None and seconds > hi:
+        return False
+    return True
diff --git a/backend/app/discover/cards.py b/backend/app/discover/cards.py
new file mode 100644
index 00000000..9b4f9da1
--- /dev/null
+++ b/backend/app/discover/cards.py
@@ -0,0 +1,134 @@
+"""The unified discover card: one row shape the Home surface can render without knowing which
+module it came from.
+
+This is deliberately a THIN mapping, not a new serializer layer. The per-source projections already
+exist and are already the shapes the frontend's `VideoCard` and `PlexPosterCard` consume —
+`feedquery.serialize` for YouTube, `plex.cards.movie_card`/`show_card`/`leaf_card` for Plex — so each
+unified card carries the native payload under `video` / `plex` alongside the handful of fields the
+rail chrome itself needs (title, thumbnail, progress, runtime). The Home page therefore renders with
+the existing card components rather than a third one, and a field added to a feed row or a Plex card
+still gets added in exactly one place. CLAUDE.md's "not wanted" list names a base serializer for
+precisely this reason.
+
+`aspect` is the one genuinely new field. A rail has to know whether a tile is a 16:9 still or a 2:3
+poster, and that is a property of the ARTWORK, not of the module: a Plex episode ships a landscape
+still exactly like a YouTube video does, while a movie and a show ship posters. Emitting it lets the
+rail layer stay domain-free instead of branching on `source`.
+
+`subtitle` only ever carries DATA — a channel name, a show title, a year. Nothing here composes an
+English phrase: this repo is HU+EN with zero key drift, and a sentence assembled in the backend is a
+string no translation file can reach. Anything that needs wording (season counts, "N videó") is left
+to the client, which has both the native payload and the translations.
+"""
+
+from app.discover.budget import remaining_seconds
+from app.models import Channel, Playlist
+
+# Plex artwork is a poster for the things you browse and a landscape still for the things you play.
+_POSTER_KINDS = ("movie", "show")
+
+
+def _base(
+    *,
+    source: str,  # "youtube" | "plex"
+    kind: str,  # "video" | "movie" | "show" | "episode" | "channel" | "playlist"
+    id: str,
+    title: str | None,
+    aspect: str,  # "wide" (16:9) | "poster" (2:3)
+    subtitle: str | None = None,
+    thumb: str | None = None,
+    duration_seconds: int | None = None,
+    position_seconds: int = 0,
+    status: str = "new",
+) -> dict:
+    return {
+        "source": source,
+        "kind": kind,
+        "id": id,
+        "title": title or "",
+        "subtitle": subtitle,
+        "thumb": thumb,
+        "aspect": aspect,
+        "duration_seconds": duration_seconds,
+        "position_seconds": position_seconds,
+        "remaining_seconds": remaining_seconds(duration_seconds, position_seconds),
+        "status": status,
+        # Native payloads — at most one is ever set. The client opens the player from these.
+        "video": None,
+        "plex": None,
+    }
+
+
+def from_video(row: dict) -> dict:
+    """`row` is a `feedquery.serialize` dict — the same object the feed grid renders."""
+    card = _base(
+        source="youtube",
+        kind="video",
+        id=row["id"],
+        title=row["title"],
+        aspect="wide",
+        subtitle=row.get("channel_title"),
+        thumb=row.get("thumbnail_url"),
+        duration_seconds=row.get("duration_seconds"),
+        position_seconds=row.get("position_seconds") or 0,
+        status=row.get("status") or "new",
+    )
+    card["video"] = row
+    return card
+
+
+def from_plex(row: dict) -> dict:
+    """`row` is a `plex.cards` dict (movie / show / leaf). Its own `type` is the kind."""
+    kind = row.get("type") or "movie"
+    if kind == "episode":
+        subtitle = row.get("show_title")
+    elif kind == "show":
+        subtitle = None  # season count lives in the native payload; the client words it
+    else:
+        subtitle = str(row["year"]) if row.get("year") else None
+
+    card = _base(
+        source="plex",
+        kind=kind,
+        id=row["id"],
+        title=row.get("title"),
+        aspect="poster" if kind in _POSTER_KINDS else "wide",
+        subtitle=subtitle,
+        thumb=row.get("thumb"),
+        duration_seconds=row.get("duration_seconds"),
+        position_seconds=row.get("position_seconds") or 0,
+        status=row.get("status") or "new",
+    )
+    card["plex"] = row
+    return card
+
+
+def from_channel(ch: Channel, unwatched: int) -> dict:
+    """A pinned channel. It is not itself playable — the shelf resolves it to its newest unwatched
+    video on click (see `resolve.py`) — so the tile advertises what is waiting there instead of a
+    runtime it does not have."""
+    card = _base(
+        source="youtube",
+        kind="channel",
+        id=ch.id,
+        title=ch.title,
+        aspect="wide",
+        thumb=ch.thumbnail_url,
+    )
+    card["unwatched_count"] = unwatched
+    return card
+
+
+def from_playlist(pl: Playlist, item_count: int, thumb: str | None) -> dict:
+    """A pinned playlist. `thumb` is its first item's thumbnail — a shelf of blank tiles would defeat
+    the point of a surface you are supposed to be able to scan at a glance."""
+    card = _base(
+        source="youtube",
+        kind="playlist",
+        id=str(pl.id),
+        title=pl.name,
+        aspect="wide",
+        thumb=thumb,
+    )
+    card["item_count"] = item_count
+    return card
diff --git a/backend/app/discover/hero.py b/backend/app/discover/hero.py
new file mode 100644
index 00000000..b9b7b086
--- /dev/null
+++ b/backend/app/discover/hero.py
@@ -0,0 +1,71 @@
+"""The hero: the single suggestion the Home surface leads with.
+
+One rule, in one function, because the page's entire promise is that it answers rather than offers.
+Three tiers, tried in order:
+
+1. **resume** — something already started whose REMAINING time fits the budget. Almost always the
+   right answer: it is the thing the user themselves chose most recently, and finishing it costs no
+   decision at all.
+2. **comfort** — a pinned shelf entry that fits. The user's own declared fallback, which beats
+   anything we could infer.
+3. **surprise** — a seeded draw from what is unseen and fits.
+
+The tier travels with the card so the client can label it honestly ("Ott hagytad abba" vs "A
+polcodról" vs "Meglepetés") instead of guessing from the shape of the payload — a kicker that says
+"where you left off" over a film the user has never opened is exactly the kind of quiet lie the house
+rules call a dishonest state.
+
+Budgets are matched against REMAINING time throughout (`budget.remaining_seconds`, which equals the
+runtime for anything not yet started): with eight minutes left, a 90-minute film is a ten-minute
+answer.
+"""
+
+from sqlalchemy.orm import Session
+
+from app.discover import cards, pins, queries, rails
+from app.discover.budget import fits
+from app.models import PlexShow, User
+from app.plex.lookups import by_rating_key
+
+HERO_TIERS = ("resume", "comfort", "surprise")
+
+
+def _playable(db: Session, uid: int, card: dict) -> dict | None:
+    """A hero has to have a Play button, so a Plex SHOW is resolved to the episode that would start.
+    Everything else already plays itself. None when the show has no episodes mirrored yet."""
+    if not (card.get("source") == "plex" and card.get("kind") == "show"):
+        return card
+    sh = by_rating_key(db, PlexShow, card["id"])
+    if sh is None:
+        return None
+    ep = queries.next_episode(db, uid, sh.id)
+    return cards.from_plex(ep) if ep else None
+
+
+def pick(
+    db: Session,
+    user: User,
+    budget: str,
+    seed: int,
+    *,
+    resume: list[dict] | None = None,
+    shelf: list[dict] | None = None,
+) -> dict | None:
+    """`{"tier": ..., "card": ...}`, or None when nothing at all fits.
+
+    `resume` and `shelf` may be passed pre-computed so the consolidated `/api/discover` response does
+    not build the same two rails twice; the reroll endpoint passes neither and pays only for what it
+    reaches."""
+    for card in rails.continue_watching(db, user) if resume is None else resume:
+        playable = _playable(db, user.id, card)
+        if playable is not None and fits(playable["remaining_seconds"], budget):
+            return {"tier": "resume", "card": playable}
+
+    for card in pins.playables(pins.shelf(db, user) if shelf is None else shelf):
+        if fits(card["remaining_seconds"], budget):
+            return {"tier": "comfort", "card": card}
+
+    for card in rails.surprise(db, user, budget, seed, limit=1):
+        return {"tier": "surprise", "card": card}
+
+    return None
diff --git a/backend/app/discover/pins.py b/backend/app/discover/pins.py
new file mode 100644
index 00000000..3152cc29
--- /dev/null
+++ b/backend/app/discover/pins.py
@@ -0,0 +1,154 @@
+"""The comfort shelf: pin a thing you can always fall back to, and have it be one click from playing.
+
+Five kinds are pinnable — a video, a YouTube channel, a Plex show, a Plex movie, a playlist — because
+comfort media is almost always "put on X", a series or a channel, and only sometimes one particular
+video. `models.ComfortPin` carries that union as five nullable FKs under a `num_nonnulls(...) = 1`
+check; this module is where the union is spoken about in one vocabulary, so no call site invents a
+second one.
+
+Every shelf card carries a resolved `next`: the thing that actually plays when you click the tile —
+a show's in-progress-or-next episode, a channel's newest unwatched video, a playlist's first
+unwatched item. It costs a query per non-leaf pin, which is affordable precisely because this shelf is
+short and hand-curated, and it buys the property the whole surface exists for: one click, no
+intermediate page, no second decision.
+
+Domain module, so nothing here raises HTTP. `resolve_ref` returns None for "no such thing, or not
+yours" and the route turns that into its 404 — core never imports routes.
+"""
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.discover import cards, queries
+from app.models import Channel, ComfortPin, PlexItem, PlexShow, Playlist, User, Video
+from app.plex import rollup
+from app.plex.cards import leaf_card, show_card
+from app.plex.lookups import by_rating_key
+
+# kind -> the ComfortPin column that holds it. The client's vocabulary and the schema's, in one place.
+PIN_COLUMNS: dict[str, str] = {
+    "video": "video_id",
+    "channel": "channel_id",
+    "plex_item": "plex_item_id",
+    "plex_show": "plex_show_id",
+    "playlist": "playlist_id",
+}
+PIN_KINDS = tuple(PIN_COLUMNS)
+
+
+def resolve_ref(db: Session, user: User, kind: str, ref: str) -> tuple[str, object] | None:
+    """Translate the client's (kind, ref) into the (column, value) the row stores, or None.
+
+    The client speaks the ids the rest of the API speaks: a video id, a channel id, a Plex
+    `rating_key`, a playlist id. Plex's internal integer keys never leave the backend, so the
+    translation belongs here rather than in five call sites."""
+    if kind == "video":
+        return ("video_id", ref) if db.get(Video, ref) is not None else None
+    if kind == "channel":
+        return ("channel_id", ref) if db.get(Channel, ref) is not None else None
+    if kind == "plex_item":
+        it = by_rating_key(db, PlexItem, ref)
+        return ("plex_item_id", it.id) if it is not None else None
+    if kind == "plex_show":
+        sh = by_rating_key(db, PlexShow, ref)
+        return ("plex_show_id", sh.id) if sh is not None else None
+    if kind == "playlist":
+        try:
+            pid = int(ref)
+        except ValueError:
+            return None
+        pl = db.get(Playlist, pid)
+        # A playlist is per-user, unlike the other four (which are shared catalog rows whose PRIVACY
+        # lives in the state tables). Someone else's list must read as "no such thing", not as a 403 —
+        # the same reason `owned_or_404` 404s rather than 403s.
+        return ("playlist_id", pl.id) if pl is not None and pl.user_id == user.id else None
+    return None
+
+
+def list_pins(db: Session, user: User) -> list[ComfortPin]:
+    return list(
+        db.execute(
+            select(ComfortPin)
+            .where(ComfortPin.user_id == user.id)
+            .order_by(ComfortPin.position.asc(), ComfortPin.id.asc())
+        )
+        .scalars()
+        .all()
+    )
+
+
+def next_position(db: Session, user: User) -> int:
+    """New pins land at the end of the shelf."""
+    return len(list_pins(db, user))
+
+
+def _shelf_card(db: Session, user: User, pin: ComfortPin) -> dict | None:
+    """One pin as a unified card with its resolved `next`. None when the pinned row has gone (the FK
+    cascade removes the pin itself, so this is only the window between a delete and a refetch)."""
+    card: dict | None = None
+    nxt: dict | None = None
+
+    if pin.video_id is not None:
+        row = queries.yt_video_row(db, user, pin.video_id)
+        if row is None:
+            return None
+        card = cards.from_video(row)
+        nxt = dict(card)  # a video plays itself
+
+    elif pin.channel_id is not None:
+        ch = db.get(Channel, pin.channel_id)
+        if ch is None:
+            return None
+        card = cards.from_channel(ch, queries.channel_unwatched_count(db, user, ch.id))
+        row = queries.channel_next_video(db, user, ch.id)
+        nxt = cards.from_video(row) if row else None
+
+    elif pin.plex_item_id is not None:
+        it = db.get(PlexItem, pin.plex_item_id)
+        if it is None:
+            return None
+        st = rollup.item_state_map(db, user.id, [it.id]).get(it.id)
+        card = cards.from_plex(leaf_card(it, st))
+        nxt = dict(card)  # a movie or a single episode plays itself
+
+    elif pin.plex_show_id is not None:
+        sh = db.get(PlexShow, pin.plex_show_id)
+        if sh is None:
+            return None
+        status = rollup.state_map(db, user.id, [sh.id]).get(sh.id, "new")
+        card = cards.from_plex(show_card(sh, status))
+        ep = queries.next_episode(db, user.id, sh.id)
+        nxt = cards.from_plex(ep) if ep else None
+
+    elif pin.playlist_id is not None:
+        pl = db.get(Playlist, pin.playlist_id)
+        if pl is None:
+            return None
+        count, row = queries.playlist_summary(db, user, pl.id)
+        card = cards.from_playlist(pl, count, row.get("thumbnail_url") if row else None)
+        nxt = cards.from_video(row) if row else None
+
+    if card is None:
+        return None
+    # A COPY, never the card itself: a self-playing pin (a video, a movie) would otherwise make
+    # `card["next"] is card`, which serializes as a circular reference and 500s the whole shelf. The
+    # copy is taken before `pin_id`/`next` are attached, so `next` stays a plain playable card.
+    card["pin_id"] = pin.id
+    card["next"] = nxt
+    return card
+
+
+def shelf(db: Session, user: User) -> list[dict]:
+    """The whole shelf, in the user's own order."""
+    out = []
+    for pin in list_pins(db, user):
+        card = _shelf_card(db, user, pin)
+        if card is not None:
+            out.append(card)
+    return out
+
+
+def playables(shelf_cards: list[dict]) -> list[dict]:
+    """The `next` of every shelf entry that has one — what the hero may draw from. A pinned channel
+    with nothing unwatched left, or an empty playlist, simply is not an answer right now."""
+    return [c["next"] for c in shelf_cards if c.get("next")]
diff --git a/backend/app/discover/queries.py b/backend/app/discover/queries.py
new file mode 100644
index 00000000..c1136cc2
--- /dev/null
+++ b/backend/app/discover/queries.py
@@ -0,0 +1,309 @@
+"""The row fetchers behind the Home surface — one place where "what is startable right now" is
+expressed as SQL, for both sources.
+
+Two rules shape this module:
+
+**Reuse the query layer, not the route.** The YouTube half calls `feedquery.filtered_query` /
+`sort_keys`, so the Home surface inherits the feed's own definitions of subscribed, blocked,
+explore-visible and in-progress for free, and cannot drift from them. The Plex half calls
+`plex.rollup` and the `plex.cards` serializers rather than `routes/plex/discover.py`: that route
+narrows EVERY rail by the Plex sidebar's `LibraryFilters`, which is right for a surface that HAS a
+sidebar and wrong for a landing page that does not.
+
+**One seed drives both halves.** The feed orders a shuffle by `md5(id || seed)` so a keyset cursor
+stays valid across pages; the Plex surprise rail uses `ORDER BY random()`, which is fine for a single
+un-paged draw but cannot be reproduced. A cross-source reroll needs both halves to move together —
+otherwise "Mást kérek" re-draws the Plex picks while the YouTube ones stand still — so the Plex side
+here takes the feed's seeded-md5 idiom instead of `random()`.
+"""
+
+from sqlalchemy import and_, func, or_, select
+from sqlalchemy.orm import Session, aliased
+
+from app.discover.budget import band
+from app.feedquery import feed_columns, filtered_query, order_by, serialize, sort_keys
+from app.models import (
+    Channel,
+    PlaylistItem,
+    PlexItem,
+    PlexShow,
+    PlexState,
+    User,
+    Video,
+    VideoState,
+)
+from app.plex import rollup
+from app.plex.cards import leaf_card, movie_card, show_card
+from app.plex.query import apply_show_state_filter, content_libs
+
+# How many rows a single rail may carry. Matches `routes/plex/discover.py`'s RAIL_LIMIT on purpose —
+# the two surfaces should feel like the same app, and a rail is a rail.
+RAIL_LIMIT = 20
+
+# The feed parameters a Home rail runs with. Spelled out rather than borrowed from the route's
+# `_feed_params` defaults, because those are an HTTP contract that may change for the feed's own
+# reasons; the landing page's scope should not move with them silently.
+_FEED_BASE: dict = {
+    "tags": [],
+    "tag_modes": {},
+    "untagged": False,
+    "channel_ids": [],
+    "q": None,
+    "min_duration": None,
+    "max_duration": None,
+    "max_age_days": None,
+    "published_after": None,
+    "published_before": None,
+    "show_normal": True,
+    "include_shorts": False,
+    "include_live": False,
+    "show": "unwatched",
+    "scope": "my",
+    "library_source": "organic",
+}
+
+
+def _seeded(col, seed: int):
+    """The feed's deterministic shuffle expression, reused for Plex so one seed moves both halves."""
+    return func.md5(func.concat(col, str(seed)))
+
+
+# ── YouTube ───────────────────────────────────────────────────────────────────────────────────────
+
+
+def yt_in_progress(db: Session, user: User, limit: int = RAIL_LIMIT) -> list[tuple[object, dict]]:
+    """Started-but-not-finished videos, most recently progressed first, as (progressed_at, row).
+
+    The predicate is `feedquery`'s `show="in_progress"` verbatim (a stored resume position, status not
+    watched/hidden) — but the query is built here rather than through `filtered_query` because this
+    rail wants an ordering by `progress_updated_at`, and that column lives on the state alias the feed
+    query keeps private. Driving from `video_states` is also the faster plan: it is a per-user table of
+    a few hundred rows, against a catalog of many thousands."""
+    st = aliased(VideoState)
+    q = (
+        select(*feed_columns(user, st), st.progress_updated_at.label("progressed_at"))
+        .join(Channel, Channel.id == Video.channel_id)
+        .join(st, and_(st.video_id == Video.id, st.user_id == user.id))
+        .where(
+            st.position_seconds > 0,
+            st.status.notin_(("watched", "hidden")),
+            Video.unavailable_since.is_(None),
+        )
+        .order_by(st.progress_updated_at.desc().nullslast())
+        .limit(limit)
+    )
+    return [(r.progressed_at, serialize(r)) for r in db.execute(q).all()]
+
+
+def yt_surprise(db: Session, user: User, budget: str, seed: int, limit: int) -> list[dict]:
+    """Unwatched videos from the user's own feed scope, in seeded-shuffle order, inside the budget.
+
+    Shorts are opened up only for the shortest band: "I have ten minutes" is the one answer a Short is
+    actually a good response to, and the feed's own default keeps them out otherwise."""
+    lo, hi = band(budget)
+    params = dict(_FEED_BASE, min_duration=lo, max_duration=hi, include_shorts=budget == "short")
+    query, rank_expr = filtered_query(db, user, **params)
+    keys = sort_keys("shuffle", seed, rank_expr)
+    # Over-fetch, then drop anything already started. The feed's `unwatched` scope keeps watched and
+    # hidden out but deliberately keeps in-progress rows in (they belong in the feed). Here they must
+    # go: the resume tier has already had its chance at them, and it may have passed precisely because
+    # the REMAINING time did not fit the budget — offering the same item back as a surprise, whose
+    # runtime does fit, would label a half-finished thing "Meglepetés" and contradict the question.
+    rows = db.execute(query.order_by(*order_by(keys)).limit(limit * 4)).all()
+    out = [serialize(r) for r in rows]
+    return [r for r in out if not r["position_seconds"]][:limit]
+
+
+def channel_next_video(db: Session, user: User, channel_id: str) -> dict | None:
+    """The newest unwatched video of a pinned channel — what "put on that channel" resolves to."""
+    # `library_source="all"`: a pinned channel means the whole channel, including videos that only
+    # ever arrived through a search. The feed's "organic" default is a feed-scoping choice, not a
+    # statement about what belongs to a channel.
+    params = dict(_FEED_BASE, channel_ids=[channel_id], scope="all", library_source="all")
+    query, rank_expr = filtered_query(db, user, **params)
+    keys = sort_keys("newest", 0, rank_expr)
+    row = db.execute(query.order_by(*order_by(keys)).limit(1)).first()
+    return serialize(row) if row is not None else None
+
+
+def yt_video_row(db: Session, user: User, video_id: str) -> dict | None:
+    """One video as a feed row, with this user's watch state.
+
+    Built directly rather than through `filtered_query` on purpose: a PINNED video must keep working
+    even when it falls outside the feed's scope — the user unsubscribed from the channel, or blocked
+    it, or the video only ever arrived through a search. The shelf is the surface you fall back to;
+    it silently losing an entry because a filter moved is exactly the failure it must not have."""
+    st = aliased(VideoState)
+    row = db.execute(
+        select(*feed_columns(user, st))
+        .join(Channel, Channel.id == Video.channel_id)
+        .outerjoin(st, and_(st.video_id == Video.id, st.user_id == user.id))
+        .where(Video.id == video_id)
+    ).first()
+    return serialize(row) if row is not None else None
+
+
+def channel_unwatched_count(db: Session, user: User, channel_id: str) -> int:
+    """How many unwatched videos a pinned channel is holding — the tile's only honest promise, since a
+    channel has no runtime of its own."""
+    params = dict(_FEED_BASE, channel_ids=[channel_id], scope="all", library_source="all")
+    query, _ = filtered_query(db, user, **params)
+    return int(db.execute(select(func.count()).select_from(query.subquery())).scalar() or 0)
+
+
+def playlist_summary(db: Session, user: User, playlist_id: int) -> tuple[int, dict | None]:
+    """(item count, the row to play next) for a pinned playlist — its first not-yet-watched item, or
+    its first item once the whole list is watched. The same row doubles as the tile's artwork, so a
+    shelf of playlists is scannable instead of a row of blank boxes."""
+    st = aliased(VideoState)
+    rows = db.execute(
+        select(*feed_columns(user, st), PlaylistItem.position)
+        .join(PlaylistItem, PlaylistItem.video_id == Video.id)
+        .join(Channel, Channel.id == Video.channel_id)
+        .outerjoin(st, and_(st.video_id == Video.id, st.user_id == user.id))
+        .where(PlaylistItem.playlist_id == playlist_id)
+        .order_by(PlaylistItem.position.asc())
+    ).all()
+    if not rows:
+        return 0, None
+    nxt = next((r for r in rows if (r.status or "new") not in ("watched", "hidden")), rows[0])
+    return len(rows), serialize(nxt)
+
+
+# ── Plex ──────────────────────────────────────────────────────────────────────────────────────────
+
+
+def _unseen_movie_conds(uid: int):
+    """A movie the user has neither finished nor hidden. A LEFT-JOIN-IS-NULL would also drop rows that
+    merely carry a resume position, which are exactly the ones Continue watching wants back."""
+    seen = (
+        select(1)
+        .select_from(PlexState)
+        .where(
+            PlexState.item_id == PlexItem.id,
+            PlexState.user_id == uid,
+            or_(
+                PlexState.status.in_(("watched", "hidden")),
+                # Started counts as "not a surprise" too — see the note in `yt_surprise`.
+                PlexState.position_seconds > 0,
+            ),
+        )
+        .exists()
+    )
+    return ~seen
+
+
+def plex_in_progress(db: Session, uid: int, limit: int = RAIL_LIMIT) -> list[tuple[object, dict]]:
+    """In-progress movies + partially-watched shows as (progressed_at, card).
+
+    Mirrors `routes/plex/discover.py::_continue_watching` — same `PlexState` predicate for movies (a
+    NULL status still counts as started, so `notin_` alone would drop it), and the same shared
+    `rollup.agg_subquery` + `apply_show_state_filter` for shows, so the two rails cannot disagree on
+    what "in progress" means. What it does NOT inherit is that route's `LibraryFilters` narrowing:
+    this surface has no Plex sidebar to read a filter from."""
+    movie_libs, show_libs = content_libs(db)
+    out: list[tuple[object, dict]] = []
+
+    if movie_libs:
+        mq = (
+            select(PlexItem, PlexState)
+            .join(PlexState, and_(PlexState.item_id == PlexItem.id, PlexState.user_id == uid))
+            .where(
+                PlexItem.kind == "movie",
+                PlexItem.library_id.in_(movie_libs),
+                PlexState.position_seconds > 0,
+                or_(
+                    PlexState.status.is_(None),
+                    PlexState.status.notin_(("watched", "hidden")),
+                ),
+            )
+            .order_by(PlexState.progress_updated_at.desc().nullslast())
+            .limit(limit)
+        )
+        for it, st in db.execute(mq).all():
+            out.append((st.progress_updated_at, movie_card(it, st)))
+
+    if show_libs:
+        prog = rollup.agg_subquery(db, uid, include_resumed=True).subquery()
+        sq = apply_show_state_filter(
+            select(PlexShow, prog.c.resumed_at).join(prog, prog.c.show_id == PlexShow.id),
+            prog.c.total,
+            prog.c.watched,
+            prog.c.inprog,
+            "in_progress",
+        ).where(PlexShow.library_id.in_(show_libs))
+        for sh, resumed_at in db.execute(
+            sq.order_by(prog.c.resumed_at.desc().nullslast()).limit(limit)
+        ).all():
+            out.append((resumed_at, show_card(sh, "in_progress")))
+
+    return out
+
+
+def plex_surprise(db: Session, uid: int, budget: str, seed: int, limit: int) -> list[dict]:
+    """Unseen playable movies inside the budget, in seeded-shuffle order.
+
+    Movies only. A show is not a thing you can start playing, and resolving every candidate show to
+    its next episode just to draw three cards would be a query per row for a rail whose whole value is
+    that it answers instantly — shows reach this surface through Continue watching and the comfort
+    shelf, which is where "put the next episode on" actually belongs.
+
+    No `playable` filter, matching the Plex surprise rail: `PlexItem.playable` is a playback CLASS
+    ("direct" / "remux" / "transcode"), not a yes/no, and NULL there means "not probed yet" rather than
+    "will not play". Excluding un-probed items would quietly hide a fresh library from the one surface
+    that is supposed to have an answer. The class travels on the card, so the client still shows it."""
+    movie_libs, _ = content_libs(db)
+    if not movie_libs:
+        return []
+    lo, hi = band(budget)
+    q = (
+        select(PlexItem)
+        .where(
+            PlexItem.kind == "movie",
+            PlexItem.library_id.in_(movie_libs),
+            _unseen_movie_conds(uid),
+        )
+        .order_by(_seeded(PlexItem.rating_key, seed))
+        .limit(limit)
+    )
+    if lo is not None:
+        q = q.where(PlexItem.duration_s >= lo)
+    if hi is not None:
+        q = q.where(PlexItem.duration_s <= hi)
+    items = db.execute(q).scalars().all()
+    states = rollup.item_state_map(db, uid, [it.id for it in items])
+    return [movie_card(it, states.get(it.id)) for it in items]
+
+
+def next_episode(db: Session, uid: int, show_id: int) -> dict | None:
+    """What "put that show on" means: the episode in progress, else the first unwatched one, else the
+    first — in broadcast order, the same `(season, episode)` ordering `detail.py::_episode_nav` uses.
+
+    Returned as a `leaf_card` so it carries the show title the episode row itself does not."""
+    rows = db.execute(
+        select(PlexItem, PlexState)
+        .outerjoin(PlexState, and_(PlexState.item_id == PlexItem.id, PlexState.user_id == uid))
+        .where(PlexItem.show_id == show_id, PlexItem.kind == "episode")
+        .order_by(
+            PlexItem.season_number.asc().nullsfirst(),
+            PlexItem.episode_number.asc().nullsfirst(),
+        )
+    ).all()
+    if not rows:
+        return None
+
+    def _started(st: PlexState | None) -> bool:
+        return bool(st and st.position_seconds > 0 and st.status not in ("watched", "hidden"))
+
+    def _unseen(st: PlexState | None) -> bool:
+        return st is None or st.status not in ("watched", "hidden")
+
+    pick = next((r for r in rows if _started(r[1])), None)
+    pick = pick or next((r for r in rows if _unseen(r[1])), None)
+    pick = pick or rows[0]
+
+    ep, st = pick
+    show = db.get(PlexShow, show_id)
+    return leaf_card(
+        ep, st, show_title=show.title if show else None, show_rk=show.rating_key if show else None
+    )
diff --git a/backend/app/discover/rails.py b/backend/app/discover/rails.py
new file mode 100644
index 00000000..eac51240
--- /dev/null
+++ b/backend/app/discover/rails.py
@@ -0,0 +1,71 @@
+"""The Home surface's rails: cross-source lists, assembled from `queries.py` and emitted as unified
+cards.
+
+The whole point of this module is that a rail is ONE list, not a YouTube list beside a Plex list. Two
+side-by-side rails would just move the choice the user could not make ("which half do I look at?") one
+level up, which is the opposite of what this epic is for.
+"""
+
+from datetime import datetime, timezone
+
+from sqlalchemy.orm import Session
+
+from app import sysconfig
+from app.discover import cards, queries
+from app.discover.queries import RAIL_LIMIT
+from app.models import User
+
+# Sort sentinel for a row whose timestamp is missing. A plain `(ts is not None, ts)` key looks like it
+# handles NULLs and does not: two NULL rows compare equal on the flag and then compare `None < None`,
+# which raises. A rail must not be able to 500 the landing page over missing metadata.
+_NO_TS = datetime.min.replace(tzinfo=timezone.utc)
+
+
+def plex_available(db: Session) -> bool:
+    """Plex is a deployment-level toggle, not a per-user one (see `routes/me.py`). With it off, the
+    Home rails are YouTube-only and say nothing about Plex."""
+    return sysconfig.get_bool(db, "plex_enabled")
+
+
+def continue_watching(db: Session, user: User, limit: int = RAIL_LIMIT) -> list[dict]:
+    """One resume rail across both sources, most recently progressed first.
+
+    A Plex SHOW enters as a show card (that is what the Plex rail shows too, and it is what the user
+    recognises); resolving it to an episode is the hero's job, where something actually has to play."""
+    rows: list[tuple[object, dict]] = [
+        (ts, cards.from_video(r)) for ts, r in queries.yt_in_progress(db, user, limit)
+    ]
+    if plex_available(db):
+        rows += [(ts, cards.from_plex(c)) for ts, c in queries.plex_in_progress(db, user.id, limit)]
+    rows.sort(key=lambda r: r[0] or _NO_TS, reverse=True)
+    return [card for _, card in rows[:limit]]
+
+
+def surprise(db: Session, user: User, budget: str, seed: int, limit: int = 3) -> list[dict]:
+    """The hero's draw pool: unseen things that fit the budget, from both sources under ONE seed.
+
+    Interleaved rather than concatenated — the same alternation `routes/plex/discover.py` uses for its
+    movie/show split — so a short pool never comes back all-YouTube or all-Plex just because one side
+    happened to be queried first.
+
+    WHICH source leads is decided by the seed, not by a hardcoded order. The hero draws a pool of ONE,
+    so a fixed order would mean the Plex library could never answer "egy este" while a single YouTube
+    video also fit — the reroll would look broken and the cross-source promise would quietly be a
+    YouTube promise. Parity on the same seed that orders the rows keeps it deterministic: the same
+    seed always gives the same answer, and pressing "Mást kérek" moves both halves together."""
+    halves = [
+        (cards.from_video, queries.yt_surprise(db, user, budget, seed, limit)),
+        (
+            cards.from_plex,
+            queries.plex_surprise(db, user.id, budget, seed, limit) if plex_available(db) else [],
+        ),
+    ]
+    if seed % 2:
+        halves.reverse()
+
+    out: list[dict] = []
+    for i in range(max(len(rows) for _, rows in halves)):
+        for to_card, rows in halves:
+            if i < len(rows):
+                out.append(to_card(rows[i]))
+    return out[:limit]
diff --git a/backend/app/main.py b/backend/app/main.py
index 1cfeccf8..1386116c 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -33,6 +33,7 @@ from app.routes import (
     auth as auth_routes,
     channels,
     config as config_routes,
+    discover as discover_routes,
     downloads,
     feed,
     health,
@@ -187,6 +188,7 @@ app.include_router(auth_routes.router)
 app.include_router(sync.router)
 app.include_router(tags.router)
 app.include_router(feed.router)
+app.include_router(discover_routes.router)
 app.include_router(search_routes.router)
 app.include_router(me.router)
 app.include_router(notifications.router)
diff --git a/backend/app/models.py b/backend/app/models.py
index 8d16706f..02a2895a 100644
--- a/backend/app/models.py
+++ b/backend/app/models.py
@@ -3,6 +3,7 @@ from datetime import date, datetime
 from sqlalchemy import (
     BigInteger,
     Boolean,
+    CheckConstraint,
     Computed,
     Date,
     DateTime,
@@ -1406,3 +1407,52 @@ class PlexPlaylistItem(Base):
         ForeignKey("plex_items.id", ondelete="CASCADE"), index=True
     )
     position: Mapped[int] = mapped_column(Integer, default=0, server_default="0", nullable=False)
+
+
+class ComfortPin(Base, TimestampMixin):
+    """One entry on a user's comfort shelf — the anti-paralysis fallback: "just put on X" (E24).
+
+    Deliberately NOT a boolean on `VideoState`/`PlexState`, and NOT a polymorphic `(kind, ref)` string
+    pair. A flag cannot ORDER a shelf and cannot hold a channel or a show — and comfort media is
+    usually "put on X", a series or a channel, not one particular video. A string ref would order and
+    hold anything, but it would also lose the foreign key: a Plex item dropped from the library, or an
+    unsubscribed channel, would leave a dead tile on the one surface whose whole job is to be
+    reliable. So instead: one nullable CASCADE-ing FK per pinnable kind, and a check constraint that
+    exactly one of them is set.
+
+    That constraint is the house rule applied to a five-way union — the invariant lives in the schema,
+    not in a comment every insert site has to remember. `kind`/`ref` below derive FROM the columns, so
+    there is no second vocabulary to keep in sync.
+
+    NULLs are DISTINCT in a Postgres unique index, so the five per-kind unique constraints each stop a
+    double-pin of the same thing without colliding across kinds."""
+
+    __tablename__ = "comfort_pins"
+    __table_args__ = (
+        CheckConstraint(
+            "num_nonnulls(video_id, channel_id, plex_item_id, plex_show_id, playlist_id) = 1",
+            name="ck_comfort_pin_one_ref",
+        ),
+        UniqueConstraint("user_id", "video_id", name="uq_comfort_pin_video"),
+        UniqueConstraint("user_id", "channel_id", name="uq_comfort_pin_channel"),
+        UniqueConstraint("user_id", "plex_item_id", name="uq_comfort_pin_plex_item"),
+        UniqueConstraint("user_id", "plex_show_id", name="uq_comfort_pin_plex_show"),
+        UniqueConstraint("user_id", "playlist_id", name="uq_comfort_pin_playlist"),
+    )
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
+
+    video_id: Mapped[str | None] = mapped_column(ForeignKey("videos.id", ondelete="CASCADE"))
+    channel_id: Mapped[str | None] = mapped_column(ForeignKey("channels.id", ondelete="CASCADE"))
+    plex_item_id: Mapped[int | None] = mapped_column(
+        ForeignKey("plex_items.id", ondelete="CASCADE")
+    )
+    plex_show_id: Mapped[int | None] = mapped_column(
+        ForeignKey("plex_shows.id", ondelete="CASCADE")
+    )
+    playlist_id: Mapped[int | None] = mapped_column(ForeignKey("playlists.id", ondelete="CASCADE"))
+
+    # 0-based shelf order. The shelf is short and hand-curated, so a dense integer the reorder
+    # endpoint rewrites wholesale beats a fractional index nobody would ever exercise here.
+    position: Mapped[int] = mapped_column(Integer, default=0, server_default="0", nullable=False)
diff --git a/backend/app/routes/discover.py b/backend/app/routes/discover.py
new file mode 100644
index 00000000..8ba888f2
--- /dev/null
+++ b/backend/app/routes/discover.py
@@ -0,0 +1,165 @@
+"""E24 · the Home surface's HTTP shell.
+
+Thin by construction: every decision — what counts as in progress, what fits a time budget, which
+tier the hero comes from — lives in `app/discover/`, so the scheduler, a future CLI or a test can ask
+the same questions without a request. This file validates, gates and serializes.
+
+The consolidated `GET /api/discover` mirrors the Plex Discover route's one-payload shape (E5 S4): a
+landing page that fires four requests to answer one question would spend its whole budget on the
+thing it exists to remove.
+"""
+
+from typing import Literal
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
+from sqlalchemy import select, update
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.auth import current_user, require_human
+from app.db import get_db
+from app.discover import hero as hero_mod
+from app.discover import pins, rails
+from app.discover.budget import Budget, DEFAULT_BUDGET
+from app.models import ComfortPin, User
+from app.routes._common import owned_or_404
+
+router = APIRouter(prefix="/api/discover", tags=["discover"])
+
+
+class PinIn(BaseModel):
+    kind: Literal["video", "channel", "plex_item", "plex_show", "playlist"]
+    # The id the rest of the API speaks for that kind: a video id, a channel id, a Plex rating_key,
+    # or a playlist id. `pins.resolve_ref` does the translation to the row's internal key.
+    ref: str
+
+
+class ReorderIn(BaseModel):
+    ids: list[int]
+
+
+@router.get("")
+def get_discover(
+    budget: Budget = DEFAULT_BUDGET,
+    seed: int = 0,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    """Everything the Home page renders, in one pass.
+
+    `resume` and `shelf` are built once and handed to the hero, so the two rails the hero draws from
+    are not queried twice for a single page load."""
+    resume = rails.continue_watching(db, user)
+    shelf = pins.shelf(db, user)
+    return {
+        "budget": budget,
+        "hero": hero_mod.pick(db, user, budget, seed, resume=resume, shelf=shelf),
+        "continue_watching": resume,
+        "comfort": shelf,
+    }
+
+
+@router.get("/pick")
+def get_pick(
+    budget: Budget = DEFAULT_BUDGET,
+    seed: int = 0,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    """The "Mást kérek" reroll: a hero on its own, for a new seed and/or a new budget.
+
+    One seed drives BOTH sources (see `discover/queries.py`) — otherwise a reroll would re-draw the
+    Plex half while the YouTube half stood still, and the button would look broken half the time."""
+    return {"budget": budget, "hero": hero_mod.pick(db, user, budget, seed)}
+
+
+@router.get("/pins")
+def list_pins(
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    """The shelf on its own — for the shelf editor, which needs it without the rest of the page."""
+    return {"items": pins.shelf(db, user)}
+
+
+@router.post("/pins")
+def add_pin(
+    body: PinIn,
+    user: User = Depends(require_human),
+    db: Session = Depends(get_db),
+) -> dict:
+    """Pin something to the comfort shelf.
+
+    `require_human`, like every other personal-state write: the shared demo account must not be able
+    to curate a communal shelf. Re-pinning something already on the shelf is a no-op that returns the
+    existing pin rather than a 409 — the gesture is "make sure this is on my shelf", and the unique
+    constraint is the mechanism, not the message."""
+    resolved = pins.resolve_ref(db, user, body.kind, body.ref)
+    if resolved is None:
+        raise HTTPException(status_code=404, detail="Unknown item")
+    column, value = resolved
+
+    existing = db.execute(
+        select(ComfortPin).where(
+            ComfortPin.user_id == user.id,
+            getattr(ComfortPin, column) == value,
+        )
+    ).scalar_one_or_none()
+    if existing is not None:
+        return {"ok": True, "id": existing.id, "created": False}
+
+    pin = ComfortPin(user_id=user.id, position=pins.next_position(db, user), **{column: value})
+    db.add(pin)
+    try:
+        db.commit()
+    except IntegrityError:
+        # A concurrent pin of the same thing from a second tab. The unique constraint did its job;
+        # the user's intent ("be on my shelf") is satisfied either way, so report the winner rather
+        # than a 500 the user can do nothing with.
+        db.rollback()
+        winner = db.execute(
+            select(ComfortPin).where(
+                ComfortPin.user_id == user.id,
+                getattr(ComfortPin, column) == value,
+            )
+        ).scalar_one_or_none()
+        if winner is None:
+            raise
+        return {"ok": True, "id": winner.id, "created": False}
+    return {"ok": True, "id": pin.id, "created": True}
+
+
+@router.delete("/pins/{pin_id}")
+def remove_pin(
+    pin_id: int,
+    user: User = Depends(require_human),
+    db: Session = Depends(get_db),
+) -> dict:
+    pin = owned_or_404(db, ComfortPin, pin_id, user)
+    db.delete(pin)
+    db.commit()
+    return {"deleted": pin_id}
+
+
+@router.post("/pins/reorder")
+def reorder_pins(
+    body: ReorderIn,
+    user: User = Depends(require_human),
+    db: Session = Depends(get_db),
+) -> dict:
+    """Rewrite the shelf order from a full list of pin ids.
+
+    Whole-list rather than a move-one delta: the shelf is short, and a full rewrite cannot leave two
+    tiles claiming the same slot the way an interrupted sequence of single moves can. Ids that are not
+    this user's are ignored rather than 404'd — a stale tab replaying an order that includes a pin
+    deleted meanwhile should reorder what still exists, not fail wholesale."""
+    owned = {p.id for p in pins.list_pins(db, user)}
+    for position, pin_id in enumerate(i for i in body.ids if i in owned):
+        db.execute(
+            update(ComfortPin)
+            .where(ComfortPin.id == pin_id, ComfortPin.user_id == user.id)
+            .values(position=position)
+        )
+    db.commit()
+    return {"ok": True}
diff --git a/backend/tests/test_discover.py b/backend/tests/test_discover.py
new file mode 100644
index 00000000..318f5301
--- /dev/null
+++ b/backend/tests/test_discover.py
@@ -0,0 +1,529 @@
+"""E24 S1 — the Home surface's read-model.
+
+Covers the four things that can silently rot here: the time-budget bands (pure), the cross-source
+resume rail's merge order, the hero's tier ladder, and the comfort shelf (its schema-level one-of
+invariant, its cascade, and the resolved `next` that makes one click enough).
+
+DB lane (`db` fixture) except for the budget block — skipped without Postgres.
+"""
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import delete, func, select, text
+from sqlalchemy.exc import IntegrityError
+
+from app.discover import hero, pins, queries, rails
+from app.discover.budget import BANDS, fits, remaining_seconds
+from app.models import (
+    Channel,
+    ComfortPin,
+    PlexItem,
+    PlexLibrary,
+    PlexShow,
+    PlexState,
+    Playlist,
+    PlaylistItem,
+    Subscription,
+    User,
+    Video,
+    VideoState,
+)
+
+NOW = datetime(2026, 6, 1, tzinfo=timezone.utc)
+
+
+# ── budget (pure) ─────────────────────────────────────────────────────────────────────────────────
+
+
+def test_bands_cover_every_runtime_and_overlap_deliberately():
+    """Every runtime has a home in `any`, and the bands overlap so "30 perc" is not answered with a
+    two-minute Short just because it fits under the cap."""
+    assert fits(3, "any") and fits(10_000, "any")
+    assert fits(2 * 60, "short") and not fits(2 * 60, "medium")  # a Short is not a 30-minute answer
+    assert fits(30 * 60, "medium") and not fits(30 * 60, "short")
+    assert fits(110 * 60, "long") and not fits(110 * 60, "medium")
+
+
+def test_unknown_duration_only_fits_the_unbounded_band():
+    """We cannot promise an unknown-length item finishes inside ten minutes, so it is not offered as
+    an answer to a bounded question — the one control on the page has to mean what it says."""
+    for budget in BANDS:
+        assert fits(None, budget) is (budget == "any")
+
+
+def test_budget_is_matched_against_what_is_LEFT():
+    """A 90-minute film with 8 minutes to go is a ten-minute answer."""
+    assert remaining_seconds(90 * 60, 82 * 60) == 8 * 60
+    assert fits(remaining_seconds(90 * 60, 82 * 60), "short")
+    assert not fits(remaining_seconds(90 * 60, 0), "short")
+    assert remaining_seconds(None, 30) is None
+    assert remaining_seconds(600, 900) == 0  # a position past the end never goes negative
+
+
+# ── fixtures ──────────────────────────────────────────────────────────────────────────────────────
+
+
+def _seed_yt(db) -> tuple[User, Channel]:
+    """A subscribed channel with three videos of distinct lengths, so a budget can pick between them."""
+    user = User(email="discover-e24@example.com")
+    ch = Channel(id="UCe24", title="Comfort Channel", handle="@comfort")
+    db.add_all([user, ch])
+    db.flush()
+    db.add(Subscription(user_id=user.id, channel_id=ch.id))
+    db.add_all(
+        [
+            Video(
+                id="vshort",
+                channel_id=ch.id,
+                title="Ten minute video",
+                duration_seconds=9 * 60,
+                published_at=NOW - timedelta(days=1),
+            ),
+            Video(
+                id="vmed",
+                channel_id=ch.id,
+                title="Half hour video",
+                duration_seconds=30 * 60,
+                published_at=NOW - timedelta(days=2),
+            ),
+            Video(
+                id="vlong",
+                channel_id=ch.id,
+                title="Feature length video",
+                duration_seconds=95 * 60,
+                published_at=NOW - timedelta(days=3),
+            ),
+        ]
+    )
+    db.commit()
+    return user, ch
+
+
+def _seed_plex(db, user: User) -> tuple[PlexItem, PlexShow, PlexItem]:
+    lib_m = PlexLibrary(plex_key="90", title="Movies", kind="movie", enabled=True)
+    lib_s = PlexLibrary(plex_key="91", title="Shows", kind="show", enabled=True)
+    db.add_all([lib_m, lib_s])
+    db.flush()
+    movie = PlexItem(
+        rating_key="pm1",
+        library_id=lib_m.id,
+        kind="movie",
+        title="A Movie",
+        duration_s=100 * 60,
+        playable="direct",
+    )
+    show = PlexShow(rating_key="ps1", library_id=lib_s.id, title="A Show", child_count=1)
+    db.add_all([movie, show])
+    db.flush()
+    ep1 = PlexItem(
+        rating_key="ps1e1",
+        library_id=lib_s.id,
+        kind="episode",
+        title="Ep 1",
+        show_id=show.id,
+        season_number=1,
+        episode_number=1,
+        duration_s=45 * 60,
+    )
+    ep2 = PlexItem(
+        rating_key="ps1e2",
+        library_id=lib_s.id,
+        kind="episode",
+        title="Ep 2",
+        show_id=show.id,
+        season_number=1,
+        episode_number=2,
+        duration_s=45 * 60,
+    )
+    db.add_all([ep1, ep2])
+    db.commit()
+    return movie, show, ep1
+
+
+def _enable_plex(db):
+    from app import sysconfig
+
+    sysconfig.set_value(db, "plex_enabled", True)
+    db.commit()
+
+
+# ── continue watching ─────────────────────────────────────────────────────────────────────────────
+
+
+def test_continue_watching_merges_both_sources_by_recency(db):
+    """One rail, not two lists side by side — and ordered by when the user last touched each thing,
+    across sources. Two side-by-side rails would just move the decision up a level."""
+    user, _ = _seed_yt(db)
+    movie, _, _ = _seed_plex(db, user)
+    _enable_plex(db)
+    db.add_all(
+        [
+            VideoState(
+                user_id=user.id,
+                video_id="vmed",
+                status="new",
+                position_seconds=300,
+                progress_updated_at=NOW - timedelta(hours=5),
+            ),
+            PlexState(
+                user_id=user.id,
+                item_id=movie.id,
+                status="new",
+                position_seconds=600,
+                progress_updated_at=NOW - timedelta(hours=1),
+            ),
+        ]
+    )
+    db.commit()
+
+    got = rails.continue_watching(db, user)
+    assert [(c["source"], c["id"]) for c in got] == [("plex", "pm1"), ("youtube", "vmed")]
+    assert got[0]["remaining_seconds"] == 100 * 60 - 600
+
+
+def test_continue_watching_is_youtube_only_when_plex_is_off(db):
+    """Plex is a deployment toggle; with it off the surface says nothing about Plex at all."""
+    user, _ = _seed_yt(db)
+    movie, _, _ = _seed_plex(db, user)
+    db.add_all(
+        [
+            VideoState(
+                user_id=user.id,
+                video_id="vmed",
+                status="new",
+                position_seconds=300,
+                progress_updated_at=NOW,
+            ),
+            PlexState(
+                user_id=user.id,
+                item_id=movie.id,
+                status="new",
+                position_seconds=600,
+                progress_updated_at=NOW,
+            ),
+        ]
+    )
+    db.commit()
+    assert {c["source"] for c in rails.continue_watching(db, user)} == {"youtube"}
+
+
+def test_a_null_timestamp_sorts_last_instead_of_raising(db):
+    """The `(ts is not None, ts)` key this rail's Plex sibling used compares `None < None` when two
+    rows both lack a timestamp, which raises. A missing progress timestamp must sort last, not 500
+    the landing page."""
+    user, _ = _seed_yt(db)
+    db.add_all(
+        [
+            VideoState(
+                user_id=user.id, video_id="vmed", status="new", position_seconds=1
+            ),  # no progress_updated_at
+            VideoState(
+                user_id=user.id, video_id="vlong", status="new", position_seconds=1
+            ),  # nor here
+            VideoState(
+                user_id=user.id,
+                video_id="vshort",
+                status="new",
+                position_seconds=1,
+                progress_updated_at=NOW,
+            ),
+        ]
+    )
+    db.commit()
+    got = rails.continue_watching(db, user)
+    assert got[0]["id"] == "vshort"
+    assert len(got) == 3
+
+
+# ── hero ──────────────────────────────────────────────────────────────────────────────────────────
+
+
+def test_hero_prefers_resume_and_labels_the_tier(db):
+    user, _ = _seed_yt(db)
+    db.add(
+        VideoState(
+            user_id=user.id,
+            video_id="vmed",
+            status="new",
+            position_seconds=300,
+            progress_updated_at=NOW,
+        )
+    )
+    db.commit()
+    got = hero.pick(db, user, "any", seed=1)
+    assert got["tier"] == "resume"
+    assert got["card"]["id"] == "vmed"
+
+
+def test_hero_skips_a_resume_that_does_not_fit_the_budget(db):
+    """ "I have ten minutes" must not be answered with the 80 minutes left on a film, even though
+    that film is the most recent thing the user touched."""
+    user, _ = _seed_yt(db)
+    db.add(
+        VideoState(
+            user_id=user.id,
+            video_id="vlong",
+            status="new",
+            position_seconds=60,
+            progress_updated_at=NOW,
+        )
+    )
+    db.commit()
+    got = hero.pick(db, user, "short", seed=1)
+    assert got["tier"] == "surprise"
+    assert got["card"]["duration_seconds"] <= BANDS["short"][1]
+
+
+def test_hero_falls_back_to_the_shelf_before_guessing(db):
+    """The user's own declared fallback beats anything we could infer."""
+    user, ch = _seed_yt(db)
+    db.add(ComfortPin(user_id=user.id, video_id="vmed", position=0))
+    db.commit()
+    got = hero.pick(db, user, "medium", seed=1)
+    assert got["tier"] == "comfort"
+    assert got["card"]["id"] == "vmed"
+
+
+def test_hero_resolves_a_show_to_the_episode_that_would_play(db):
+    """A show has no Play button. The hero of a partially-watched show is its next episode."""
+    user, _ = _seed_yt(db)
+    _, show, ep1 = _seed_plex(db, user)
+    _enable_plex(db)
+    db.add(
+        PlexState(
+            user_id=user.id,
+            item_id=ep1.id,
+            status="watched",
+            watched_at=NOW,
+            progress_updated_at=NOW,
+        )
+    )
+    db.commit()
+    got = hero.pick(db, user, "any", seed=1)
+    assert got["tier"] == "resume"
+    assert got["card"]["kind"] == "episode"
+    assert got["card"]["id"] == "ps1e2"  # ep1 is finished → ep2 is what plays
+
+
+def test_hero_is_none_when_nothing_fits(db):
+    """An empty answer is an honest answer: the page renders its own empty state rather than being
+    handed a card that does not answer what was asked. Here the whole library is under 40 minutes,
+    so "egy este" has nothing to offer."""
+    user, ch = _seed_yt(db)
+    db.execute(text("DELETE FROM videos WHERE id = 'vlong'"))
+    db.commit()
+    assert hero.pick(db, user, "long", seed=1) is None
+
+
+def test_surprise_is_reproducible_for_a_seed(db):
+    """The reroll button depends on this: one seed drives both sources, so the same seed must give
+    the same answer and a different seed must be free to give another."""
+    user, _ = _seed_yt(db)
+    a = rails.surprise(db, user, "any", seed=7, limit=3)
+    b = rails.surprise(db, user, "any", seed=7, limit=3)
+    assert [c["id"] for c in a] == [c["id"] for c in b]
+    seeds = {tuple(c["id"] for c in rails.surprise(db, user, "any", s, 3)) for s in range(12)}
+    assert len(seeds) > 1  # the seed actually reaches the ordering
+
+
+def test_surprise_never_offers_something_already_started(db):
+    """The resume tier may have skipped an item because its REMAINING time did not fit; handing the
+    same item back as "Meglepetés" — where the full runtime does fit — would contradict the question
+    the user just answered."""
+    user, _ = _seed_yt(db)
+    db.add(
+        VideoState(
+            user_id=user.id,
+            video_id="vlong",
+            status="new",
+            position_seconds=90 * 60,
+            progress_updated_at=NOW,
+        )
+    )
+    db.commit()
+    assert "vlong" not in {c["id"] for c in rails.surprise(db, user, "long", seed=3, limit=5)}
+
+
+def test_the_seed_decides_which_source_leads(db):
+    """With a pool of one, a fixed source order would mean the Plex library could never answer "egy
+    este" whenever a YouTube video also fit — the cross-source promise would quietly be a YouTube
+    promise."""
+    user, _ = _seed_yt(db)
+    _seed_plex(db, user)
+    _enable_plex(db)
+    leaders = {rails.surprise(db, user, "any", s, limit=1)[0]["source"] for s in range(6)}
+    assert leaders == {"youtube", "plex"}
+
+
+# ── comfort shelf ─────────────────────────────────────────────────────────────────────────────────
+
+
+def test_a_pin_must_reference_exactly_one_thing(db):
+    """The union invariant lives in the schema, not in a comment at every insert site."""
+    user, ch = _seed_yt(db)
+    for kwargs in ({}, {"video_id": "vmed", "channel_id": ch.id}):
+        db.add(ComfortPin(user_id=user.id, position=0, **kwargs))
+        with pytest.raises(IntegrityError):
+            db.flush()
+        db.rollback()
+
+
+def test_deleting_the_pinned_row_removes_the_pin(db):
+    """A film dropped from the library must not leave a dead tile on the one surface whose whole job
+    is to be reliable — which is why the shelf holds foreign keys, not strings."""
+    user, _ = _seed_yt(db)
+    db.add(ComfortPin(user_id=user.id, video_id="vmed", position=0))
+    db.commit()
+    db.execute(text("DELETE FROM videos WHERE id = 'vmed'"))
+    db.commit()
+    assert db.execute(select(func.count()).select_from(ComfortPin)).scalar() == 0
+
+
+def test_shelf_resolves_a_channel_to_its_newest_unwatched_video(db):
+    """ "Put that channel on" has to be one click, so the tile carries what would play."""
+    user, ch = _seed_yt(db)
+    db.add(ComfortPin(user_id=user.id, channel_id=ch.id, position=0))
+    db.commit()
+    (card,) = pins.shelf(db, user)
+    assert card["kind"] == "channel"
+    assert card["unwatched_count"] == 3
+    assert card["next"]["id"] == "vshort"  # newest
+
+
+def test_shelf_resolves_a_playlist_to_its_first_unwatched_item(db):
+    user, _ = _seed_yt(db)
+    pl = Playlist(user_id=user.id, name="Elalvós")
+    db.add(pl)
+    db.flush()
+    db.add_all(
+        [
+            PlaylistItem(playlist_id=pl.id, video_id="vshort", position=0),
+            PlaylistItem(playlist_id=pl.id, video_id="vmed", position=1),
+        ]
+    )
+    db.add(VideoState(user_id=user.id, video_id="vshort", status="watched", watched_at=NOW))
+    db.add(ComfortPin(user_id=user.id, playlist_id=pl.id, position=0))
+    db.commit()
+    (card,) = pins.shelf(db, user)
+    assert card["kind"] == "playlist"
+    assert card["item_count"] == 2
+    assert card["next"]["id"] == "vmed"
+
+
+def test_a_pinned_video_survives_leaving_the_feed_scope(db):
+    """The shelf is the fallback surface. An entry silently vanishing because the user unsubscribed
+    from the channel is exactly the failure it must not have."""
+    user, ch = _seed_yt(db)
+    db.add(ComfortPin(user_id=user.id, video_id="vmed", position=0))
+    db.commit()
+    db.execute(delete(Subscription))
+    db.commit()
+    (card,) = pins.shelf(db, user)
+    assert card["id"] == "vmed"
+    assert card["next"]["id"] == "vmed"
+
+
+def test_a_self_playing_pin_does_not_reference_itself(db):
+    """`next` must be a distinct dict even when the pin plays itself. Aliasing the card into its own
+    `next` is a cycle, and the serializer turns the whole shelf into a 500 over it."""
+    user, _ = _seed_yt(db)
+    db.add(ComfortPin(user_id=user.id, video_id="vmed", position=0))
+    db.commit()
+    (card,) = pins.shelf(db, user)
+    assert card["next"] is not card
+    assert "next" not in card["next"]
+
+
+def test_next_episode_prefers_the_one_in_progress(db):
+    """In progress beats "next unwatched": resuming is cheaper than starting."""
+    user, _ = _seed_yt(db)
+    _, show, ep1 = _seed_plex(db, user)
+    ep2 = db.execute(select(PlexItem).where(PlexItem.rating_key == "ps1e2")).scalar_one()
+    db.add_all(
+        [
+            PlexState(user_id=user.id, item_id=ep1.id, status="watched", watched_at=NOW),
+            PlexState(user_id=user.id, item_id=ep2.id, status="new", position_seconds=120),
+        ]
+    )
+    db.commit()
+    assert queries.next_episode(db, user.id, show.id)["id"] == "ps1e2"
+
+
+# ── HTTP surface ──────────────────────────────────────────────────────────────────────────────────
+
+
+def test_discover_endpoint_answers_in_one_pass(db, as_user):
+    user, _ = _seed_yt(db)
+    db.add(
+        VideoState(
+            user_id=user.id,
+            video_id="vmed",
+            status="new",
+            position_seconds=300,
+            progress_updated_at=NOW,
+        )
+    )
+    db.commit()
+    r = as_user(user).get("/api/discover?budget=medium")
+    assert r.status_code == 200
+    body = r.json()
+    assert body["budget"] == "medium"
+    assert body["hero"]["tier"] == "resume"
+    assert [c["id"] for c in body["continue_watching"]] == ["vmed"]
+    assert body["comfort"] == []
+
+
+def test_an_unknown_budget_is_refused_rather_than_silently_widened(db, as_user):
+    user, _ = _seed_yt(db)
+    assert as_user(user).get("/api/discover?budget=whenever").status_code == 422
+
+
+def test_pinning_twice_is_the_same_shelf(db, as_user):
+    """The gesture is "make sure this is on my shelf", so a re-pin is a no-op, not a 409."""
+    user, _ = _seed_yt(db)
+    c = as_user(user)
+    first = c.post("/api/discover/pins", json={"kind": "video", "ref": "vmed"}).json()
+    second = c.post("/api/discover/pins", json={"kind": "video", "ref": "vmed"}).json()
+    assert first["created"] is True and second["created"] is False
+    assert first["id"] == second["id"]
+    assert len(c.get("/api/discover/pins").json()["items"]) == 1
+
+
+def test_pinning_something_that_does_not_exist_is_a_404(db, as_user):
+    user, _ = _seed_yt(db)
+    r = as_user(user).post("/api/discover/pins", json={"kind": "video", "ref": "nope"})
+    assert r.status_code == 404
+
+
+def test_another_users_playlist_is_not_pinnable(db, as_user):
+    """404, not 403 — a 403 would confirm the id exists to a stranger."""
+    user, _ = _seed_yt(db)
+    other = User(email="other-e24@example.com")
+    db.add(other)
+    db.flush()
+    pl = Playlist(user_id=other.id, name="Theirs")
+    db.add(pl)
+    db.commit()
+    r = as_user(user).post("/api/discover/pins", json={"kind": "playlist", "ref": str(pl.id)})
+    assert r.status_code == 404
+
+
+def test_reorder_ignores_ids_that_are_not_yours(db, as_user):
+    """A stale tab replaying an order that includes a since-deleted pin should reorder what still
+    exists rather than fail wholesale."""
+    user, ch = _seed_yt(db)
+    c = as_user(user)
+    a = c.post("/api/discover/pins", json={"kind": "video", "ref": "vmed"}).json()["id"]
+    b = c.post("/api/discover/pins", json={"kind": "channel", "ref": ch.id}).json()["id"]
+    assert c.post("/api/discover/pins/reorder", json={"ids": [b, 9999, a]}).status_code == 200
+    assert [p["pin_id"] for p in c.get("/api/discover/pins").json()["items"]] == [b, a]
+
+
+def test_deleting_a_pin_reports_what_went(db, as_user):
+    user, _ = _seed_yt(db)
+    c = as_user(user)
+    pin_id = c.post("/api/discover/pins", json={"kind": "video", "ref": "vmed"}).json()["id"]
+    r = c.delete(f"/api/discover/pins/{pin_id}")
+    assert r.status_code == 200 and r.json() == {"deleted": pin_id}
+    assert c.get("/api/discover/pins").json()["items"] == []