diff --git a/.env.example b/.env.example
index fd17c2d5..95b3ad55 100644
--- a/.env.example
+++ b/.env.example
@@ -76,10 +76,10 @@ SCHEDULER_ENABLED=true
 # --- Download center ---
 # The Download Center adds a `worker` container (runs the yt-dlp/ffmpeg job loop) and a small
 # `bgutil-pot` sidecar (mints YouTube tokens) — both come up automatically with docker compose.
-# Downloaded media defaults to a Docker-managed named volume. Set DOWNLOAD_HOST_PATH to a host
+# Downloaded media defaults to a Docker-managed named volume. Set FILES_HOST_PATH to a host
 # directory instead — e.g. one your Plex server can read — to keep the Plex-style tree there.
 # The path must be writable by the container user (uid of `appuser`, 1000): chown 1000:1000 <dir>.
-# DOWNLOAD_HOST_PATH=/mnt/media/youtube
+# FILES_HOST_PATH=/mnt/media/youtube
 
 # --- Plex integration (optional) ---
 # The optional Plex module browses/plays your Plex library, playing the LOCAL physical file
@@ -89,6 +89,6 @@ SCHEDULER_ENABLED=true
 # to /plex-media inside the container (the local side of the admin "path map", whose default is
 # /data=/plex-media). Leave unset if you don't use Plex.
 # PLEX_MEDIA_HOST_PATH=/mnt/media
-# Optional: scratch dir for on-the-fly HLS remux segments. Defaults to DOWNLOAD_ROOT; point it at
+# Optional: scratch dir for on-the-fly HLS remux segments. Defaults to FILES_ROOT; point it at
 # fast local storage if your download volume is slow.
 # PLEX_HLS_DIR=/var/tmp/plex-hls
diff --git a/.gitignore b/.gitignore
index 04718ff8..af6667dc 100644
--- a/.gitignore
+++ b/.gitignore
@@ -29,9 +29,9 @@ frontend/e2e/.auth/
 *.sql.gz
 backups/
 
-# Download center: locally downloaded media (the DOWNLOAD_ROOT bind mount at the repo root).
+# Download center: locally downloaded media (the FILES_ROOT bind mount at the repo root).
 # Anchored with a leading slash so it does NOT also ignore the backend/app/downloads package.
-/downloads/
+/files/
 
 # OS / editor
 .DS_Store
diff --git a/Dockerfile b/Dockerfile
index 2a31b361..b04e23be 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -39,13 +39,13 @@ ENV PYTHONDONTWRITEBYTECODE=1 \
 WORKDIR /app
 
 # Create the app user BEFORE the copies so files land already-owned via COPY --chown (a trailing
-# `chown -R /app` would rewrite every file into a duplicate layer). /downloads must be appuser-owned
+# `chown -R /app` would rewrite every file into a duplicate layer). /files must be appuser-owned
 # so a fresh named volume mounted there inherits that ownership (Docker copies the image dir's
 # ownership into a new empty volume) — the worker/API can then write without a manual chown. A bind
 # mount to a host path instead needs that path writable by this uid (see docs/self-hosting.md).
 RUN adduser --disabled-password --gecos "" appuser \
-    && mkdir -p /downloads \
-    && chown appuser /app /downloads
+    && mkdir -p /files \
+    && chown appuser /app /files
 
 # ffmpeg: merge separate video+audio streams + postprocessing, and the Plex VAAPI transcode.
 # intel-media-va-driver (iHD): the VA driver ffmpeg's h264_vaapi needs — libva2 ships in the base but
diff --git a/backend/.basedpyright/baseline.json b/backend/.basedpyright/baseline.json
index cc5f74f1..8f2e3b98 100644
--- a/backend/.basedpyright/baseline.json
+++ b/backend/.basedpyright/baseline.json
@@ -1196,16 +1196,6 @@
                 }
             }
         ],
-        "./app/routes/public.py": [
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 57,
-                    "endColumn": 71,
-                    "lineCount": 1
-                }
-            }
-        ],
         "./app/routes/scheduler.py": [
             {
                 "code": "reportArgumentType",
diff --git a/backend/alembic/versions/0050_asset_poster.py b/backend/alembic/versions/0050_asset_poster.py
index 7c210b7b..3059ea2b 100644
--- a/backend/alembic/versions/0050_asset_poster.py
+++ b/backend/alembic/versions/0050_asset_poster.py
@@ -6,7 +6,7 @@ Create Date: 2026-07-07
 
 Some sources (a direct media URL like a reddit v.redd.it HLS manifest) carry no thumbnail, so the
 download shows a blank image box. The worker now extracts a representative frame with ffmpeg and
-saves it as a poster sidecar; `poster_path` (relative to download_root) records it so the card /
+saves it as a poster sidecar; `poster_path` (relative to files_root) records it so the card /
 watch page / link preview can fall back to it. Nullable — only set when there was no thumbnail.
 """
 
diff --git a/backend/alembic/versions/0102_file_folders.py b/backend/alembic/versions/0102_file_folders.py
new file mode 100644
index 00000000..30b965a6
--- /dev/null
+++ b/backend/alembic/versions/0102_file_folders.py
@@ -0,0 +1,193 @@
+"""The Files library: folders, and a file's retention on the user's job
+
+Revision ID: 0102_file_folders
+Revises: 0101_plex_text_plain
+Create Date: 2026-10-11
+
+Every download, clip and upload of a user now sits in a folder of their Files library
+(`app.downloads.folders`): a source folder made per origin (YouTube, Plex, uploads, one per other
+site) or one of their own. How long a file stays moves from the shared asset to each user's job
+(`download_jobs.expires_at`, stamped from the folder's `retention_days`), so one user keeping a file
+no longer depends on another's copy; the asset's own `expires_at` / `gc_notified` go.
+
+This places every existing job in its source folder and carries each done job's expiry over from
+the asset it holds (an upload keeps none). The source-key rule is copied here, frozen, from
+`folders.source_key_for` as it is today — a migration must not follow the code's later changes.
+"""
+
+from urllib.parse import urlsplit
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0102_file_folders"
+down_revision: str | None = "0101_plex_text_plain"
+branch_labels: str | None = None
+depends_on: str | None = None
+
+_FRONT_DOORS = ("www.", "m.", "mobile.", "web.")
+_YOUTUBE_HOSTS = {"youtube.com", "youtu.be", "youtube-nocookie.com", "music.youtube.com"}
+_DEFAULT_RETENTION_DAYS = 30  # config.download_retention_days on the day this was written
+
+
+def _site_key(url: str) -> str:
+    host = (urlsplit(url).hostname or "").lower().rstrip(".")
+    for door in _FRONT_DOORS:
+        if host.startswith(door) and host.count(".") > 1:
+            host = host[len(door) :]
+            break
+    if host in _YOUTUBE_HOSTS:
+        return "youtube"
+    return f"site:{host or 'unknown'}"
+
+
+def _asset_key(kind: str | None, ref: str | None) -> str:
+    if kind == "youtube":
+        return "youtube"
+    if kind == "url" and ref:
+        return _site_key(ref)
+    return "upload"
+
+
+def _key(job, jobs_by_id, assets) -> str:
+    kind = job.source_kind
+    if kind == "youtube":
+        return "youtube"
+    if kind == "upload":
+        return "upload"
+    if kind == "clip":
+        spec = job.edit_spec or {}
+        return "plex" if spec.get("source") == "plex" else "youtube"
+    if kind == "url":
+        return _site_key(job.source_ref)
+    if kind == "edit":
+        parent = jobs_by_id.get(job.source_job_id)
+        if parent is not None and parent.id != job.id:
+            return _key(parent, jobs_by_id, assets)
+        asset = assets.get(job.source_asset_id)
+        if asset is not None:
+            return _asset_key(asset.source_kind, asset.source_ref)
+    return "upload"
+
+
+def upgrade() -> None:
+    op.create_table(
+        "file_folders",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column(
+            "user_id",
+            sa.Integer(),
+            sa.ForeignKey("users.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("kind", sa.String(16), nullable=False),
+        sa.Column("source_key", sa.String(260)),  # "site:" + a host (<= 253)
+        sa.Column("name", sa.String(255)),
+        sa.Column("retention_days", sa.Integer()),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+        sa.UniqueConstraint("user_id", "source_key", name="uq_file_folder_source"),
+        sa.CheckConstraint(
+            "(kind = 'source' AND source_key IS NOT NULL AND name IS NULL)"
+            " OR (kind = 'own' AND name IS NOT NULL AND source_key IS NULL)",
+            name="ck_file_folder_kind",
+        ),
+    )
+    op.create_index("ix_file_folders_user_id", "file_folders", ["user_id"])
+    op.create_index(
+        "uq_file_folder_own_name",
+        "file_folders",
+        ["user_id", sa.text("lower(name)")],
+        unique=True,
+        postgresql_where=sa.text("name IS NOT NULL"),
+    )
+
+    op.add_column(
+        "download_jobs",
+        sa.Column(
+            "folder_id",
+            sa.Integer(),
+            sa.ForeignKey("file_folders.id", ondelete="RESTRICT"),
+        ),
+    )
+    op.add_column(
+        "download_jobs",
+        sa.Column("kept", sa.Boolean(), server_default="false", nullable=False),
+    )
+    op.add_column("download_jobs", sa.Column("expires_at", sa.DateTime(timezone=True)))
+    op.add_column(
+        "download_jobs",
+        sa.Column("expiry_warned", sa.Boolean(), server_default="false", nullable=False),
+    )
+    op.create_index("ix_download_jobs_folder_id", "download_jobs", ["folder_id"])
+    op.create_index("ix_download_jobs_expires_at", "download_jobs", ["expires_at"])
+
+    conn = op.get_bind()
+    raw = conn.execute(
+        sa.text("SELECT value FROM system_config WHERE key = 'download_retention_days'")
+    ).scalar()
+    retention = int(raw) if raw not in (None, "") else _DEFAULT_RETENTION_DAYS
+
+    jobs = conn.execute(
+        sa.text(
+            "SELECT j.id, j.user_id, j.source_kind, j.source_ref, j.edit_spec, j.source_job_id,"
+            " j.source_asset_id, j.status, j.asset_id, a.expires_at AS asset_expires_at"
+            " FROM download_jobs j LEFT JOIN media_assets a ON a.id = j.asset_id"
+        )
+    ).fetchall()
+    jobs_by_id = {j.id: j for j in jobs}
+    assets = {
+        a.id: a
+        for a in conn.execute(sa.text("SELECT id, source_kind, source_ref FROM media_assets"))
+    }
+
+    folder_ids: dict[tuple[int, str], int] = {}
+    for job in jobs:
+        key = _key(job, jobs_by_id, assets)
+        slot = (job.user_id, key)
+        if slot not in folder_ids:
+            folder_ids[slot] = conn.execute(
+                sa.text(
+                    "INSERT INTO file_folders (user_id, kind, source_key, retention_days)"
+                    " VALUES (:u, 'source', :k, :r) RETURNING id"
+                ),
+                {"u": job.user_id, "k": key, "r": None if key == "upload" else retention},
+            ).scalar_one()
+        # A done file keeps the expiry it had (its asset's); an upload had none and keeps none.
+        expires = job.asset_expires_at if job.status == "done" and key != "upload" else None
+        conn.execute(
+            sa.text("UPDATE download_jobs SET folder_id = :f, expires_at = :e WHERE id = :id"),
+            {"f": folder_ids[slot], "e": expires, "id": job.id},
+        )
+
+    op.drop_column("media_assets", "expires_at")
+    op.drop_column("media_assets", "gc_notified")
+
+
+def downgrade() -> None:
+    op.add_column("media_assets", sa.Column("expires_at", sa.DateTime(timezone=True)))
+    op.add_column(
+        "media_assets",
+        sa.Column("gc_notified", sa.Boolean(), server_default="false", nullable=False),
+    )
+    # The asset goes when its last holder would: the latest of its holders' expiries (a holder that
+    # never expires keeps the asset unexpired).
+    op.execute(
+        "UPDATE media_assets a SET expires_at = sub.e FROM ("
+        " SELECT asset_id, CASE WHEN bool_or(expires_at IS NULL) THEN NULL ELSE max(expires_at) END"
+        " AS e FROM download_jobs WHERE asset_id IS NOT NULL AND status = 'done' GROUP BY asset_id"
+        ") sub WHERE a.id = sub.asset_id AND a.source_kind <> 'upload'"
+    )
+    op.drop_index("ix_download_jobs_expires_at", table_name="download_jobs")
+    op.drop_index("ix_download_jobs_folder_id", table_name="download_jobs")
+    op.drop_column("download_jobs", "expiry_warned")
+    op.drop_column("download_jobs", "expires_at")
+    op.drop_column("download_jobs", "kept")
+    op.drop_column("download_jobs", "folder_id")
+    op.drop_index("uq_file_folder_own_name", table_name="file_folders")
+    op.drop_index("ix_file_folders_user_id", table_name="file_folders")
+    op.drop_table("file_folders")
diff --git a/backend/app/config.py b/backend/app/config.py
index 8edf93aa..4007159e 100644
--- a/backend/app/config.py
+++ b/backend/app/config.py
@@ -173,7 +173,7 @@ class Settings(BaseSettings):
     # --- Download center (yt-dlp) ---
     # Root directory (inside the container) for downloaded media — a Docker volume/bind mount.
     # Env-only (a mount path can't live in the DB). The worker lays a Plex-style tree under it.
-    download_root: str = "/downloads"
+    files_root: str = "/files"
     # Whether THIS process runs the download worker loop. On for the dedicated worker
     # container, off for the API (which keeps the scheduler). Mirrors scheduler_enabled.
     worker_enabled: bool = False
@@ -345,8 +345,8 @@ class Settings(BaseSettings):
     plex_transcode_crf: int = 20
     plex_transcode_max_height: int = 1080
     # Where on-the-fly HLS remux segments are written (transient, reaped). Empty → a `.plex-hls`
-    # dir under download_root. On localdev the download_root is a slow Windows bind-mount, so point
-    # this at a fast container-local path (e.g. /var/tmp/plex-hls); prod's download_root is fast
+    # dir under files_root. On localdev the files_root is a slow Windows bind-mount, so point
+    # this at a fast container-local path (e.g. /var/tmp/plex-hls); prod's files_root is fast
     # local disk, so the default is fine there.
     plex_hls_dir: str = ""
 
diff --git a/backend/app/downloads/backfill.py b/backend/app/downloads/backfill.py
index b793a770..13d035ab 100644
--- a/backend/app/downloads/backfill.py
+++ b/backend/app/downloads/backfill.py
@@ -89,7 +89,7 @@ def run() -> dict:
             if a.poster_path is None:
                 # Record a self-hosted poster for every download (the reliable og:image source):
                 # ensure_poster returns the existing <base>.jpg thumbnail sidecar, or cuts a frame.
-                rel = editmod.ensure_poster(a, settings.download_root)
+                rel = editmod.ensure_poster(a, settings.files_root)
                 if rel:
                     a.poster_path = rel
                     filled_poster += 1
diff --git a/backend/app/downloads/edit.py b/backend/app/downloads/edit.py
index 21b82983..2ce485b1 100644
--- a/backend/app/downloads/edit.py
+++ b/backend/app/downloads/edit.py
@@ -732,7 +732,7 @@ def thumbnail_to_jpg(src: Path | None, dest: Path, timeout_s: int = 30) -> Path
     return dest if dest.exists() else None
 
 
-def ensure_poster(asset, download_root: str, timeout_s: int = 30) -> str | None:
+def ensure_poster(asset, files_root: str, timeout_s: int = 30) -> str | None:
     """Generate a poster frame for a ready asset that has no thumbnail (e.g. a direct media URL
     like a reddit HLS manifest, which yt-dlp can't get a thumbnail for). Written as the media's
     `<base>.jpg` sidecar — so Plex uses it too and `delete_asset_files` cleans it up — and returned
@@ -740,7 +740,7 @@ def ensure_poster(asset, download_root: str, timeout_s: int = 30) -> str | None:
     short clip) dodges black intro frames."""
     if not asset.rel_path:
         return None
-    root = Path(download_root)
+    root = Path(files_root)
     src = root / asset.rel_path
     if not src.exists():
         return None
@@ -756,7 +756,7 @@ def ensure_poster(asset, download_root: str, timeout_s: int = 30) -> str | None:
     return rel if dest.exists() else None
 
 
-def ensure_storyboard(asset, download_root: str, timeout_s: int = 90) -> dict | None:
+def ensure_storyboard(asset, files_root: str, timeout_s: int = 90) -> dict | None:
     """Return the filmstrip meta for a ready source asset, generating the sprite on first use.
 
     Best-effort + time-bounded: for a very long/high-res source the one-pass decode can be slow,
@@ -768,7 +768,7 @@ def ensure_storyboard(asset, download_root: str, timeout_s: int = 90) -> dict |
     if not asset.rel_path:
         return None
     geom = storyboard_geometry(asset.duration_s)
-    root = Path(download_root)
+    root = Path(files_root)
     rel = f".storyboards/asset-{asset.id}.jpg"
     dest = root / rel
 
diff --git a/backend/app/downloads/folders.py b/backend/app/downloads/folders.py
new file mode 100644
index 00000000..b962bf65
--- /dev/null
+++ b/backend/app/downloads/folders.py
@@ -0,0 +1,139 @@
+"""The folders of the Files library — where each of a user's files sits.
+
+A folder is a row, never a directory: on disk a file stays where the download layout put it (one
+shared asset can be held by several users), so a rename or a move is a column update. Two kinds:
+
+- a SOURCE folder is made automatically, one per origin a user has files from (YouTube, Plex clips,
+  uploads, one per other site). Its `source_key` names the origin; the client renders its name
+  from the key, in the reader's language.
+- an OWN folder is the user's, with the name they gave it.
+
+Every file of a user is in exactly one folder. The folder a new file lands in is decided here
+(`source_key_for` + `source_folder`), at every place a job is created, so the rule lives once.
+
+Retention is the folder's: a source folder keeps a file `download_retention_days` (the value when the
+folder was made; the user can change it), Uploads and own folders keep it until it is deleted. A
+done file carries its own `expires_at`, stamped by `done_expiry` when it finishes — unless the user
+kept it.
+"""
+
+from datetime import datetime, timedelta, timezone
+from enum import StrEnum
+from urllib.parse import urlsplit
+
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app import sysconfig
+from app.models import DownloadJob, FileFolder, MediaAsset
+
+
+class FolderKind(StrEnum):
+    SOURCE = "source"
+    OWN = "own"
+
+
+class SourceKey(StrEnum):
+    """The fixed origins. Any other site is `site:<host>` (see `site_key`)."""
+
+    YOUTUBE = "youtube"
+    PLEX = "plex"
+    UPLOAD = "upload"
+
+
+SITE_PREFIX = "site:"
+
+# A host's leading label that only says "the same site, another front door".
+_FRONT_DOORS = ("www.", "m.", "mobile.", "web.")
+_YOUTUBE_HOSTS = {"youtube.com", "youtu.be", "youtube-nocookie.com", "music.youtube.com"}
+
+
+def site_key(url: str) -> str:
+    """The source key of a page on some site: `site:<host>` without its front door (www., m.),
+    or the YouTube key for a YouTube address that arrived as a plain URL."""
+    host = (urlsplit(url).hostname or "").lower().rstrip(".")
+    for door in _FRONT_DOORS:
+        if host.startswith(door) and host.count(".") > 1:
+            host = host[len(door) :]
+            break
+    if host in _YOUTUBE_HOSTS:
+        return SourceKey.YOUTUBE
+    return f"{SITE_PREFIX}{host or 'unknown'}"
+
+
+def source_key_for(db: Session, job: DownloadJob) -> str:
+    """The origin a job's file came from — the source folder it belongs in.
+
+    youtube → YouTube; a clip → the clip's source (YouTube or Plex); an upload → Uploads; a URL →
+    its site; an edit → wherever its parent's file came from (its parent job, or, once that is
+    deleted, the parent's asset)."""
+    kind = job.source_kind
+    if kind == "youtube":
+        return SourceKey.YOUTUBE
+    if kind == "upload":
+        return SourceKey.UPLOAD
+    if kind == "clip":
+        return (
+            SourceKey.PLEX if (job.edit_spec or {}).get("source") == "plex" else SourceKey.YOUTUBE
+        )
+    if kind == "url":
+        return site_key(job.source_ref)
+    if kind == "edit":
+        parent = db.get(DownloadJob, job.source_job_id) if job.source_job_id else None
+        if parent is not None and parent.id != job.id:
+            return source_key_for(db, parent)
+        asset = db.get(MediaAsset, job.source_asset_id) if job.source_asset_id else None
+        if asset is not None:
+            if asset.source_kind == "youtube":
+                return SourceKey.YOUTUBE
+            if asset.source_kind == "upload":
+                return SourceKey.UPLOAD
+            if asset.source_kind == "url":
+                return site_key(asset.source_ref)
+    return SourceKey.UPLOAD
+
+
+def source_folder(db: Session, user_id: int, key: str) -> FileFolder:
+    """The user's source folder for `key`, made on first use. A race on the unique (user, key) is
+    settled by re-selecting, inside a SAVEPOINT so the caller's own pending writes survive."""
+    stmt = select(FileFolder).where(FileFolder.user_id == user_id, FileFolder.source_key == key)
+    folder = db.execute(stmt).scalar_one_or_none()
+    if folder is not None:
+        return folder
+    retention = (
+        None if key == SourceKey.UPLOAD else sysconfig.get_int(db, "download_retention_days")
+    )
+    folder = FileFolder(
+        user_id=user_id, kind=FolderKind.SOURCE, source_key=key, retention_days=retention
+    )
+    try:
+        with db.begin_nested():
+            db.add(folder)
+            db.flush()
+    # silent-ok: the expected outcome of the race the savepoint exists for — another request made
+    # the same folder first; re-selecting it is the whole handling.
+    except IntegrityError:
+        folder = db.execute(stmt).scalar_one()
+    return folder
+
+
+def place(db: Session, job: DownloadJob) -> None:
+    """Put a new job's file in its source folder. Called wherever a job is created; the job must
+    carry its `user_id`, `source_kind` and source refs already."""
+    job.folder_id = source_folder(db, job.user_id, source_key_for(db, job)).id
+
+
+def done_expiry(db: Session, folder_id: int | None, kept: bool) -> datetime | None:
+    """When a file that finishes now goes: its folder's retention from now, or never (kept, or a
+    folder that keeps files)."""
+    if kept or folder_id is None:
+        return None
+    days = db.execute(select(FileFolder.retention_days).where(FileFolder.id == folder_id)).scalar()
+    return None if days is None else datetime.now(timezone.utc) + timedelta(days=days)
+
+
+def stamp_expiry(db: Session, job: DownloadJob) -> None:
+    """Stamp a job that just became done (in this session) with its expiry."""
+    job.expires_at = done_expiry(db, job.folder_id, job.kept)
+    job.expiry_warned = False
diff --git a/backend/app/downloads/gc.py b/backend/app/downloads/gc.py
index 7a9c2afd..6b22c2ea 100644
--- a/backend/app/downloads/gc.py
+++ b/backend/app/downloads/gc.py
@@ -1,12 +1,13 @@
 """Download retention garbage collector (a scheduler job).
 
 Five passes each run:
-  1. Pre-expiry warning — for ready assets expiring within the grace window, notify each holder
-     once (gc_notified flag) so a download isn't silently deleted.
-  2. Expiry deletion — delete assets past their TTL (files + row). The FK SET NULL nulls the
-     holders' jobs' asset_id, so those jobs surface as "expired" (re-downloadable) in the UI.
-  3. LRU eviction — if a total cache cap is set and exceeded, delete least-recently-accessed
-     ready assets (never an upload) until back under it.
+  1. Pre-expiry warning — a done file whose `expires_at` (its folder's retention, `folders`) falls
+     within the grace window, and that its user did not keep, warns its user once (`expiry_warned`).
+  2. Expiry — such a file past its `expires_at` leaves the library: the job is deleted and its hold
+     released, so the shared file goes with its last holder.
+  3. LRU eviction — if a total cache cap is set and exceeded, delete least-recently-accessed ready
+     assets (never an upload, never one a user kept) until back under it; each holder's file leaves
+     their library, with a notice.
   4. Orphan reap — delete errored, fileless, unreferenced asset rows (they carry no TTL, so
      passes 1-3 skip them; benign row-count hygiene).
   5. Orphan FILE sweep — delete on-disk files/dirs no surviving asset owns (cruft left when a
@@ -14,7 +15,7 @@ Five passes each run:
      It skips the dot-prefixed system trees (all but `.uploads`) and anything newer than an hour,
      so an in-flight download is never touched.
 
-Runs in the API process (which mounts DOWNLOAD_ROOT). Pure disk/DB work — no YouTube quota.
+Runs in the API process (which mounts FILES_ROOT). Pure disk/DB work — no YouTube quota.
 """
 
 import time
@@ -26,7 +27,8 @@ from sqlalchemy.orm import Session
 
 from app import logs, sysconfig
 from app.config import settings
-from app.downloads import quota, storage
+from app.downloads import quota, service, storage
+from app.downloads.status import JobStatus
 from app.models import DownloadJob, MediaAsset
 from app.notifications import create_notification
 
@@ -37,42 +39,39 @@ _UPLOAD_TREE = ".uploads"
 _ORPHAN_MIN_AGE_S = 3600  # only sweep files older than this, so an in-flight download is never hit
 
 
-def _holders(db: Session, asset_id: int) -> list[int]:
-    return list(
-        db.execute(
-            select(DownloadJob.user_id).where(DownloadJob.asset_id == asset_id).distinct()
-        ).scalars()
+def _notify(db: Session, job: DownloadJob, asset: MediaAsset | None, ntype: str) -> None:
+    title = job.display_name or (asset.title if asset else None) or "download"
+    create_notification(
+        db,
+        user_id=job.user_id,
+        type=ntype,
+        title=title,  # English fallback; the client renders a translated message from data
+        data={
+            "kind": ntype,
+            "title": title,
+            "video_id": asset.source_ref if asset else None,
+            "job_id": job.id,
+        },
+        commit=False,
     )
 
 
-def _notify_holders(db: Session, asset: MediaAsset, ntype: str) -> None:
-    title = asset.title or "download"
-    for uid in _holders(db, asset.id):
-        create_notification(
-            db,
-            user_id=uid,
-            type=ntype,
-            title=title,  # English fallback; the client renders a translated message from data
-            data={"kind": ntype, "title": asset.title, "video_id": asset.source_ref},
-            commit=False,
-        )
+def _drop_job(db: Session, job: DownloadJob, ntype: str) -> None:
+    """Take a file out of its user's library: tell them, release the job's hold on the asset (the
+    file goes with the last holder), delete the row (its shares and links cascade)."""
+    _notify(db, job, db.get(MediaAsset, job.asset_id) if job.asset_id else None, ntype)
+    service.release_asset(db, job)
+    db.delete(job)
 
 
-def _delete_asset(db: Session, asset: MediaAsset, ntype: str) -> None:
-    if asset.rel_path:
-        storage.delete_asset_files(settings.download_root, asset.rel_path)
-    _notify_holders(db, asset, ntype)
-    db.delete(asset)  # jobs.asset_id -> NULL via FK ON DELETE SET NULL
-
-
-def _sweep_orphan_files(db: Session, download_root: str) -> int:
-    """Delete media/sidecar files (and now-empty dirs) under DOWNLOAD_ROOT that no `ready` asset
+def _sweep_orphan_files(db: Session, files_root: str) -> int:
+    """Delete media/sidecar files (and now-empty dirs) under FILES_ROOT that no `ready` asset
     owns — cruft left when a re-download changes an asset's on-disk name (e.g. an old accented
     channel dir, or a poster from a superseded naming scheme) — or, under `.uploads`, that no upload
     row names at all. Skips the other dot-prefixed system trees and only removes files older than
     `_ORPHAN_MIN_AGE_S`, so a just-completed or in-flight download is never deleted, and only files
     NOT owned by a surviving asset."""
-    root = Path(download_root)
+    root = Path(files_root)
     if not root.is_dir():
         return 0
     keep: set[Path] = set()
@@ -151,47 +150,52 @@ def run_download_gc(db: Session) -> dict:
     grace_days = sysconfig.get_int(db, "download_gc_grace_days")
     warned = expired = evicted = reaped = 0
 
-    # 1. Pre-expiry warnings.
+    # 1. Pre-expiry warnings — one per file, the first run inside the grace window.
     warn_before = now + timedelta(days=grace_days)
     soon = (
         db.execute(
-            select(MediaAsset).where(
-                MediaAsset.status == "ready",
-                MediaAsset.gc_notified.is_(False),
-                MediaAsset.expires_at.is_not(None),
-                MediaAsset.expires_at > now,
-                MediaAsset.expires_at <= warn_before,
+            select(DownloadJob).where(
+                DownloadJob.status == JobStatus.DONE,
+                DownloadJob.kept.is_(False),
+                DownloadJob.expiry_warned.is_(False),
+                DownloadJob.expires_at.is_not(None),
+                DownloadJob.expires_at > now,
+                DownloadJob.expires_at <= warn_before,
             )
         )
         .scalars()
         .all()
     )
-    for asset in soon:
-        _notify_holders(db, asset, "download_expiring")
-        asset.gc_notified = True
+    for job in soon:
+        _notify(
+            db, job, db.get(MediaAsset, job.asset_id) if job.asset_id else None, "download_expiring"
+        )
+        job.expiry_warned = True
         warned += 1
     if soon:
         db.commit()
 
-    # 2. Expiry deletion (ready assets past TTL, and any errored leftovers with a TTL).
+    # 2. Expiry — a file past its folder's retention leaves the library (warned in pass 1).
     dead = (
         db.execute(
-            select(MediaAsset).where(
-                MediaAsset.status.in_(("ready", "error")),
-                MediaAsset.expires_at.is_not(None),
-                MediaAsset.expires_at <= now,
+            select(DownloadJob).where(
+                DownloadJob.status == JobStatus.DONE,
+                DownloadJob.kept.is_(False),
+                DownloadJob.expires_at.is_not(None),
+                DownloadJob.expires_at <= now,
             )
         )
         .scalars()
         .all()
     )
-    for asset in dead:
-        _delete_asset(db, asset, "download_expired")
+    for job in dead:
+        _drop_job(db, job, "download_expired")
         expired += 1
     if dead:
         db.commit()
 
-    # 3. LRU eviction over the total cache cap.
+    # 3. LRU eviction over the total cache cap: the disk's safety net. Never an upload (no source to
+    #    fetch again) and never a file anyone kept; an evicted file leaves every holder's library.
     cap = sysconfig.get_int(db, "download_total_max_bytes")
     if cap > 0:
         total = (
@@ -201,12 +205,17 @@ def run_download_gc(db: Session) -> dict:
             or 0
         )
         if total > cap:
-            # An upload counts toward the total (it is disk like any other) but is never evicted:
-            # it has no source to fetch again, and it lives until its owner deletes it.
+            kept_assets = select(DownloadJob.asset_id).where(
+                DownloadJob.kept.is_(True), DownloadJob.asset_id.is_not(None)
+            )
             candidates = (
                 db.execute(
                     select(MediaAsset)
-                    .where(MediaAsset.status == "ready", MediaAsset.source_kind != "upload")
+                    .where(
+                        MediaAsset.status == "ready",
+                        MediaAsset.source_kind != "upload",
+                        MediaAsset.id.not_in(kept_assets),
+                    )
                     .order_by(MediaAsset.last_access_at.asc().nulls_first())
                 )
                 .scalars()
@@ -216,7 +225,17 @@ def run_download_gc(db: Session) -> dict:
                 if total <= cap:
                     break
                 total -= asset.size_bytes or 0
-                _delete_asset(db, asset, "download_evicted")
+                holders = (
+                    db.execute(select(DownloadJob).where(DownloadJob.asset_id == asset.id))
+                    .scalars()
+                    .all()
+                )
+                for job in holders:
+                    _notify(db, job, asset, "download_evicted")
+                    db.delete(job)
+                if asset.rel_path:
+                    storage.delete_asset_files(settings.files_root, asset.rel_path)
+                db.delete(asset)
                 evicted += 1
             db.commit()
 
@@ -257,7 +276,7 @@ def run_download_gc(db: Session) -> dict:
     )
     for asset in unheld:
         if asset.rel_path:  # no holder is left to tell, so no notice — just the file and the row
-            storage.delete_asset_files(settings.download_root, asset.rel_path)
+            storage.delete_asset_files(settings.files_root, asset.rel_path)
         db.delete(asset)
         reaped += 1
     if unheld:
@@ -265,7 +284,7 @@ def run_download_gc(db: Session) -> dict:
 
     # 5. Orphan FILE sweep — remove on-disk cruft no surviving asset owns (old accented dirs, stale
     #    posters). Runs last so the keep-set reflects the assets that survived passes 2-4.
-    swept = _sweep_orphan_files(db, settings.download_root)
+    swept = _sweep_orphan_files(db, settings.files_root)
 
     result = {
         "warned": warned,
diff --git a/backend/app/downloads/jobdb.py b/backend/app/downloads/jobdb.py
index a0935749..a8e95b29 100644
--- a/backend/app/downloads/jobdb.py
+++ b/backend/app/downloads/jobdb.py
@@ -68,8 +68,8 @@ def _reset_asset_pending(asset_id: int) -> None:
 
 
 def _staging_dir(asset_id: int) -> Path:
-    return Path(settings.download_root) / ".staging" / f"asset-{asset_id}"
+    return Path(settings.files_root) / ".staging" / f"asset-{asset_id}"
 
 
 def _edit_staging_dir(asset_id: int) -> Path:
-    return Path(settings.download_root) / ".staging" / f"edit-{asset_id}"
+    return Path(settings.files_root) / ".staging" / f"edit-{asset_id}"
diff --git a/backend/app/downloads/og.py b/backend/app/downloads/og.py
index 2bc7bf47..99cd2154 100644
--- a/backend/app/downloads/og.py
+++ b/backend/app/downloads/og.py
@@ -135,7 +135,7 @@ def _watch_tags(token: str, db: Session) -> str | None:
         if image_dims is None and image.endswith("poster.jpg") and asset.poster_path:
             # safe_abs_path returns None if the poster file is missing (gc'd) or escapes the root —
             # guard it, else _jpeg_size(None) would raise an uncaught AttributeError on this public page.
-            poster_abs = storage.safe_abs_path(settings.download_root, asset.poster_path)
+            poster_abs = storage.safe_abs_path(settings.files_root, asset.poster_path)
             image_dims = _jpeg_size(poster_abs) if poster_abs else None
         if image_dims:
             tags.append(_tag("og:image:width", str(image_dims[0])))
diff --git a/backend/app/downloads/pipeline.py b/backend/app/downloads/pipeline.py
index 2b937438..3e792513 100644
--- a/backend/app/downloads/pipeline.py
+++ b/backend/app/downloads/pipeline.py
@@ -3,7 +3,7 @@
 import re
 import shutil
 import time
-from datetime import date, datetime, timedelta, timezone
+from datetime import date, datetime, timezone
 from pathlib import Path
 
 import yt_dlp
@@ -434,14 +434,12 @@ def _extract_with_fallback(
     raise last_exc  # unreachable (loop either returns or raises)
 
 
-def _download_settings() -> tuple[str, int]:
-    """Layout + retention days from the DB config (admin-overridable on the Configuration page),
-    falling back to the env defaults. Read here, not from `settings`, so an admin edit takes effect."""
+def _download_layout() -> str:
+    """The on-disk layout from the DB config (admin-overridable on the Configuration page), falling
+    back to the env default. Read here, not from `settings`, so an admin edit takes effect. (How long
+    a file stays is its folder's, stamped on the job when it is done — `folders.done_expiry`.)"""
     with SessionLocal() as db:
-        return (
-            sysconfig.get_str(db, "download_layout"),
-            sysconfig.get_int(db, "download_retention_days"),
-        )
+        return sysconfig.get_str(db, "download_layout")
 
 
 def _ensure_phone_safe(
@@ -588,19 +586,18 @@ def _download(
             thumbnail_url=info.get("thumbnail"),
         )
 
-        layout, retention_days = _download_settings()
+        layout = _download_layout()
         rel = storage.rel_path(meta, ext, layout, asset_id=asset_id)
         thumb = _find_thumbnail(staging, produced)
         # Convert the raw thumbnail (webp/png/jpeg) to a real JPEG ourselves — robustly — for the
         # poster + og:image sidecar, now that yt-dlp's fatal thumbnail convertor is gone. On failure
         # this is None and write_sidecars writes no poster, so ensure_poster cuts a frame instead.
         poster_jpg = editmod.thumbnail_to_jpg(thumb, staging / "poster.jpg")
-        storage.place_file(produced, settings.download_root, rel)
-        nfo_ok = storage.write_sidecars(settings.download_root, rel, meta, poster_jpg)
+        storage.place_file(produced, settings.files_root, rel)
+        nfo_ok = storage.write_sidecars(settings.files_root, rel, meta, poster_jpg)
 
-        final = storage.abs_path(settings.download_root, rel)
+        final = storage.abs_path(settings.files_root, rel)
         size = final.stat().st_size if final.exists() else None
-        expires = datetime.now(timezone.utc) + timedelta(days=retention_days)
 
         with SessionLocal() as db:
             asset = db.get(MediaAsset, asset_id)
@@ -626,12 +623,11 @@ def _download(
                 # (og:image) source — a remote thumbnail, esp. Facebook's signed CDN URL, expires
                 # and can't be relied on cross-origin. write_sidecars already wrote <base>.jpg from
                 # the thumbnail; ensure_poster returns that, or cuts a frame if there was none.
-                asset.poster_path = editmod.ensure_poster(asset, settings.download_root)
+                asset.poster_path = editmod.ensure_poster(asset, settings.files_root)
                 asset.source_webpage_url = info.get("webpage_url")
                 asset.nfo_written = nfo_ok
                 asset.error = None
                 asset.last_access_at = datetime.now(timezone.utc)
-                asset.expires_at = expires
             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
@@ -840,7 +836,7 @@ def _process_edit(job_id: int, asset_id: int) -> None:
     if not src_ready or not src_rel:
         _fail_edit(job_id, asset_id, "The source download is no longer available.")
         return
-    src_path = storage.abs_path(settings.download_root, src_rel)
+    src_path = storage.abs_path(settings.files_root, src_rel)
     if not src_path.exists():
         _fail_edit(job_id, asset_id, "The source file is missing.")
         return
@@ -887,13 +883,11 @@ def _process_edit(job_id: int, asset_id: int) -> None:
                 raise StatedFailure(dlerrors.CLIP_TOO_BIG_MESSAGE)
 
         rel = storage.edit_rel_path(user_id, asset_id, out_ext)
-        storage.place_file(dest_staging, settings.download_root, rel)
-        final = storage.abs_path(settings.download_root, rel)
+        storage.place_file(dest_staging, settings.files_root, rel)
+        final = storage.abs_path(settings.files_root, rel)
         size = final.stat().st_size if final.exists() else None
         # The clip's real frame: a re-encode evens an odd crop or source (`editmod.EVEN_FRAME`).
         _, _, out_w, out_h, _ = probe.probe_media(final)
-        _, retention_days = _download_settings()
-        expires = datetime.now(timezone.utc) + timedelta(days=retention_days)
 
         with SessionLocal() as db:
             asset = db.get(MediaAsset, asset_id)
@@ -912,7 +906,6 @@ def _process_edit(job_id: int, asset_id: int) -> None:
                     asset.width, asset.height = src_w, src_h
                 asset.error = None
                 asset.last_access_at = datetime.now(timezone.utc)
-                asset.expires_at = expires
             job = db.get(DownloadJob, job_id)
             # Capture owner + title before commit expires the rows (avoids a refresh SELECT).
             owner_id = job.user_id if job else None
@@ -1086,10 +1079,9 @@ def _process_clip(job_id: int, asset_id: int) -> None:
             raise _OwnerRefused(dlerrors.CLIP_TOO_BIG_MESSAGE)
         ext = out.suffix.lstrip(".").lower() or "mp4"
         rel = storage.edit_rel_path(user_id, asset_id, ext)
-        storage.place_file(out, settings.download_root, rel)
-        final = storage.abs_path(settings.download_root, rel)
+        storage.place_file(out, settings.files_root, rel)
+        final = storage.abs_path(settings.files_root, rel)
         vcodec, acodec, width, height, _ = probe.probe_media(final)
-        _, retention_days = _download_settings()
 
         with SessionLocal() as db:
             asset = db.get(MediaAsset, asset_id)
@@ -1108,10 +1100,9 @@ def _process_clip(job_id: int, asset_id: int) -> None:
                 asset.media_kind = mediakind.derive(vcodec, acodec)
                 # The clip's own frame as its poster: the link preview shows the moment, and a Plex
                 # item has no public thumbnail at all.
-                asset.poster_path = editmod.ensure_poster(asset, settings.download_root)
+                asset.poster_path = editmod.ensure_poster(asset, settings.files_root)
                 asset.error = None
                 asset.last_access_at = datetime.now(timezone.utc)
-                asset.expires_at = datetime.now(timezone.utc) + timedelta(days=retention_days)
             fields: dict = {"status": JobStatus.DONE, "progress": 100, "phase": None}
             # A video outside the catalog was queued as "Clip (clip)": the fetch has its title now.
             placeholder = service.clip_title(None, None)
@@ -1230,7 +1221,7 @@ def _free_unheld_upload(asset_id: int) -> bool:
         if asset is None or asset.ref_count != 0:
             return False
         if asset.rel_path:
-            storage.delete_asset_files(settings.download_root, asset.rel_path)
+            storage.delete_asset_files(settings.files_root, asset.rel_path)
         db.delete(asset)
         db.commit()
     return True
@@ -1253,7 +1244,7 @@ def _convert_upload(job_id: int, asset_id: int) -> None:
         duration = asset.duration_s if asset else None
         job = db.get(DownloadJob, job_id)
         owner_id = job.user_id if job else None
-    src = storage.safe_abs_path(settings.download_root, old_rel) if old_rel else None
+    src = storage.safe_abs_path(settings.files_root, old_rel) if old_rel else None
     if old_rel is None or src is None or not src.exists():
         _fail_edit(job_id, asset_id, dlerrors.UPLOAD_MISSING_MESSAGE, what="upload")
         return
@@ -1267,10 +1258,10 @@ def _convert_upload(job_id: int, asset_id: int) -> None:
         new_rel = storage.with_ext(old_rel, old_ext, ext)
         if safe != src:
             _check_converted_fits(owner_id, old_size, safe.stat().st_size)
-            storage.place_file(safe, settings.download_root, new_rel)
+            storage.place_file(safe, settings.files_root, new_rel)
             if new_rel != old_rel:
-                placed = storage.abs_path(settings.download_root, new_rel)
-        final = storage.abs_path(settings.download_root, new_rel)
+                placed = storage.abs_path(settings.files_root, new_rel)
+        final = storage.abs_path(settings.files_root, new_rel)
         vcodec, acodec, width, height, dur = probe.probe_media(final)
         with SessionLocal() as db:
             asset = db.get(MediaAsset, asset_id)
diff --git a/backend/app/downloads/service.py b/backend/app/downloads/service.py
index 96770af0..fd91d10f 100644
--- a/backend/app/downloads/service.py
+++ b/backend/app/downloads/service.py
@@ -19,7 +19,7 @@ from app.downloads import adapters
 from app.downloads import clip as clipmod
 from app.downloads import credentials
 from app.downloads import edit as editmod
-from app.downloads import formats, mediakind, quota, storage
+from app.downloads import folders, formats, mediakind, quota, storage
 from app.downloads.status import HOLDING, JobKind, JobStatus
 from app.models import (
     DISPLAY_TEXT_MAX,
@@ -198,12 +198,14 @@ def enqueue(
         queue_pos=next_queue_pos(db),
     )
     db.add(job)
+    folders.place(db, job)
     incr_ref_count(db, asset.id)
 
     # Cache hit: the file already exists — no worker round-trip needed.
     if asset.status == "ready":
         job.status = JobStatus.DONE
         job.progress = 100
+        folders.stamp_expiry(db, job)
         asset.last_access_at = datetime.now(timezone.utc)
 
     db.commit()
@@ -277,11 +279,13 @@ def enqueue_edit(
         queue_pos=next_queue_pos(db),
     )
     db.add(job)
+    folders.place(db, job)
     incr_ref_count(db, asset.id)
 
     if asset.status == "ready":  # identical clip already produced for this user → instant
         job.status = JobStatus.DONE
         job.progress = 100
+        folders.stamp_expiry(db, job)
         asset.last_access_at = datetime.now(timezone.utc)
 
     db.commit()
@@ -364,11 +368,13 @@ def enqueue_clip(
         queue_pos=next_queue_pos(db),
     )
     db.add(job)
+    folders.place(db, job)
     incr_ref_count(db, asset.id)
 
     if asset.status == "ready":  # the same range was already cut for this user → instant
         job.status = JobStatus.DONE
         job.progress = 100
+        folders.stamp_expiry(db, job)
         asset.last_access_at = datetime.now(timezone.utc)
 
     db.commit()
@@ -392,6 +398,17 @@ def finalize_if_running(db: Session, job_id: int, **fields) -> bool:
     if not fields:
         return False
     db.flush()  # asset-then-job lock order: push pending asset writes before we lock the job row
+    if fields.get("status") == JobStatus.DONE:
+        # Done is when a file's retention starts: its folder's, unless the user kept it.
+        held = db.execute(
+            select(DownloadJob.folder_id, DownloadJob.kept).where(DownloadJob.id == job_id)
+        ).one_or_none()
+        if held is not None:
+            fields = {
+                **fields,
+                "expires_at": folders.done_expiry(db, held.folder_id, held.kept),
+                "expiry_warned": False,
+            }
     # A Core UPDATE with the guard as a WHERE — same atomic "only while still running" write as the
     # rest of the codebase's 2.0 statements, no hand-built SQL string. synchronize_session=False: this
     # is a terminal write, so we don't pay to sync the (about-to-be-discarded) ORM identity map. Bound
@@ -436,7 +453,7 @@ def release_asset(db: Session, job: DownloadJob) -> None:
     )
     if new_count == 0 and freeable:
         if asset.rel_path:
-            storage.delete_asset_files(settings.download_root, asset.rel_path)
+            storage.delete_asset_files(settings.files_root, asset.rel_path)
         db.delete(asset)
 
 
@@ -565,4 +582,6 @@ def redownload(db: Session, job: DownloadJob) -> None:
     # name would survive the rebuild).
     job.display_name = None
     job.queue_pos = next_queue_pos(db)  # tail of the queue, not its stale original slot
+    job.expires_at = None  # stamped again when the rebuild is done
+    job.expiry_warned = False
     db.commit()
diff --git a/backend/app/downloads/storage.py b/backend/app/downloads/storage.py
index 4a3ea17e..d2ec540e 100644
--- a/backend/app/downloads/storage.py
+++ b/backend/app/downloads/storage.py
@@ -122,7 +122,7 @@ def display_filename(name: str) -> str:
 
 
 def rel_path(meta: MediaMeta, ext: str, layout: str = "plex", asset_id: int | None = None) -> str:
-    """Path of the media file relative to DOWNLOAD_ROOT (forward slashes).
+    """Path of the media file relative to FILES_ROOT (forward slashes).
 
     `asset_id` disambiguates the file. A MediaAsset is unique on (source_kind, source_ref,
     format_sig), so the SAME video downloaded in two formats is two distinct assets — but without
@@ -144,7 +144,7 @@ def rel_path(meta: MediaMeta, ext: str, layout: str = "plex", asset_id: int | No
 
 
 def edit_rel_path(user_id: int, asset_id: int, ext: str) -> str:
-    """On-disk path (relative to DOWNLOAD_ROOT) for a phase-2 editor derivative.
+    """On-disk path (relative to FILES_ROOT) for a phase-2 editor derivative.
 
     Kept in a per-user `.edits/` tree — device-downloadable via the file endpoint, but NOT part of
     the Plex library (these are user clips, not canonical episodes; there are no .nfo/poster
@@ -161,7 +161,7 @@ def _clean_upload_ext(filename: str) -> str:
 
 
 def upload_rel_path(user_id: int, asset_id: int, filename: str) -> str:
-    """On-disk path (relative to DOWNLOAD_ROOT) for a locally-uploaded file.
+    """On-disk path (relative to FILES_ROOT) for a locally-uploaded file.
 
     Kept in a per-user `.uploads/` tree — device-downloadable via the file endpoint, but NOT part of
     the Plex library (an upload is arbitrary user content, not a canonical episode; no .nfo/poster
@@ -250,24 +250,27 @@ def download_filename(display_name: str, container: str | None, path: Path) -> s
     return f"{base}.{ext}"
 
 
-def abs_path(download_root: str, rel: str) -> Path:
-    return Path(download_root) / rel
+def abs_path(files_root: str, rel: str) -> Path:
+    return Path(files_root) / rel
 
 
-def safe_abs_path(download_root: str, rel: str) -> Path | None:
-    """The resolved absolute path for `rel`, but ONLY if it stays inside DOWNLOAD_ROOT and exists on
-    disk — otherwise None. Centralizes the path-traversal + existence guard every file-serving endpoint
-    shares, so the containment check lives in exactly one place."""
-    root = Path(download_root).resolve()
-    path = abs_path(download_root, rel).resolve()
+def safe_abs_path(files_root: str, rel: str | None) -> Path | None:
+    """The resolved absolute path for `rel`, but ONLY if it stays inside FILES_ROOT and exists on
+    disk — otherwise None (no `rel` at all included: an asset without a file has nothing to serve).
+    Centralizes the path-traversal + existence guard every file-serving endpoint shares, so the
+    containment check lives in exactly one place."""
+    if not rel:
+        return None
+    root = Path(files_root).resolve()
+    path = abs_path(files_root, rel).resolve()
     if root not in path.parents or not path.exists():
         return None
     return path
 
 
-def place_file(staging_file: Path, download_root: str, rel: str) -> None:
-    """Move a completed staging file into its final location under DOWNLOAD_ROOT."""
-    dest = abs_path(download_root, rel)
+def place_file(staging_file: Path, files_root: str, rel: str) -> None:
+    """Move a completed staging file into its final location under FILES_ROOT."""
+    dest = abs_path(files_root, rel)
     dest.parent.mkdir(parents=True, exist_ok=True)
     shutil.move(str(staging_file), str(dest))
 
@@ -301,9 +304,9 @@ def _sidecar(media: Path, ext: str) -> Path:
     return media.with_suffix(ext)
 
 
-def write_sidecars(download_root: str, rel: str, meta: MediaMeta, thumb_src: Path | None) -> bool:
+def write_sidecars(files_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)
+    media = abs_path(files_root, rel)
     try:
         _sidecar(media, ".nfo").write_text(_nfo_xml(meta), encoding="utf-8")
         if thumb_src and thumb_src.exists():
@@ -319,9 +322,9 @@ def write_sidecars(download_root: str, rel: str, meta: MediaMeta, thumb_src: Pat
         return False
 
 
-def delete_asset_files(download_root: str, rel: str) -> None:
+def delete_asset_files(files_root: str, rel: str) -> None:
     """Remove the media file and its sidecars; prune now-empty parent dirs (best effort)."""
-    media = abs_path(download_root, rel)
+    media = abs_path(files_root, rel)
     for p in (media, _sidecar(media, ".nfo"), _sidecar(media, ".jpg")):
         try:
             p.unlink()
@@ -335,7 +338,7 @@ def delete_asset_files(download_root: str, rel: str) -> None:
                 extra=logs.event("download.asset.file_unlink_failed", reason=type(exc).__name__),
             )
     # Prune empty Season/Channel dirs up to (but not including) the root.
-    root = Path(download_root).resolve()
+    root = Path(files_root).resolve()
     parent = media.parent
     while parent != root and root in parent.parents:
         try:
diff --git a/backend/app/downloads/upload.py b/backend/app/downloads/upload.py
index 7612c4d3..c22ec9c6 100644
--- a/backend/app/downloads/upload.py
+++ b/backend/app/downloads/upload.py
@@ -33,7 +33,7 @@ from sqlalchemy.orm import Session
 
 from app.config import settings
 from app.downloads import edit as editmod
-from app.downloads import mediakind, playable, probe, quota, service, storage
+from app.downloads import folders, mediakind, playable, probe, quota, service, storage
 from app.downloads.status import JobKind, JobStatus
 from app.models import DownloadJob, MediaAsset, display_text
 
@@ -81,14 +81,14 @@ def ingest_upload(
     db.flush()  # need asset.id for the on-disk path
 
     rel = storage.upload_rel_path(user_id, asset.id, filename)
-    storage.place_file(src_path, settings.download_root, rel)
+    storage.place_file(src_path, settings.files_root, rel)
     # From here the file is on disk under `.uploads/`. If the transaction fails before it commits, the
     # asset/job rows roll back and nothing names the placed file: the per-user delete path keys off
     # asset.rel_path (which is gone), and the GC's `.uploads` sweep reaches it only an hour later. So
     # roll back the on-disk side too — unless the commit itself raised after it landed (see below).
     asset_id = asset.id
     try:
-        final = storage.abs_path(settings.download_root, rel)
+        final = storage.abs_path(settings.files_root, rel)
         actual_size = final.stat().st_size if final.exists() else size
 
         asset.status = "pending" if convert else "ready"
@@ -108,7 +108,7 @@ def ingest_upload(
             # A frame poster gives the library row a thumbnail + the share link an og:image. Audio has
             # no frame and a generic file no player, so neither gets one. Best-effort (ffmpeg decodes
             # server side even for a codec the browser can't play).
-            asset.poster_path = editmod.ensure_poster(asset, settings.download_root)
+            asset.poster_path = editmod.ensure_poster(asset, settings.files_root)
 
         job = DownloadJob(
             user_id=user_id,
@@ -123,6 +123,9 @@ def ingest_upload(
             queue_pos=service.next_queue_pos(db),
         )
         db.add(job)
+        folders.place(db, job)
+        if not convert:
+            folders.stamp_expiry(db, job)
         service.incr_ref_count(db, asset.id)
         db.commit()
     except Exception:
@@ -139,7 +142,7 @@ def ingest_upload(
         # return would skip that orphan. unlink is missing_ok, so the poster path is a harmless no-op
         # when none was generated (audio/file, or a video whose poster never started).
         for leaked in (rel, str(Path(rel).with_suffix(".jpg"))):
-            storage.abs_path(settings.download_root, leaked).unlink(missing_ok=True)
+            storage.abs_path(settings.files_root, leaked).unlink(missing_ok=True)
         raise
     db.refresh(job)
     return job
diff --git a/backend/app/models.py b/backend/app/models.py
index ec4c780c..45d57e29 100644
--- a/backend/app/models.py
+++ b/backend/app/models.py
@@ -18,6 +18,7 @@ from sqlalchemy import (
     Text,
     UniqueConstraint,
     func,
+    text,
 )
 from sqlalchemy.dialects.postgresql import JSONB, TSVECTOR
 from sqlalchemy.orm import Mapped, mapped_column, relationship, validates
@@ -826,12 +827,13 @@ class MediaAsset(Base, TimestampMixin):
     links to this ready asset instantly — no re-download, no cost. Only the SOURCE download is
     shared; per-user edited derivatives (phase 2) are never cached here.
 
-    `rel_path` is relative to `settings.download_root` (the mount), so the physical filename is
+    `rel_path` is relative to `settings.files_root` (the mount), so the physical filename is
     system-decided and bound to the source id — never renamed (the user's custom name is display
     metadata on `download_job`, applied only when downloading to their own machine). The naming
     fields (`title`/`uploader`/`upload_date`) are denormalized so the Plex-tree path can be built
-    uniformly for both catalog videos and ad-hoc URLs. `ref_count` (active jobs pointing here)
-    plus `last_access_at`/`expires_at` drive retention + LRU eviction in the GC job."""
+    uniformly for both catalog videos and ad-hoc URLs. `ref_count` (active jobs pointing here) frees
+    the file with its last holder, and `last_access_at` orders the GC's LRU eviction. How long a file
+    stays is not the asset's: each holder's job carries its own `expires_at` (its library folder's)."""
 
     __tablename__ = "media_assets"
     __table_args__ = (
@@ -847,10 +849,10 @@ class MediaAsset(Base, TimestampMixin):
     status: Mapped[str] = mapped_column(
         String(16), default="pending", server_default="pending", index=True
     )  # "pending" | "ready" | "error"
-    rel_path: Mapped[str | None] = mapped_column(Text)  # relative to download_root
+    rel_path: Mapped[str | None] = mapped_column(Text)  # relative to files_root
     storyboard_path: Mapped[str | None] = mapped_column(Text)  # phase-2 filmstrip mosaic
     # A locally-generated poster frame (ffmpeg), set only when the source had NO thumbnail so the
-    # card / watch page / link preview can still show an image. Relative to download_root.
+    # card / watch page / link preview can still show an image. Relative to files_root.
     poster_path: Mapped[str | None] = mapped_column(Text)
     size_bytes: Mapped[int | None] = mapped_column(BigInteger)
     duration_s: Mapped[int | None] = mapped_column(Integer)
@@ -866,8 +868,6 @@ class MediaAsset(Base, TimestampMixin):
     # earlier ad-hoc `vcodec is None and acodec` check lived only in links.py; consolidated here).
     media_kind: Mapped[str | None] = mapped_column(String(16))
     nfo_written: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
-    # Set once the GC has warned holders of the upcoming expiry, so it doesn't warn every run.
-    gc_notified: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
     # Denormalized naming/metadata for the Plex tree + ad-hoc display.
     title: Mapped[str | None] = mapped_column(Text)
     uploader: Mapped[str | None] = mapped_column(String(255))
@@ -883,7 +883,6 @@ class MediaAsset(Base, TimestampMixin):
     error: Mapped[str | None] = mapped_column(Text)
     ref_count: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
     last_access_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
-    expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
 
 
 class DownloadProfile(Base, TimestampMixin):
@@ -907,6 +906,40 @@ class DownloadProfile(Base, TimestampMixin):
     sort_order: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
 
 
+class FileFolder(Base, TimestampMixin):
+    """A folder of the Files library (`app.downloads.folders`): a row, never a directory.
+
+    A SOURCE folder (`source_key` set: youtube / plex / upload / site:<host>) is made the first time a
+    user gets a file from that origin; an OWN folder (`name` set) is one the user made. Every file —
+    a `DownloadJob` — sits in exactly one (`DownloadJob.folder_id`). `retention_days` is how long a
+    file stays once done (NULL = until deleted); a file's own `expires_at` is stamped from it."""
+
+    __tablename__ = "file_folders"
+    __table_args__ = (
+        UniqueConstraint("user_id", "source_key", name="uq_file_folder_source"),
+        Index(
+            "uq_file_folder_own_name",
+            "user_id",
+            func.lower(text("name")),
+            unique=True,
+            postgresql_where=text("name IS NOT NULL"),
+        ),
+        CheckConstraint(
+            "(kind = 'source' AND source_key IS NOT NULL AND name IS NULL)"
+            " OR (kind = 'own' AND name IS NOT NULL AND source_key IS NULL)",
+            name="ck_file_folder_kind",
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
+    kind: Mapped[str] = mapped_column(String(16))  # folders.FolderKind
+    # "site:" + a host name (DNS caps one at 253 characters).
+    source_key: Mapped[str | None] = mapped_column(String(260))
+    name: Mapped[str | None] = mapped_column(String(DISPLAY_TEXT_MAX))
+    retention_days: Mapped[int | None] = mapped_column(Integer)
+
+
 class DownloadJob(Base, TimestampMixin, UpdatedAtMixin):
     """A user's request to download one source — the per-user layer over `MediaAsset`.
 
@@ -974,6 +1007,16 @@ class DownloadJob(Base, TimestampMixin, UpdatedAtMixin):
     # The host-specific failure-catalog entry on top of the category (`facebook.login_wall`), or NULL.
     error_detail: Mapped[str | None] = mapped_column(String(48))  # see downloads.errors.diagnose
     queue_pos: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
+    # The Files library: the folder this file sits in (`downloads.folders.place` at creation; an own
+    # folder is emptied back into the source folders before it is deleted, hence RESTRICT), whether
+    # the user keeps it past its folder's retention, and when it goes (stamped once done; NULL =
+    # never). `expiry_warned` marks the one warning the GC sends before it deletes the file.
+    folder_id: Mapped[int | None] = mapped_column(
+        ForeignKey("file_folders.id", ondelete="RESTRICT"), index=True
+    )
+    kept: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
+    expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), index=True)
+    expiry_warned: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
 
 
 class DownloadShare(Base, TimestampMixin):
diff --git a/backend/app/plex/cache.py b/backend/app/plex/cache.py
index b1aec046..1836f713 100644
--- a/backend/app/plex/cache.py
+++ b/backend/app/plex/cache.py
@@ -10,4 +10,4 @@ from pathlib import Path
 
 from app.config import settings
 
-CACHE_DIR = Path(settings.download_root) / ".plex-img-cache"
+CACHE_DIR = Path(settings.files_root) / ".plex-img-cache"
diff --git a/backend/app/plex/stream.py b/backend/app/plex/stream.py
index 1039f6af..4616bae0 100644
--- a/backend/app/plex/stream.py
+++ b/backend/app/plex/stream.py
@@ -41,7 +41,7 @@ log = logs.get_logger(__name__)
 _HLS_ROOT = (
     Path(settings.plex_hls_dir)
     if settings.plex_hls_dir
-    else Path(settings.download_root) / ".plex-hls"
+    else Path(settings.files_root) / ".plex-hls"
 )
 _SEG_SECONDS = 6
 # Cold-start warm-up. A freshly-spawned ffmpeg has produced only the first segment (or two) by the
diff --git a/backend/app/routes/downloads.py b/backend/app/routes/downloads.py
index 3e3af777..ff9acc32 100644
--- a/backend/app/routes/downloads.py
+++ b/backend/app/routes/downloads.py
@@ -145,6 +145,10 @@ def _serialize(job: DownloadJob, asset: MediaAsset | None) -> dict:
         "edit_spec": job.edit_spec,
         "can_download": ready and job.status == JobStatus.DONE,
         "expired": job.asset_id is None and job.status == JobStatus.DONE,
+        # The Files library: its folder, whether the user keeps it, and when it goes (NULL = never).
+        "folder_id": job.folder_id,
+        "kept": job.kept,
+        "expires_at": iso(job.expires_at),
         "asset": None
         if asset is None
         else {
@@ -163,7 +167,6 @@ def _serialize(job: DownloadJob, asset: MediaAsset | None) -> dict:
             "media_kind": asset.media_kind,
             "uploader_url": asset.uploader_url,
             "thumbnail_url": asset.thumbnail_url,
-            "expires_at": iso(asset.expires_at),
         },
     }
 
@@ -495,7 +498,7 @@ async def spool_upload(
     # already an int or a header in hand: touching an ORM attribute below would reload `user` and
     # check a connection straight back out.
     release_read_connection(db)
-    staging = Path(settings.download_root) / ".staging" / "uploads"
+    staging = Path(settings.files_root) / ".staging" / "uploads"
     staging.mkdir(parents=True, exist_ok=True)
     fd, tmp = tempfile.mkstemp(dir=str(staging))
     tmp_path = Path(tmp)
@@ -871,7 +874,7 @@ def download_file(
     asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
     if asset is None or asset.status != "ready" or not asset.rel_path:
         raise refused(409, "download_not_ready")
-    path = storage.safe_abs_path(settings.download_root, asset.rel_path)
+    path = storage.safe_abs_path(settings.files_root, asset.rel_path)
     if path is None:
         raise not_found("file")
 
@@ -946,7 +949,7 @@ def storyboard_meta(
     asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
     if asset is None or asset.status != "ready" or not asset.rel_path:
         raise refused(409, "download_not_ready")
-    meta = editmod.ensure_storyboard(asset, settings.download_root)
+    meta = editmod.ensure_storyboard(asset, settings.files_root)
     if meta is None:
         return {"available": False}
     db.commit()  # persist storyboard_path if it was just generated
@@ -967,7 +970,7 @@ def poster_image(job_id: int, user: User = Depends(current_user), db: Session =
     asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
     if asset is None or not asset.poster_path:
         raise not_found("poster")
-    path = storage.safe_abs_path(settings.download_root, asset.poster_path)
+    path = storage.safe_abs_path(settings.files_root, asset.poster_path)
     if path is None:
         raise not_found("poster")
     return FileResponse(path, media_type="image/jpeg")
@@ -981,7 +984,7 @@ def storyboard_image(
     asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
     if asset is None or not asset.storyboard_path:
         raise not_found("filmstrip")
-    path = storage.safe_abs_path(settings.download_root, asset.storyboard_path)
+    path = storage.safe_abs_path(settings.files_root, asset.storyboard_path)
     if path is None:
         raise not_found("filmstrip")
     return FileResponse(path, media_type="image/jpeg")
diff --git a/backend/app/routes/public.py b/backend/app/routes/public.py
index 42099a93..ddb6e8da 100644
--- a/backend/app/routes/public.py
+++ b/backend/app/routes/public.py
@@ -73,7 +73,7 @@ def _ready_target(db: Session, link: DownloadLink) -> tuple[DownloadJob, MediaAs
         raise refused(404, "video_gone")
     # Verify the file is actually on disk here (not just flagged ready), so the watch page never
     # loads a player for a missing file — meta and file agree.
-    path = storage.safe_abs_path(settings.download_root, asset.rel_path)
+    path = storage.safe_abs_path(settings.files_root, asset.rel_path)
     if path is None:
         raise refused(404, "video_gone")
     return job, asset
@@ -158,7 +158,7 @@ def watch_file(token: str, g: str | None = None, db: Session = Depends(get_db)):
     if link.password_hash and not linksmod.check_grant(token, g, link.password_version):
         raise refused(403, "share_password_required")
     asset = _ready_asset(db, link)
-    path = storage.safe_abs_path(settings.download_root, asset.rel_path)
+    path = storage.safe_abs_path(settings.files_root, asset.rel_path)
     if path is None:
         raise refused(404, "video_gone")
 
@@ -194,7 +194,7 @@ def watch_poster(token: str, g: str | None = None, db: Session = Depends(get_db)
     job, asset = _ready_target(db, link)
     if not asset.poster_path:
         raise not_found("poster")
-    path = storage.safe_abs_path(settings.download_root, asset.poster_path)
+    path = storage.safe_abs_path(settings.files_root, asset.poster_path)
     if path is None:
         raise not_found("poster")
     return FileResponse(path, media_type="image/jpeg")
diff --git a/backend/app/worker.py b/backend/app/worker.py
index dbff8e35..5f16d46b 100644
--- a/backend/app/worker.py
+++ b/backend/app/worker.py
@@ -715,7 +715,7 @@ def main() -> None:
     _target = max(1, settings.download_worker_concurrency)  # used only if the DB can't answer
     threads: dict[int, threading.Thread] = {}
     level = _sync_knobs(threads, None)
-    log.info("Download worker started (concurrency=%d, root=%s).", _target, settings.download_root)
+    log.info("Download worker started (concurrency=%d, root=%s).", _target, settings.files_root)
     while not _stop.wait(_KNOB_SECONDS):
         level = _sync_knobs(threads, level)
     log.info("Download worker stopping…")
diff --git a/backend/tests/test_clip.py b/backend/tests/test_clip.py
index 8566ca0d..9778e2cc 100644
--- a/backend/tests/test_clip.py
+++ b/backend/tests/test_clip.py
@@ -395,7 +395,7 @@ def test_a_clip_job_reaches_the_clip_entry(monkeypatch):
 def worker(db, tmp_path, monkeypatch):
     """The worker path with its external programs replaced: ffmpeg writes the file it was given, the
     probe answers what the test says, the phone-safe step and the poster pass through."""
-    monkeypatch.setattr(settings, "download_root", str(tmp_path / "root"))
+    monkeypatch.setattr(settings, "files_root", str(tmp_path / "root"))
     seen: dict = {}
 
     def ffmpeg(cmd, *_a, **_k):
diff --git a/backend/tests/test_download_failure_catalog.py b/backend/tests/test_download_failure_catalog.py
index ed585e64..10ea902d 100644
--- a/backend/tests/test_download_failure_catalog.py
+++ b/backend/tests/test_download_failure_catalog.py
@@ -80,7 +80,7 @@ def _asset_and_job(db, user, *, sig, asset_status, job_status, error=None):
 
 
 def _fail_the_child(monkeypatch, tmp_path, *, cookie: bool):
-    monkeypatch.setattr(settings, "download_root", str(tmp_path))
+    monkeypatch.setattr(settings, "files_root", str(tmp_path))
     monkeypatch.setattr(pipeline.urlguard, "check_url", lambda url: None)
     cookiefile = tmp_path / "cookies.txt"
     cookiefile.write_text("# Netscape HTTP Cookie File\n")
@@ -230,7 +230,7 @@ def test_a_download_yt_dlp_gave_no_duration_stores_the_files_duration(db, monkey
     # row lost its trim action. Through the real finalize, with a real 3-second file.
     u = _user(db, "fb-duration@example.com")
     a, j = _asset_and_job(db, u, sig="shared", asset_status="downloading", job_status="running")
-    monkeypatch.setattr(settings, "download_root", str(tmp_path))
+    monkeypatch.setattr(settings, "files_root", str(tmp_path))
     monkeypatch.setattr(pipeline.urlguard, "check_url", lambda url: None)
 
     def extract(job_id, asset_id, url, spec, staging, adapter, **kwargs):
@@ -350,7 +350,7 @@ def test_a_signed_in_facebook_download_is_titled_with_its_posts_text(db, monkeyp
     # file name and the share preview all said "Video". The post's text is in the description.
     u = _user(db, "fb-title@example.com")
     a, j = _asset_and_job(db, u, sig="shared", asset_status="downloading", job_status="running")
-    monkeypatch.setattr(settings, "download_root", str(tmp_path))
+    monkeypatch.setattr(settings, "files_root", str(tmp_path))
     monkeypatch.setattr(pipeline.urlguard, "check_url", lambda url: None)
 
     def extract(job_id, asset_id, url, spec, staging, adapter, **kwargs):
@@ -401,7 +401,7 @@ def test_an_mp3_download_is_stored_as_an_mp3(db, monkeypatch, tmp_path):
     # video/mp4, which iPad's WebKit will not play. It stays an .mp3 (served as audio/mpeg).
     u = _user(db, "fb-mp3@example.com")
     a, j = _asset_and_job(db, u, sig="shared", asset_status="downloading", job_status="running")
-    monkeypatch.setattr(settings, "download_root", str(tmp_path))
+    monkeypatch.setattr(settings, "files_root", str(tmp_path))
     monkeypatch.setattr(pipeline.urlguard, "check_url", lambda url: None)
 
     def extract(job_id, asset_id, url, spec, staging, adapter, **kwargs):
@@ -430,7 +430,7 @@ def test_a_video_download_is_still_made_browser_playable(db, monkeypatch, tmp_pa
     a, j = _asset_and_job(
         db, u, sig=f"mkv-{mode}", asset_status="downloading", job_status="running"
     )
-    monkeypatch.setattr(settings, "download_root", str(tmp_path))
+    monkeypatch.setattr(settings, "files_root", str(tmp_path))
     monkeypatch.setattr(pipeline.urlguard, "check_url", lambda url: None)
 
     def extract(job_id, asset_id, url, spec, staging, adapter, **kwargs):
diff --git a/backend/tests/test_download_limits.py b/backend/tests/test_download_limits.py
index d2dfdc32..5ddbb948 100644
--- a/backend/tests/test_download_limits.py
+++ b/backend/tests/test_download_limits.py
@@ -259,7 +259,7 @@ def test_a_refusal_for_one_user_does_not_poison_the_shared_asset(db, monkeypatch
     that shared row — correct for an upstream failure, wrong for "YOUR storage is full", which
     `worker._process_job` would then hand verbatim to someone with an empty library.
     """
-    monkeypatch.setattr(settings, "download_root", str(tmp_path))
+    monkeypatch.setattr(settings, "files_root", str(tmp_path))
     over, other = User(email="over@example.com"), User(email="other@example.com")
     db.add_all([over, other])
     db.flush()
@@ -306,7 +306,7 @@ def test_a_download_over_the_space_left_says_so_and_releases_the_shared_asset(
     # yt-dlp refuses a file over `max_filesize` by saying so and returning with no file (pinned in
     # test_formats). That is the owner's storage, not a processing fault: their refusal, in words
     # true of a cap with space still left, and the shared asset goes back for the next job.
-    monkeypatch.setattr(settings, "download_root", str(tmp_path))
+    monkeypatch.setattr(settings, "files_root", str(tmp_path))
     u = User(email="nearcap@example.com")
     db.add(u)
     db.flush()
@@ -352,7 +352,7 @@ def test_a_download_that_outgrows_the_space_left_unannounced_is_refused_and_not_
     # yt-dlp's `max_filesize` only stops a plain HTTP file that announces its size: YouTube's
     # fragmented formats never meet it (measured on localdev: 718 MB fetched with 1 MB left). The
     # file it produced is held to what is left by its real size, as a clip is.
-    monkeypatch.setattr(settings, "download_root", str(tmp_path))
+    monkeypatch.setattr(settings, "files_root", str(tmp_path))
     u = User(email="unannounced@example.com")
     db.add(u)
     db.flush()
diff --git a/backend/tests/test_edit_cap.py b/backend/tests/test_edit_cap.py
index 8f78c28d..fbc52efa 100644
--- a/backend/tests/test_edit_cap.py
+++ b/backend/tests/test_edit_cap.py
@@ -21,7 +21,7 @@ pytestmark = pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="ffmpeg n
 def test_an_editor_clip_larger_than_what_is_left_is_refused_and_not_placed(
     db, make_user, tmp_path, monkeypatch
 ):
-    monkeypatch.setattr(settings, "download_root", str(tmp_path))
+    monkeypatch.setattr(settings, "files_root", str(tmp_path))
     src = tmp_path / "in.mp4"
     subprocess.run(
         [
diff --git a/backend/tests/test_failure_lines.py b/backend/tests/test_failure_lines.py
index 431fc8d8..3ade4eee 100644
--- a/backend/tests/test_failure_lines.py
+++ b/backend/tests/test_failure_lines.py
@@ -258,7 +258,7 @@ def test_an_edit_that_crashes_is_said_once(sink, finalized, monkeypatch, tmp_pat
     """The line `_process_edit`'s handler used to log before `_fail_edit` logged it again."""
     (tmp_path / "src.mp4").write_bytes(b"x")
     monkeypatch.setattr(pipeline, "SessionLocal", _EditDb)
-    monkeypatch.setattr(pipeline.settings, "download_root", str(tmp_path))
+    monkeypatch.setattr(pipeline.settings, "files_root", str(tmp_path))
     monkeypatch.setattr(pipeline, "_edit_staging_dir", lambda asset_id: tmp_path / "staging")
 
     def odd_spec(*a, **k):
@@ -300,7 +300,7 @@ def test_a_failed_download_line_does_not_put_back_what_the_adapter_took_out(sink
         RuntimeError("dictionary changed size during iteration"),
         RecursionError("maximum recursion depth exceeded"),
         NotImplementedError(),
-        FileNotFoundError(2, "No such file", "/downloads/Some_Channel/Season_2024/A_Private_Title"),
+        FileNotFoundError(2, "No such file", "/files/Some_Channel/Season_2024/A_Private_Title"),
     ],
     ids=lambda e: type(e).__name__,
 )
@@ -499,11 +499,9 @@ def test_a_failed_poster_or_storyboard_names_no_file(sink, tmp_path):
 @pytest.mark.parametrize(
     "err",
     [
-        subprocess.TimeoutExpired(
-            ["ffmpeg", "-i", "/downloads/Some_Channel/A_Private_Title.mp4"], 30
-        ),
-        FileNotFoundError(2, "No such file", "/downloads/Some_Channel/A_Private_Title.jpg"),
-        subprocess.CalledProcessError(1, ["ffmpeg", "-i", "/downloads/Some_Channel/x.mp4"]),
+        subprocess.TimeoutExpired(["ffmpeg", "-i", "/files/Some_Channel/A_Private_Title.mp4"], 30),
+        FileNotFoundError(2, "No such file", "/files/Some_Channel/A_Private_Title.jpg"),
+        subprocess.CalledProcessError(1, ["ffmpeg", "-i", "/files/Some_Channel/x.mp4"]),
     ],
     ids=lambda e: type(e).__name__,
 )
@@ -523,9 +521,7 @@ def test_a_thumbnail_that_cannot_be_copied_names_no_file(sink, monkeypatch, tmp_
     src.write_bytes(b"\xff\xd8\xff rest")
 
     def refuse(a, b):
-        raise PermissionError(
-            13, "Permission denied", "/downloads/Some_Channel/A_Private_Title.jpg"
-        )
+        raise PermissionError(13, "Permission denied", "/files/Some_Channel/A_Private_Title.jpg")
 
     monkeypatch.setattr(editmod.shutil, "copyfile", refuse)
     assert editmod.thumbnail_to_jpg(src, tmp_path / "out.jpg") is None
diff --git a/backend/tests/test_file_folders.py b/backend/tests/test_file_folders.py
new file mode 100644
index 00000000..7bd6681e
--- /dev/null
+++ b/backend/tests/test_file_folders.py
@@ -0,0 +1,263 @@
+"""The Files library's folders and a file's retention: which folder a new file lands in, when a done
+file goes, and the GC that takes it — warning once, never a file its user kept, never a file another
+user still holds.
+
+Pure-logic for the source-key rule; DB-backed for the rest.
+"""
+
+import hashlib
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import func, select
+
+from app import sysconfig
+from app.config import settings
+from app.downloads import folders, formats, service
+from app.downloads import gc as gcmod
+from app.downloads.status import JobStatus
+from app.models import DownloadJob, FileFolder, MediaAsset, Notification
+
+NOW = datetime.now(timezone.utc)
+
+
+@pytest.mark.parametrize(
+    ("url", "key"),
+    [
+        ("https://www.tiktok.com/@a/video/1", "site:tiktok.com"),
+        ("https://m.facebook.com/groups/x/posts/2", "site:facebook.com"),
+        ("https://vimeo.com/123", "site:vimeo.com"),
+        ("https://youtu.be/abcdefghijk", "youtube"),
+        ("https://music.youtube.com/watch?v=abcdefghijk", "youtube"),
+        ("https://www.youtube.com/watch?v=abcdefghijk", "youtube"),
+        # A front door is only stripped when a real host remains behind it.
+        ("https://m.co/x", "site:m.co"),
+        ("not a url", "site:unknown"),
+    ],
+)
+def test_a_url_lands_in_its_sites_folder(url, key):
+    assert folders.site_key(url) == key
+
+
+@pytest.fixture
+def _root(tmp_path, monkeypatch):
+    monkeypatch.setattr(settings, "files_root", str(tmp_path))
+    return tmp_path
+
+
+def _asset(db, kind="youtube", ref="abcdefghijk", status="ready", rel="a.mp4", size=10):
+    asset = MediaAsset(
+        source_kind=kind,
+        source_ref=ref,
+        format_sig=hashlib.sha256(f"{kind}:{ref}".encode()).hexdigest()[:32],
+        status=status,
+        rel_path=rel,
+        size_bytes=size,
+        ref_count=0,
+        title="A title",
+    )
+    db.add(asset)
+    db.flush()
+    return asset
+
+
+def _job(db, user_id, asset, *, kind="youtube", ref=None, status=JobStatus.DONE, **kw):
+    job = DownloadJob(
+        user_id=user_id,
+        asset_id=asset.id if asset else None,
+        source_kind=kind,
+        source_ref=ref or (asset.source_ref if asset else "x"),
+        profile_snapshot={},
+        status=status,
+        **kw,
+    )
+    db.add(job)
+    db.flush()
+    folders.place(db, job)
+    if asset is not None:
+        service.incr_ref_count(db, asset.id)
+    db.commit()
+    return job
+
+
+def _folder(db, job) -> FileFolder:
+    return db.get(FileFolder, job.folder_id)
+
+
+def test_every_kind_of_job_lands_in_its_source_folder(db, make_user):
+    uid = make_user().id
+    yt = _job(db, uid, _asset(db))
+    site = _job(db, uid, _asset(db, "url", "https://www.tiktok.com/@a/video/1"), kind="url")
+    up = _job(db, uid, _asset(db, "upload", "u1"), kind="upload")
+    plex = _job(
+        db, uid, _asset(db, "clip", "c1"), kind="clip", edit_spec={"source": "plex", "ref": "9"}
+    )
+    ytclip = _job(
+        db, uid, _asset(db, "clip", "c2"), kind="clip", edit_spec={"source": "youtube", "ref": "v"}
+    )
+    # An edit of a Plex clip: only its parent knows it came from Plex (the clip's asset says "clip").
+    edit = _job(
+        db, uid, _asset(db, "edit", "e1"), kind="edit", source_job_id=plex.id,
+        source_asset_id=plex.asset_id,
+    )  # fmt: skip
+    keys = {j.id: _folder(db, j).source_key for j in (yt, site, up, plex, ytclip, edit)}
+    assert keys == {
+        yt.id: "youtube",
+        site.id: "site:tiktok.com",
+        up.id: "upload",
+        plex.id: "plex",
+        ytclip.id: "youtube",
+        edit.id: "plex",  # an edit sits with the file it was cut from
+    }
+    # One folder per origin per user: the two YouTube files share theirs.
+    assert _folder(db, yt).id == _folder(db, ytclip).id
+
+
+def test_an_edit_whose_parent_is_gone_follows_the_parents_asset(db, make_user):
+    uid = make_user().id
+    src = _asset(db, "url", "https://vimeo.com/1")
+    edit = DownloadJob(
+        user_id=uid, source_kind="edit", source_ref="e", profile_snapshot={},
+        source_job_id=None, source_asset_id=src.id, status=JobStatus.DONE,
+    )  # fmt: skip
+    assert folders.source_key_for(db, edit) == "site:vimeo.com"
+
+
+def test_folders_are_per_user(db, make_user):
+    a, b = make_user().id, make_user().id
+    asset = _asset(db)
+    ja, jb = _job(db, a, asset), _job(db, b, asset)
+    assert ja.folder_id != jb.folder_id
+    assert (_folder(db, ja).user_id, _folder(db, jb).user_id) == (a, b)
+
+
+def test_a_source_folder_keeps_files_the_retention_days_and_uploads_forever(db, make_user):
+    sysconfig.set_value(db, "download_retention_days", 12)
+    db.commit()
+    uid = make_user().id
+    yt = _job(db, uid, _asset(db))
+    up = _job(db, uid, _asset(db, "upload", "u1"), kind="upload")
+    assert _folder(db, yt).retention_days == 12
+    assert _folder(db, up).retention_days is None
+
+
+def test_enqueue_places_the_job_and_a_cache_hit_is_stamped_at_once(db, make_user):
+    uid = make_user().id
+    spec = formats.normalize({})
+    first = service.enqueue(db, uid, "youtube", "abcdefghijk", spec)
+    assert _folder(db, first).source_key == "youtube"
+    assert first.expires_at is None  # queued: its retention starts when it is done
+    asset = db.get(MediaAsset, first.asset_id)
+    asset.status, asset.rel_path = "ready", "a.mp4"
+    db.commit()
+    other = make_user().id
+    hit = service.enqueue(db, other, "youtube", "abcdefghijk", spec)
+    assert hit.status == JobStatus.DONE
+    days = _folder(db, hit).retention_days
+    assert hit.expires_at is not None
+    assert abs(hit.expires_at - (NOW + timedelta(days=days))) < timedelta(minutes=5)
+
+
+def test_a_finishing_worker_stamps_the_expiry_unless_kept(db, make_user):
+    uid = make_user().id
+    asset = _asset(db)
+    plain = _job(db, uid, asset, status=JobStatus.RUNNING)
+    kept = _job(db, uid, asset, status=JobStatus.RUNNING, kept=True)
+    for job in (plain, kept):
+        assert service.finalize_if_running(db, job.id, status=JobStatus.DONE, progress=100)
+    db.commit()
+    db.refresh(plain)
+    db.refresh(kept)
+    assert plain.expires_at is not None
+    assert kept.expires_at is None
+
+
+def _gc(db):
+    return gcmod.run_download_gc(db)
+
+
+def _notices(db, uid, kind):
+    return db.execute(
+        select(func.count())
+        .select_from(Notification)
+        .where(Notification.user_id == uid, Notification.type == kind)
+    ).scalar()
+
+
+def test_the_gc_warns_once_then_takes_the_file_out_of_the_library(db, make_user, _root):
+    (_root / "a.mp4").write_bytes(b"x")
+    uid = make_user().id
+    job = _job(db, uid, _asset(db), expires_at=NOW + timedelta(hours=12))
+    _gc(db)
+    _gc(db)
+    assert _notices(db, uid, "download_expiring") == 1
+    job.expires_at = NOW - timedelta(minutes=1)
+    db.commit()
+    assert _gc(db)["expired"] == 1
+    assert db.get(DownloadJob, job.id) is None
+    assert _notices(db, uid, "download_expired") == 1
+    # Its last holder gone, the shared file and its row go too.
+    assert not (_root / "a.mp4").exists()
+    assert db.execute(select(func.count()).select_from(MediaAsset)).scalar() == 0
+
+
+def test_the_gc_never_takes_a_kept_file(db, make_user, _root):
+    (_root / "a.mp4").write_bytes(b"x")
+    uid = make_user().id
+    job = _job(db, uid, _asset(db), kept=True, expires_at=NOW - timedelta(days=1))
+    assert _gc(db)["expired"] == 0
+    assert db.get(DownloadJob, job.id) is not None
+
+
+def test_one_users_expiry_leaves_another_users_copy(db, make_user, _root):
+    (_root / "a.mp4").write_bytes(b"x")
+    a, b = make_user().id, make_user().id
+    asset = _asset(db)
+    gone = _job(db, a, asset, expires_at=NOW - timedelta(minutes=1))
+    stays = _job(db, b, asset, expires_at=NOW + timedelta(days=20))
+    _gc(db)
+    assert db.get(DownloadJob, gone.id) is None
+    assert db.get(DownloadJob, stays.id) is not None
+    assert (_root / "a.mp4").exists()
+
+
+def test_the_cache_cap_never_evicts_a_kept_file_and_an_evicted_one_leaves_every_library(
+    db, make_user, _root
+):
+    sysconfig.set_value(db, "download_total_max_bytes", 15)
+    db.commit()
+    a, b = make_user().id, make_user().id
+    for name in ("old.mp4", "new.mp4"):
+        (_root / name).write_bytes(b"x")
+    old = _asset(db, ref="oldoldoldol", rel="old.mp4", size=10)
+    new = _asset(db, ref="newnewnewne", rel="new.mp4", size=10)
+    old.last_access_at = NOW - timedelta(days=9)
+    new.last_access_at = NOW - timedelta(days=1)
+    kept = _job(db, a, old, kept=True)  # the older file, but kept: the newer one goes instead
+    ja, jb = _job(db, a, new), _job(db, b, new)
+    assert _gc(db)["evicted"] == 1
+    assert db.get(DownloadJob, kept.id) is not None
+    assert db.get(DownloadJob, ja.id) is None and db.get(DownloadJob, jb.id) is None
+    assert _notices(db, a, "download_evicted") == 1
+    assert _notices(db, b, "download_evicted") == 1
+    assert (_root / "old.mp4").exists() and not (_root / "new.mp4").exists()
+
+
+def test_deleting_a_user_takes_their_folders_and_files_with_them(db, make_user):
+    # Both tables cascade from users; the files' folder FK refuses deleting a folder that still holds
+    # files, which must not stop the account's own deletion.
+    user = make_user()
+    _job(db, user.id, _asset(db))
+    db.delete(user)
+    db.commit()
+    assert db.execute(select(func.count()).select_from(FileFolder)).scalar() == 0
+    assert db.execute(select(func.count()).select_from(DownloadJob)).scalar() == 0
+
+
+def test_the_longest_host_has_room_for_its_folder(db, make_user):
+    label = "a" * 63
+    host = ".".join([label, label, label, "b" * 58]) + ".hu"  # 253 characters, the DNS maximum
+    assert len(host) == 253
+    uid = make_user().id
+    job = _job(db, uid, _asset(db, "url", f"https://{host}/v"), kind="url")
+    assert _folder(db, job).source_key == f"site:{host}"
diff --git a/backend/tests/test_migration_0102.py b/backend/tests/test_migration_0102.py
new file mode 100644
index 00000000..384bbcb7
--- /dev/null
+++ b/backend/tests/test_migration_0102.py
@@ -0,0 +1,73 @@
+"""Migration 0102's data step, DB-lane: every existing job lands in its source folder, and a done
+file's expiry moves from the shared asset to the user's job.
+
+The schema half is checked by test_migrations.py; a row placed in the wrong folder is invisible there.
+"""
+
+from alembic import command
+from alembic.config import Config
+from sqlalchemy import text
+
+from app.db import engine
+
+
+def test_existing_jobs_are_placed_and_keep_their_expiry(db, make_user):
+    alice, bob = make_user().id, make_user().id
+    cfg = Config("alembic.ini")
+    command.downgrade(cfg, "0101_plex_text_plain")
+    try:
+        with engine.begin() as conn:
+
+            def asset(kind, ref, expires=None):
+                return conn.execute(
+                    text(
+                        "INSERT INTO media_assets (source_kind, source_ref, format_sig, status,"
+                        " ref_count, expires_at) VALUES (:k, :r, :sig, 'ready', 1, :e)"
+                        " RETURNING id"
+                    ),
+                    {"k": kind, "r": ref, "sig": kind + ref, "e": expires},
+                ).scalar_one()
+
+            def job(user, kind, ref, asset_id, status="done", spec=None, parent=None):
+                return conn.execute(
+                    text(
+                        "INSERT INTO download_jobs (user_id, asset_id, source_kind, source_ref,"
+                        " profile_snapshot, status, edit_spec, source_job_id)"
+                        " VALUES (:u, :a, :k, :r, '{}', :s, CAST(:spec AS json), :p) RETURNING id"
+                    ),
+                    {"u": user, "a": asset_id, "k": kind, "r": ref, "s": status,
+                     "spec": spec, "p": parent},
+                ).scalar_one()  # fmt: skip
+
+            yt = asset("youtube", "abcdefghijk", "2030-01-02T00:00:00+00:00")
+            tt = asset("url", "https://www.tiktok.com/@a/video/1", "2030-01-03T00:00:00+00:00")
+            up = asset("upload", "u1")
+            cl = asset("clip", "c1", "2030-01-04T00:00:00+00:00")
+            j_yt = job(alice, "youtube", "abcdefghijk", yt)
+            j_yt_bob = job(bob, "youtube", "abcdefghijk", yt, status="queued")
+            j_tt = job(alice, "url", "https://www.tiktok.com/@a/video/1", tt)
+            j_up = job(alice, "upload", "u1", up)
+            j_cl = job(alice, "clip", "c1", cl, spec='{"source": "plex", "ref": "9"}')
+            j_ed = job(alice, "edit", "e1", cl, parent=j_tt)
+    finally:
+        command.upgrade(cfg, "head")
+
+    with engine.connect() as conn:
+        rows = conn.execute(
+            text(
+                "SELECT j.id, f.user_id, f.source_key, f.retention_days, j.expires_at::date"
+                " FROM download_jobs j JOIN file_folders f ON f.id = j.folder_id"
+            )
+        ).all()
+        folders = conn.execute(text("SELECT count(*) FROM file_folders")).scalar()
+    got = {r[0]: (r[1], r[2], r[3], str(r[4]) if r[4] else None) for r in rows}
+    assert got == {
+        j_yt: (alice, "youtube", 30, "2030-01-02"),
+        j_yt_bob: (bob, "youtube", 30, None),  # not done yet: no expiry until it is
+        j_tt: (alice, "site:tiktok.com", 30, "2030-01-03"),
+        j_up: (alice, "upload", None, None),  # an upload never expires
+        j_cl: (alice, "plex", 30, "2030-01-04"),
+        j_ed: (alice, "site:tiktok.com", 30, "2030-01-04"),
+    }
+    # alice: youtube, tiktok, upload, plex; bob: youtube.
+    assert folders == 5
diff --git a/backend/tests/test_odd_frame.py b/backend/tests/test_odd_frame.py
index 13f13561..78b1cffe 100644
--- a/backend/tests/test_odd_frame.py
+++ b/backend/tests/test_odd_frame.py
@@ -46,7 +46,7 @@ def test_an_odd_sized_4_4_4_video_converts_to_the_even_frame_below_it(tmp_path):
 
 
 def test_an_odd_crop_makes_a_clip_and_stores_the_frame_it_has(db, make_user, tmp_path, monkeypatch):
-    monkeypatch.setattr(settings, "download_root", str(tmp_path))
+    monkeypatch.setattr(settings, "files_root", str(tmp_path))
     src = tmp_path / "in.mp4"
     _ffmpeg(
         "-f", "lavfi", "-i", "testsrc2=size=160x120:rate=10:duration=2",
diff --git a/backend/tests/test_phone_safe_contract.py b/backend/tests/test_phone_safe_contract.py
index a15e1476..c40220fe 100644
--- a/backend/tests/test_phone_safe_contract.py
+++ b/backend/tests/test_phone_safe_contract.py
@@ -127,7 +127,7 @@ _FILES = {
 def media_root(tmp_path, monkeypatch):
     root = tmp_path / "media"
     root.mkdir()
-    monkeypatch.setattr(settings, "download_root", str(root))
+    monkeypatch.setattr(settings, "files_root", str(root))
     return root
 
 
diff --git a/backend/tests/test_public_unlock_limit.py b/backend/tests/test_public_unlock_limit.py
index d4fc12d6..f5603cbc 100644
--- a/backend/tests/test_public_unlock_limit.py
+++ b/backend/tests/test_public_unlock_limit.py
@@ -19,7 +19,7 @@ _PW = "open sesame"
 
 @pytest.fixture
 def unlock_url(as_user, make_user, tmp_path, monkeypatch, upload_headers):
-    monkeypatch.setattr(settings, "download_root", str(tmp_path))
+    monkeypatch.setattr(settings, "files_root", str(tmp_path))
     video = {"codec_type": "video", "codec_name": "h264", "pix_fmt": "yuv420p"}
     audio = {"codec_type": "audio", "codec_name": "aac"}
     data = {"streams": [video, audio], "format": {"format_name": "mov,mp4", "duration": "5"}}
diff --git a/backend/tests/test_spa_entries.py b/backend/tests/test_spa_entries.py
index 7fab4d24..95837020 100644
--- a/backend/tests/test_spa_entries.py
+++ b/backend/tests/test_spa_entries.py
@@ -27,7 +27,7 @@ def pages(tmp_path, monkeypatch):
 
 @pytest.fixture
 def link_token(as_user, make_user, tmp_path, monkeypatch, upload_headers):
-    monkeypatch.setattr(settings, "download_root", str(tmp_path / "media"))
+    monkeypatch.setattr(settings, "files_root", str(tmp_path / "media"))
     video = {"codec_type": "video", "codec_name": "h264", "pix_fmt": "yuv420p"}
     audio = {"codec_type": "audio", "codec_name": "aac"}
     data = {"streams": [video, audio], "format": {"format_name": "mov,mp4", "duration": "5"}}
diff --git a/backend/tests/test_storage.py b/backend/tests/test_storage.py
index 108c26bc..d679e5ff 100644
--- a/backend/tests/test_storage.py
+++ b/backend/tests/test_storage.py
@@ -100,7 +100,7 @@ class TestDownloadFilename:
 
 
 class TestSafeAbsPath:
-    """The path-traversal guard: only paths that resolve INSIDE download_root AND exist are returned."""
+    """The path-traversal guard: only paths that resolve INSIDE files_root AND exist are returned."""
 
     def test_returns_the_path_for_a_file_inside_the_root(self, tmp_path: Path):
         (tmp_path / "sub").mkdir()
diff --git a/backend/tests/test_upload.py b/backend/tests/test_upload.py
index a240e67f..4d551232 100644
--- a/backend/tests/test_upload.py
+++ b/backend/tests/test_upload.py
@@ -5,7 +5,7 @@ generic file (force-download so uploaded HTML/SVG can't execute inline), quota a
 size known up front, the per-file cap enforced while streaming, filename sanitization, and the two
 guards that keep an upload out of paths that assume a re-fetchable source (edit / redownload).
 
-DB-backed (real Postgres) — skipped when no database is reachable. `settings.download_root` is
+DB-backed (real Postgres) — skipped when no database is reachable. `settings.files_root` is
 pointed at a per-test tmp dir so ingest's on-disk moves are isolated. The ingest/route tests
 monkeypatch the shared `probe.probe_json` so they don't shell out to ffprobe; the classifier's OWN
 parsing (image → file, cover-art skip, dims, duration) is covered directly by `TestProbeClassifier`
@@ -51,7 +51,7 @@ def test_derive_media_kind():
 
 @pytest.fixture
 def _root(tmp_path, monkeypatch):
-    monkeypatch.setattr(settings, "download_root", str(tmp_path))
+    monkeypatch.setattr(settings, "files_root", str(tmp_path))
     return tmp_path
 
 
@@ -761,7 +761,7 @@ def test_a_refused_upload_leaves_nothing_in_the_staging_directory(
 
     sysconfig.set_value(db, "download_upload_max_bytes", 8)
     db.commit()
-    staging = Path(settings.download_root) / ".staging" / "uploads"
+    staging = Path(settings.files_root) / ".staging" / "uploads"
     c = as_user(make_user())
 
     r = c.post("/api/downloads/upload", content=b"x" * 4096, headers=upload_headers("leaky.bin"))
diff --git a/backend/tests/test_upload_conversion.py b/backend/tests/test_upload_conversion.py
index 8a8407f0..b4d84521 100644
--- a/backend/tests/test_upload_conversion.py
+++ b/backend/tests/test_upload_conversion.py
@@ -29,7 +29,7 @@ from sqlalchemy.orm import Session
 
 @pytest.fixture
 def _root(tmp_path, monkeypatch):
-    monkeypatch.setattr(settings, "download_root", str(tmp_path))
+    monkeypatch.setattr(settings, "files_root", str(tmp_path))
     return tmp_path
 
 
diff --git a/backend/tests/test_worker_dispatch.py b/backend/tests/test_worker_dispatch.py
index 8dc74b96..2593a306 100644
--- a/backend/tests/test_worker_dispatch.py
+++ b/backend/tests/test_worker_dispatch.py
@@ -32,7 +32,7 @@ def test_a_clip_the_phone_safe_step_changed_is_stored_under_its_new_extension(
 ):
     # A fast trim stream-copies its source's container; when the phone-safe step remuxes it (an
     # audio clip out of an .mp4 becomes an .m4a), the stored path and container follow the new file.
-    monkeypatch.setattr(settings, "download_root", str(tmp_path))
+    monkeypatch.setattr(settings, "files_root", str(tmp_path))
     data = {
         "streams": [{"codec_type": "audio", "codec_name": "aac"}],
         "format": {"format_name": "mov,mp4", "duration": "10"},
diff --git a/docker-compose.home.yml.example b/docker-compose.home.yml.example
index 69c40b84..fc30bfd8 100644
--- a/docker-compose.home.yml.example
+++ b/docker-compose.home.yml.example
@@ -17,7 +17,7 @@
 # TOKEN_ENCRYPTION_KEY (Fernet), GOOGLE_CLIENT_ID, GOOGLE_CLIENT_SECRET,
 # OAUTH_REDIRECT_URL=https://siftlode.example.com/auth/callback, ALLOWED_EMAILS, ADMIN_EMAILS,
 # YOUTUBE_API_KEY. APP_VERSION is exported by the deploy step (selects the image tag).
-# Optional: DOWNLOAD_HOST_PATH (host dir for the download tree) and PLEX_MEDIA_HOST_PATH (host dir
+# Optional: FILES_HOST_PATH (host dir for the download tree) and PLEX_MEDIA_HOST_PATH (host dir
 # holding the Plex library, mounted read-only for the optional Plex module).
 
 name: siftlode
@@ -73,7 +73,7 @@ services:
       SCHEDULER_ENABLED: "true"
       # Download center: the API serves finished files + builds filmstrips (read-only root is fine —
       # it only writes to the /downloads mount + tmpfs). The yt-dlp job loop runs in `worker` below.
-      DOWNLOAD_ROOT: /downloads
+      FILES_ROOT: /files
       WORKER_ENABLED: "false"
     depends_on:
       db:
@@ -84,9 +84,9 @@ services:
       # (10.x.x.x:8080) and on the trusted home LAN. Never forwarded to the public internet.
       - "8080:8000"
     volumes:
-      # Downloaded media. Defaults to a Docker-managed named volume; set DOWNLOAD_HOST_PATH in .env
+      # Downloaded media. Defaults to a Docker-managed named volume; set FILES_HOST_PATH in .env
       # to a host directory (e.g. one your Plex server reads) to keep the Plex-style tree there.
-      - ${DOWNLOAD_HOST_PATH:-siftlode_downloads}:/downloads
+      - ${FILES_HOST_PATH:-siftlode_files}:/files
       # Plex media, read-only. Only used when the optional Plex module is enabled: set
       # PLEX_MEDIA_HOST_PATH in .env to the host directory holding the Plex library (the local side
       # of plex_path_map, default /plex-media). Left unset → an empty named volume (harmless; the
@@ -124,7 +124,7 @@ services:
       DATABASE_URL: postgresql+psycopg://${POSTGRES_USER:-siftlode}:${POSTGRES_PASSWORD}@db:5432/${POSTGRES_DB:-siftlode}
       # The API owns the scheduler; the worker must not also run it.
       SCHEDULER_ENABLED: "false"
-      DOWNLOAD_ROOT: /downloads
+      FILES_ROOT: /files
       WORKER_ENABLED: "true"
     depends_on:
       db:
@@ -137,7 +137,7 @@ services:
         condition: service_started
     networks: [internal]
     volumes:
-      - ${DOWNLOAD_HOST_PATH:-siftlode_downloads}:/downloads
+      - ${FILES_HOST_PATH:-siftlode_files}:/files
       # Plex media (read-only) — same mount as the api; used by Plex clip jobs (a later phase).
       - ${PLEX_MEDIA_HOST_PATH:-siftlode_plex_media}:/plex-media:ro
     mem_limit: 2g
@@ -167,7 +167,7 @@ services:
 
 volumes:
   siftlode_pgdata:
-  siftlode_downloads:
+  siftlode_files:
   # Empty fallback for the Plex media mount when PLEX_MEDIA_HOST_PATH is unset (Plex module off).
   siftlode_plex_media:
 
diff --git a/docker-compose.localdev.yml b/docker-compose.localdev.yml
index 202ca5b8..c2be617e 100644
--- a/docker-compose.localdev.yml
+++ b/docker-compose.localdev.yml
@@ -21,21 +21,21 @@ services:
     environment:
       # This instance owns its scheduler (its own DB, so no double-write concern).
       SCHEDULER_ENABLED: "true"
-      # Plex HLS remux scratch on fast container-local storage (the /downloads bind-mount is a slow
+      # Plex HLS remux scratch on fast container-local storage (the /files bind-mount is a slow
       # Windows Docker Desktop mount in dev; prod uses fast local disk so it keeps the default).
       PLEX_HLS_DIR: /var/tmp/plex-hls
     security_opt:
       - apparmor:unconfined
     volumes:
       # Downloaded media lives on the host so you can inspect the generated Plex tree. Gitignored.
-      - ./downloads:/downloads
+      - ./files:/files
       - plex_media:/plex-media:ro
 
   worker:
     security_opt:
       - apparmor:unconfined
     volumes:
-      - ./downloads:/downloads
+      - ./files:/files
       - plex_media:/plex-media:ro
 
   bgutil-pot:
diff --git a/docker-compose.selfhost.yml b/docker-compose.selfhost.yml
index d3087d9d..2095000e 100644
--- a/docker-compose.selfhost.yml
+++ b/docker-compose.selfhost.yml
@@ -49,7 +49,7 @@ services:
       SCHEDULER_ENABLED: "true"
       # Download center: the API serves finished files + builds filmstrips; the worker (below) runs
       # the yt-dlp job loop. Read-only root is fine — it only writes to the /downloads mount + tmpfs.
-      DOWNLOAD_ROOT: /downloads
+      FILES_ROOT: /files
       WORKER_ENABLED: "false"
     depends_on:
       db:
@@ -60,9 +60,9 @@ services:
       # front for HTTPS / public exposure — and set OAUTH_REDIRECT_URL to the https URL.
       - "${HTTP_PORT:-8080}:8000"
     volumes:
-      # Downloaded media. Defaults to a Docker-managed named volume; set DOWNLOAD_HOST_PATH in .env
+      # Downloaded media. Defaults to a Docker-managed named volume; set FILES_HOST_PATH in .env
       # to a host directory (e.g. one your Plex server reads) to keep the Plex-style tree there.
-      - ${DOWNLOAD_HOST_PATH:-siftlode_downloads}:/downloads
+      - ${FILES_HOST_PATH:-siftlode_files}:/files
       # Plex library, read-only. The Plex module streams files itself, so it needs to SEE them: set
       # PLEX_MEDIA_HOST_PATH in .env to the host directory holding the library (the local side of the
       # admin "path map", whose default is /data=/plex-media). Left unset it falls back to an empty
@@ -98,7 +98,7 @@ services:
     environment:
       DATABASE_URL: postgresql+psycopg://${POSTGRES_USER:-siftlode}:${POSTGRES_PASSWORD}@db:5432/${POSTGRES_DB:-siftlode}
       SCHEDULER_ENABLED: "false"
-      DOWNLOAD_ROOT: /downloads
+      FILES_ROOT: /files
       WORKER_ENABLED: "true"
     depends_on:
       db:
@@ -111,7 +111,7 @@ services:
         condition: service_started
     networks: [internal]
     volumes:
-      - ${DOWNLOAD_HOST_PATH:-siftlode_downloads}:/downloads
+      - ${FILES_HOST_PATH:-siftlode_files}:/files
       # Same read-only Plex library as the API — the worker resolves library paths too.
       - ${PLEX_MEDIA_HOST_PATH:-siftlode_plex_media}:/plex-media:ro
     security_opt:
@@ -136,7 +136,7 @@ services:
 
 volumes:
   siftlode_pgdata:
-  siftlode_downloads:
+  siftlode_files:
   # Empty fallback for the Plex media mount when PLEX_MEDIA_HOST_PATH is unset (Plex module off).
   siftlode_plex_media:
 
diff --git a/docker-compose.yml b/docker-compose.yml
index fe760858..136fe5ca 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -37,7 +37,7 @@ services:
       DATABASE_URL: postgresql+psycopg://${POSTGRES_USER:-siftlode}:${POSTGRES_PASSWORD:-siftlode}@db:5432/${POSTGRES_DB:-siftlode}
       # Download center: the API serves finished files + builds filmstrips; the worker (below) runs
       # the yt-dlp job loop.
-      DOWNLOAD_ROOT: /downloads
+      FILES_ROOT: /files
       WORKER_ENABLED: "false"
     depends_on:
       db:
@@ -45,9 +45,9 @@ services:
     ports:
       - "${APP_PORT:-8080}:8000"
     volumes:
-      # Downloaded media. Defaults to a Docker-managed named volume; set DOWNLOAD_HOST_PATH in .env
+      # Downloaded media. Defaults to a Docker-managed named volume; set FILES_HOST_PATH in .env
       # to a host directory (e.g. one your Plex server reads) to keep the Plex-style tree there.
-      - ${DOWNLOAD_HOST_PATH:-downloads}:/downloads
+      - ${FILES_HOST_PATH:-files}:/files
       # Plex library, read-only. The Plex module streams files itself, so it needs to SEE them: set
       # PLEX_MEDIA_HOST_PATH in .env to the host directory holding the library (the local side of the
       # admin "path map", whose default is /data=/plex-media). Left unset it falls back to an empty
@@ -80,7 +80,7 @@ services:
     environment:
       DATABASE_URL: postgresql+psycopg://${POSTGRES_USER:-siftlode}:${POSTGRES_PASSWORD:-siftlode}@db:5432/${POSTGRES_DB:-siftlode}
       SCHEDULER_ENABLED: "false"
-      DOWNLOAD_ROOT: /downloads
+      FILES_ROOT: /files
       WORKER_ENABLED: "true"
     depends_on:
       db:
@@ -91,7 +91,7 @@ services:
       bgutil-pot:
         condition: service_started
     volumes:
-      - ${DOWNLOAD_HOST_PATH:-downloads}:/downloads
+      - ${FILES_HOST_PATH:-files}:/files
       # Same read-only Plex library as the API — the worker resolves library paths too.
       - ${PLEX_MEDIA_HOST_PATH:-plex_media}:/plex-media:ro
     logging: *logging
@@ -107,6 +107,6 @@ services:
 
 volumes:
   pgdata:
-  downloads:
+  files:
   # Empty fallback for the Plex media mount when PLEX_MEDIA_HOST_PATH is unset (Plex module off).
   plex_media:
diff --git a/docs/self-hosting.md b/docs/self-hosting.md
index b6265c73..1cb68b4d 100644
--- a/docs/self-hosting.md
+++ b/docs/self-hosting.md
@@ -105,12 +105,12 @@ them. It runs two extra containers that come up automatically — a `worker` (th
 loop) and a small `bgutil-pot` sidecar (mints YouTube tokens so downloads aren't bot-blocked). No
 configuration is required; per-user storage quotas are set on the admin **Downloads → System** page.
 
-By default the media lives in a Docker-managed volume (`siftlode_downloads`). To keep it somewhere
+By default the media lives in a Docker-managed volume (`siftlode_files`). To keep it somewhere
 you can reach from other apps — e.g. a folder your **Plex** server indexes — point it at a host
 directory by adding this to `.env` and re-running `up -d`:
 
 ```bash
-DOWNLOAD_HOST_PATH=/mnt/media/youtube
+FILES_HOST_PATH=/mnt/media/youtube
 ```
 
 The directory must be writable by the container user (uid `1000`): `sudo chown -R 1000:1000 <dir>`.
@@ -164,8 +164,8 @@ then cannot read its own secrets.
    token and every stored refresh token are encrypted with it *inside* the database. Restore a dump
    against a different key and those values are lost — you would have to clear the affected rows from
    `system_config` and re-enter every secret by hand. Keep `.env` beside the dump, and just as private.
-3. **Downloaded media** *(only if you use the Download Center)* — the `siftlode_downloads` volume, or
-   whatever `DOWNLOAD_HOST_PATH` points at. Not needed to bring the instance back; the jobs and their
+3. **Downloaded media** *(only if you use the Download Center)* — the `siftlode_files` volume, or
+   whatever `FILES_HOST_PATH` points at. Not needed to bring the instance back; the jobs and their
    metadata are in the database, the files themselves are not.
 
 Dump and restore the database with these two commands. They also exist as scripts
@@ -254,6 +254,18 @@ Then `pull` + `up -d` stays on that version until you change the line. Released
 changed in them are listed in the app itself, under **Release Notes** in the account menu, and in
 [`CHANGELOG.md`](../CHANGELOG.md).
 
+**The media mount was renamed** (the Files library): `DOWNLOAD_HOST_PATH` is now `FILES_HOST_PATH`,
+mounted at `/files`, and the default volume is `siftlode_files` instead of `siftlode_downloads`. If you
+set a host directory, rename the variable in `.env`. If you used the default volume, copy it over once
+before you start the new version (Docker cannot rename a volume), then remove the old one when the
+library looks right:
+
+```bash
+docker compose -f docker-compose.selfhost.yml down
+docker volume create siftlode_files
+docker run --rm -v siftlode_downloads:/from -v siftlode_files:/to alpine cp -a /from/. /to/
+```
+
 **The registry keeps only the newest few releases, not every version the changelog lists.** A pin is
 "decide when you take the next hop", not "stay here forever": once you are several releases behind,
 your pinned tag can no longer be pulled — which matters the day you rebuild the host, prune your
diff --git a/frontend/src/components/DownloadCenter.confirm.dom.test.tsx b/frontend/src/components/DownloadCenter.confirm.dom.test.tsx
index 30833de2..f3ca2fc5 100644
--- a/frontend/src/components/DownloadCenter.confirm.dom.test.tsx
+++ b/frontend/src/components/DownloadCenter.confirm.dom.test.tsx
@@ -74,6 +74,9 @@ const job = (id: number, status: DownloadJob["status"]): DownloadJob => ({
   extra_links: [],
   can_download: status === "done",
   expired: false,
+  folder_id: null,
+  kept: false,
+  expires_at: null,
   asset: null,
 });
 
diff --git a/frontend/src/components/LibraryPlayer.dom.test.tsx b/frontend/src/components/LibraryPlayer.dom.test.tsx
index 1cae2383..17fbc086 100644
--- a/frontend/src/components/LibraryPlayer.dom.test.tsx
+++ b/frontend/src/components/LibraryPlayer.dom.test.tsx
@@ -54,6 +54,9 @@ const baseJob: DownloadJob = {
   extra_links: [],
   can_download: true,
   expired: false,
+  folder_id: null,
+  kept: false,
+  expires_at: null,
   asset: null,
 };
 
@@ -73,7 +76,6 @@ const withAsset = (over: Partial<NonNullable<DownloadJob["asset"]>>): DownloadJo
     container: null,
     media_kind: "video",
     thumbnail_url: null,
-    expires_at: null,
     ...over,
   },
 });
diff --git a/frontend/src/components/downloadRow.dom.test.tsx b/frontend/src/components/downloadRow.dom.test.tsx
index 39df6c0c..1acf0243 100644
--- a/frontend/src/components/downloadRow.dom.test.tsx
+++ b/frontend/src/components/downloadRow.dom.test.tsx
@@ -52,6 +52,9 @@ const baseJob: DownloadJob = {
   extra_links: [],
   can_download: true,
   expired: false,
+  folder_id: null,
+  kept: false,
+  expires_at: null,
   asset: null,
 };
 
@@ -74,7 +77,6 @@ const withAsset = (
     container,
     media_kind: "video",
     thumbnail_url: null,
-    expires_at: null,
     ...over,
   },
 });
diff --git a/frontend/src/lib/api/downloads.ts b/frontend/src/lib/api/downloads.ts
index 805bf043..35cb5607 100644
--- a/frontend/src/lib/api/downloads.ts
+++ b/frontend/src/lib/api/downloads.ts
@@ -40,7 +40,6 @@ interface DownloadAsset {
   // while still in flight (set at finalize). See backend downloads/mediakind.py.
   media_kind: "video" | "audio" | "file" | null;
   thumbnail_url: string | null;
-  expires_at: string | null;
 }
 
 export type DownloadStatus = "queued" | "running" | "paused" | "done" | "error" | "canceled";
@@ -107,6 +106,11 @@ export interface DownloadJob {
   edit_spec?: EditSpec | Omit<ClipRequest, "display_name"> | null;
   can_download: boolean;
   expired: boolean;
+  // The Files library (backend downloads/folders): its folder, whether the user keeps it past the
+  // folder's retention, and when it goes (null = never).
+  folder_id: number | null;
+  kept: boolean;
+  expires_at: string | null;
   asset: DownloadAsset | null;
   shared?: boolean;
   user_email?: string;
diff --git a/install.ps1 b/install.ps1
index c3069dee..cf160749 100644
--- a/install.ps1
+++ b/install.ps1
@@ -31,7 +31,7 @@ OAUTH_REDIRECT_URL=$url/auth/callback
 
 # Optional: store Download Center media in a host folder (e.g. one your Plex server reads) instead
 # of a Docker volume. Must be writable by uid 1000 (chown -R 1000:1000 <dir>).
-# DOWNLOAD_HOST_PATH=/mnt/media/youtube
+# FILES_HOST_PATH=/mnt/media/youtube
 "@ | Set-Content -Path .env -Encoding ascii
   Write-Host "Wrote .env (secrets generated)."
 }
diff --git a/install.sh b/install.sh
index c6ad65be..b275e9de 100755
--- a/install.sh
+++ b/install.sh
@@ -29,7 +29,7 @@ OAUTH_REDIRECT_URL=$URL/auth/callback
 
 # Optional: store Download Center media in a host folder (e.g. one your Plex server reads) instead
 # of a Docker volume. Must be writable by uid 1000 (sudo chown -R 1000:1000 <dir>).
-# DOWNLOAD_HOST_PATH=/mnt/media/youtube
+# FILES_HOST_PATH=/mnt/media/youtube
 EOF
   chmod 600 .env
   echo "Wrote .env (secrets generated)."