diff --git a/backend/alembic/versions/0004_feed_state_prefs.py b/backend/alembic/versions/0004_feed_state_prefs.py
new file mode 100644
index 0000000..a8ca407
--- /dev/null
+++ b/backend/alembic/versions/0004_feed_state_prefs.py
@@ -0,0 +1,52 @@
+"""user preferences + per-user video watch state
+
+Revision ID: 0004_feed_state_prefs
+Revises: 0003_tags
+Create Date: 2026-06-11
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "0004_feed_state_prefs"
+down_revision: Union[str, None] = "0003_tags"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column("users", sa.Column("preferences", sa.JSON(), nullable=True))
+
+    op.create_table(
+        "video_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(
+            "video_id",
+            sa.String(length=16),
+            sa.ForeignKey("videos.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("status", sa.String(length=12), nullable=False, server_default="new"),
+        sa.Column("watched_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+        sa.UniqueConstraint("user_id", "video_id", name="uq_user_video"),
+    )
+    op.create_index("ix_video_states_user_id", "video_states", ["user_id"])
+    op.create_index("ix_video_states_video_id", "video_states", ["video_id"])
+
+
+def downgrade() -> None:
+    op.drop_table("video_states")
+    op.drop_column("users", "preferences")
diff --git a/backend/app/main.py b/backend/app/main.py
index 0b97348..d129ab4 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -9,7 +9,7 @@ from starlette.middleware.sessions import SessionMiddleware
 
 from app import auth
 from app.config import settings
-from app.routes import health, sync, tags
+from app.routes import feed, health, me, sync, tags
 from app.scheduler import shutdown_scheduler, start_scheduler
 
 
@@ -44,6 +44,8 @@ app.include_router(health.router)
 app.include_router(auth.router)
 app.include_router(sync.router)
 app.include_router(tags.router)
+app.include_router(feed.router)
+app.include_router(me.router)
 
 STATIC_DIR = Path(__file__).parent / "static"
 app.mount("/assets", StaticFiles(directory=STATIC_DIR / "assets"), name="assets")
diff --git a/backend/app/models.py b/backend/app/models.py
index c5f9012..70f93af 100644
--- a/backend/app/models.py
+++ b/backend/app/models.py
@@ -28,6 +28,8 @@ class User(Base):
     display_name: Mapped[str | None] = mapped_column(String(255))
     avatar_url: Mapped[str | None] = mapped_column(String(1024))
     role: Mapped[str] = mapped_column(String(16), default="user", server_default="user")
+    # Free-form UI preferences (theme, color scheme, font scale, default filters…).
+    preferences: Mapped[dict | None] = mapped_column(JSON)
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now()
     )
@@ -201,6 +203,28 @@ class ChannelTag(Base):
     confidence: Mapped[float | None] = mapped_column(Float)
 
 
+class VideoState(Base):
+    """Per-user watch state for a video. Rows exist only for non-default states; the
+    absence of a row means 'new/unseen'."""
+
+    __tablename__ = "video_states"
+    __table_args__ = (UniqueConstraint("user_id", "video_id", name="uq_user_video"),)
+
+    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] = mapped_column(
+        ForeignKey("videos.id", ondelete="CASCADE"), index=True
+    )
+    # new | watched | saved | hidden
+    status: Mapped[str] = mapped_column(String(12), default="new", server_default="new")
+    watched_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
+    )
+
+
 class ApiQuotaUsage(Base):
     """Tracks YouTube Data API units spent per Pacific-time day (the quota resets at
     midnight Pacific). The whole app shares one daily budget."""
diff --git a/backend/app/routes/feed.py b/backend/app/routes/feed.py
new file mode 100644
index 0000000..6ff013f
--- /dev/null
+++ b/backend/app/routes/feed.py
@@ -0,0 +1,171 @@
+from datetime import datetime, timezone
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy import and_, func, or_, select
+from sqlalchemy.orm import Session, aliased
+
+from app.auth import current_user
+from app.db import get_db
+from app.models import Channel, ChannelTag, User, Video, VideoState
+
+router = APIRouter(prefix="/api", tags=["feed"])
+
+VALID_STATES = {"new", "watched", "saved", "hidden"}
+HIDDEN_LIVE = ("live", "upcoming")
+
+
+def _serialize(row) -> dict:
+    return {
+        "id": row.id,
+        "title": row.title,
+        "channel_id": row.channel_id,
+        "channel_title": row.channel_title,
+        "channel_thumbnail": row.channel_thumbnail,
+        "published_at": row.published_at.isoformat() if row.published_at else None,
+        "thumbnail_url": row.thumbnail_url,
+        "duration_seconds": row.duration_seconds,
+        "view_count": row.view_count,
+        "is_short": row.is_short,
+        "live_status": row.live_status,
+        "status": row.status or "new",
+        "watch_url": f"https://www.youtube.com/watch?v={row.id}",
+    }
+
+
+@router.get("/feed")
+def get_feed(
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+    tags: list[int] = Query(default=[]),
+    tag_mode: str = "or",
+    channel_id: str | None = None,
+    q: str | None = None,
+    min_duration: int | None = None,
+    max_duration: int | None = None,
+    max_age_days: int | None = None,
+    include_shorts: bool = False,
+    include_live: bool = False,
+    show: str = "unwatched",  # all | unwatched | watched | saved | hidden
+    sort: str = "newest",
+    seed: int = 0,
+    limit: int = Query(default=60, le=200),
+    offset: int = 0,
+) -> dict:
+    state = aliased(VideoState)
+    status_expr = func.coalesce(state.status, "new").label("status")
+
+    query = (
+        select(
+            Video.id,
+            Video.title,
+            Video.channel_id,
+            Channel.title.label("channel_title"),
+            Channel.thumbnail_url.label("channel_thumbnail"),
+            Video.published_at,
+            Video.thumbnail_url,
+            Video.duration_seconds,
+            Video.view_count,
+            Video.is_short,
+            Video.live_status,
+            status_expr,
+        )
+        .join(Channel, Channel.id == Video.channel_id)
+        .outerjoin(
+            state, and_(state.video_id == Video.id, state.user_id == user.id)
+        )
+    )
+
+    if not include_shorts:
+        query = query.where(Video.is_short.is_(False))
+    if not include_live:
+        query = query.where(Video.live_status.notin_(HIDDEN_LIVE))
+    if channel_id:
+        query = query.where(Video.channel_id == channel_id)
+    if min_duration is not None:
+        query = query.where(Video.duration_seconds >= min_duration)
+    if max_duration is not None:
+        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(
+            Video.published_at >= datetime.fromtimestamp(cutoff, tz=timezone.utc)
+        )
+    if q:
+        like = f"%{q}%"
+        query = query.where(or_(Video.title.ilike(like), Video.description.ilike(like)))
+
+    if tags:
+        visible = or_(ChannelTag.user_id.is_(None), ChannelTag.user_id == user.id)
+        if tag_mode == "and":
+            sub = (
+                select(ChannelTag.channel_id)
+                .where(ChannelTag.tag_id.in_(tags), visible)
+                .group_by(ChannelTag.channel_id)
+                .having(func.count(func.distinct(ChannelTag.tag_id)) == len(set(tags)))
+            )
+        else:
+            sub = select(ChannelTag.channel_id).where(
+                ChannelTag.tag_id.in_(tags), visible
+            )
+        query = query.where(Video.channel_id.in_(sub))
+
+    # Watch-state visibility.
+    if show == "unwatched":
+        query = query.where(status_expr.notin_(("watched", "hidden")))
+    elif show == "watched":
+        query = query.where(status_expr == "watched")
+    elif show == "saved":
+        query = query.where(status_expr == "saved")
+    elif show == "hidden":
+        query = query.where(status_expr == "hidden")
+    else:  # all
+        query = query.where(status_expr != "hidden")
+
+    sorts = {
+        "newest": Video.published_at.desc().nulls_last(),
+        "oldest": Video.published_at.asc().nulls_last(),
+        "views": Video.view_count.desc().nulls_last(),
+        "duration_desc": Video.duration_seconds.desc().nulls_last(),
+        "duration_asc": Video.duration_seconds.asc().nulls_last(),
+        "shuffle": func.md5(func.concat(Video.id, str(seed))),
+    }
+    query = query.order_by(sorts.get(sort, sorts["newest"]))
+
+    rows = db.execute(query.offset(offset).limit(limit + 1)).all()
+    has_more = len(rows) > limit
+    items = [_serialize(r) for r in rows[:limit]]
+    return {"items": items, "has_more": has_more, "offset": offset, "limit": limit}
+
+
+@router.post("/videos/{video_id}/state")
+def set_video_state(
+    video_id: str,
+    payload: dict,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    status = payload.get("status")
+    if status not in VALID_STATES:
+        raise HTTPException(status_code=400, detail=f"status must be one of {VALID_STATES}")
+    if db.get(Video, video_id) is None:
+        raise HTTPException(status_code=404, detail="Unknown video")
+
+    row = db.execute(
+        select(VideoState).where(
+            VideoState.user_id == user.id, VideoState.video_id == video_id
+        )
+    ).scalar_one_or_none()
+
+    if status == "new":
+        if row is not None:
+            db.delete(row)
+            db.commit()
+        return {"video_id": video_id, "status": "new"}
+
+    if row is None:
+        row = VideoState(user_id=user.id, video_id=video_id)
+        db.add(row)
+    row.status = status
+    row.watched_at = datetime.now(timezone.utc) if status == "watched" else row.watched_at
+    db.commit()
+    return {"video_id": video_id, "status": status}
diff --git a/backend/app/routes/me.py b/backend/app/routes/me.py
new file mode 100644
index 0000000..9919aed
--- /dev/null
+++ b/backend/app/routes/me.py
@@ -0,0 +1,35 @@
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+
+from app.auth import current_user
+from app.db import get_db
+from app.models import User
+
+router = APIRouter(prefix="/api/me", tags=["me"])
+
+
+@router.get("")
+def get_me(user: User = Depends(current_user)) -> dict:
+    return {
+        "id": user.id,
+        "email": user.email,
+        "display_name": user.display_name,
+        "avatar_url": user.avatar_url,
+        "role": user.role,
+        "preferences": user.preferences or {},
+    }
+
+
+@router.put("/preferences")
+def update_preferences(
+    preferences: dict,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    # Merge so partial updates don't wipe other keys.
+    merged = dict(user.preferences or {})
+    merged.update(preferences)
+    user.preferences = merged
+    db.add(user)
+    db.commit()
+    return {"preferences": merged}