diff --git a/.github/workflows/unit-tests.yml b/.github/workflows/unit-tests.yml
index 357769c12..75e6e3cdb 100644
--- a/.github/workflows/unit-tests.yml
+++ b/.github/workflows/unit-tests.yml
@@ -77,6 +77,7 @@ jobs:
             tests/test_curated_config.py \
             tests/test_check_oci_refs.py \
             tests/test_skill_docs.py \
+            tests/test_ffmpeg_thumbnailer.py \
             -v
 
       - name: Run pytest (Hidamari movie generator)
diff --git a/Justfile b/Justfile
index d748acba7..58e585457 100644
--- a/Justfile
+++ b/Justfile
@@ -5,7 +5,7 @@ just := just_executable()
 # reason; tests/test_suite_registration.bats enforces that.
 # Run unit tests (pytest for hooks.py, bats for shell scripts)
 test:
-    python3 -m pytest tests/test_hooks.py tests/test_check_oci_refs.py tests/test_curated_config.py tests/test_skill_docs.py tests/test_chairlift_config.py tests/test_renovate_config.py tests/test_render_hidamari_movie.py -v --cov=tests --cov-report=term-missing
+    python3 -m pytest tests/test_hooks.py tests/test_check_oci_refs.py tests/test_curated_config.py tests/test_skill_docs.py tests/test_chairlift_config.py tests/test_renovate_config.py tests/test_render_hidamari_movie.py tests/test_ffmpeg_thumbnailer.py -v --cov=tests --cov-report=term-missing
     bats tests/test_libsetup.bats
     bats tests/test_setup_scripts.bats
     bats tests/test_privileged_setup.bats
diff --git a/system_files/shared/usr/lib/environment.d/10-ffmpeg-thumbnailer.conf b/system_files/shared/usr/lib/environment.d/10-ffmpeg-thumbnailer.conf
new file mode 100644
index 000000000..d858f8a42
--- /dev/null
+++ b/system_files/shared/usr/lib/environment.d/10-ffmpeg-thumbnailer.conf
@@ -0,0 +1,5 @@
+# Prepend Bluefin custom thumbnailer data directory so ffmpeg-thumbnailer
+# takes precedence over /usr/share/thumbnailers/gst-video-thumbnailer.thumbnailer.
+# GNOME Desktop loads thumbnailers in XDG_DATA_DIRS order; the first registered
+# thumbnailer for each MIME type wins.
+XDG_DATA_DIRS=/usr/share/ublue-os/thumbnailer:${XDG_DATA_DIRS:-/usr/local/share:/usr/share}
diff --git a/system_files/shared/usr/lib/systemd/user/ffmpeg-thumbnailer-daemon.service b/system_files/shared/usr/lib/systemd/user/ffmpeg-thumbnailer-daemon.service
new file mode 100644
index 000000000..574d4ff1a
--- /dev/null
+++ b/system_files/shared/usr/lib/systemd/user/ffmpeg-thumbnailer-daemon.service
@@ -0,0 +1,23 @@
+[Unit]
+Description=FFmpeg Video Thumbnailer Host Daemon
+Documentation=https://github.com/projectbluefin/common
+After=graphical-session.target ffmpeg-thumbnailer.service ffmpeg-thumbnailer-nvidia.service
+PartOf=graphical-session.target
+
+[Service]
+Type=simple
+ExecStart=/usr/libexec/ffmpeg-thumbnailer/ffmpeg-thumbnailer-daemon
+Restart=always
+RestartSec=3s
+Slice=background.slice
+CPUWeight=20
+IOWeight=20
+MemoryHigh=512M
+MemoryMax=1024M
+TasksMax=512
+ManagedOOMMemoryPressure=kill
+Nice=10
+IOSchedulingClass=idle
+
+[Install]
+WantedBy=graphical-session.target default.target
diff --git a/system_files/shared/usr/libexec/ffmpeg-thumbnailer/ffmpeg-thumbnailer-daemon b/system_files/shared/usr/libexec/ffmpeg-thumbnailer/ffmpeg-thumbnailer-daemon
new file mode 100755
index 000000000..89389aa8f
--- /dev/null
+++ b/system_files/shared/usr/libexec/ffmpeg-thumbnailer/ffmpeg-thumbnailer-daemon
@@ -0,0 +1,270 @@
+#!/usr/bin/env python3
+"""
+Host-side daemon for ffmpeg-thumbnailer.
+
+Listens on a UNIX domain socket inside the GStreamer plugin cache directory
+(~/.cache/gnome-desktop-thumbnailer/gstreamer-1.0/ffmpeg-thumbnailer/sock),
+which is the one writable host path that GNOME's thumbnailer bubblewrap sandbox
+binds through at its real host location. The socket lives in a subdirectory
+because gnome-desktop's clean_gst_registry_dir() runs g_remove() on every
+top-level entry after each thumbnail; remove() cannot delete a non-empty
+directory, so the subdirectory and the socket inside it survive.
+
+The in-sandbox shim opens the input file and output thumbnail and passes their
+file descriptors over SCM_RIGHTS. This daemon invokes:
+    podman exec --preserve-fd=N,M <container_name> timeout -s KILL <sec> ffmpeg -threads 1 ...
+passing the descriptors directly to the container process.
+
+Resource-managed by systemd under background.slice with CPUWeight, IOWeight,
+MemoryHigh, MemoryMax, and TasksMax constraints.
+"""
+
+import os
+import selectors
+import signal
+import socket
+import socketserver
+import subprocess
+import sys
+import threading
+import time
+
+CONTAINER_NAME = os.environ.get("FFT_CONTAINER", "ffmpeg-thumbnailer")
+
+GST_CACHE_DIR = os.path.join(
+    os.environ.get("XDG_CACHE_HOME") or os.path.expanduser("~/.cache"),
+    "gnome-desktop-thumbnailer",
+    "gstreamer-1.0",
+)
+
+SOCK_PATH = os.environ.get(
+    "FFT_SOCKET",
+    os.path.join(GST_CACHE_DIR, "ffmpeg-thumbnailer", "sock"),
+)
+
+_default_inflight = max(1, min(4, (os.cpu_count() or 2) // 2))
+MAX_INFLIGHT = int(os.environ.get("FFT_MAX_INFLIGHT", str(_default_inflight)))
+FFMPEG_TIMEOUT = int(os.environ.get("FFT_TIMEOUT", "15"))
+HWACCEL = os.environ.get("FFT_HWACCEL", "none").strip().lower()
+
+_slots = threading.Semaphore(MAX_INFLIGHT)
+
+
+def log(msg: str) -> None:
+    sys.stderr.write(f"ffmpeg-thumbnailer-daemon: {msg}\n")
+    sys.stderr.flush()
+
+
+def run_container_ffmpeg(seek: str, size: int, in_fd: int, out_fd: int) -> bool:
+    vf = (
+        f"scale=min({size}\\,iw):min({size}\\,ih)"
+        ":force_original_aspect_ratio=decrease"
+    )
+    hwaccel_opts = []
+    if HWACCEL in ("vaapi", "cuda", "auto"):
+        hwaccel_opts = ["-hwaccel", HWACCEL]
+
+    # Enforce deadline inside the container with timeout(1), and pin threads to 1
+    # so concurrent decodes do not multiply core saturation.
+    cmd = [
+        "podman", "exec",
+        f"--preserve-fd={in_fd},{out_fd}",
+        CONTAINER_NAME,
+        "timeout", "-s", "KILL", str(FFMPEG_TIMEOUT),
+        "ffmpeg", "-y", "-loglevel", "error",
+        "-threads", "1",
+        *hwaccel_opts,
+        "-ss", seek,
+        "-i", f"/proc/self/fd/{in_fd}",
+        "-vf", vf,
+        "-frames:v", "1",
+        "-f", "image2",
+        "-c:v", "png",
+        "-update", "1",
+        f"/proc/self/fd/{out_fd}",
+    ]
+
+    try:
+        res = subprocess.run(
+            cmd,
+            timeout=FFMPEG_TIMEOUT + 5,
+            pass_fds=(in_fd, out_fd),
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.PIPE,
+        )
+        if res.returncode == 125:
+            # Podman exit 125 = container does not exist or not running
+            return "UNAVAILABLE"
+        return res.returncode == 0
+    except (subprocess.TimeoutExpired, FileNotFoundError, OSError) as exc:
+        log(f"podman exec failed ({exc})")
+        return False
+
+
+def make_thumbnail(size: int, in_fd: int, out_fd: int) -> str:
+    # Security: validate that in_fd and out_fd are regular files and out_fd is owned by the current user
+    try:
+        in_st = os.fstat(in_fd)
+        out_st = os.fstat(out_fd)
+        import stat
+        if not (stat.S_ISREG(in_st.st_mode) and stat.S_ISREG(out_st.st_mode)):
+            return "ERR: invalid file descriptor type"
+        if out_st.st_uid != os.getuid():
+            return "ERR: output file owner mismatch"
+    except OSError as err:
+        return f"ERR: fd fstat failed ({err})"
+
+    with _slots:
+        for seek in ("00:00:03", "00:00:00"):
+            try:
+                os.lseek(in_fd, 0, os.SEEK_SET)
+                os.ftruncate(out_fd, 0)
+                os.lseek(out_fd, 0, os.SEEK_SET)
+            except OSError as err:
+                return f"ERR: seek/truncate failed ({err})"
+
+            ret = run_container_ffmpeg(seek, size, in_fd, out_fd)
+            if ret == "UNAVAILABLE":
+                return "ERR_UNAVAILABLE"
+            if ret is True:
+                try:
+                    if os.fstat(out_fd).st_size > 0:
+                        return "OK"
+                except OSError:
+                    pass
+    return "ERR: ffmpeg produced no frame"
+
+
+def handle_request(data: bytes, fds: list[int]) -> str:
+    parts = data.decode("utf-8", errors="replace").strip().split("\t")
+    verb = parts[0]
+
+    if verb == "PING":
+        return "OK"
+
+    if verb != "THUMB" or len(parts) != 2:
+        return "ERR: invalid request"
+    if len(fds) != 2:
+        return "ERR: expected input and output descriptors"
+
+    try:
+        size = int(parts[1])
+    except ValueError:
+        size = 256
+    if size <= 0:
+        size = 256
+
+    return make_thumbnail(size, fds[0], fds[1])
+
+
+class Handler(socketserver.BaseRequestHandler):
+    def handle(self):
+        conn = self.request
+        conn.settimeout(FFMPEG_TIMEOUT * 2 + 5)
+        fds = []
+        try:
+            data, fds, _flags, _addr = socket.recv_fds(conn, 4096, 2)
+            if not data:
+                return
+            try:
+                status = handle_request(data, fds)
+            except Exception as exc:
+                log(f"request handler error: {exc!r}")
+                status = f"ERR: {exc}"
+            try:
+                conn.sendall(f"{status}\n".encode("utf-8"))
+            except OSError:
+                pass
+        finally:
+            for fd in fds:
+                try:
+                    os.close(fd)
+                except OSError:
+                    pass
+
+
+class DaemonServer(socketserver.ThreadingUnixStreamServer):
+    daemon_threads = True
+    request_queue_size = 64
+
+    def handle_error(self, request, client_address):
+        log(f"connection error: {sys.exc_info()[1]!r}")
+
+
+def serve_recovering(sock_path, handler, stop, poll_interval=0.5):
+    """Serve requests at sock_path, rebinding if the path is removed.
+
+    The socket's parent subdirectory survives gnome-desktop's
+    clean_gst_registry_dir(), but the path can still disappear (for example
+    when the user clears ~/.cache). The old listener would then keep serving
+    on a dead path and every new connection would fail. Detect the missing
+    path and rebind at the same path so the next connection succeeds without
+    a full daemon restart. Request handling runs in per-request threads
+    (ThreadingUnixStreamServer), so this poll loop never blocks on a long
+    request.
+    """
+    state = {"server": None}
+
+    def make_server():
+        os.makedirs(os.path.dirname(sock_path), mode=0o700, exist_ok=True)
+        # Remove any stale socket at the path before rebinding so the new
+        # listener owns a fresh inode.
+        if os.path.exists(sock_path):
+            try:
+                os.unlink(sock_path)
+            except OSError:
+                pass
+        server = DaemonServer(sock_path, handler)
+        os.chmod(sock_path, 0o600)
+        state["server"] = server
+
+    try:
+        while not stop.is_set():
+            if state["server"] is None or not os.path.exists(sock_path):
+                if state["server"] is not None:
+                    state["server"].server_close()
+                make_server()
+            sel = selectors.DefaultSelector()
+            sel.register(state["server"].socket, selectors.EVENT_READ)
+            if sel.select(poll_interval):
+                state["server"]._handle_request_noblock()
+    finally:
+        if state["server"] is not None:
+            state["server"].server_close()
+
+
+def main():
+    rundir = os.path.dirname(SOCK_PATH)
+    os.makedirs(rundir, exist_ok=True)
+
+    stop = threading.Event()
+
+    def on_term(signum, frame):
+        stop.set()
+        try:
+            os.unlink(SOCK_PATH)
+        except OSError:
+            pass
+        sys.exit(0)
+
+    signal.signal(signal.SIGTERM, on_term)
+    signal.signal(signal.SIGINT, on_term)
+
+    log(
+        f"listening on {SOCK_PATH} "
+        f"(container={CONTAINER_NAME}, max_inflight={MAX_INFLIGHT})"
+    )
+
+    try:
+        serve_recovering(SOCK_PATH, Handler, stop)
+    except KeyboardInterrupt:
+        pass
+    finally:
+        try:
+            os.unlink(SOCK_PATH)
+        except OSError:
+            pass
+
+
+if __name__ == "__main__":
+    main()
diff --git a/system_files/shared/usr/libexec/ffmpeg-thumbnailer/ffmpeg-video-thumbnailer-shim b/system_files/shared/usr/libexec/ffmpeg-thumbnailer/ffmpeg-video-thumbnailer-shim
new file mode 100755
index 000000000..7bbd77cc9
--- /dev/null
+++ b/system_files/shared/usr/libexec/ffmpeg-thumbnailer/ffmpeg-video-thumbnailer-shim
@@ -0,0 +1,146 @@
+#!/usr/bin/env python3
+"""
+ffmpeg-video-thumbnailer-shim
+
+Runs inside GNOME's thumbnailer bubblewrap sandbox, where no container runtime
+can start (--unshare-all plus seccomp filter denying unshare/mount/CLONE_NEWUSER).
+Hands the thumbnailing job to the ffmpeg-thumbnailer-daemon over a UNIX socket
+located in ~/.cache/gnome-desktop-thumbnailer/gstreamer-1.0/ffmpeg-thumbnailer/,
+inside the one writable host directory GNOME binds through at its real host path.
+
+Usage: ffmpeg-video-thumbnailer-shim [-s SIZE] <INPUT> <OUTPUT>
+"""
+
+import argparse
+import os
+import shutil
+import socket
+import subprocess
+import sys
+import urllib.parse
+
+
+def get_socket_path() -> str | None:
+    cache_home = os.environ.get("XDG_CACHE_HOME") or os.path.expanduser("~/.cache")
+    candidates = [
+        os.environ.get("FFT_SOCKET"),
+        os.path.join(
+            cache_home,
+            "gnome-desktop-thumbnailer",
+            "gstreamer-1.0",
+            "ffmpeg-thumbnailer",
+            "sock",
+        ),
+        os.path.join(os.environ.get("XDG_RUNTIME_DIR", ""), "ffmpeg-thumbnailer.sock"),
+        f"/run/user/{os.getuid()}/ffmpeg-thumbnailer.sock",
+    ]
+    for c in candidates:
+        if c and os.path.exists(c):
+            return c
+    return None
+
+
+def fallback_local_ffmpeg(input_path: str, output_path: str, size: int) -> bool:
+    ffmpeg_bin = shutil.which("ffmpeg")
+    if not ffmpeg_bin:
+        return False
+
+    vf = (
+        f"scale=min({size}\\,iw):min({size}\\,ih)"
+        ":force_original_aspect_ratio=decrease"
+    )
+    for seek in ("00:00:03", "00:00:00"):
+        cmd = [
+            ffmpeg_bin, "-y", "-loglevel", "error",
+            "-threads", "1",
+            "-ss", seek,
+            "-i", input_path,
+            "-vf", vf,
+            "-vframes", "1",
+            "-f", "image2",
+            "-c:v", "png",
+            "-update", "1",
+            output_path,
+        ]
+        try:
+            res = subprocess.run(
+                cmd,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+                timeout=15,
+            )
+            if res.returncode == 0 and os.path.exists(output_path) and os.path.getsize(output_path) > 0:
+                return True
+        except (subprocess.TimeoutExpired, OSError):
+            continue
+    return False
+
+
+def main():
+    parser = argparse.ArgumentParser(description="GNOME thumbnailer shim for ffmpeg-thumbnailer")
+    parser.add_argument("-s", "--size", type=int, default=256, help="Thumbnail max dimension")
+    parser.add_argument("input", help="Input video file or URI")
+    parser.add_argument("output", help="Output thumbnail path")
+    args = parser.parse_args()
+
+    input_path = args.input
+    if input_path.startswith("file://"):
+        parsed = urllib.parse.urlparse(input_path)
+        input_path = urllib.parse.unquote(parsed.path)
+
+    output_path = args.output
+    outdir = os.path.dirname(output_path)
+    if outdir:
+        os.makedirs(outdir, exist_ok=True)
+
+    sock_path = get_socket_path()
+    daemon_attempted = False
+
+    if sock_path:
+        in_fd = None
+        out_fd = None
+        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        sock.settimeout(60.0)
+        try:
+            in_fd = os.open(input_path, os.O_RDONLY)
+            out_fd = os.open(output_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o644)
+            sock.connect(sock_path)
+            daemon_attempted = True
+            socket.send_fds(sock, [f"THUMB\t{args.size}\n".encode("utf-8")], [in_fd, out_fd])
+            resp = sock.recv(4096).decode("utf-8", errors="replace").strip()
+            if resp == "OK" and os.path.exists(output_path) and os.path.getsize(output_path) > 0:
+                sys.exit(0)
+            sys.stderr.write(f"ffmpeg-thumbnailer: daemon returned '{resp}'\n")
+            if resp == "ERR_UNAVAILABLE":
+                # Container is pulling, offline, or starting up: allow local fallback
+                daemon_attempted = False
+            else:
+                # Video decode failed: do not fan out with local ffmpeg
+                sys.exit(1)
+        except (ConnectionRefusedError, FileNotFoundError):
+            # Daemon socket exists but daemon is not listening; fall through to local fallback
+            daemon_attempted = False
+        except Exception as exc:
+            sys.stderr.write(f"ffmpeg-thumbnailer: daemon communication failed: {exc}\n")
+            # If connect succeeded but communication timed out/failed, do not fan out
+            if daemon_attempted:
+                sys.exit(1)
+        finally:
+            for fd in (in_fd, out_fd):
+                if fd is not None:
+                    try:
+                        os.close(fd)
+                    except OSError:
+                        pass
+            sock.close()
+
+    # Local fallback only runs if daemon was completely unreachable
+    if not daemon_attempted and fallback_local_ffmpeg(input_path, output_path, args.size):
+        sys.exit(0)
+
+    sys.stderr.write(f"ffmpeg-thumbnailer: failed to thumbnail {input_path}\n")
+    sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/system_files/shared/usr/share/containers/systemd/users/ffmpeg-thumbnailer-nvidia.container b/system_files/shared/usr/share/containers/systemd/users/ffmpeg-thumbnailer-nvidia.container
new file mode 100644
index 000000000..aabaefe36
--- /dev/null
+++ b/system_files/shared/usr/share/containers/systemd/users/ffmpeg-thumbnailer-nvidia.container
@@ -0,0 +1,28 @@
+[Unit]
+Description=FFmpeg Video Thumbnailer Container (NVIDIA CUDA / NVDEC)
+Documentation=https://github.com/projectbluefin/common
+# Select NVIDIA CUDA image on x86-64 hosts with active NVIDIA driver module
+ConditionArchitecture=x86-64
+ConditionPathExists=/sys/module/nvidia/version
+
+[Container]
+ContainerName=ffmpeg-thumbnailer
+Image=docker.io/jrottenberg/ffmpeg:9.0-nvidia2404@sha256:76b1f3212f453945580853cda19f09c61fc08d17502c1cf67473e070c017165d
+Entrypoint=/bin/sleep
+Exec=infinity
+UserNS=keep-id
+RunInit=true
+Network=none
+ReadOnly=true
+AddDevice=nvidia.com/gpu=all
+AddDevice=-/dev/dri
+
+[Service]
+Slice=background.slice
+TimeoutStartSec=900
+CPUWeight=20
+IOWeight=20
+MemoryHigh=512M
+MemoryMax=1024M
+TasksMax=64
+ManagedOOMMemoryPressure=kill
diff --git a/system_files/shared/usr/share/containers/systemd/users/ffmpeg-thumbnailer.container b/system_files/shared/usr/share/containers/systemd/users/ffmpeg-thumbnailer.container
new file mode 100644
index 000000000..3ce4c4da2
--- /dev/null
+++ b/system_files/shared/usr/share/containers/systemd/users/ffmpeg-thumbnailer.container
@@ -0,0 +1,27 @@
+[Unit]
+Description=FFmpeg Video Thumbnailer Container (Alpine / Software / AMD / Intel)
+Documentation=https://github.com/projectbluefin/common
+# Select Alpine image on all non-NVIDIA hosts, and on aarch64 NVIDIA hosts (where CUDA image is unavailable)
+ConditionPathExists=|!/sys/module/nvidia/version
+ConditionArchitecture=|!x86-64
+
+[Container]
+ContainerName=ffmpeg-thumbnailer
+Image=docker.io/jrottenberg/ffmpeg:9.0-alpine320@sha256:2c3d71587b80173373ff52f2573ec5be9f527f2ba4fe601de6641299b1d68211
+Entrypoint=/bin/sleep
+Exec=infinity
+UserNS=keep-id
+RunInit=true
+Network=none
+ReadOnly=true
+AddDevice=-/dev/dri
+
+[Service]
+Slice=background.slice
+TimeoutStartSec=900
+CPUWeight=20
+IOWeight=20
+MemoryHigh=512M
+MemoryMax=1024M
+TasksMax=64
+ManagedOOMMemoryPressure=kill
diff --git a/system_files/shared/usr/share/ublue-os/thumbnailer/thumbnailers/ffmpeg-thumbnailer.thumbnailer b/system_files/shared/usr/share/ublue-os/thumbnailer/thumbnailers/ffmpeg-thumbnailer.thumbnailer
new file mode 100644
index 000000000..783de8e89
--- /dev/null
+++ b/system_files/shared/usr/share/ublue-os/thumbnailer/thumbnailers/ffmpeg-thumbnailer.thumbnailer
@@ -0,0 +1,4 @@
+[Thumbnailer Entry]
+TryExec=/usr/libexec/ffmpeg-thumbnailer/ffmpeg-video-thumbnailer-shim
+Exec=/usr/libexec/ffmpeg-thumbnailer/ffmpeg-video-thumbnailer-shim -s %s %u %o
+MimeType=application/mxf;image/vnd.rn-realpix;image/x-pict;video/3gp;video/3gpp;video/3gpp2;video/dv;video/divx;video/fli;video/flv;video/matroska;video/mp2t;video/mp4;video/mp4v-es;video/mpeg;video/mpeg-system;video/msvideo;video/ogg;video/quicktime;video/vivo;video/vnd.avi;video/vnd.divx;video/vnd.rn-realvideo;video/vnd.vivo;video/webm;video/x-anim;video/x-avi;video/x-flc;video/x-fli;video/x-flic;video/x-flv;video/x-m4v;video/x-matroska;video/x-mjpeg;video/x-mpeg;video/x-mpeg2;video/x-ms-asf;video/x-ms-asf-plugin;video/x-ms-asx;video/x-msvideo;video/x-ms-wm;video/x-ms-wmv;video/x-ms-wmx;video/x-ms-wvx;video/x-nsv;video/x-ogm+ogg;video/x-theora;video/x-theora+ogg;video/x-totem-stream;
diff --git a/tests/test_ffmpeg_thumbnailer.py b/tests/test_ffmpeg_thumbnailer.py
new file mode 100644
index 000000000..b7232c86a
--- /dev/null
+++ b/tests/test_ffmpeg_thumbnailer.py
@@ -0,0 +1,266 @@
+"""
+Tests for ffmpeg-thumbnailer-daemon and ffmpeg-video-thumbnailer-shim.
+"""
+
+from importlib.machinery import SourceFileLoader
+import importlib.util
+import os
+from pathlib import Path
+import socket
+import threading
+import time
+from unittest import mock
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+DAEMON_PATH = REPO_ROOT / "system_files/shared/usr/libexec/ffmpeg-thumbnailer/ffmpeg-thumbnailer-daemon"
+SHIM_PATH = REPO_ROOT / "system_files/shared/usr/libexec/ffmpeg-thumbnailer/ffmpeg-video-thumbnailer-shim"
+
+
+@pytest.fixture(scope="module")
+def daemon_mod():
+    loader = SourceFileLoader("daemon", str(DAEMON_PATH))
+    spec = importlib.util.spec_from_loader(loader.name, loader)
+    mod = importlib.util.module_from_spec(spec)
+    loader.exec_module(mod)
+    return mod
+
+
+@pytest.fixture(scope="module")
+def shim_mod():
+    loader = SourceFileLoader("shim", str(SHIM_PATH))
+    spec = importlib.util.spec_from_loader(loader.name, loader)
+    mod = importlib.util.module_from_spec(spec)
+    loader.exec_module(mod)
+    return mod
+
+
+def test_daemon_handle_request_ping(daemon_mod):
+    assert daemon_mod.handle_request(b"PING\n", []) == "OK"
+
+
+def test_daemon_handle_request_invalid_verb(daemon_mod):
+    assert "invalid request" in daemon_mod.handle_request(b"UNKNOWN\n", [])
+
+
+def test_daemon_handle_request_missing_fds(daemon_mod):
+    assert "expected input and output descriptors" in daemon_mod.handle_request(b"THUMB\t256\n", [])
+
+
+def test_daemon_cmd_structure(daemon_mod):
+    with mock.patch("subprocess.run") as mock_run:
+        mock_run.return_value = mock.Mock(returncode=0)
+        daemon_mod.run_container_ffmpeg("00:00:03", 256, 3, 4)
+        assert mock_run.called
+        cmd = mock_run.call_args[0][0]
+        assert "timeout" in cmd
+        assert "-threads" in cmd and "1" in cmd
+        assert "--preserve-fd=3,4" in cmd[2]
+
+
+def test_daemon_handle_request_valid_invocation(daemon_mod, tmp_path):
+    in_file = tmp_path / "in.mp4"
+    in_file.write_bytes(b"dummy")
+    out_file = tmp_path / "out.png"
+
+    in_fd = os.open(str(in_file), os.O_RDONLY)
+    out_fd = os.open(str(out_file), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o644)
+
+    try:
+        with mock.patch.object(daemon_mod, "run_container_ffmpeg") as mock_ffmpeg:
+            def fake_ffmpeg(seek, size, in_fd, out_fd):
+                os.write(out_fd, b"\x89PNGfake")
+                return True
+
+            mock_ffmpeg.side_effect = fake_ffmpeg
+            res = daemon_mod.handle_request(b"THUMB\t256\n", [in_fd, out_fd])
+            assert res == "OK"
+            assert out_file.read_bytes() == b"\x89PNGfake"
+    finally:
+        os.close(in_fd)
+        os.close(out_fd)
+
+
+def test_shim_uri_parsing_and_arg_forwarding(shim_mod, tmp_path):
+    out_file = str(tmp_path / "out.png")
+    with mock.patch("sys.argv", ["shim", "-s", "512", "file:///tmp/my%20video.mp4", out_file]):
+        with mock.patch.object(shim_mod, "get_socket_path", return_value=None):
+            with mock.patch.object(shim_mod, "fallback_local_ffmpeg") as mock_fallback:
+                mock_fallback.return_value = True
+                with pytest.raises(SystemExit) as exc:
+                    shim_mod.main()
+                assert exc.value.code == 0
+                mock_fallback.assert_called_once_with("/tmp/my video.mp4", out_file, 512)
+
+
+def test_daemon_handle_request_invalid_fd_type(daemon_mod, tmp_path):
+    # Pass a non-regular file descriptor (e.g. socket) to verify security rejection
+    s1, s2 = socket.socketpair()
+    try:
+        res = daemon_mod.handle_request(b"THUMB\t256\n", [s1.fileno(), s2.fileno()])
+        assert "invalid file descriptor type" in res
+    finally:
+        s1.close()
+        s2.close()
+
+
+def test_daemon_returns_unavailable_when_container_missing(daemon_mod, tmp_path):
+    in_file = tmp_path / "in.mp4"
+    in_file.write_bytes(b"dummy")
+    out_file = tmp_path / "out.png"
+
+    in_fd = os.open(str(in_file), os.O_RDONLY)
+    out_fd = os.open(str(out_file), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o644)
+
+    try:
+        with mock.patch("subprocess.run") as mock_run:
+            mock_run.return_value = mock.Mock(returncode=125)
+            res = daemon_mod.handle_request(b"THUMB\t256\n", [in_fd, out_fd])
+            assert res == "ERR_UNAVAILABLE"
+    finally:
+        os.close(in_fd)
+        os.close(out_fd)
+
+
+def test_shim_falls_back_on_err_unavailable(shim_mod, tmp_path):
+    out_file = str(tmp_path / "out.png")
+    with mock.patch("sys.argv", ["shim", "-s", "256", "file:///tmp/vid.mp4", out_file]):
+        with mock.patch.object(shim_mod, "get_socket_path", return_value="/tmp/fake.sock"):
+            with mock.patch("socket.socket") as mock_sock_cls:
+                mock_sock = mock.MagicMock()
+                mock_sock.recv.return_value = b"ERR_UNAVAILABLE\n"
+                mock_sock_cls.return_value = mock_sock
+                with mock.patch("os.open", return_value=10):
+                    with mock.patch("os.close"):
+                        with mock.patch.object(shim_mod, "fallback_local_ffmpeg", return_value=True) as mock_fb:
+                            with pytest.raises(SystemExit) as exc:
+                                shim_mod.main()
+                            assert exc.value.code == 0
+                            assert mock_fb.called
+
+
+def test_quadlet_condition_complementarity():
+    # Verify that Alpine and NVIDIA quadlet conditions form a partition over (arch, nvidia_present)
+    import configparser
+    repo_root = Path(__file__).resolve().parent.parent
+    alpine_path = repo_root / "system_files/shared/usr/share/containers/systemd/users/ffmpeg-thumbnailer.container"
+    nvidia_path = repo_root / "system_files/shared/usr/share/containers/systemd/users/ffmpeg-thumbnailer-nvidia.container"
+
+    # Both must target the identical container name so the daemon config is invariant
+    alpine_cfg = configparser.ConfigParser(strict=False)
+    alpine_cfg.read(alpine_path)
+    nvidia_cfg = configparser.ConfigParser(strict=False)
+    nvidia_cfg.read(nvidia_path)
+
+    assert alpine_cfg["Container"]["ContainerName"] == "ffmpeg-thumbnailer"
+    assert nvidia_cfg["Container"]["ContainerName"] == "ffmpeg-thumbnailer"
+
+    # Verify quadlet options
+    assert alpine_cfg["Container"]["RunInit"] == "true"
+    assert nvidia_cfg["Container"]["RunInit"] == "true"
+    with open(nvidia_path) as nf:
+        nvidia_raw = nf.read()
+    assert "AddDevice=nvidia.com/gpu=all" in nvidia_raw
+
+
+def test_serve_recovering_rebinds_after_socket_deleted(daemon_mod, tmp_path):
+    # GNOME's clean_gst_registry_dir() deletes the socket path out from under
+    # the running daemon. serve_recovering must rebind at the same path so
+    # subsequent connections succeed without a full daemon restart.
+    sock_path = str(tmp_path / "rebind.sock")
+    stop = threading.Event()
+
+    thread = threading.Thread(
+        target=daemon_mod.serve_recovering,
+        args=(sock_path, daemon_mod.Handler, stop),
+        daemon=True,
+    )
+    thread.start()
+
+    # Wait for the daemon to bind before connecting (avoid a startup race).
+    deadline = time.monotonic() + 5
+    while not os.path.exists(sock_path) and time.monotonic() < deadline:
+        time.sleep(0.05)
+
+    def ping_ok():
+        try:
+            conn = socket.socket(socket.AF_UNIX)
+            conn.settimeout(2)
+            conn.connect(sock_path)
+            conn.sendall(b"PING\n")
+            return conn.recv(64) == b"OK\n"
+        except OSError:
+            return False
+
+    assert ping_ok(), "daemon should answer PING before the socket is deleted"
+
+    # Simulate gnome-desktop-thumbnailer removing the socket path.
+    os.unlink(sock_path)
+    assert not os.path.exists(sock_path)
+
+    # The loop polls every ~0.5s; wait for it to notice and rebind.
+    deadline = time.monotonic() + 5
+    while time.monotonic() < deadline and not os.path.exists(sock_path):
+        time.sleep(0.05)
+    assert os.path.exists(sock_path), "socket path should be recreated"
+    assert ping_ok(), "daemon should answer PING after rebinding"
+
+    stop.set()
+    thread.join(timeout=5)
+    assert not thread.is_alive(), "serve_recovering should stop when signaled"
+
+
+def test_socket_survives_gnome_gst_registry_cleanup(daemon_mod, shim_mod, tmp_path):
+    # After every thumbnail, gnome-desktop's clean_gst_registry_dir() calls
+    # g_remove() (C remove()) on each top-level entry of the gst cache dir
+    # except gstreamer-1.0.registry. The daemon and shim must agree on a socket
+    # path that this non-recursive cleanup cannot delete, otherwise the next
+    # back-to-back thumbnail finds no socket.
+    import ctypes
+    import tempfile
+
+    # Keep the path short: AF_UNIX paths are capped at 108 bytes and pytest's
+    # tmp_path is too deep to hold the real layout.
+    short_root = tempfile.TemporaryDirectory(prefix="fft", dir="/tmp")
+    cache_home = Path(short_root.name) / "cache"
+    gst_dir = cache_home / "gnome-desktop-thumbnailer" / "gstreamer-1.0"
+    rel = os.path.relpath(daemon_mod.SOCK_PATH, daemon_mod.GST_CACHE_DIR)
+    sock_path = str(gst_dir / rel)
+    # gnome-desktop creates this directory (create_gst_cache_dir()).
+    gst_dir.mkdir(parents=True)
+
+    with mock.patch.dict(os.environ, {"XDG_CACHE_HOME": str(cache_home)}):
+        os.environ.pop("FFT_SOCKET", None)
+        stop = threading.Event()
+        thread = threading.Thread(
+            target=daemon_mod.serve_recovering,
+            args=(sock_path, daemon_mod.Handler, stop),
+            daemon=True,
+        )
+        thread.start()
+        try:
+            deadline = time.monotonic() + 5
+            while not os.path.exists(sock_path) and time.monotonic() < deadline:
+                time.sleep(0.05)
+            assert shim_mod.get_socket_path() == sock_path
+            (gst_dir / "gstreamer-1.0.registry").write_bytes(b"")
+            inode = os.stat(sock_path).st_ino
+
+            libc = ctypes.CDLL(None, use_errno=True)
+            for name in os.listdir(gst_dir):
+                if name != "gstreamer-1.0.registry":
+                    libc.remove(str(gst_dir / name).encode())
+
+            # No wait: the very next thumbnail must still reach the daemon.
+            assert os.stat(sock_path).st_ino == inode
+            conn = socket.socket(socket.AF_UNIX)
+            conn.settimeout(2)
+            conn.connect(sock_path)
+            conn.sendall(b"PING\n")
+            assert conn.recv(64) == b"OK\n"
+            conn.close()
+        finally:
+            stop.set()
+            thread.join(timeout=5)
+            short_root.cleanup()