Video download history: permanent archive snapshotted at terminal status (phase 1)

video_downloads is a transient queue (hard-deleted on cleanup), so there was no record
of what SoulSync actually grabbed. Add a permanent video_download_history table +
capture: the monitor snapshots every terminal download (completed/import_failed/
cancelled/failed) into it, with rich metadata (title, year, S/E from search_ctx,
release, source, size, quality + parsed resolution/codec, dest path, poster, outcome,
timestamps). Idempotent per (download_id, outcome, dest_path).

DB methods: record_download_history, query_download_history (paged/kind/search),
download_history_detail, download_history_counts, latest_completed_download(media_type)
— the last is the probe target for the upcoming smart post-download scan. Schema v17.

core/video/download_monitor.py

@@ -235,6 +235,15 @@ def _apply_candidate(db, dl_id, row, cand, rest) -> bool:
     return True
 
 
+def _archive_history(db, dl, upd) -> None:
+    """Snapshot a terminal download into the permanent history. Best-effort — a
+    history failure must never disturb the download pipeline."""
+    try:
+        db.record_download_history({**dl, **upd})
+    except Exception:
+        logger.exception("video download %s: history snapshot failed", dl.get("id"))
+
+
 def _fail_or_retry(db, dl, error_msg) -> None:
     """A download just failed/disappeared. Try the next candidate inline; if none,
     hand off to a requery thread; if nothing left, mark it failed for real."""
@@ -246,8 +255,10 @@ def _fail_or_retry(db, dl, error_msg) -> None:
         db.update_video_download(dl["id"], status="searching", error=None)
         _spawn_requery(dl["id"])
         return
-    db.update_video_download(dl["id"], status="failed", error=error_msg or "Download failed",
+    err = error_msg or "Download failed"
-                             completed_at=_now())
+    completed = _now()
+    db.update_video_download(dl["id"], status="failed", error=err, completed_at=completed)
+    _archive_history(db, dl, {"status": "failed", "error": err, "completed_at": completed})
 
 
 def _search_for_retry(query, max_seconds=22):
@@ -378,6 +389,10 @@ def _tick(db) -> None:
             upd.setdefault("completed_at", _now())
         try:
             db.update_video_download(dl["id"], **upd)
+            # Snapshot terminal outcomes into the permanent history (survives the
+            # queue cleanup; powers the History modal + smart post-download scan).
+            if upd.get("status") in ("completed", "cancelled", "import_failed"):
+                _archive_history(db, dl, upd)
         except Exception:
             logger.exception("video download %s: failed to persist update", dl.get("id"))
     for k in [k for k in _misses if k not in live_ids]:

database/video_database.py

@@ -31,7 +31,7 @@ logger = get_logger("video_database")
 
 # Bump when video_schema.sql changes in a way worth recording. Stored in
 # PRAGMA user_version as a backstop indicator (nothing gates on it yet).
-SCHEMA_VERSION = 16
+SCHEMA_VERSION = 17
 
 _DEFAULT_DB_PATH = "database/video_library.db"
 _SCHEMA_FILE = Path(__file__).resolve().parent / "video_schema.sql"
@@ -1311,6 +1311,144 @@ class VideoDatabase:
         finally:
             conn.close()
 
+    # ── download history (permanent archive; survives the queue cleanup) ───────
+    @staticmethod
+    def _parse_resolution(*texts) -> str | None:
+        """Best-effort resolution/codec sniff from a release/file name."""
+        import re
+        blob = " ".join(t for t in texts if t)
+        for pat, label in ((r"\b2160p\b|\b4k\b|\buhd\b", "2160p"), (r"\b1080p\b", "1080p"),
+                           (r"\b720p\b", "720p"), (r"\b480p\b", "480p")):
+            if re.search(pat, blob, re.I):
+                return label
+        return None
+
+    @staticmethod
+    def _codec(*texts) -> str | None:
+        import re
+        blob = " ".join(t for t in texts if t)
+        for pat, label in ((r"\b[xh]?265\b|\bhevc\b", "x265"), (r"\b[xh]?264\b|\bavc\b", "x264"),
+                           (r"\bav1\b", "AV1"), (r"\bxvid\b", "XviD")):
+            if re.search(pat, blob, re.I):
+                return label
+        return None
+
+    def record_download_history(self, row: dict) -> int:
+        """Snapshot a finished download (the merged final ``video_downloads`` record)
+        into the permanent history. ``outcome`` is derived from its status. Idempotent
+        per (download_id, outcome, dest_path) so a re-persist / restart never dupes.
+        Returns the new history id, or 0 if it was a duplicate / not worth recording."""
+        if not row:
+            return 0
+        status = row.get("status") or "completed"
+        outcome = {"completed": "completed", "import_failed": "import_failed",
+                   "failed": "failed", "cancelled": "cancelled"}.get(status, status)
+        kind = row.get("kind") or "movie"
+        media_type = "show" if kind == "show" else ("movie" if kind == "movie" else kind)
+        # season/episode live in the retry search_ctx JSON for episode grabs.
+        sn = en = None
+        ctx = row.get("search_ctx")
+        if ctx:
+            try:
+                ctx = json.loads(ctx) if isinstance(ctx, str) else (ctx or {})
+                sn, en = ctx.get("season"), ctx.get("episode")
+            except (ValueError, TypeError):
+                pass
+        rel, fn = row.get("release_title"), row.get("filename")
+        conn = self._get_connection()
+        try:
+            cur = conn.execute(
+                """INSERT OR IGNORE INTO video_download_history
+                       (download_id, kind, media_type, title, year, season_number, episode_number,
+                        release_title, source, username, filename, dest_path, size_bytes,
+                        quality_label, resolution, video_codec, media_id, media_source, poster_url,
+                        outcome, error, grabbed_at, completed_at)
+                   VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
+                (row.get("id"), kind, media_type, row.get("title"), row.get("year"), sn, en,
+                 rel, row.get("source"), row.get("username"), fn, row.get("dest_path"),
+                 int(row.get("size_bytes") or 0), row.get("quality_label"),
+                 self._parse_resolution(rel, fn, row.get("quality_label")), self._codec(rel, fn),
+                 row.get("media_id"), row.get("media_source"), row.get("poster_url"),
+                 outcome, row.get("error"), row.get("created_at"),
+                 row.get("completed_at")))
+            conn.commit()
+            return cur.lastrowid or 0
+        except Exception:
+            logger.exception("record_download_history failed for download %s", row.get("id"))
+            conn.rollback()
+            return 0
+        finally:
+            conn.close()
+
+    def query_download_history(self, *, kind=None, search=None, outcome=None,
+                               page=1, limit=40) -> dict:
+        """Paged history slice for the modal. ``kind`` ∈ movie|show (None=all);
+        ``outcome`` filters by result. Newest first. {items, pagination}."""
+        try:
+            page = max(1, int(page or 1)); limit = max(1, min(200, int(limit or 40)))
+        except (TypeError, ValueError):
+            page, limit = 1, 40
+        where, args = ["1=1"], []
+        if kind in ("movie", "show"):
+            where.append("kind = ?"); args.append(kind)
+        if outcome:
+            where.append("outcome = ?"); args.append(outcome)
+        s = (search or "").strip()
+        if s:
+            where.append("(title LIKE ? OR release_title LIKE ?) COLLATE NOCASE")
+            args += ["%" + s + "%", "%" + s + "%"]
+        wsql = " WHERE " + " AND ".join(where)
+        conn = self._get_connection()
+        try:
+            total = conn.execute("SELECT COUNT(*) c FROM video_download_history" + wsql, args).fetchone()["c"]
+            rows = conn.execute(
+                "SELECT * FROM video_download_history" + wsql +
+                " ORDER BY COALESCE(completed_at, created_at) DESC, id DESC LIMIT ? OFFSET ?",
+                args + [limit, (page - 1) * limit]).fetchall()
+            items = [dict(r) for r in rows]
+        finally:
+            conn.close()
+        total_pages = max(1, (total + limit - 1) // limit)
+        return {"items": items, "pagination": {
+            "page": min(page, total_pages), "total_pages": total_pages, "total_count": total,
+            "has_prev": page > 1, "has_next": page < total_pages}}
+
+    def download_history_detail(self, history_id: int) -> dict | None:
+        conn = self._get_connection()
+        try:
+            row = conn.execute("SELECT * FROM video_download_history WHERE id=?", (int(history_id),)).fetchone()
+            return dict(row) if row else None
+        finally:
+            conn.close()
+
+    def download_history_counts(self) -> dict:
+        """{movie, show, total} of completed grabs (for the modal tabs/badge)."""
+        conn = self._get_connection()
+        try:
+            rows = conn.execute(
+                "SELECT kind, COUNT(*) c FROM video_download_history "
+                "WHERE outcome='completed' GROUP BY kind").fetchall()
+            by = {r["kind"]: r["c"] for r in rows}
+            movie, show = by.get("movie", 0), by.get("show", 0)
+            return {"movie": movie, "show": show, "total": movie + show}
+        finally:
+            conn.close()
+
+    def latest_completed_download(self, media_type: str = "all") -> dict | None:
+        """The most recently completed grab of a type — the probe target for the smart
+        post-download scan. ``media_type`` ∈ movie|show|all."""
+        where, args = ["outcome = 'completed'"], []
+        if media_type in ("movie", "show"):
+            where.append("kind = ?"); args.append(media_type)
+        conn = self._get_connection()
+        try:
+            row = conn.execute(
+                "SELECT * FROM video_download_history WHERE " + " AND ".join(where) +
+                " ORDER BY COALESCE(completed_at, created_at) DESC, id DESC LIMIT 1", args).fetchone()
+            return dict(row) if row else None
+        finally:
+            conn.close()
+
     # ── library mapping (which server library is Movies / TV) ─────────────────
     def get_library_selection(self, server: str) -> dict:
         return {

database/video_schema.sql

@@ -549,3 +549,41 @@ CREATE TABLE IF NOT EXISTS video_downloads (
     completed_at  TEXT
 );
 CREATE INDEX IF NOT EXISTS idx_video_downloads_status ON video_downloads(status);
+
+-- video_download_history — a PERMANENT record of every grab SoulSync completed
+-- (movies + episodes). video_downloads is the transient working queue (cleaned when
+-- finished); this is the archive that powers the Download History modal AND the
+-- smart post-download scan (newest completed item per library → probe the server).
+CREATE TABLE IF NOT EXISTS video_download_history (
+    id             INTEGER PRIMARY KEY AUTOINCREMENT,
+    download_id    INTEGER,                       -- original video_downloads.id (transient)
+    kind           TEXT NOT NULL,                 -- movie | show
+    media_type     TEXT,                          -- movie | show (normalized; the scan scope)
+    title          TEXT,                          -- movie/show title
+    year           INTEGER,
+    season_number  INTEGER,                       -- episodes only
+    episode_number INTEGER,
+    episode_title  TEXT,
+    release_title  TEXT,                          -- the release/file grabbed
+    source         TEXT,                          -- soulseek | torrent | usenet
+    username       TEXT,                          -- uploader (soulseek)
+    filename       TEXT,                          -- remote filename grabbed
+    dest_path      TEXT,                          -- final placed path
+    size_bytes     INTEGER DEFAULT 0,
+    quality_label  TEXT,                          -- e.g. "1080p", "2160p HDR"
+    resolution     TEXT,                          -- parsed from the release name, best-effort
+    video_codec    TEXT,
+    media_id       TEXT,                          -- movie/show id for the detail deep-link
+    media_source   TEXT,                          -- library | tmdb
+    poster_url     TEXT,
+    outcome        TEXT NOT NULL DEFAULT 'completed', -- completed | import_failed | failed | cancelled
+    error          TEXT,                          -- failure reason (non-completed)
+    grabbed_at     TEXT,                          -- when the download started
+    completed_at   TEXT,                          -- terminal time
+    created_at     TEXT DEFAULT (datetime('now'))
+);
+CREATE INDEX IF NOT EXISTS idx_vdl_history_kind ON video_download_history(kind);
+CREATE INDEX IF NOT EXISTS idx_vdl_history_completed ON video_download_history(completed_at);
+-- one history row per terminal download (idempotent re-persist / restart-safe)
+CREATE UNIQUE INDEX IF NOT EXISTS idx_vdl_history_dedup
+    ON video_download_history(download_id, outcome, dest_path);

tests/test_video_download_history.py

@@ -0,0 +1,99 @@
+"""Permanent video download history — the archive that powers the History modal
+and the smart post-download scan. video_downloads is the transient queue; this
+table survives the cleanup, so it's snapshotted at terminal status."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from database.video_database import VideoDatabase
+
+
+@pytest.fixture()
+def db(tmp_path):
+    return VideoDatabase(database_path=str(tmp_path / "video_library.db"))
+
+
+def _movie(**over):
+    row = {"id": 1, "kind": "movie", "title": "Dune", "year": 2024, "status": "completed",
+           "release_title": "Dune.2024.2160p.UHD.BluRay.x265-GRP", "source": "soulseek",
+           "username": "bob", "filename": "Dune.2024.2160p.x265.mkv",
+           "dest_path": "/movies/Dune (2024)/Dune (2024).mkv", "size_bytes": 9_000_000_000,
+           "quality_label": "2160p", "media_id": "55", "media_source": "library",
+           "poster_url": "/p/dune.jpg", "created_at": "2026-06-20 10:00:00",
+           "completed_at": "2026-06-20 10:30:00"}
+    row.update(over)
+    return row
+
+
+def _episode(**over):
+    row = {"id": 2, "kind": "show", "title": "Severance", "year": 2025, "status": "completed",
+           "release_title": "Severance.S02E05.1080p.WEB.h264", "source": "soulseek",
+           "dest_path": "/tv/Severance/Season 02/Severance - S02E05.mkv", "size_bytes": 2_000_000_000,
+           "search_ctx": json.dumps({"scope": "episode", "title": "Severance", "season": 2, "episode": 5}),
+           "media_id": "9", "media_source": "library", "completed_at": "2026-06-21 02:00:00"}
+    row.update(over)
+    return row
+
+
+def test_records_a_completed_movie_with_parsed_quality(db):
+    hid = db.record_download_history(_movie())
+    assert hid > 0
+    d = db.download_history_detail(hid)
+    assert d["title"] == "Dune" and d["outcome"] == "completed" and d["media_type"] == "movie"
+    assert d["resolution"] == "2160p" and d["video_codec"] == "x265"   # sniffed from the release name
+    assert d["size_bytes"] == 9_000_000_000 and d["dest_path"].endswith("Dune (2024).mkv")
+
+
+def test_records_an_episode_with_season_episode_from_search_ctx(db):
+    hid = db.record_download_history(_episode())
+    d = db.download_history_detail(hid)
+    assert (d["kind"], d["media_type"]) == ("show", "show")
+    assert d["season_number"] == 2 and d["episode_number"] == 5
+    assert d["resolution"] == "1080p" and d["video_codec"] == "x264"
+
+
+def test_history_is_idempotent_per_terminal_download(db):
+    first = db.record_download_history(_movie())
+    again = db.record_download_history(_movie())     # same download_id/outcome/dest_path
+    assert first > 0 and again == 0                  # INSERT OR IGNORE → no dupe
+    assert db.download_history_counts()["movie"] == 1
+
+
+def test_query_filters_by_kind_and_search(db):
+    db.record_download_history(_movie())
+    db.record_download_history(_episode())
+    assert db.query_download_history(kind="movie")["pagination"]["total_count"] == 1
+    assert db.query_download_history(kind="show")["items"][0]["title"] == "Severance"
+    hits = db.query_download_history(search="dune")["items"]
+    assert len(hits) == 1 and hits[0]["title"] == "Dune"
+
+
+def test_counts_only_count_completed(db):
+    db.record_download_history(_movie())
+    db.record_download_history(_movie(id=3, status="failed", dest_path=None,
+                                      error="no release found"))
+    c = db.download_history_counts()
+    assert c == {"movie": 1, "show": 0, "total": 1}   # the failed one isn't counted
+
+
+def test_latest_completed_download_is_the_probe_target(db):
+    db.record_download_history(_movie(id=1, completed_at="2026-06-20 10:30:00"))
+    db.record_download_history(_movie(id=4, title="Wicked",
+                                      dest_path="/movies/Wicked (2024)/Wicked.mkv",
+                                      completed_at="2026-06-22 09:00:00"))
+    db.record_download_history(_episode())
+    assert db.latest_completed_download("movie")["title"] == "Wicked"   # newest movie
+    assert db.latest_completed_download("show")["title"] == "Severance"
+    assert db.latest_completed_download("all")["title"] == "Wicked"     # newest overall
+
+
+def test_newest_first_ordering_in_the_feed(db):
+    db.record_download_history(_movie(id=1, title="Old", dest_path="/m/old.mkv",
+                                      completed_at="2026-01-01 00:00:00"))
+    db.record_download_history(_movie(id=2, title="New", dest_path="/m/new.mkv",
+                                      completed_at="2026-06-01 00:00:00"))
+    titles = [i["title"] for i in db.query_download_history()["items"]]
+    assert titles == ["New", "Old"]