Standalone mediapreview project

.gitignore

@@ -0,0 +1,6 @@
+.*
+*.lock
+!.gitignore
+__pycache__/
+*.egg-info/
+/dist

README.md

@@ -0,0 +1,58 @@
+# mediapreview
+
+Low-level media preview converters plus an optional persistent worker pool
+framework. All converters produce AVIF output.
+
+## Layout
+
+| Module | Purpose |
+|--------|---------|
+| `mediapreview.backends.image` | Image → AVIF via pyvips (ffmpeg for HEIC/HEIF/AVIF) |
+| `mediapreview.backends.video` | Video frame → AVIF via PyAV (HDR preserved) |
+| `mediapreview.backends.pdf` | PDF/XPS/EPUB page → AVIF via PyMuPDF + pyvips |
+| `mediapreview.backends` | `dispatch()` — pick a backend by path/mimetype |
+| `mediapreview.formats` | Suffix sets, previewability and priority classification |
+| `mediapreview.office` | OnlyOffice Document Server client + Docker bootstrap |
+| `mediapreview.docker/` | Patched OnlyOffice image build context (ships in the wheel) |
+| `mediapreview.pool` | Async persistent subprocess worker pool (optional) |
+| `mediapreview.worker` | Worker subprocess entry point (framed stdin/stdout protocol) |
+| `mediapreview.protocol` | msgspec wire structs (`PreviewRequest` / `PreviewResponse`) |
+| `mediapreview.cache` | Thread-safe LRU cache for preview responses |
+
+## Extras
+
+```bash
+pip install mediapreview            # image converter only (pyvips)
+pip install mediapreview[pdf]       # + PDF/XPS/EPUB (pymupdf)
+pip install mediapreview[video]     # + video (av, numpy)
+pip install mediapreview[office]    # + OnlyOffice client (httpx, pyjwt)
+pip install mediapreview[worker]    # + worker pool (blake3, tracerite)
+pip install mediapreview[standard]  # everything
+```
+
+## Usage
+
+Low-level, in-process:
+
+```python
+from mediapreview import dispatch
+
+avif_bytes, resp = dispatch(path, quality=60, maxsize=512, maxzoom=2.0)
+```
+
+Worker pool (isolates heavy imports and native crashes from the async loop):
+
+```python
+from mediapreview.pool import start_preview_workers, shutdown_preview_workers
+```
+
+## OnlyOffice Docker bootstrap
+
+A patched OnlyOffice image (configurable converter worker count) ships as
+package data and can be built/started with:
+
+```python
+from mediapreview.office import setup_docker
+
+setup_docker()  # builds + runs "onlyoffice-mediapreview" on port 8988
+```

mediapreview/__init__.py

@@ -0,0 +1,34 @@
+"""Media preview framework — low-level converters and a worker pool.
+
+Optional functionality is gated by extras:
+
+    pip install mediapreview[standard]   # all preview backends + worker pool
+    pip install mediapreview[worker]     # persistent subprocess worker pool
+    pip install mediapreview[pdf]        # PDF previews
+    pip install mediapreview[video]      # video previews
+    pip install mediapreview[office]     # OnlyOffice document conversion
+"""
+
+from mediapreview.backends import (
+    dispatch,
+    process_image,
+    process_image_buffer,
+    process_pdf,
+    process_video,
+)
+from mediapreview.cache import CachedPreview, PreviewCache
+from mediapreview.formats import is_previewable_path
+from mediapreview.protocol import PreviewRequest, PreviewResponse
+
+__all__ = [
+    "CachedPreview",
+    "PreviewCache",
+    "PreviewRequest",
+    "PreviewResponse",
+    "dispatch",
+    "is_previewable_path",
+    "process_image",
+    "process_image_buffer",
+    "process_pdf",
+    "process_video",
+]

mediapreview/__main__.py

@@ -0,0 +1,10 @@
+"""CLI entry point: delegates to the preview worker subprocess.
+
+A richer CLI for one-shot preview generation and pool management can be
+added later.
+"""
+
+from mediapreview.worker import main
+
+if __name__ == "__main__":
+    main()

mediapreview/backends/__init__.py

@@ -0,0 +1,57 @@
+"""Low-level preview conversion backends.
+
+Each backend module exposes a `process_*` function taking a path (or buffer)
+plus quality/size parameters and returning `(avif_bytes, PreviewResponse)`.
+`dispatch` picks the right backend for a path.
+"""
+
+import logging
+import mimetypes
+
+from mediapreview.backends.image import (
+    process_image,
+    process_image_buffer,
+    process_image_pyvips,
+)
+from mediapreview.backends.pdf import process_pdf
+from mediapreview.backends.video import process_video
+from mediapreview.formats import DOC_PREVIEW_SUFFIXES
+from mediapreview.protocol import PreviewResponse
+
+__all__ = [
+    "dispatch",
+    "process_image",
+    "process_image_buffer",
+    "process_image_pyvips",
+    "process_pdf",
+    "process_video",
+]
+
+logger = logging.getLogger(__name__)
+
+
+def dispatch(path, quality, maxsize, maxzoom, data=None):
+    backend = "unknown"
+    try:
+        if data:
+            backend = "pyvips"
+            return process_image_buffer(
+                data, quality=quality, maxsize=maxsize, maxzoom=maxzoom
+            )
+        suffix = path.suffix.lower()
+        if suffix in DOC_PREVIEW_SUFFIXES:
+            backend = "pdf"
+            return process_pdf(path, quality=quality, maxsize=maxsize, maxzoom=maxzoom)
+        mime_type, _ = mimetypes.guess_type(path.name)
+        if mime_type and mime_type.startswith("video/"):
+            backend = "video"
+            return process_video(path, quality=quality, maxsize=maxsize)
+        if mime_type and mime_type.startswith("image/"):
+            backend = "pyvips"
+            return process_image(path, quality=quality, maxsize=maxsize)
+    except ValueError as e:
+        return None, PreviewResponse(ok=False, backend=backend, error=str(e))
+    except Exception as e:
+        logger.exception("Preview dispatch failed for %s", path)
+        return None, PreviewResponse(ok=False, backend=backend, error=str(e))
+    return None, PreviewResponse(ok=False, backend=backend, error="preview unsupported")

mediapreview/backends/image.py

@@ -0,0 +1,188 @@
+"""Image preview conversion via pyvips (and ffmpeg for HEIC/HEIF/AVIF)."""
+
+import shlex
+import subprocess
+import tempfile
+from pathlib import Path
+from time import perf_counter
+
+import pyvips
+
+from mediapreview.protocol import PreviewResponse
+
+AVIF_FAST_EFFORT = 0
+
+
+def process_image(path, *, maxsize, quality):
+    return process_image_pyvips(path, maxsize=maxsize, quality=quality)
+
+
+def _get_image_dimensions(path: Path) -> tuple[int, int] | None:
+    """Probe image dimensions.
+
+    pyvips can read the header of most formats (including HEIC) without
+    fully decoding the image.
+    """
+    try:
+        img = pyvips.Image.new_from_file(str(path))
+        img = img.autorot()
+    except pyvips.error.Error:
+        return None
+    else:
+        return img.width, img.height
+
+
+def _image_via_ffmpeg(path: Path, maxsize: int, quality: int) -> bytes:
+    """Convert any image to AVIF using ffmpeg CLI.
+
+    ffmpeg handles HEIC tile assembly, HDR metadata and ICC profile embedding
+    automatically. Note: -vf cannot be used here — HEIC tile assembly feeds
+    the stream from a complex filtergraph, which conflicts with simple -vf
+    filtering; scaling must use the -s output option instead.
+    """
+    dims = _get_image_dimensions(path)
+    crf = int(63 * (1 - quality / 100) ** 2)
+    with tempfile.NamedTemporaryFile(suffix=".avif", delete=False) as tmp_f:
+        tmp_path = tmp_f.name
+    cmd = [
+        "ffmpeg",
+        # Keep error messages, drop the banner/config/stream-mapping spam.
+        "-hide_banner",
+        "-loglevel",
+        "error",
+        "-nostats",
+        # No interactive keyboard prompts ("Press [q] to stop ...").
+        "-nostdin",
+        "-y",
+        "-i",
+        str(path),
+        "-frames:v",
+        "1",
+        "-c:v",
+        "av1",
+        "-crf",
+        str(crf),
+        "-cpu-used",
+        "8",
+        tmp_path,
+    ]
+    if dims is not None:
+        w, h = dims
+        if max(w, h) > maxsize:
+            scale = min(maxsize / w, maxsize / h)
+            new_w = int(w * scale)
+            new_h = int(h * scale)
+            # insert -s <wxh> right after the input file
+            input_index = cmd.index(str(path)) + 1
+            cmd.insert(input_index, "-s")
+            cmd.insert(input_index + 1, f"{new_w}x{new_h}")
+    try:
+        try:
+            # stdin=DEVNULL is critical: ffmpeg must not inherit the worker's
+            # stdin, which carries the framed request protocol. An inherited
+            # stdin lets ffmpeg eat protocol bytes and, if the worker is
+            # killed mid-conversion, keeps the orphaned ffmpeg holding the
+            # pipe open so the parent's proc.wait() hangs forever.
+            subprocess.run(  # noqa: S603
+                cmd,
+                capture_output=True,
+                check=True,
+                shell=False,
+                stdin=subprocess.DEVNULL,
+            )
+        except subprocess.CalledProcessError as e:
+            shell_cmd = shlex.join(cmd)
+            stderr = (e.stderr or b"").decode(errors="replace").strip()
+            if stderr:
+                raise RuntimeError(
+                    f"ffmpeg failed (exit {e.returncode}):\n{shell_cmd}\n{stderr}"
+                ) from e
+            raise RuntimeError(
+                f"ffmpeg failed (exit {e.returncode}):\n{shell_cmd}"
+            ) from e
+        with Path(tmp_path).open("rb") as f:
+            return f.read()
+    finally:
+        Path(tmp_path).unlink(missing_ok=True)
+
+
+def process_image_pyvips(path, *, maxsize, quality):
+    t_start = perf_counter()
+    suffix = path.suffix.lower()
+
+    # HEIC/HEIF/AVIF: ffmpeg handles tile assembly and HDR correctly;
+    # skip pyvips entirely (pyvips drops CICP colour metadata, turning
+    # HDR sources into washed-out SDR previews).
+    if suffix in (".heic", ".heif", ".avif"):
+        dims = _get_image_dimensions(path)
+        width, height = dims or (None, None)
+        ret = _image_via_ffmpeg(path, maxsize, quality)
+        t_end = perf_counter()
+        return ret, PreviewResponse(
+            ok=True,
+            mime="image/avif",
+            backend="ffmpeg",
+            timings=[round((t_end - t_start) * 1000, 1)],
+            width=width,
+            height=height,
+        )
+
+    # Other image formats: pyvips only. ffmpeg is not a useful fallback
+    # here — when pyvips cannot decode a file, ffmpeg's image decoders
+    # cannot either, and their failure output is far noisier.
+    try:
+        img = pyvips.Image.new_from_file(str(path), access="sequential")
+        if img.get_typeof("orientation") and img.get("orientation") != 1:
+            # autorot's rot90 reads pixels out of order, which sequential
+            # access cannot do — reopen with random access when rotating.
+            img = pyvips.Image.new_from_file(str(path)).autorot()
+        orig_w, orig_h = img.width, img.height
+        scale = min(maxsize / img.width, maxsize / img.height, 1.0)
+        if scale < 1.0:
+            img = img.resize(scale)
+        ret = img.write_to_buffer(
+            ".avif",
+            Q=quality,
+            effort=AVIF_FAST_EFFORT,
+            keep="none",
+        )
+    except pyvips.error.Error as e:
+        raise ValueError(f"cannot decode image: {e}") from e
+    backend = "pyvips"
+    t_end = perf_counter()
+
+    return ret, PreviewResponse(
+        ok=True,
+        mime="image/avif",
+        backend=backend,
+        timings=[round((t_end - t_start) * 1000, 1)],
+        width=orig_w,
+        height=orig_h,
+    )
+
+
+def process_image_buffer(data: bytes, *, quality, maxsize, maxzoom):
+    _ = maxzoom
+    t_start = perf_counter()
+    img = pyvips.Image.new_from_buffer(data, "")
+    img = img.autorot()
+    orig_w, orig_h = img.width, img.height
+    scale = min(maxsize / img.width, maxsize / img.height, 1.0)
+    if scale < 1.0:
+        img = img.resize(scale)
+    ret = img.write_to_buffer(
+        ".avif",
+        Q=quality,
+        effort=AVIF_FAST_EFFORT,
+        keep="none",
+    )
+    t_end = perf_counter()
+
+    return ret, PreviewResponse(
+        ok=True,
+        mime="image/avif",
+        backend="pyvips",
+        timings=[round((t_end - t_start) * 1000, 1)],
+        width=orig_w,
+        height=orig_h,
+    )

mediapreview/backends/pdf.py

@@ -0,0 +1,48 @@
+"""PDF/XPS/EPUB preview conversion via PyMuPDF + pyvips."""
+
+from time import perf_counter
+
+import pyvips
+
+from mediapreview.backends.image import AVIF_FAST_EFFORT
+from mediapreview.protocol import PreviewResponse
+
+try:
+    import pymupdf
+except ImportError:  # pragma: no cover - optional pdf extra
+    pymupdf = None
+
+
+def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0):
+    if pymupdf is None:
+        raise ImportError(
+            "PDF previews require the 'pdf' extra: pip install mediapreview[pdf]"
+        )
+    t_load_start = perf_counter()
+    with pymupdf.open(path) as pdf:
+        page = pdf.load_page(page_number)
+        w, h = page.rect[2:4]
+        zoom = min(maxsize / w, maxsize / h, maxzoom)
+        mat = pymupdf.Matrix(zoom, zoom)
+        pix = page.get_pixmap(matrix=mat)
+        t_load_end = perf_counter()
+
+        t_save_start = perf_counter()
+        img = pyvips.Image.new_from_memory(
+            pix.samples_mv, pix.width, pix.height, pix.n, "uchar"
+        )
+    ret = img.write_to_buffer(".avif", Q=quality, effort=AVIF_FAST_EFFORT, keep="none")
+    backend = "pdf+pyvips"
+    t_save_end = perf_counter()
+
+    return ret, PreviewResponse(
+        ok=True,
+        mime="image/avif",
+        backend=backend,
+        timings=[
+            round((t_load_end - t_load_start) * 1000, 1),
+            round((t_save_end - t_save_start) * 1000, 1),
+        ],
+        width=round(w),
+        height=round(h),
+    )

mediapreview/backends/video.py

@@ -0,0 +1,154 @@
+"""Video preview conversion via PyAV (AV1/AVIF output, HDR preserved)."""
+
+import gc
+import io
+import os
+
+# Silence the SVT-AV1 encoder's stderr spam, set log level ERROR
+os.environ.setdefault("SVT_LOG", "1")
+
+from time import perf_counter
+
+from mediapreview.protocol import PreviewResponse
+
+try:
+    import av
+    import numpy as np
+except ImportError:  # pragma: no cover - optional video extra
+    av = None
+    np = None
+
+
+def _rotate_frame_yuv(frame, k):
+    """Rotate a planar YUV420 frame by k*90° counter-clockwise, keeping its format.
+
+    Rotating each plane independently preserves the pixel format (including
+    10-bit HDR formats like yuv420p10le, which PyAV exposes as uint16 planes)
+    and the source colorspace.
+    """
+    if av is None or np is None:
+        raise ImportError(
+            "Video previews require the 'video' extra: pip install mediapreview[video]"
+        )
+    fmt = frame.format
+    w, h = frame.width, frame.height
+    if (
+        not fmt.is_planar
+        or fmt.chroma_width(w) * 2 != w
+        or fmt.chroma_height(h) * 2 != h
+    ):
+        raise ValueError(f"unsupported format for YUV rotation: {fmt.name}")
+    planes = frame.to_ndarray()
+    y, u, v = (
+        planes[:h],
+        planes[h : h + h // 4].reshape(h // 2, w // 2),
+        planes[h + h // 4 :].reshape(h // 2, w // 2),
+    )
+    planes = np.hstack(
+        [p.flat for p in (np.rot90(y, k), np.rot90(u, k), np.rot90(v, k))]
+    )
+    new_width = w if k % 2 == 0 else h
+    return av.VideoFrame.from_ndarray(planes.reshape(-1, new_width), format=fmt.name)
+
+
+def process_video(path, *, maxsize, quality):
+    if av is None or np is None:
+        raise ImportError(
+            "Video previews require the 'video' extra: pip install mediapreview[video]"
+        )
+    frame = None
+    imgdata = io.BytesIO()
+    istream = ostream = icc = occ = frame = None
+    t_load_start = perf_counter()
+    # Initialize to avoid "possibly unbound" in static analysis when exceptions occur
+    t_load_end = t_load_start
+    t_save_start = t_load_start
+    t_save_end = t_load_start
+    with (
+        av.open(
+            str(path),
+            options={
+                "analyzeduration": "1000000",  # 1 second (in microseconds)
+                "fflags": "fastseek",
+            },
+        ) as icontainer,
+        av.open(imgdata, "w", format="avif") as ocontainer,
+    ):
+        istream = icontainer.streams.video[0]
+        istream.codec_context.skip_frame = "NONKEY"
+        icontainer.seek((icontainer.duration or 0) // 8)
+        for frame in icontainer.decode(istream):
+            if frame.dts is not None:
+                break
+        else:
+            raise RuntimeError("No frames found in video")
+
+        # Resize frame to thumbnail size
+        # Capture display dimensions before resize (accounting for rotation)
+        disp_w = frame.width
+        disp_h = frame.height
+        if frame.rotation in (90, 270):
+            disp_w, disp_h = disp_h, disp_w
+        if frame.width > maxsize or frame.height > maxsize:
+            scale_factor = min(maxsize / frame.width, maxsize / frame.height)
+            new_width = int(frame.width * scale_factor)
+            new_height = int(frame.height * scale_factor)
+            frame = frame.reformat(width=new_width, height=new_height)
+
+        # Apply display-matrix rotation if present
+        if frame.rotation:
+            # frame.rotation indicates clockwise rotation needed to display correctly
+            k = (frame.rotation // 90) % 4  # Convert to counter-clockwise rotations
+            frame = _rotate_frame_yuv(frame, k)
+
+        # libsvtav1 rejects full-range JPEG-style YUV pixel formats such as
+        # yuvj420p, so normalize them before opening the encoder.
+        if frame.format.name.startswith("yuvj"):
+            frame = frame.reformat(format="yuv420p")
+        t_load_end = perf_counter()
+
+        t_save_start = perf_counter()
+        crf = str(int(63 * (1 - quality / 100) ** 2))  # Closely matching PIL quality-%
+        ostream = ocontainer.add_stream(
+            "av1",
+            options={
+                "crf": crf,
+                "usage": "realtime",
+                "cpu-used": "8",
+                "threads": "1",
+            },
+        )
+        if not isinstance(ostream, av.VideoStream):
+            raise TypeError("failed to initialize AV1 video stream")
+        ostream.width = frame.width
+        ostream.height = frame.height
+        ostream.pix_fmt = frame.format.name
+        icc = istream.codec_context
+        occ = ostream.codec_context
+
+        # Copy HDR metadata from input video stream
+        occ.color_primaries = icc.color_primaries
+        occ.color_trc = icc.color_trc
+        occ.colorspace = icc.colorspace
+        occ.color_range = icc.color_range
+
+        ocontainer.mux(ostream.encode(frame))
+        ocontainer.mux(ostream.encode(None))  # Flush the stream
+        t_save_end = perf_counter()
+
+    # Capture result before cleanup
+    ret = imgdata.getvalue()
+    resp = PreviewResponse(
+        ok=True,
+        mime="image/avif",
+        backend="video",
+        timings=[
+            round((t_load_end - t_load_start) * 1000, 1),
+            round((t_save_end - t_save_start) * 1000, 1),
+        ],
+        width=disp_w,
+        height=disp_h,
+    )
+    del imgdata, istream, ostream, icc, occ, frame
+    gc.collect()
+    return ret, resp

mediapreview/cache.py

@@ -0,0 +1,44 @@
+"""In-memory LRU cache for preview responses."""
+
+from collections import OrderedDict
+from dataclasses import dataclass
+from threading import Lock
+
+
+@dataclass(slots=True)
+class CachedPreview:
+    """Cached preview with headers and body."""
+
+    headers: dict[str, str]
+    body: bytes
+
+
+class PreviewCache:
+    """Thread-safe LRU cache for preview responses."""
+
+    def __init__(self, capacity: int = 500):
+        self.capacity = capacity
+        self._cache: OrderedDict[str, CachedPreview] = OrderedDict()
+        self._lock = Lock()
+
+    def get(self, key: str) -> CachedPreview | None:
+        """Get cached preview, moving it to end (most recently used)."""
+        with self._lock:
+            if key in self._cache:
+                self._cache.move_to_end(key)
+                return self._cache[key]
+        return None
+
+    def set(self, key: str, value: CachedPreview) -> None:
+        """Cache preview, evicting oldest if at capacity."""
+        with self._lock:
+            if key in self._cache:
+                self._cache.move_to_end(key)
+            else:
+                if len(self._cache) >= self.capacity:
+                    self._cache.popitem(last=False)
+                self._cache[key] = value
+
+    def __len__(self) -> int:
+        with self._lock:
+            return len(self._cache)

mediapreview/docker/Dockerfile

@@ -0,0 +1,70 @@
+# Patched OnlyOffice Document Server with configurable converter worker count.
+#
+# The Community Edition hardcodes the document converter to 1 worker,
+# which creates a severe bottleneck under concurrent load.
+# This image patches the open-source license.js to spawn a configurable
+# number of converter workers (default 8).
+#
+# Build:
+#   docker build -t onlyoffice-mediapreview mediapreview/mediapreview/docker
+#
+# Run:
+#   docker run -d -p 8988:80 \
+#     -e WORKERS=16 \
+#     -e JWT_SECRET=your-strong-secret \
+#     --name onlyoffice-mediapreview onlyoffice-mediapreview
+#
+# JWT:
+#   Set JWT_SECRET to the same value you pass to Cista as ONLYOFFICE_JWT_SECRET.
+#   OnlyOffice will enable token validation automatically.
+#
+# The ONLYOFFICE_VERSION build arg lets you target a specific release.
+
+ARG ONLYOFFICE_VERSION=9.3.1
+
+FROM onlyoffice/documentserver:${ONLYOFFICE_VERSION}
+
+# Prevent interactive apt prompts
+ENV DEBIAN_FRONTEND=noninteractive
+
+# Install Node.js, npm, and git so we can run the FileConverter from source.
+RUN apt-get update -qq && \
+    apt-get install -y -qq --no-install-recommends \
+        nodejs \
+        npm \
+        git \
+        ca-certificates && \
+    rm -rf /var/lib/apt/lists/*
+
+# Clone the open-source server components (shallow, ~15 MB).
+# The master branch is used because the Linux/web tags are not published
+# in the server repo; the license.js file has been stable for years.
+RUN git clone --depth 1 https://github.com/ONLYOFFICE/server.git /opt/oo-server
+
+# Patch license.js so the converter worker count is read from an env var
+# instead of being hardcoded to 1.
+RUN sed -i \
+    's/count: 1,/count: parseInt(process.env.WORKERS, 10) || 8,/' \
+    /opt/oo-server/Common/sources/license.js
+
+# Install npm dependencies for the modules the FileConverter touches.
+# DocService deps are also needed because converter.js pulls in baseConnector.
+RUN cd /opt/oo-server/Common      && npm ci --no-audit --no-fund
+RUN cd /opt/oo-server/FileConverter && npm ci --no-audit --no-fund
+RUN cd /opt/oo-server/DocService  && npm ci --no-audit --no-fund
+
+# Back up the compiled pkg binary and replace it with our wrapper.
+RUN mv /var/www/onlyoffice/documentserver/server/FileConverter/converter \
+       /var/www/onlyoffice/documentserver/server/FileConverter/converter.orig
+
+COPY converter-wrapper.sh /var/www/onlyoffice/documentserver/server/FileConverter/converter
+RUN chmod +x /var/www/onlyoffice/documentserver/server/FileConverter/converter
+
+# Default worker count (override at runtime with -e WORKERS=16).
+ENV WORKERS=8
+
+# Use our custom entrypoint to persist the env var to a file that the
+# non-root converter process (user=ds) can read.
+COPY entrypoint.sh /app/ds/run-document-server-patched.sh
+RUN chmod +x /app/ds/run-document-server-patched.sh
+ENTRYPOINT ["/app/ds/run-document-server-patched.sh"]

mediapreview/docker/converter-wrapper.sh

@@ -0,0 +1,19 @@
+#!/bin/bash
+# Wrapper that runs the OnlyOffice FileConverter from patched Node.js source.
+# Replaces the compiled pkg binary shipped with the Community Edition.
+
+# The env var is not passed through supervisor to the 'ds' user, so we read
+# it from a file written by the custom entrypoint.
+if [ -z "${WORKERS}" ] && [ -r /tmp/oo-converter-workers.txt ]; then
+  export WORKERS=$(cat /tmp/oo-converter-workers.txt)
+fi
+
+cd /opt/oo-server/FileConverter || exit 1
+
+export NODE_ENV=production-linux
+export NODE_CONFIG_DIR=/etc/onlyoffice/documentserver
+export NODE_DISABLE_COLORS=1
+export APPLICATION_NAME=onlyoffice
+export LD_LIBRARY_PATH=/var/www/onlyoffice/documentserver/server/FileConverter/bin
+
+exec node sources/convertermaster.js "$@"

mediapreview/docker/entrypoint.sh

@@ -0,0 +1,8 @@
+#!/bin/bash
+# Custom entrypoint that persists WORKERS to a file readable by
+# the non-root user that supervisor uses to run the converter.
+
+echo "${WORKERS:-8}" > /tmp/oo-converter-workers.txt
+chmod 644 /tmp/oo-converter-workers.txt
+
+exec /app/ds/run-document-server.sh "$@"

mediapreview/formats.py

@@ -0,0 +1,94 @@
+"""File-type classification for preview generation.
+
+Suffix sets and helpers that decide which backend handles a given path,
+how previewable it is, and how jobs should be prioritised.
+"""
+
+import mimetypes
+from pathlib import Path
+
+DOC_PREVIEW_SUFFIXES = {".pdf", ".xps", ".epub", ".mobi"}
+
+OFFICE_PREVIEW_SUFFIXES = {
+    ".doc",
+    ".dot",
+    ".docx",
+    ".docm",
+    ".dotx",
+    ".dotm",
+    ".rtf",
+    ".odt",
+    ".ott",
+    ".txt",
+    ".md",
+    ".mhtml",
+    ".mht",
+    ".html",
+    ".htm",
+    ".xml",
+    ".wps",
+    ".wri",
+    # Spreadsheets
+    ".xls",
+    ".xlsx",
+    ".xlsm",
+    ".xlsb",
+    ".xltx",
+    ".xltm",
+    ".ods",
+    ".ots",
+    ".csv",
+    # Presentations
+    ".ppt",
+    ".pptx",
+    ".pptm",
+    ".pps",
+    ".ppsx",
+    ".pot",
+    ".potx",
+    ".odp",
+    ".otp",
+}
+
+
+def is_previewable_path(path) -> bool:
+    suffix = path.suffix.lower()
+    if suffix in DOC_PREVIEW_SUFFIXES or suffix in OFFICE_PREVIEW_SUFFIXES:
+        return True
+    mime_type, _ = mimetypes.guess_type(path.name)
+    if not mime_type:
+        return False
+    return mime_type.startswith(("image/", "video/"))
+
+
+def preview_job_priority(path) -> int:
+    """Return priority for preview job (lower=higher priority).
+
+    Priority order: images (0) < video (1) < PDF (2) < office (3) < unknown (4)
+    """
+    suffix = path.suffix.lower()
+    if suffix in DOC_PREVIEW_SUFFIXES:
+        return 2
+    if suffix in OFFICE_PREVIEW_SUFFIXES:
+        return 3
+    mime_type, _ = mimetypes.guess_type(path.name)
+    if mime_type and mime_type.startswith("image/"):
+        return 0
+    if mime_type and mime_type.startswith("video/"):
+        return 1
+    return 4
+
+
+def expected_backend(path: Path) -> str:
+    """Best-effort backend label used for timeout/access logging."""
+    suffix = path.suffix.lower()
+    if suffix in OFFICE_PREVIEW_SUFFIXES:
+        return "onlyoffice"
+    if suffix in DOC_PREVIEW_SUFFIXES:
+        return "pdf"
+    mime_type, _ = mimetypes.guess_type(path.name)
+    if mime_type and mime_type.startswith("video/"):
+        return "video"
+    if mime_type and mime_type.startswith("image/"):
+        return "pyvips"
+    return "preview"

mediapreview/office.py

@@ -0,0 +1,422 @@
+"""OnlyOffice Document Server integration for office document preview.
+
+Provides server-side conversion of office documents to PNG via the
+OnlyOffice Document Server /ConvertService.ashx API. The resulting PNG
+is passed through pyvips for AVIF compression.
+
+Environment requirements:
+    - OnlyOffice Document Server must be running and reachable.
+    - If Document Server runs in Docker, the callback host IP must be
+      reachable from the container (usually the docker bridge IP).
+"""
+
+import asyncio
+import json
+import logging
+import os
+import secrets
+import socket
+import socketserver
+import subprocess
+import threading
+import urllib.error
+import urllib.request
+from functools import lru_cache, partial
+from http.server import SimpleHTTPRequestHandler
+from multiprocessing import cpu_count
+from pathlib import Path
+from time import perf_counter
+from urllib.parse import quote
+
+try:
+    import httpx
+    import jwt
+except ImportError:  # pragma: no cover - optional office extra
+    httpx = None
+    jwt = None
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Configuration helpers
+# ---------------------------------------------------------------------------
+
+
+_httpx_client: httpx.AsyncClient | None = None
+
+
+def _get_onlyoffice_url() -> str:
+    return os.environ.get(
+        "ONLYOFFICE_URL",
+        os.environ.get("ONLYOFFICE_CISTA_URL", "http://localhost:8988"),
+    )
+
+
+def _get_jwt_secret() -> str:
+    return os.environ.get("ONLYOFFICE_JWT_SECRET", "")
+
+
+@lru_cache(maxsize=1)
+def _get_callback_host() -> str:
+    """Return the host IP that OnlyOffice (usually in Docker) can use to reach us."""
+    if host := os.environ.get("ONLYOFFICE_CALLBACK_HOST"):
+        return host
+    # Try to auto-detect docker bridge IP
+    try:
+        result = subprocess.run(
+            ["/sbin/ip", "-4", "addr", "show", "docker0"],
+            capture_output=True,
+            text=True,
+            timeout=2,
+            check=False,
+        )
+        for line in result.stdout.splitlines():
+            if "inet " in line:
+                parts = line.strip().split()
+                addr_part = parts[1]  # e.g. 172.17.0.1/16
+                return addr_part.split("/")[0]
+    except Exception:
+        logger.debug("Failed to auto-detect docker bridge IP")
+    return "127.0.0.1"
+
+
+def onlyoffice_error_short_text(detail: str) -> str:
+    """Short human-readable label for an OnlyOffice failure, for log annotation."""
+    if detail.startswith("OnlyOffice conversion error:"):
+        code = detail.rsplit(":", 1)[-1].strip()
+        return {
+            "-8": "onlyoffice jwt error",
+            "-4": "onlyoffice input error",
+            "-2": "onlyoffice timeout error",
+            "-1": "onlyoffice unknown error",
+        }.get(code, f"onlyoffice {code} error")
+    if "OnlyOffice response did not contain FileUrl" in detail:
+        return "onlyoffice no-fileurl error"
+    return "onlyoffice error"
+
+
+# ---------------------------------------------------------------------------
+# Async HTTP client
+# ---------------------------------------------------------------------------
+
+
+def get_httpx_client() -> httpx.AsyncClient:
+    """Return the shared async HTTP client for OnlyOffice requests."""
+    if httpx is None:
+        raise ImportError(
+            "OnlyOffice integration requires the 'office' extra: pip install mediapreview[office]"
+        )
+    global _httpx_client
+    if _httpx_client is None:
+        _httpx_client = httpx.AsyncClient()
+    return _httpx_client
+
+
+async def close_oo_client() -> None:
+    """Close the shared async HTTP client."""
+    global _httpx_client
+    if _httpx_client is not None:
+        await _httpx_client.aclose()
+        _httpx_client = None
+
+
+# ---------------------------------------------------------------------------
+# Availability check
+# ---------------------------------------------------------------------------
+
+
+def _probe_status() -> tuple[bool, bool, str | None]:
+    """Return (ok, responded, detail) for a lightweight reachability probe."""
+    url = _get_onlyoffice_url().rstrip("/") + "/ConvertService.ashx"
+    try:
+        with urllib.request.urlopen(url, timeout=2) as resp:  # noqa: S310
+            status = resp.status
+    except urllib.error.HTTPError as e:
+        status = e.code
+    except Exception:
+        return False, False, None
+
+    if status in (200, 405):
+        return True, True, None
+    if status >= 500:
+        return False, True, f"HTTP {status}"
+    return False, True, f"HTTP {status}"
+
+
+def log_reachable_info() -> None:
+    """Log info on success, warning on responded probe errors, silent on no-response."""
+    ok, responded, detail = _probe_status()
+    if ok:
+        logger.info("Using OnlyOffice document server at %s", _get_onlyoffice_url())
+    elif responded:
+        suffix = f": {detail}" if detail else ""
+        logger.warning("OnlyOffice probe failed%s", suffix)
+
+
+def setup_docker(name: str = "onlyoffice-mediapreview", port: int = 8988) -> int:
+    """Build and run the patched OnlyOffice Docker image.
+
+    Uses ONLYOFFICE_JWT_SECRET if set, otherwise generates a random secret.
+    The Docker build context ships inside the package at `mediapreview/docker`.
+    """
+    secret = _get_jwt_secret() or secrets.token_hex(16)
+    docker_dir = Path(__file__).parent / "docker"
+    if not docker_dir.is_dir():
+        raise FileNotFoundError(
+            f"Docker files not found at {docker_dir}. Is the package installed correctly?"
+        )
+
+    logger.info("Building OnlyOffice image")
+    build_cmd = ["docker", "build", "-t", name, str(docker_dir)]
+    logger.info("%s", " ".join(build_cmd))
+    result = subprocess.run(build_cmd, check=False, shell=False)  # noqa: S603
+    if result.returncode != 0:
+        raise RuntimeError("Failed to build OnlyOffice image")
+
+    logger.info("Starting OnlyOffice container")
+    run_cmd = [
+        "docker",
+        "run",
+        "-d",
+        "-p",
+        f"{port}:80",
+        "-e",
+        f"JWT_SECRET={secret}",
+        "-e",
+        "WORKERS=8",
+        "--name",
+        name,
+        "--restart",
+        "unless-stopped",
+        name,
+    ]
+    logger.info("%s", " ".join(run_cmd))
+    result = subprocess.run(run_cmd, check=False, shell=False)  # noqa: S603
+    if result.returncode != 0:
+        raise RuntimeError("Failed to start OnlyOffice container")
+    logger.info("OnlyOffice is running on http://localhost:%d", port)
+    return 0
+
+
+async def is_available_async(request_timeout: float = 2.0) -> bool:
+    """Return True if the configured OnlyOffice Document Server is reachable."""
+    if httpx is None:
+        raise ImportError(
+            "OnlyOffice integration requires the 'office' extra: pip install mediapreview[office]"
+        )
+    url = _get_onlyoffice_url().rstrip("/") + "/ConvertService.ashx"
+    client = get_httpx_client()
+    try:
+        response = await client.get(url, timeout=request_timeout)
+    except Exception:
+        return False
+    else:
+        return response.status_code in (200, 405)
+
+
+_oo_available_cache: tuple[bool, float] | None = None
+OO_AVAILABILITY_CACHE_TTL = 30.0
+
+
+async def is_available_cached() -> bool:
+    """Return cached OnlyOffice availability, refreshed every 30 seconds.
+
+    State transitions are logged, so an unreachable server is reported once
+    instead of on every preview attempt.
+    """
+    global _oo_available_cache
+    now = perf_counter()
+    if _oo_available_cache is not None:
+        result, timestamp = _oo_available_cache
+        if now - timestamp < OO_AVAILABILITY_CACHE_TTL:
+            return result
+    result = await is_available_async()
+    if _oo_available_cache is None or _oo_available_cache[0] != result:
+        if result:
+            logger.info(
+                "OnlyOffice document server available at %s", _get_onlyoffice_url()
+            )
+        else:
+            logger.warning(
+                "OnlyOffice document server not reachable at %s", _get_onlyoffice_url()
+            )
+    _oo_available_cache = (result, now)
+    return result
+
+
+# ---------------------------------------------------------------------------
+# Temporary HTTP server so OnlyOffice can download the file
+# ---------------------------------------------------------------------------
+
+
+class _QuietHandler(SimpleHTTPRequestHandler):
+    def log_message(self, fmt, *args) -> None:
+        pass
+
+
+def _get_free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("0.0.0.0", 0))  # noqa: S104
+        return s.getsockname()[1]
+
+
+def _serve_file_temporarily(file_path: Path):
+    """Start a temporary HTTP server for *file_path* and return (url, server)."""
+    directory = str(file_path.parent)
+    filename = file_path.name
+    port = _get_free_port()
+
+    handler = partial(_QuietHandler, directory=directory)
+    httpd = socketserver.TCPServer(("0.0.0.0", port), handler)  # noqa: S104
+    thread = threading.Thread(target=httpd.serve_forever, daemon=True)
+    thread.start()
+
+    host = _get_callback_host()
+    url = f"http://{host}:{port}/{quote(filename)}"
+    return url, httpd
+
+
+# ---------------------------------------------------------------------------
+# OnlyOffice conversion client
+# ---------------------------------------------------------------------------
+
+
+def _build_jwt_token(payload: dict) -> str | None:
+    secret = _get_jwt_secret()
+    if not secret:
+        return None
+    return jwt.encode(payload, secret, algorithm="HS256")
+
+
+async def convert_to_png_async(file_path: Path, request_timeout: float = 5.0) -> bytes:
+    """Convert *file_path* to PNG using OnlyOffice Document Server (async).
+
+    Returns the PNG bytes. Raises RuntimeError on failure.
+    """
+    if httpx is None or jwt is None:
+        raise ImportError(
+            "OnlyOffice integration requires the 'office' extra: pip install mediapreview[office]"
+        )
+    oo_url = _get_onlyoffice_url().rstrip("/")
+    convert_url = f"{oo_url}/ConvertService.ashx"
+    client = get_httpx_client()
+
+    # Start temporary HTTP server so OnlyOffice can fetch the file
+    doc_url, httpd = await asyncio.to_thread(_serve_file_temporarily, file_path)
+    try:
+        suffix = file_path.suffix.lstrip(".").lower()
+        payload = {
+            "async": False,
+            "filetype": suffix,
+            "key": f"mediapreview_{(await asyncio.to_thread(file_path.stat)).st_mtime_ns}",
+            "outputtype": "png",
+            "title": file_path.name,
+            "url": doc_url,
+        }
+
+        headers = {"Content-Type": "application/json"}
+        token = _build_jwt_token(payload)
+        if token:
+            # Conversion API expects JWT in request body when token checks are enabled.
+            payload["token"] = token
+            headers["Authorization"] = token
+
+        t_start = perf_counter()
+        response = await client.post(
+            convert_url,
+            content=json.dumps(payload).encode(),
+            headers=headers,
+            timeout=request_timeout,
+        )
+        response.raise_for_status()
+        body = response.content
+        t_end = perf_counter()
+
+        # Parse XML response
+        text = body.decode("utf-8", errors="replace")
+        if "<Error>" in text:
+            code = "unknown"
+            if "<Error>" in text and "</Error>" in text:
+                code = text.split("<Error>")[1].split("</Error>")[0]
+            raise RuntimeError(f"OnlyOffice conversion error: {code}")
+
+        if "<FileUrl>" not in text:
+            raise RuntimeError("OnlyOffice response did not contain FileUrl")
+
+        file_url = text.split("<FileUrl>")[1].split("</FileUrl>")[0]
+        file_url = file_url.replace("&amp;", "&")
+
+        logger.debug("OnlyOffice converted in %.2fs: %s", t_end - t_start, file_url)
+
+        # Download converted PNG
+        png_response = await client.get(file_url, timeout=request_timeout)
+        png_response.raise_for_status()
+        return png_response.content
+    finally:
+        await asyncio.to_thread(httpd.shutdown)
+
+
+# ---------------------------------------------------------------------------
+# Conversion manager (pulled from the preview orchestrator)
+# ---------------------------------------------------------------------------
+
+# Max concurrent OnlyOffice conversion requests. OO has its own queue;
+# we must not flood it. This is intentionally small.
+OO_MAX_CONCURRENT = max(2, min(8, cpu_count()))
+
+
+class OOConversionManager:
+    """Manages async OnlyOffice conversions with deduplication and concurrency limits."""
+
+    def __init__(self, max_concurrent: int = OO_MAX_CONCURRENT):
+        self._semaphore = asyncio.Semaphore(max_concurrent)
+        self._in_flight: dict[str, asyncio.Future[bytes]] = {}
+        self._tasks: set[asyncio.Task[None]] = set()
+        self._lock = asyncio.Lock()
+
+    async def convert(self, filepath: Path) -> bytes:
+        """Return PNG bytes for *filepath*, deduplicating concurrent requests."""
+        if not await is_available_cached():
+            raise RuntimeError("OnlyOffice server not reachable")
+        stat = await asyncio.to_thread(filepath.stat)
+        key = f"{filepath}:{stat.st_mtime_ns}"
+
+        async with self._lock:
+            if key in self._in_flight:
+                future = self._in_flight[key]
+            else:
+                future = asyncio.get_running_loop().create_future()
+                self._in_flight[key] = future
+                task = asyncio.create_task(self._do_convert(filepath, key, future))
+                self._tasks.add(task)
+                task.add_done_callback(self._tasks.discard)
+
+        return await future
+
+    async def _do_convert(
+        self, filepath: Path, key: str, future: asyncio.Future[bytes]
+    ) -> None:
+        try:
+            async with self._semaphore:
+                png_bytes = await convert_to_png_async(filepath, request_timeout=5.0)
+        except Exception as e:
+            if not future.done():
+                future.set_exception(e)
+            async with self._lock:
+                self._in_flight.pop(key, None)
+        else:
+            if not future.done():
+                future.set_result(png_bytes)
+            async with self._lock:
+                self._in_flight.pop(key, None)
+
+
+_oo_manager: OOConversionManager | None = None
+
+
+def get_oo_manager() -> OOConversionManager:
+    """Return the singleton OOConversionManager."""
+    global _oo_manager
+    if _oo_manager is None:
+        _oo_manager = OOConversionManager(max_concurrent=OO_MAX_CONCURRENT)
+    return _oo_manager

mediapreview/pool.py

@@ -0,0 +1,488 @@
+"""Async preview worker pool framework."""
+
+import asyncio
+import contextlib
+import logging
+import os
+import signal
+import struct
+import sys
+from multiprocessing import cpu_count
+from pathlib import Path
+from time import perf_counter
+
+import msgspec
+from blake3 import blake3
+
+from mediapreview.formats import (
+    expected_backend as _expected_preview_backend,
+)
+from mediapreview.formats import (
+    is_previewable_path,
+)
+from mediapreview.formats import (
+    preview_job_priority as _preview_job_priority,
+)
+from mediapreview.office import get_oo_manager
+from mediapreview.protocol import PreviewRequest, PreviewResponse
+
+__all__ = [
+    "PREVIEW_TIMEOUT",
+    "PreviewError",
+    "PreviewPoolClosedError",
+    "PreviewTimeoutError",
+    "generate_office_preview",
+    "is_previewable_path",
+    "run_preview",
+    "shutdown_preview_workers",
+    "start_preview_workers",
+]
+
+logger = logging.getLogger(__name__)
+
+
+class PreviewTimeoutError(Exception):
+    """Raised when the preview subprocess exceeds PREVIEW_TIMEOUT."""
+
+    def __init__(self, message: str, *, backend: str | None = None):
+        super().__init__(message)
+        self.backend = backend
+
+
+class PreviewError(Exception):
+    """Raised when the preview subprocess exits with a non-zero status."""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        stderr: str | None = None,
+        backend: str | None = None,
+    ):
+        super().__init__(message)
+        self.stderr = stderr
+        self.backend = backend
+
+
+class PreviewPoolClosedError(PreviewError):
+    """The preview worker pool has been shut down."""
+
+
+class WorkerChecksumError(Exception):
+    """Raised when worker response checksum does not match the packet."""
+
+
+class WorkerProtocolError(Exception):
+    """Raised when worker response packet is malformed."""
+
+
+PREVIEW_TIMEOUT = 10.0  # seconds until preview subprocess is killed
+PREVIEW_WORKERS = max(2, min(8, cpu_count()))
+WORKER_KILL_GRACE = 5.0  # max seconds to wait for a killed worker to be reaped
+WORKER_RESPAWN_DELAY = 1.0  # initial delay before retrying a failed worker spawn
+WORKER_RESPAWN_DELAY_MAX = 30.0
+WORKER_CHECKSUM_BYTES = 32
+WORKER_MAX_JSON_BYTES = 1_000_000
+
+_active_procs: set[asyncio.subprocess.Process] = set()
+_preview_pool = None
+_preview_pool_lock = asyncio.Lock()
+_pool_stopped = False
+
+
+class _PreviewWorker:
+    def __init__(self, proc: asyncio.subprocess.Process):
+        self.proc = proc
+
+    async def request(
+        self,
+        filepath,
+        quality: int,
+        maxsize: int,
+        maxzoom: float,
+        data: bytes | None = None,
+    ):
+        if self.proc.returncode is not None:
+            raise WorkerProtocolError("worker already exited")
+        if self.proc.stdin is None or self.proc.stdout is None:
+            raise WorkerProtocolError("worker streams not available")
+
+        meta = msgspec.json.encode(
+            PreviewRequest(
+                path=str(filepath),
+                quality=quality,
+                maxsize=maxsize,
+                maxzoom=maxzoom,
+            )
+        )
+        payload = data or b""
+        packet = struct.pack("<II", len(meta), len(payload)) + meta + payload
+        self.proc.stdin.write(packet)
+        await self.proc.stdin.drain()
+
+        checksum = await self.proc.stdout.readexactly(WORKER_CHECKSUM_BYTES)
+        header = await self.proc.stdout.readexactly(8)
+        json_size, data_size = struct.unpack("<II", header)
+        if json_size > WORKER_MAX_JSON_BYTES:
+            raise WorkerProtocolError(f"worker JSON too large: {json_size}")
+        meta_raw = await self.proc.stdout.readexactly(json_size)
+        payload = await self.proc.stdout.readexactly(data_size)
+        packet = header + meta_raw + payload
+        if blake3(packet).digest() != checksum:
+            raise WorkerChecksumError("worker checksum mismatch")
+
+        resp = msgspec.json.decode(meta_raw, type=PreviewResponse)
+        if not resp.ok:
+            raise PreviewError(
+                resp.error or "preview worker error",
+                stderr=resp.stderr,
+                backend=resp.backend,
+            )
+        return payload or None, resp
+
+    async def kill(self) -> None:
+        try:
+            if self.proc.returncode is None:
+                # Safe to hard-kill: the worker is stateless per request.
+                # Kill the whole process group (worker is the group leader,
+                # spawned with start_new_session) so that an in-flight ffmpeg
+                # grandchild cannot be orphaned by the worker's SIGKILL.
+                # proc.wait() must not be awaited unaided: if a pipe
+                # transport is flow-control paused (e.g. an undrained stderr
+                # pipe), asyncio may never resolve wait() even after SIGKILL,
+                # which would permanently wedge the calling dispatcher.
+                with contextlib.suppress(ProcessLookupError):
+                    os.killpg(self.proc.pid, signal.SIGKILL)
+                try:
+                    await asyncio.wait_for(self.proc.wait(), timeout=WORKER_KILL_GRACE)
+                except TimeoutError:
+                    logger.exception(
+                        "Preview worker pid=%s not reaped within %ds of kill",
+                        self.proc.pid,
+                        int(WORKER_KILL_GRACE),
+                    )
+        finally:
+            _active_procs.discard(self.proc)
+
+
+class _PreviewWorkerPool:
+    def __init__(self, size: int):
+        self.size = size
+        self._idle: asyncio.Queue[_PreviewWorker] = asyncio.Queue()
+        self._pending: asyncio.PriorityQueue[tuple[int, int, asyncio.Future, tuple]] = (
+            asyncio.PriorityQueue()
+        )
+        self._workers: set[_PreviewWorker] = set()
+        self._dispatchers: list[asyncio.Task] = []
+        self._in_flight: set[asyncio.Future] = set()
+        self._seq = 0
+        self._closed = False
+
+    async def _spawn_worker(self) -> _PreviewWorker:
+        # stderr is inherited, not piped: a piped stderr that nobody drains
+        # eventually fills its OS buffer, blocking the worker mid-request,
+        # and its flow-control-paused transport makes proc.wait() hang even
+        # after kill() — together this used to permanently wedge the pool.
+        # Inheriting sends worker diagnostics straight to the server log.
+        proc = await asyncio.create_subprocess_exec(
+            sys.executable,
+            "-m",
+            "mediapreview.worker",
+            stdin=asyncio.subprocess.PIPE,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=None,
+            # Own process group so kill() can SIGKILL the worker together with
+            # any grandchild (e.g. ffmpeg) it may have spawned.
+            start_new_session=True,
+        )
+        _active_procs.add(proc)
+        try:
+            ready = await asyncio.wait_for(proc.stdout.readexactly(1), timeout=30.0)
+        except TimeoutError as err:
+            with contextlib.suppress(ProcessLookupError):
+                os.killpg(proc.pid, signal.SIGKILL)
+            with contextlib.suppress(Exception):
+                await proc.wait()
+            raise WorkerProtocolError(
+                "preview worker failed to become ready"
+                " (worker stderr goes to the server log)"
+            ) from err
+        except asyncio.IncompleteReadError as err:
+            raise WorkerProtocolError(
+                "preview worker exited before signalling readiness"
+                " (worker stderr goes to the server log)"
+            ) from err
+        if ready != b"\x01":
+            raise WorkerProtocolError(f"preview worker ready signal invalid: {ready!r}")
+        return _PreviewWorker(proc)
+
+    async def _add_worker(self) -> None:
+        worker = await self._spawn_worker()
+        self._workers.add(worker)
+        await self._idle.put(worker)
+
+    async def _replace_worker(self, worker: _PreviewWorker) -> None:
+        self._workers.discard(worker)
+        try:
+            await worker.kill()
+        except Exception:
+            logger.exception("Failed to kill preview worker pid=%s", worker.proc.pid)
+        # Keep retrying until a replacement is up: a pool that silently
+        # shrinks degrades all preview traffic to timeouts.
+        delay = WORKER_RESPAWN_DELAY
+        while not self._closed:
+            try:
+                await self._add_worker()
+            except Exception:
+                logger.exception(
+                    "Failed to replace preview worker (pool %d/%d); retrying in %ds",
+                    len(self._workers),
+                    self.size,
+                    int(delay),
+                )
+                await asyncio.sleep(delay)
+                delay = min(delay * 2, WORKER_RESPAWN_DELAY_MAX)
+            else:
+                return
+
+    async def _dispatch_loop(self) -> None:
+        # Nothing may escape the loop body: a dispatcher that dies silently
+        # permanently shrinks pool capacity and degrades all preview
+        # traffic to timeouts.
+        while True:
+            try:
+                await self._dispatch_one()
+            except asyncio.CancelledError:
+                return
+            except Exception:
+                logger.exception("Preview dispatcher error; continuing")
+
+    async def _dispatch_one(self) -> None:
+        _priority, _seq, future, args = await self._pending.get()
+
+        if future.cancelled():
+            return
+
+        try:
+            worker = await asyncio.wait_for(self._idle.get(), timeout=PREVIEW_TIMEOUT)
+        except TimeoutError:
+            logger.warning(
+                "Preview worker unavailable (%ds) for %s",
+                int(PREVIEW_TIMEOUT),
+                args[0].name,
+            )
+            if not future.done():
+                future.set_exception(
+                    PreviewTimeoutError(
+                        args[0].name,
+                        backend=_expected_preview_backend(args[0]),
+                    )
+                )
+            return
+
+        filepath = args[0]
+        replace = False
+        try:
+            out, resp = await asyncio.wait_for(
+                worker.request(*args),
+                timeout=PREVIEW_TIMEOUT,
+            )
+            if not future.done():
+                future.set_result((out, resp))
+        except TimeoutError:
+            replace = True
+            logger.warning(
+                "Preview worker pid=%s timed out (%ds) on %s; replacing it",
+                worker.proc.pid,
+                int(PREVIEW_TIMEOUT),
+                filepath.name,
+            )
+            if not future.done():
+                future.set_exception(
+                    PreviewTimeoutError(
+                        filepath.name,
+                        backend=_expected_preview_backend(filepath),
+                    )
+                )
+        except WorkerChecksumError:
+            replace = True
+            logger.exception(
+                "Preview checksum mismatch for %s (worker pid=%s); replacing it",
+                filepath.name,
+                worker.proc.pid,
+            )
+            if not future.done():
+                future.set_exception(
+                    PreviewError(f"worker checksum mismatch for {filepath.name}")
+                )
+        except PreviewError as e:
+            if not future.done():
+                future.set_exception(e)
+        except (
+            WorkerProtocolError,
+            asyncio.IncompleteReadError,
+            BrokenPipeError,
+            ConnectionResetError,
+            OSError,
+            ValueError,
+            msgspec.DecodeError,
+        ) as e:
+            replace = True
+            logger.warning(
+                "Preview worker pid=%s protocol failure for %s: %s",
+                worker.proc.pid,
+                filepath.name,
+                e,
+            )
+            if not future.done():
+                future.set_exception(
+                    PreviewError(f"worker protocol failure for {filepath.name}: {e}")
+                )
+        except Exception:
+            replace = True
+            logger.exception("Unexpected preview worker error for %s", filepath.name)
+            if not future.done():
+                future.set_exception(
+                    PreviewError(f"unexpected worker error for {filepath.name}")
+                )
+        finally:
+            if replace:
+                await self._replace_worker(worker)
+            elif worker.proc.returncode is None:
+                await self._idle.put(worker)
+            else:
+                await self._replace_worker(worker)
+
+    async def start(self) -> None:
+        workers = await asyncio.gather(
+            *(self._spawn_worker() for _ in range(self.size))
+        )
+        for worker in workers:
+            self._workers.add(worker)
+            await self._idle.put(worker)
+        for _ in range(self.size):
+            self._dispatchers.append(asyncio.create_task(self._dispatch_loop()))
+
+    async def run(
+        self,
+        filepath,
+        quality: int,
+        maxsize: int,
+        maxzoom: float,
+        data: bytes | None = None,
+    ):
+        if self._closed:
+            raise PreviewPoolClosedError("preview worker pool closed")
+        loop = asyncio.get_running_loop()
+        future = loop.create_future()
+        self._in_flight.add(future)
+        self._seq += 1
+        await self._pending.put(
+            (
+                _preview_job_priority(filepath),
+                self._seq,
+                future,
+                (filepath, quality, maxsize, maxzoom, data),
+            )
+        )
+        try:
+            return await future
+        finally:
+            self._in_flight.discard(future)
+
+    async def close(self) -> None:
+        self._closed = True
+        for task in self._dispatchers:
+            task.cancel()
+        if self._dispatchers:
+            await asyncio.gather(*self._dispatchers, return_exceptions=True)
+        self._dispatchers.clear()
+        workers = list(self._workers)
+        self._workers.clear()
+        # Fail every future still waiting on a result — pending and in-flight
+        # alike — so request handlers finish immediately instead of waiting
+        # out their timeouts during server shutdown.
+        for future in list(self._in_flight):
+            if not future.done():
+                future.set_exception(
+                    PreviewPoolClosedError("preview worker pool closed")
+                )
+        while not self._pending.empty():
+            try:
+                _priority, _seq, future, _args = self._pending.get_nowait()
+            except asyncio.QueueEmpty:
+                break
+            if not future.done():
+                future.set_exception(
+                    PreviewPoolClosedError("preview worker pool closed")
+                )
+        while not self._idle.empty():
+            try:
+                self._idle.get_nowait()
+            except asyncio.QueueEmpty:
+                break
+        await asyncio.gather(
+            *(worker.kill() for worker in workers), return_exceptions=True
+        )
+
+
+async def start_preview_workers() -> None:
+    """Warm up persistent preview workers."""
+    global _preview_pool
+    if _preview_pool is not None or _pool_stopped:
+        return
+    async with _preview_pool_lock:
+        if _preview_pool is not None or _pool_stopped:
+            return
+        pool = _PreviewWorkerPool(PREVIEW_WORKERS)
+        await pool.start()
+        _preview_pool = pool
+        logger.info("Started %d persistent preview workers", PREVIEW_WORKERS)
+
+
+async def shutdown_preview_workers() -> None:
+    """Kill persistent preview workers."""
+    global _preview_pool, _pool_stopped
+    _pool_stopped = True
+    async with _preview_pool_lock:
+        pool = _preview_pool
+        _preview_pool = None
+    if pool is not None:
+        await pool.close()
+    if not _active_procs:
+        return
+    for proc in list(_active_procs):
+        with contextlib.suppress(ProcessLookupError):
+            os.killpg(proc.pid, signal.SIGKILL)
+    await asyncio.gather(
+        *(proc.wait() for proc in list(_active_procs)), return_exceptions=True
+    )
+    _active_procs.clear()
+
+
+async def generate_office_preview(
+    filepath: Path, quality: int, maxsize: int, maxzoom: float
+) -> tuple[bytes | None, PreviewResponse | None]:
+    """Generate a preview for an office file using OnlyOffice + worker AVIF conversion."""
+    manager = get_oo_manager()
+    t_oo_start = perf_counter()
+    png_bytes = await manager.convert(filepath)
+    t_oo_end = perf_counter()
+
+    img, resp = await run_preview(filepath, quality, maxsize, maxzoom, data=png_bytes)
+
+    if resp is not None:
+        resp.backend = "onlyoffice+" + (resp.backend or "pyvips")
+        if resp.timings:
+            resp.timings = [round((t_oo_end - t_oo_start) * 1000, 1), *resp.timings]
+    return img, resp
+
+
+async def run_preview(
+    filepath, quality: int, maxsize: int, maxzoom: float, data: bytes | None = None
+) -> tuple[bytes | None, PreviewResponse | None]:
+    """Run preview request in a persistent worker process."""
+    await start_preview_workers()
+    if _preview_pool is None:
+        raise PreviewPoolClosedError("preview worker pool closed")
+    return await _preview_pool.run(filepath, quality, maxsize, maxzoom, data)

mediapreview/protocol.py

@@ -0,0 +1,21 @@
+"""Wire protocol structs for the preview worker pool."""
+
+import msgspec
+
+
+class PreviewRequest(msgspec.Struct, omit_defaults=True):
+    path: str
+    quality: int
+    maxsize: int
+    maxzoom: float
+
+
+class PreviewResponse(msgspec.Struct, omit_defaults=True):
+    ok: bool
+    mime: str | None = None
+    backend: str | None = None
+    timings: list[float] | None = None
+    error: str | None = None
+    stderr: str | None = None
+    width: int | None = None
+    height: int | None = None

mediapreview/util/__init__.py

@@ -0,0 +1 @@
+"""Utility helpers for the mediapreview package."""

mediapreview/util/logformat.py

@@ -0,0 +1,33 @@
+"""Shared log formatting helpers."""
+
+import logging
+import unicodedata
+
+LEVEL_EMOJI = {
+    logging.DEBUG: "🔍",
+    logging.INFO: "ℹ️",  # noqa: RUF001
+    logging.WARNING: "⚠️",
+    logging.ERROR: "🛑",
+    logging.CRITICAL: "🛑",
+}
+
+
+def display_width(text: str) -> int:
+    return sum(
+        1 + (unicodedata.east_asian_width(c) in "FW")
+        for c in text
+        if unicodedata.category(c) != "Mn"
+    )
+
+
+def format_level_prefix(levelno: int) -> str:
+    emoji = LEVEL_EMOJI.get(levelno, "▪️")
+    prefix = f"{emoji} "
+    return prefix + (" " * max(0, 3 - display_width(prefix)))
+
+
+class EmojiFormatter(logging.Formatter):
+    """Compact formatter: emoji + message, no timestamp/level text/logger name."""
+
+    def format(self, record: logging.LogRecord) -> str:
+        return format_level_prefix(record.levelno) + record.getMessage()

mediapreview/worker.py

@@ -0,0 +1,175 @@
+"""Preview generation worker subprocess entry point.
+
+The actual conversion logic lives in `mediapreview.backends`; this module
+only implements the worker process shell around it.
+
+Two modes are supported:
+1) Legacy one-shot mode: argv has path/quality/maxsize/maxzoom.
+2) Long-lived mode: read framed requests from stdin and write framed responses.
+
+Framed request format (stdin):
+    (uint32 json size)(uint32 data size)(json)(binary data)
+
+Framed response format (stdout):
+    (blake3(packet))(uint32 json size)(uint32 payload size)(json)(binary payload)
+where packet = (uint32 json size)(uint32 payload size)(json)(binary payload).
+"""
+
+import contextlib
+import io
+import logging
+import os
+import signal
+import struct
+import sys
+from pathlib import Path
+
+import msgspec
+import tracerite
+from blake3 import blake3
+
+from mediapreview.backends import dispatch
+from mediapreview.protocol import PreviewRequest, PreviewResponse
+from mediapreview.util.logformat import format_level_prefix
+
+logger = logging.getLogger(__name__)
+
+
+class _WorkerLogFormatter(logging.Formatter):
+    """Emoji level prefix like the main process, tagged with the worker pid."""
+
+    def format(self, record: logging.LogRecord) -> str:
+        prefix = format_level_prefix(record.levelno)
+        return f"{prefix}worker[{os.getpid()}]: {record.getMessage()}"
+
+
+_enc = msgspec.json.Encoder()
+_dec_req = msgspec.json.Decoder(PreviewRequest)
+
+
+def _read_exactly(f, n: int) -> bytes:
+    buf = b""
+    while len(buf) < n:
+        chunk = f.read(n - len(buf))
+        if not chunk:
+            raise EOFError
+        buf += chunk
+    return buf
+
+
+def _read_request() -> tuple[PreviewRequest, bytes] | None:
+    try:
+        header = _read_exactly(sys.stdin.buffer, 8)
+    except EOFError:
+        return None
+    json_size, data_size = struct.unpack("<II", header)
+    meta_raw = _read_exactly(sys.stdin.buffer, json_size)
+    data = b""
+    if data_size:
+        data = _read_exactly(sys.stdin.buffer, data_size)
+    req = _dec_req.decode(meta_raw)
+    return req, data
+
+
+# Raw stdout buffer reserved for the binary protocol once main() redirects
+# Python-level stdout to stderr. None means "use sys.stdout.buffer as-is"
+# (CLI single-shot mode, where real stdout is wanted).
+_protocol_out = None
+
+
+def _write_response(resp: PreviewResponse, payload: bytes) -> None:
+    out = _protocol_out if _protocol_out is not None else sys.stdout.buffer
+    meta_bytes = _enc.encode(resp)
+    packet = struct.pack("<II", len(meta_bytes), len(payload)) + meta_bytes + payload
+    checksum = blake3(packet).digest()
+    out.write(checksum)
+    out.write(packet)
+    out.flush()
+
+
+def _run_once() -> None:
+    if len(sys.argv) != 5:
+        sys.stderr.write(f"Usage: {sys.argv[0]} <path> <quality> <maxsize> <maxzoom>\n")
+        sys.exit(1)
+
+    path = Path(sys.argv[1])
+    quality = int(sys.argv[2])
+    maxsize = int(sys.argv[3])
+    maxzoom = float(sys.argv[4])
+    result, _ = dispatch(path, quality, maxsize, maxzoom)
+    if result:
+        sys.stdout.buffer.write(result)
+        sys.stdout.buffer.flush()
+
+
+def _run_loop() -> None:
+    while True:
+        result = _read_request()
+        if result is None:
+            return
+        req, data = result
+        stderr_capture = io.StringIO()
+        handler = logging.StreamHandler(stderr_capture)
+        root_logger = logging.getLogger()
+        root_logger.addHandler(handler)
+        try:
+            with contextlib.redirect_stderr(stderr_capture):
+                result, resp = dispatch(
+                    Path(req.path), req.quality, req.maxsize, req.maxzoom, data
+                )
+            if not resp.ok:
+                captured = stderr_capture.getvalue().strip()
+                if captured:
+                    resp = PreviewResponse(
+                        ok=False,
+                        backend=resp.backend,
+                        error=resp.error,
+                        stderr=captured,
+                    )
+            _write_response(resp, result or b"")
+        except Exception as e:
+            logger.exception("Preview worker error for %s", req.path)
+            captured = stderr_capture.getvalue().strip()
+            _write_response(
+                PreviewResponse(ok=False, error=str(e), stderr=captured or None), b""
+            )
+        finally:
+            root_logger.removeHandler(handler)
+            handler.close()
+
+
+def main() -> None:
+    # Format tracebacks like the main process.
+    tracerite.load()
+    # Configure all log output to stderr before any imports that may emit
+    # logs. stderr is inherited by the parent, so this lands in the server
+    # log, formatted like the main process and tagged with the worker pid.
+    handler = logging.StreamHandler(sys.stderr)
+    handler.setFormatter(_WorkerLogFormatter())
+    logging.basicConfig(level=logging.INFO, handlers=[handler])
+    # pyvips is chatty at INFO ("threadpool completed ..." per operation).
+    logging.getLogger("pyvips").setLevel(logging.WARNING)
+    # NOTE: standalone package no longer depends on cista config loading.
+    # Consumers can load their own configuration before starting workers.
+    if len(sys.argv) > 1:
+        _run_once()
+        return
+    # Ctrl-C SIGINTs the whole process group; the parent pool terminates us
+    # (and our stdin EOF exits us) — don't dump KeyboardInterrupt tracebacks.
+    signal.signal(signal.SIGINT, signal.SIG_IGN)
+    # The command channel is a binary protocol on fd 1. Anything printed to
+    # stdout by Python code (e.g. a library emitting a warning via print())
+    # would corrupt the protocol, so redirect Python-level stdout to stderr
+    # (the server log) and keep the raw buffer solely for protocol traffic.
+    global _protocol_out
+    _protocol_out = sys.stdout.buffer
+    sys.stdout = sys.stderr
+    # Eagerly import heavy modules before signalling readiness so the parent
+    # does not hand us a request while we are still initialising.
+    _protocol_out.write(b"\x01")
+    _protocol_out.flush()
+    _run_loop()
+
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -0,0 +1,98 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "mediapreview"
+version = "0.1.0"
+description = "Low-level media preview converters and a worker pool framework"
+readme = "README.md"
+requires-python = ">=3.11"
+dependencies = [
+    "msgspec>=0.19.0",
+    "pyvips[binary]>=3.1.1",
+]
+
+[project.optional-dependencies]
+worker = [
+    "blake3>=1.0.5",
+    "tracerite>=2.3.1",
+]
+pdf = [
+    "pymupdf>=1.26.3",
+]
+video = [
+    "av>=15.0.0",
+    "numpy>=2.3.2",
+]
+office = [
+    "httpx>=0.28.0",
+    "pyjwt>=2.10.1",
+]
+standard = [
+    "mediapreview[worker]",
+    "mediapreview[pdf]",
+    "mediapreview[video]",
+    "mediapreview[office]",
+]
+dev = [
+    "pytest>=8.4.1",
+    "pytest-asyncio>=0.25.0",
+    "pytest-cov>=6.0.0",
+    "ruff>=0.8.0",
+    "mypy>=1.13.0",
+]
+
+[tool.hatch.build.targets.wheel]
+packages = ["mediapreview"]
+
+[tool.ruff]
+target-version = "py311"
+
+[tool.ruff.lint]
+select = ["ALL"]
+ignore = [
+    "COM812",
+    "ISC001",
+    "ANN001",
+    "ANN002",
+    "ANN003",
+    "ANN201",
+    "ANN202",
+    "ANN204",
+    "ANN205",
+    "BLE001",
+    "C901",
+    "CPY",
+    "D100",
+    "D101",
+    "D102",
+    "D103",
+    "D104",
+    "D105",
+    "D107",
+    "D200",
+    "D203",
+    "D212",
+    "D213",
+    "D400",
+    "D401",
+    "D413",
+    "D415",
+    "E501",
+    "EM101",
+    "EM102",
+    "INP001",
+    "PLR0911",
+    "PLR0912",
+    "PLR0913",
+    "PLR0915",
+    "PLR2004",
+    "PLW0603",
+    "TRY003",
+]
+isort.known-first-party = ["mediapreview"]
+per-file-ignores."tests/*" = ["S", "ANN", "D", "INP", "PLR2004", "ARG001", "SLF001"]
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"

tests/test_pool.py

@@ -0,0 +1,175 @@
+"""Tests for the preview worker pool resilience.
+
+Regression context: a piped worker stderr that nobody drains used to block
+the worker mid-request once the OS pipe buffer filled, and asyncio's
+proc.wait() then never resolved even after kill() — wedging one dispatcher
+per stuck worker until all preview traffic timed out permanently.
+"""
+
+import asyncio
+import sys
+import textwrap
+import time
+from pathlib import Path
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+
+from mediapreview import pool
+
+FAKE_WORKER = textwrap.dedent(
+    """
+    import json
+    import struct
+    import sys
+
+    import blake3
+
+
+    def read_exact(n):
+        buf = b""
+        while len(buf) < n:
+            chunk = sys.stdin.buffer.read(n - len(buf))
+            if not chunk:
+                raise EOFError
+            buf += chunk
+        return buf
+
+
+    sys.stdout.buffer.write(b"\\x01")
+    sys.stdout.buffer.flush()
+    while True:
+        header = sys.stdin.buffer.read(8)
+        if not header or len(header) < 8:
+            break
+        meta_len, payload_len = struct.unpack("<II", header)
+        meta = read_exact(meta_len)
+        read_exact(payload_len)
+        req = json.loads(meta)
+        if req["path"].endswith(".block"):
+            # Simulate a worker stuck on an undrained stderr pipe:
+            # flood stderr past the OS pipe buffer, then never respond.
+            import os
+            import time
+
+            try:
+                os.write(2, b"x" * 10_000_000)
+            except OSError:
+                pass
+            while True:
+                time.sleep(3600)
+        resp = json.dumps({"ok": True, "mime": "image/avif", "backend": "fake"}).encode()
+        payload = b"FAKEIMG"
+        packet = struct.pack("<II", len(resp), len(payload)) + resp + payload
+        sys.stdout.buffer.write(blake3.blake3(packet).digest())
+        sys.stdout.buffer.write(packet)
+        sys.stdout.buffer.flush()
+    """
+)
+
+
+@pytest.mark.asyncio
+async def test_pool_recovers_from_wedged_worker(monkeypatch, tmp_path):
+    """A worker wedged on an undrained stderr pipe must not kill the pool.
+
+    Recreates the old production setup (stderr=PIPE, never drained) and
+    verifies the request times out, the stuck worker's kill() cannot hang
+    the dispatcher, and the pool serves the next request normally.
+    """
+    monkeypatch.setattr(pool, "PREVIEW_TIMEOUT", 1.0)
+    monkeypatch.setattr(pool, "WORKER_KILL_GRACE", 0.5)
+    monkeypatch.setattr(pool, "WORKER_RESPAWN_DELAY", 0.05)
+    script = tmp_path / "fake_worker.py"
+    script.write_text(FAKE_WORKER)
+
+    async def fake_spawn(self):
+        proc = await asyncio.create_subprocess_exec(
+            sys.executable,
+            str(script),
+            stdin=asyncio.subprocess.PIPE,
+            stdout=asyncio.subprocess.PIPE,
+            # Deliberately piped-and-undrained, recreating the old
+            # production setup that wedges a worker on stderr writes.
+            stderr=asyncio.subprocess.PIPE,
+        )
+        pool._active_procs.add(proc)
+        await asyncio.wait_for(proc.stdout.readexactly(1), timeout=10)
+        return pool._PreviewWorker(proc)
+
+    monkeypatch.setattr(pool._PreviewWorkerPool, "_spawn_worker", fake_spawn)
+
+    worker_pool = pool._PreviewWorkerPool(1)
+    await worker_pool.start()
+    try:
+        with pytest.raises(pool.PreviewTimeoutError):
+            await worker_pool.run(Path("wedged.block"), 60, 512, 2.0)
+
+        out, resp = await asyncio.wait_for(
+            worker_pool.run(Path("ok.jpg"), 60, 512, 2.0), timeout=10
+        )
+        assert out == b"FAKEIMG"
+        assert resp.ok
+        assert all(not task.done() for task in worker_pool._dispatchers)
+    finally:
+        await worker_pool.close()
+
+
+@pytest.mark.asyncio
+async def test_worker_kill_grace_when_wait_hangs(monkeypatch):
+    """kill() must return even if asyncio never resolves proc.wait()."""
+    monkeypatch.setattr(pool, "WORKER_KILL_GRACE", 0.1)
+    proc = Mock()
+    proc.returncode = None
+    proc.pid = 1234
+    never = asyncio.Future()
+
+    async def wait():
+        await never
+
+    proc.wait = wait
+    worker = pool._PreviewWorker(proc)
+    pool._active_procs.add(proc)
+    start = time.monotonic()
+    await worker.kill()
+    assert time.monotonic() - start < 2
+    assert proc not in pool._active_procs
+    never.cancel()
+
+
+@pytest.mark.asyncio
+async def test_replace_worker_retries_failed_spawn(monkeypatch):
+    """A failed replacement spawn must be retried, not silently dropped."""
+    monkeypatch.setattr(pool, "WORKER_RESPAWN_DELAY", 0.01)
+    worker_pool = pool._PreviewWorkerPool(1)
+    old_worker = Mock()
+    old_worker.proc = Mock(pid=4321)
+    old_worker.kill = AsyncMock()
+    attempts = 0
+
+    async def add_worker():
+        nonlocal attempts
+        attempts += 1
+        if attempts < 3:
+            raise OSError("too many open files")
+
+    worker_pool._add_worker = add_worker
+    await worker_pool._replace_worker(old_worker)
+    assert attempts == 3
+
+
+@pytest.mark.asyncio
+async def test_dispatch_loop_survives_body_errors():
+    """Exceptions escaping a dispatch cycle must not kill the dispatcher."""
+    worker_pool = pool._PreviewWorkerPool(1)
+    calls = 0
+
+    async def dispatch_one():
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            raise RuntimeError("boom")
+        raise asyncio.CancelledError
+
+    worker_pool._dispatch_one = dispatch_one
+    await worker_pool._dispatch_loop()
+    assert calls == 2