fix: harden utility export recovery

README.md

@@ -23,7 +23,8 @@ Browser / HA / Companion
 Flask -> bounded event queue -> single Controller coordinator
                                   |        |          |
                                   |        |          +-> SQLite state/audit
-                                  |        +------------> bounded yt-dlp workers
+                                  |        +------------> Player-slot workers
+                                  |        +------------> utility-export workers
                                   +---------------------> MQTT
                                                             |
                               +-----------------------------+------------------+
@@ -411,6 +412,62 @@ state rather than a completed delete.
 Expiry never compacts slots. An expired slot becomes empty; every other slot
 keeps its number and canonical directory.
 
+## Utility exports
+
+The submission page has four explicit modes. `Player Slot` is selected by
+default, and an older client that omits `download_mode` still takes that path.
+`MP4 Export`, `M4A Audio`, and `MP3 Audio` use a separate durable job queue and
+dedicated worker pool. They never reserve or mutate a Player slot, publish
+ingest/HA/MQTT state, acquire the Audio Router, create Player metadata, or enter
+Player playback, expiry, deletion, or reconciliation gates.
+
+Player Slot remains quality-first and does not force MP4 or H.264. Its strict
+yt-dlp selector is `bv*[height<=H]+ba/b[height<=H]`: separate 1080p video and
+audio are eligible, a lower-resolution source is not upscaled, and no
+unfiltered fallback may exceed `downloads.max_height`.
+
+Utility modes behave as follows:
+
+- MP4 selects the best source at or below `downloads.export_max_height`, with
+  native H.264/AAC preferred only among otherwise equal-quality choices. It
+  probes every result with ffprobe. An existing MP4/H.264/AAC/yuv420p result is
+  moved without remuxing or transcoding; otherwise ffmpeg copies each compatible
+  stream and converts only an incompatible video stream to H.264 or audio stream
+  to AAC. The output height must equal the selected source height, so export
+  never upscales, and the final file is verified as MP4/H.264/AAC.
+- M4A selects best available audio, prefers native AAC among equal-quality
+  choices, extracts audio only, and verifies an AAC stream in an M4A/MP4
+  container. Native AAC is losslessly repackaged by yt-dlp; another source codec
+  is encoded only when required.
+- MP3 selects best available audio, extracts audio only, encodes with yt-dlp's
+  highest-quality VBR setting when required, and verifies an MP3-only result.
+
+Playlist exports preserve item order and use bounded, human-readable Unicode
+filenames with unsafe path/control characters removed and case-insensitive
+collision suffixes. Each job publishes atomically into its own directory:
+
+```text
+/mnt/Media/YTPlayer/exports/
+  video/<job-id>/*.mp4
+  audio/<job-id>/*.m4a
+  audio/<job-id>/*.mp3
+```
+
+Work is staged under `exports/.work/<job-id>`. A fsynced manifest makes the
+publication boundary recoverable: after restart, a dedicated export worker verifies and
+finishes a fully prepared or already-published job, while a known partial job
+is marked interrupted and its private staging directory is removed. It does
+not delay web readiness, MQTT reconnect, or Player coordination and does not
+resume a partial yt-dlp process. Unknown directories are left untouched.
+The web page lists live jobs before recent completed/failed/interrupted jobs and
+shows their output paths or error. It intentionally provides no file-serving,
+cancellation, or deletion endpoint.
+
+Utility output is deliberately outside slot retention. Controller never
+expires it; operators should monitor capacity and remove an entire completed
+`video/<job-id>` or `audio/<job-id>` directory according to local policy. Do not
+write untrusted files or symlinks into the media/export tree.
+
 ## Restart uncertainty and administrative reconciliation
 
 On startup, a last-known managed active slot or pending play is converted to a
@@ -552,15 +609,16 @@ Controller v2 uses one embedded SQLite database at
 `/controller-data/controller.sqlite3`. It introduces no external database
 service.
 
-Schema version 2 contains:
+Schema version 3 contains:
 
 | Table | Purpose and important constraints |
 |---|---|
-| `schema_migrations` | Applied internal schema versions; v1 databases upgrade transactionally to v2 |
+| `schema_migrations` | Applied internal schema versions; v1/v2 databases upgrade transactionally to v3 |
 | `data_migrations` | One-time import provenance, source hash, backup path, and warnings |
 | `slots` | Exactly one row for each stable slot 1-10, generation, media state, identity, timestamps, expiry, and error |
 | `slot_items` | Ordered item title/path metadata keyed by slot and position |
 | `download_jobs` | Reservation/worker lifecycle keyed by job ID and slot generation; one live job per slot |
+| `export_jobs` | Slot-free utility job mode/status, request metadata, output paths, and error; no slot foreign key |
 | `slot_guards` | Active, pending, restart-uncertain, and legacy-uncertain playback protection; one live guard per slot |
 | `player_observation` | Singleton last status/telemetry reduction, generation, active identity, pause, and reconciliation state/token |
 | `audio_router_observation` | Singleton last mode, freshness, timestamp, and warning |
@@ -615,8 +673,10 @@ discard the original or hash-named backup.
 
 Once `controller.sqlite3` exists, startup never imports YAML again and later
 YAML edits are ignored. That database-presence rule plus the `data_migrations`
-record makes the migration idempotent. An existing schema-v1 Controller SQLite
-database is upgraded transactionally rather than reimporting YAML.
+record makes the migration idempotent. Existing schema-v1 or schema-v2
+Controller SQLite databases are upgraded transactionally to schema v3 rather
+than reimporting YAML. The v2-to-v3 upgrade adds only the slot-free
+`export_jobs` table, its recent-jobs index, and migration record.
 
 Legacy state mapping is explicit:
 
@@ -686,8 +746,9 @@ The checked-in `config.yaml` is the canonical example. Important defaults are:
   queue, and 2-second QoS 1 PUBACK timeout.
 - Storage: 48-hour retention, 60-second expiry scan, and 15-minute uncertainty
   review deadline.
-- Downloads: two workers, 2-hour subprocess timeout, maximum height 1080, and
-  optional cookies.
+- Downloads: two Player-slot workers, one dedicated utility-export worker,
+  2-hour subprocess timeout, strict Player/export maximum height 1080, at most
+  20 queued/running exports, output below `exports`, and optional cookies.
 - Router: required mode `video`, 3-second confirmation timeout, unknown-state
   policy `attempt`, continue after confirmation failure, and no fallback restore
   mode.
@@ -718,8 +779,10 @@ permissions, and rotate it as a credential.
 
 The HTTP surface is intentionally small:
 
-- `GET /` renders the playlist submission form.
-- `POST /submit` accepts an HTTP(S) URL plus optional submitter and note.
+- `GET /` renders the submission form and recent utility-export status.
+- `POST /submit` accepts an HTTP(S) URL, optional submitter and note, and
+  `download_mode` equal to `player_slot`, `mp4`, `m4a`, or `mp3`. Omitting the
+  field remains equivalent to `player_slot`.
 - `GET /healthz` returns service and reconciliation state as JSON.
 
 Health includes `status`, `started`, `coordinator_alive`, `mqtt_connected`,
@@ -752,6 +815,8 @@ The production image and Compose file intentionally differ from Controller v1:
   of the old `/cookies_ro.txt` path.
 - The image includes ffmpeg and a Node 22 runtime required by the pinned yt-dlp
   YouTube JavaScript support.
+- Utility exports use the existing writable media bind mount and ffmpeg/ffprobe
+  installation; they add no service, port, volume, or external database.
 - The build currently uses `node:22.23.1-bookworm-slim` as a JavaScript-runtime
   stage and `python:3.12-slim-bookworm` as the final base. Runtime Python
   requirements are exact pins, including Flask 3.1.3, Gunicorn 26.0.0,
@@ -850,11 +915,22 @@ retry only while no final SQLite database exists. Returning to the recorded v1
 image with the untouched v1 data/media snapshot remains safe because migration
 does not modify them.
 
+### Upgrade from an earlier Controller v2 build
+
+Stop Controller and back up the complete data directory before installing this
+build. On first start, schema v1 or v2 is upgraded in one transaction to schema
+v3; existing slots, guards, observations, leases, jobs, and audit rows are not
+rewritten. Confirm `exports/.work`, `exports/video`, and `exports/audio` can be
+created by the configured container UID/GID, then submit one controlled export
+and inspect the path shown on the web page. The legacy `playlists.yaml` import
+does not rerun when the SQLite database already exists.
+
 ## Rollback
 
 Code rollback and state rollback are different operations. Controller v2 does
 not write changes back to `playlists.yaml`; v1 does not understand
-`controller.sqlite3`. After any v2 download, delete, expiry, or slot reuse, the
+`controller.sqlite3`. Earlier Controller v2 builds also do not understand
+schema v3. After any v2 download, delete, expiry, or slot reuse, the
 pre-migration YAML and current media tree may no longer agree.
 
 1. Stop Controller v2 cleanly. Leave Player Node unchanged.
@@ -881,6 +957,14 @@ For a migration-only failure before final SQLite installation, simply keep v2
 stopped and restart the recorded v1 deployment against the untouched original
 data and media; no v2 database state needs conversion.
 
+To roll back only this utility-export extension to an earlier v2 build, stop the
+service and restore the complete schema-v2 data-directory backup made before
+upgrade; do not point the older binary at the schema-v3 database. Utility files
+already published below `exports` are independent of slots and may be archived
+or left in place, but the older build will not list or manage them. Rolling
+forward again requires restoring a matched newer database backup or allowing a
+fresh schema-v2 backup to upgrade once more.
+
 ## Development and tests
 
 Application imports are side-effect free. Constructing the Flask app does not
@@ -904,7 +988,10 @@ Player generation, pathless busy/error handling and pause-timeout reduction,
 durable playback plus connection-scoped cleanup reconciliation, Audio Router
 timeouts and ownership loss, service integration, HA compatibility, HTTP
 health, and Docker/runtime assets. It uses fakes and temporary directories; a
-real-broker/container smoke test is still required for a deployment.
+real-broker/container smoke test is still required for a deployment. Utility
+coverage additionally exercises actual pinned yt-dlp selection semantics,
+Player/export isolation, queue bounds, codec command paths, safe names,
+failure/restart publication recovery, and schema-v2-to-v3 migration.
 
 Useful static and container checks are:
 
@@ -965,6 +1052,12 @@ PYTHONDONTWRITEBYTECODE=1 .venv/bin/python -m unittest discover -s tests -v
   timeout. The current job fails instead of resuming. A container kill before
   that bookkeeping completes is recovered as `interrupted`/`error`; queued
   deletion work remains durably `deleting` for the evidence-gated retry path.
+- Utility exports are not subject to slot retention and have no automatic disk
+  quota or cleanup policy. Monitor the media filesystem and remove completed
+  job directories administratively. Interrupted partial jobs are not resumed.
+- Export-path checks reject configured slot/trash overlap and encountered
+  symlinks, but the writable media tree is not a hostile multi-user sandbox.
+  Only trusted host accounts should be able to modify it while Controller runs.
 - The container is read-only and non-root, but the data and media mounts must be
   writable. Controller can intentionally delete canonical expired or
   administratively deleted slot directories; maintain independent backups.

app.py

@@ -18,7 +18,7 @@ HTML_FORM = """
 <html>
   <head>
     <meta charset="utf-8">
-    <title>YouTube Playlist Submit</title>
+    <title>YouTube Media Submit</title>
 
     <!-- Make it behave on mobile -->
     <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1">
@@ -96,6 +96,33 @@ HTML_FORM = """
         font-size: 1rem;
       }
 
+      input[type="radio"] {
+        width: auto;
+        margin: 0;
+      }
+
+      fieldset {
+        border: 0;
+        padding: 0;
+      }
+
+      legend {
+        margin-bottom: 0.35rem;
+        font-size: 0.92rem;
+      }
+
+      .download-options {
+        display: grid;
+        gap: 0.4rem;
+      }
+
+      .download-options label {
+        display: flex;
+        align-items: center;
+        gap: 0.55rem;
+        margin: 0;
+      }
+
       input::placeholder,
       textarea::placeholder {
         color: #6b7280;
@@ -139,6 +166,26 @@ HTML_FORM = """
         color: #9ca3af;
       }
 
+      .export-status {
+        margin-top: 1rem;
+        padding-top: 0.9rem;
+        border-top: 1px solid rgba(148, 163, 184, 0.25);
+      }
+
+      .export-job {
+        margin-top: 0.65rem;
+        padding: 0.65rem 0.75rem;
+        border-radius: 0.55rem;
+        background: rgba(30, 41, 59, 0.7);
+        font-size: 0.82rem;
+      }
+
+      .export-path,
+      .export-error {
+        overflow-wrap: anywhere;
+        color: #cbd5e1;
+      }
+
       @media (max-width: 480px) {
         .page-wrapper {
           padding: 1rem;
@@ -167,9 +214,9 @@ HTML_FORM = """
   <body>
     <div class="page-wrapper">
       <div class="card">
-        <h2>Submit YouTube Playlist</h2>
+        <h2>Submit YouTube Media</h2>
         <p class="subtitle">
-          Paste a playlist link below. It’ll be queued for download and assigned to the next free slot.
+          Paste a YouTube video or playlist link and choose a download type.
         </p>
 
         {% with messages = get_flashed_messages() %}
@@ -212,12 +259,40 @@ HTML_FORM = """
             ></textarea>
           </div>
 
-          <button type="submit">Submit Playlist</button>
+          <fieldset class="form-group">
+            <legend>Download as:</legend>
+            <div class="download-options">
+              <label><input type="radio" name="download_mode" value="player_slot" checked> Player Slot</label>
+              <label><input type="radio" name="download_mode" value="mp4"> MP4 Export</label>
+              <label><input type="radio" name="download_mode" value="m4a"> M4A Audio</label>
+              <label><input type="radio" name="download_mode" value="mp3"> MP3 Audio</label>
+            </div>
+          </fieldset>
+
+          <button type="submit">Submit</button>
         </form>
 
         <p class="helper">
           Playlists are kept for roughly 48 hours after they’re downloaded.
+          Utility exports are kept until an operator removes them.
         </p>
+
+        {% if export_jobs %}
+          <section class="export-status">
+            <strong>Recent utility exports</strong>
+            {% for job in export_jobs %}
+              <div class="export-job">
+                <div>{{ download_mode_labels[job.mode] }} — {{ export_status_labels[job.status] }} (job {{ job.job_id[:8] }})</div>
+                {% for path in job.output_paths %}
+                  <div class="export-path">{{ path }}</div>
+                {% endfor %}
+                {% if job.error %}
+                  <div class="export-error">{{ job.error }}</div>
+                {% endif %}
+              </div>
+            {% endfor %}
+          </section>
+        {% endif %}
       </div>
     </div>
   </body>
@@ -238,6 +313,19 @@ MAX_URL_LENGTH = 2048
 MAX_SUBMITTER_LENGTH = 200
 MAX_NOTE_LENGTH = 1000
 MAX_REQUEST_CONTENT_LENGTH = 32 * 1024
+DOWNLOAD_MODE_LABELS = {
+    "player_slot": "Player Slot",
+    "mp4": "MP4 Export",
+    "m4a": "M4A Audio",
+    "mp3": "MP3 Audio",
+}
+EXPORT_STATUS_LABELS = {
+    "queued": "queued",
+    "running": "downloading",
+    "completed": "completed",
+    "failed": "failed",
+    "interrupted": "interrupted",
+}
 
 
 def _valid_media_url(value: str) -> bool:
@@ -258,6 +346,20 @@ def _valid_media_url(value: str) -> bool:
 
 def _submission_message(result: dict[str, object]) -> str:
     status = result.get("status")
+    if result.get("kind") == "export":
+        mode = str(result.get("mode") or "")
+        label = DOWNLOAD_MODE_LABELS.get(mode, "Utility export")
+        if status == "queue_full":
+            return "The utility export queue is full; please try again later."
+        if status == "shutting_down":
+            return "The Controller is shutting down; please try again shortly."
+        if status == "worker_error":
+            return f"The {label} job was reserved, but its worker could not start."
+        if status != "queued":
+            return f"The Controller could not queue that {label} job; please try again."
+        job_id = str(result.get("job_id") or "")[:8]
+        identity = f" (job {job_id})" if job_id else ""
+        return f"{label} queued{identity}. Refresh this page to see its result."
     if status == "queue_full":
         seconds = max(0, int(result.get("oldest_expires_in_seconds") or 0))
         hours = max(1, (seconds + 3599) // 3600)
@@ -308,7 +410,19 @@ def create_app(
 
     @web_app.get("/")
     def index():
-        return render_template_string(HTML_FORM)
+        try:
+            export_jobs = service.list_export_jobs(
+                limit=min(100, config.downloads.max_queued_exports + 10)
+            )
+        except Exception as exc:
+            LOG.warning("could not load utility export status: %s", exc)
+            export_jobs = []
+        return render_template_string(
+            HTML_FORM,
+            export_jobs=export_jobs,
+            download_mode_labels=DOWNLOAD_MODE_LABELS,
+            export_status_labels=EXPORT_STATUS_LABELS,
+        )
 
     @web_app.errorhandler(413)
     def _submission_too_large(_error):
@@ -320,6 +434,8 @@ def create_app(
         url = request.form.get("url", "").strip()
         submitter = request.form.get("submitter", "").strip() or None
         note = request.form.get("note", "").strip() or None
+        raw_mode = request.form.get("download_mode", "player_slot")
+        download_mode = raw_mode.strip() if isinstance(raw_mode, str) else ""
 
         if not _valid_media_url(url):
             flash(
@@ -333,9 +449,20 @@ def create_app(
         if note is not None and len(note) > MAX_NOTE_LENGTH:
             flash(f"The note must be {MAX_NOTE_LENGTH} characters or fewer.")
             return redirect(url_for("index"))
+        if download_mode not in DOWNLOAD_MODE_LABELS:
+            flash("Invalid download type. Choose one of the available options.")
+            return redirect(url_for("index"))
 
         try:
+            if download_mode == "player_slot":
                 result = service.enqueue_download(url, submitter, note)
+            else:
+                result = service.enqueue_export(
+                    url,
+                    download_mode,
+                    submitter,
+                    note,
+                )
         except ControllerUnavailableError as exc:
             LOG.warning("playlist submission was rejected: %s", exc)
             flash("The Controller is unavailable; please try again shortly.")

config.yaml

@@ -48,6 +48,13 @@ downloads:
   max_workers: 2
   process_timeout_seconds: 7200
   max_height: 1080
+  # Utility exports live below media_base and never participate in slots.
+  export_directory: exports
+  export_max_height: 1080
+  # Dedicated capacity prevents utility exports from starving Player work.
+  export_max_workers: 1
+  # Bounds queued + running utility work; Player slots retain their own bound.
+  max_queued_exports: 20
 
 router:
   required_mode: video

controller/config.py

@@ -63,6 +63,10 @@ class DownloadConfig:
     max_workers: int = 2
     process_timeout_seconds: float = 2 * 3600.0
     max_height: int = 1080
+    export_directory: Path = Path("exports")
+    export_max_height: int = 1080
+    export_max_workers: int = 1
+    max_queued_exports: int = 20
 
 
 @dataclass(frozen=True)
@@ -172,6 +176,23 @@ def _optional_path(value: Any, name: str) -> Path | None:
     return Path(text)
 
 
+def _relative_directory(value: Any, name: str) -> Path:
+    text = _text(value, name)
+    assert text is not None
+    path = Path(text)
+    if (
+        path.is_absolute()
+        or not path.parts
+        or any(part in {".", ".."} for part in path.parts)
+        or "\\" in text
+    ):
+        raise ConfigError(f"{name} must be a safe relative directory")
+    reserved_roots = {f"slot{slot}" for slot in SLOT_IDS} | {".controller-trash"}
+    if path.parts[0].casefold() in reserved_roots:
+        raise ConfigError(f"{name} must not overlap Controller slot storage")
+    return path
+
+
 def _env_value(env: Mapping[str, str], key: str, fallback: Any) -> Any:
     return env[key] if key in env else fallback
 
@@ -277,6 +298,27 @@ def load_config(
         max_workers=_integer(downloads_raw.get("max_workers", 2), "downloads.max_workers", maximum=10),
         process_timeout_seconds=_number(downloads_raw.get("process_timeout_seconds", 2 * 3600), "downloads.process_timeout_seconds", minimum=1.0),
         max_height=_integer(downloads_raw.get("max_height", 1080), "downloads.max_height"),
+        export_directory=_relative_directory(
+            downloads_raw.get("export_directory", "exports"),
+            "downloads.export_directory",
+        ),
+        export_max_height=_integer(
+            downloads_raw.get("export_max_height", 1080),
+            "downloads.export_max_height",
+        ),
+        export_max_workers=_integer(
+            downloads_raw.get("export_max_workers", 1),
+            "downloads.export_max_workers",
+            maximum=10,
+        ),
+        max_queued_exports=_integer(
+            downloads_raw.get("max_queued_exports", 20),
+            "downloads.max_queued_exports",
+            # The web status endpoint deliberately returns at most 100 rows.
+            # Keep the queue bound within that view so every active job remains
+            # visible even when the configured queue is full.
+            maximum=100,
+        ),
     )
     if downloads.cookie_required and downloads.cookie_path is None:
         raise ConfigError("downloads.cookie_path is required when cookie_required is true")

controller/downloads.py

@@ -2,14 +2,18 @@
 
 from __future__ import annotations
 
+import json
 import logging
 import os
+import re
 import signal
 import shutil
+import stat
 import subprocess
 import tempfile
 import threading
 import time
+import unicodedata
 import uuid
 from collections.abc import Iterator
 from contextlib import contextmanager
@@ -20,12 +24,31 @@ from typing import Any, Callable
 
 from .config import ControllerConfig
 from .contracts import SLOT_IDS
-from .models import DeleteToken, DownloadReservation, SlotProtectedError, StaleOperationError
+from .models import (
+    DeleteToken,
+    DownloadReservation,
+    ExportReservation,
+    SlotProtectedError,
+    StaleOperationError,
+    UTILITY_EXPORT_MODES,
+)
 from .store import ControllerStore
 
 
 LOG = logging.getLogger(__name__)
 MEDIA_EXTENSIONS = frozenset({".mp4", ".mkv", ".webm", ".mov", ".m4v"})
+VIDEO_EXPORT_INPUT_EXTENSIONS = frozenset(
+    {*MEDIA_EXTENSIONS, ".avi", ".flv", ".ts"}
+)
+EXPORT_AUDIO_EXTENSIONS = {"m4a": ".m4a", "mp3": ".mp3"}
+MAX_EXPORT_STEM_BYTES = 180
+EXPORT_MANIFEST_NAME = "manifest.json"
+_INVALID_FILENAME_CHARACTERS = frozenset('<>:"/\\|?*')
+_WINDOWS_RESERVED_NAMES = frozenset(
+    {"CON", "PRN", "AUX", "NUL"}
+    | {f"COM{index}" for index in range(1, 10)}
+    | {f"LPT{index}" for index in range(1, 10)}
+)
 
 
 @dataclass(frozen=True)
@@ -43,6 +66,16 @@ class DownloadEvent:
     reused_slot: bool = False
 
 
+@dataclass(frozen=True)
+class MediaInspection:
+    format_names: frozenset[str]
+    video_codec: str | None
+    audio_codec: str | None
+    pixel_format: str | None
+    width: int | None
+    height: int | None
+
+
 EventCallback = Callable[[DownloadEvent], None]
 Runner = Callable[..., subprocess.CompletedProcess[str]]
 
@@ -57,6 +90,60 @@ def canonical_playlist_path(media_base: Path, slot: int) -> Path:
     return canonical_slot_directory(media_base, slot) / "playlist.m3u"
 
 
+def player_format_selector(max_height: int) -> str:
+    """Select the best Player media without exceeding its configured height."""
+
+    return f"bv*[height<={max_height}]+ba/b[height<={max_height}]"
+
+
+def mp4_export_format_selector(max_height: int) -> str:
+    """Select quality first while making native AVC/AAC the equal-quality preference."""
+
+    return f"bv*[height<={max_height}]+ba/b[height<={max_height}]"
+
+
+def safe_export_filename(
+    stem: str,
+    extension: str,
+    *,
+    used_casefolded: set[str] | None = None,
+) -> str:
+    """Return a bounded Unicode filename that cannot introduce a path component."""
+
+    if extension not in {".mp4", ".m4a", ".mp3"}:
+        raise ValueError("unsupported export filename extension")
+    normalized = unicodedata.normalize("NFC", str(stem))
+    cleaned = "".join(
+        "_"
+        if character in _INVALID_FILENAME_CHARACTERS
+        or ord(character) < 32
+        or ord(character) == 127
+        else character
+        for character in normalized
+    )
+    cleaned = re.sub(r"\s+", " ", cleaned).strip(" .") or "untitled"
+    if cleaned.split(".", 1)[0].upper() in _WINDOWS_RESERVED_NAMES:
+        cleaned = f"_{cleaned}"
+
+    def bounded(value: str, suffix: str) -> str:
+        available = MAX_EXPORT_STEM_BYTES - len(suffix.encode("utf-8"))
+        candidate = value
+        while candidate and len(candidate.encode("utf-8")) > available:
+            candidate = candidate[:-1]
+        return (candidate.rstrip(" .") or "untitled") + suffix
+
+    used = used_casefolded if used_casefolded is not None else set()
+    counter = 1
+    while True:
+        collision_suffix = "" if counter == 1 else f" ({counter})"
+        filename = bounded(cleaned, collision_suffix) + extension
+        folded = filename.casefold()
+        if folded not in used:
+            used.add(folded)
+            return filename
+        counter += 1
+
+
 def title_from_filename(path: Path) -> str:
     stem = path.stem
     if " - " in stem:
@@ -77,6 +164,7 @@ class DownloadManager:
         on_event: EventCallback | None = None,
         runner: Runner | None = None,
         executor: ThreadPoolExecutor | None = None,
+        export_executor: ThreadPoolExecutor | None = None,
     ):
         self.config = config
         self.store = store
@@ -87,6 +175,19 @@ class DownloadManager:
             thread_name_prefix="yt-download",
         )
         self._owns_executor = executor is None
+        if export_executor is not None:
+            self._export_executor = export_executor
+            self._owns_export_executor = False
+        elif executor is not None:
+            # Deterministic callers/tests may inject one executor explicitly.
+            self._export_executor = executor
+            self._owns_export_executor = False
+        else:
+            self._export_executor = ThreadPoolExecutor(
+                max_workers=config.downloads.export_max_workers,
+                thread_name_prefix="yt-export",
+            )
+            self._owns_export_executor = True
         self._lock = threading.Lock()
         self._futures: set[Future[Any]] = set()
         self._processes: set[subprocess.Popen[str]] = set()
@@ -161,13 +262,72 @@ class DownloadManager:
         future.add_done_callback(self._forget_future)
         return reservation
 
+    def enqueue_export(
+        self,
+        url: str,
+        mode: str,
+        submitter: str | None,
+        note: str | None,
+    ) -> ExportReservation:
+        """Reserve and queue a utility export without touching Player slots."""
+
+        if mode not in UTILITY_EXPORT_MODES:
+            raise ValueError("unsupported utility export mode")
+        with self._lock:
+            if self._stopping:
+                return ExportReservation(status="shutting_down")
+        reservation = self.store.reserve_export_job(
+            url,
+            mode,
+            submitter,
+            note,
+            max_active=self.config.downloads.max_queued_exports,
+        )
+        if reservation.status != "queued":
+            return reservation
+        assert reservation.job_id is not None
+        try:
+            future = self._export_executor.submit(
+                self._export,
+                url,
+                mode,
+                reservation.job_id,
+            )
+        except BaseException as exc:
+            message = f"could not start export worker: {exc}"
+            self.store.fail_export_job(reservation.job_id, message)
+            return ExportReservation(
+                status="worker_error",
+                job_id=reservation.job_id,
+            )
+        with self._lock:
+            self._futures.add(future)
+        future.add_done_callback(self._forget_future)
+        return reservation
+
+    def schedule_export_reconciliation(self) -> bool:
+        """Queue crash-artifact recovery without delaying core service startup."""
+
+        with self._lock:
+            if self._stopping:
+                return False
+        try:
+            future = self._export_executor.submit(self.reconcile_export_artifacts)
+        except BaseException:
+            LOG.exception("could not schedule utility export reconciliation")
+            return False
+        with self._lock:
+            self._futures.add(future)
+        future.add_done_callback(self._forget_future)
+        return True
+
     def _forget_future(self, future: Future[Any]) -> None:
         with self._lock:
             self._futures.discard(future)
         try:
             future.result()
         except BaseException:
-            LOG.exception("download worker escaped its failure boundary")
+            LOG.exception("media worker escaped its failure boundary")
 
     def _emit(self, event: DownloadEvent) -> None:
         try:
@@ -297,10 +457,7 @@ class DownloadManager:
                         / "%(playlist_index)03d - %(title)s [%(id)s].%(ext)s"
                     ),
                     "-f",
-                    (
-                        f"bv*[height<={self.config.downloads.max_height}]+ba/"
-                        f"b[height<={self.config.downloads.max_height}]/b"
-                    ),
+                    player_format_selector(self.config.downloads.max_height),
                     url,
                 ]
                 self._run_command(
@@ -371,12 +528,643 @@ class DownloadManager:
                 )
             )
 
+    def _export(
+        self,
+        url: str,
+        mode: str,
+        job_id: str,
+    ) -> None:
+        if not self.store.mark_export_running(job_id):
+            return
+        work_directory: Path | None = None
+        published = False
+        completed = False
+        try:
+            work_directory, source_directory, ready_directory, final_parent = (
+                self._prepare_export_directories(job_id, mode)
+            )
+            with self._yt_dlp_arguments() as base_arguments:
+                arguments = base_arguments + self._export_download_arguments(
+                    mode,
+                    source_directory,
+                    url,
+                )
+                self._run_command(
+                    arguments,
+                    check=True,
+                    text=True,
+                    timeout=self.config.downloads.process_timeout_seconds,
+                )
+
+            sources = self._export_source_files(source_directory, mode)
+            if not sources:
+                raise RuntimeError("yt-dlp completed without producing export media")
+            used_names: set[str] = set()
+            ready_outputs: list[Path] = []
+            for source in sources:
+                extension = ".mp4" if mode == "mp4" else EXPORT_AUDIO_EXTENSIONS[mode]
+                output = ready_directory / safe_export_filename(
+                    source.stem,
+                    extension,
+                    used_casefolded=used_names,
+                )
+                if mode == "mp4":
+                    self._produce_compatible_mp4(source, output)
+                else:
+                    inspection = self._inspect_media(source)
+                    self._validate_audio_export(inspection, mode)
+                    os.replace(source, output)
+                self._fsync_file(output)
+                ready_outputs.append(output)
+
+            self._fsync_directory(ready_directory)
+            self._write_export_manifest(
+                work_directory,
+                job_id=job_id,
+                mode=mode,
+                filenames=[output.name for output in ready_outputs],
+            )
+            published_directory = final_parent / job_id
+            if os.path.lexists(published_directory):
+                raise FileExistsError(
+                    f"refusing to overwrite existing export directory: {published_directory}"
+                )
+            os.rename(ready_directory, published_directory)
+            self._fsync_directory(final_parent)
+            published = True
+            media_base = self.config.storage.media_base.resolve()
+            output_paths = tuple(
+                str((published_directory / output.name).resolve().relative_to(media_base))
+                for output in ready_outputs
+            )
+            if not self.store.complete_export_job(job_id, output_paths):
+                raise StaleOperationError(
+                    "published export could not transition its durable job"
+                )
+            completed = True
+        except BaseException as exc:
+            message = self._failure_message(exc)
+            try:
+                self.store.fail_export_job(job_id, message)
+            except BaseException:
+                LOG.exception("could not persist utility export failure for %s", job_id)
+            if published:
+                LOG.error(
+                    "utility export %s was published but needs restart reconciliation: %s",
+                    job_id,
+                    message,
+                )
+            else:
+                LOG.warning("utility export %s failed: %s", job_id, message)
+        finally:
+            if work_directory is not None and (not published or completed):
+                self._remove_export_work_directory(work_directory)
+
+    def _export_download_arguments(
+        self,
+        mode: str,
+        source_directory: Path,
+        url: str,
+    ) -> list[str]:
+        arguments = [
+            "--yes-playlist",
+            "--no-overwrites",
+            "--no-progress",
+            "--windows-filenames",
+            "--trim-filenames",
+            str(MAX_EXPORT_STEM_BYTES),
+            "-o",
+            str(
+                source_directory
+                / "%(playlist_index)03d - %(title)s [%(id)s].%(ext)s"
+            ),
+        ]
+        if mode == "mp4":
+            arguments.extend(
+                (
+                    "-f",
+                    mp4_export_format_selector(
+                        self.config.downloads.export_max_height
+                    ),
+                    "-S",
+                    (
+                        f"res:{self.config.downloads.export_max_height},"
+                        "vcodec:h264,acodec:aac"
+                    ),
+                )
+            )
+        elif mode == "m4a":
+            arguments.extend(
+                (
+                    "-f",
+                    "ba/b",
+                    "-S",
+                    "abr,acodec:aac",
+                    "-x",
+                    "--audio-format",
+                    "m4a",
+                    "--audio-quality",
+                    "0",
+                )
+            )
+        elif mode == "mp3":
+            arguments.extend(
+                (
+                    "-f",
+                    "ba/b",
+                    "-x",
+                    "--audio-format",
+                    "mp3",
+                    "--audio-quality",
+                    "0",
+                )
+            )
+        else:
+            raise ValueError("unsupported utility export mode")
+        arguments.append(url)
+        return arguments
+
+    def _prepare_export_directories(
+        self,
+        job_id: str,
+        mode: str,
+    ) -> tuple[Path, Path, Path, Path]:
+        if not re.fullmatch(r"[0-9a-f]{32}", job_id):
+            raise ValueError("invalid export job identity")
+        media_base = self.config.storage.media_base
+        media_base.mkdir(parents=True, exist_ok=True)
+        export_root = media_base
+        for part in self.config.downloads.export_directory.parts:
+            export_root = export_root / part
+            self._ensure_real_directory(export_root)
+        work_root = export_root / ".work"
+        final_parent = export_root / ("video" if mode == "mp4" else "audio")
+        self._ensure_real_directory(work_root)
+        self._ensure_real_directory(final_parent)
+        work_directory = work_root / job_id
+        work_directory.mkdir(mode=0o750, exist_ok=False)
+        source_directory = work_directory / "source"
+        ready_directory = work_directory / "ready"
+        source_directory.mkdir(mode=0o750)
+        ready_directory.mkdir(mode=0o750)
+        self._fsync_directory(work_root)
+        return work_directory, source_directory, ready_directory, final_parent
+
+    @staticmethod
+    def _ensure_real_directory(path: Path) -> None:
+        try:
+            path.mkdir(mode=0o750)
+        except FileExistsError:
+            pass
+        metadata = path.lstat()
+        if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
+            raise RuntimeError(f"refusing unsafe export directory: {path}")
+
+    def _write_export_manifest(
+        self,
+        work_directory: Path,
+        *,
+        job_id: str,
+        mode: str,
+        filenames: list[str],
+    ) -> None:
+        if not filenames:
+            raise ValueError("export manifest must contain output filenames")
+        payload = json.dumps(
+            {
+                "version": 1,
+                "job_id": job_id,
+                "mode": mode,
+                "filenames": filenames,
+            },
+            ensure_ascii=False,
+            sort_keys=True,
+        )
+        temporary = work_directory / f".{EXPORT_MANIFEST_NAME}.tmp"
+        manifest = work_directory / EXPORT_MANIFEST_NAME
+        try:
+            with temporary.open("x", encoding="utf-8", newline="\n") as stream:
+                stream.write(payload + "\n")
+                stream.flush()
+                os.fsync(stream.fileno())
+            os.replace(temporary, manifest)
+            self._fsync_directory(work_directory)
+        finally:
+            temporary.unlink(missing_ok=True)
+
+    @staticmethod
+    def _export_source_files(source_directory: Path, mode: str) -> list[Path]:
+        allowed = (
+            VIDEO_EXPORT_INPUT_EXTENSIONS
+            if mode == "mp4"
+            else frozenset({EXPORT_AUDIO_EXTENSIONS[mode]})
+        )
+        media: list[Path] = []
+        for candidate in source_directory.iterdir():
+            if candidate.is_symlink() or not candidate.is_file():
+                raise RuntimeError(
+                    f"yt-dlp produced an unsafe export entry: {candidate.name}"
+                )
+            if candidate.suffix.lower() not in allowed:
+                raise RuntimeError(
+                    f"yt-dlp produced an unexpected export file: {candidate.name}"
+                )
+            media.append(candidate)
+        return sorted(media, key=lambda path: path.name.casefold())
+
+    def _inspect_media(self, path: Path) -> MediaInspection:
+        result = self._run_command(
+            [
+                "ffprobe",
+                "-v",
+                "error",
+                "-show_streams",
+                "-show_format",
+                "-of",
+                "json",
+                str(path),
+            ],
+            check=True,
+            capture_output=True,
+            text=True,
+            timeout=min(120.0, self.config.downloads.process_timeout_seconds),
+        )
+        try:
+            payload = json.loads(result.stdout or "")
+            streams = payload["streams"]
+            format_name = payload["format"]["format_name"]
+        except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
+            raise RuntimeError("ffprobe returned invalid media metadata") from exc
+        if not isinstance(streams, list) or not isinstance(format_name, str):
+            raise RuntimeError("ffprobe returned invalid media metadata")
+        video = next(
+            (
+                stream
+                for stream in streams
+                if isinstance(stream, dict) and stream.get("codec_type") == "video"
+            ),
+            None,
+        )
+        audio = next(
+            (
+                stream
+                for stream in streams
+                if isinstance(stream, dict) and stream.get("codec_type") == "audio"
+            ),
+            None,
+        )
+
+        def positive_integer(value: Any) -> int | None:
+            return (
+                value
+                if isinstance(value, int)
+                and not isinstance(value, bool)
+                and value > 0
+                else None
+            )
+
+        return MediaInspection(
+            format_names=frozenset(
+                value.strip().lower()
+                for value in format_name.split(",")
+                if value.strip()
+            ),
+            video_codec=(
+                video.get("codec_name") if isinstance(video, dict) else None
+            ),
+            audio_codec=(
+                audio.get("codec_name") if isinstance(audio, dict) else None
+            ),
+            pixel_format=(
+                video.get("pix_fmt") if isinstance(video, dict) else None
+            ),
+            width=positive_integer(video.get("width") if isinstance(video, dict) else None),
+            height=positive_integer(video.get("height") if isinstance(video, dict) else None),
+        )
+
+    def _produce_compatible_mp4(self, source: Path, output: Path) -> None:
+        source_info = self._inspect_media(source)
+        if source_info.video_codec is None or source_info.audio_codec is None:
+            raise RuntimeError("MP4 export requires both video and audio")
+        if source_info.height is None:
+            raise RuntimeError("MP4 export source has no verifiable video height")
+        if source_info.height > self.config.downloads.export_max_height:
+            raise RuntimeError("MP4 export source exceeds configured maximum height")
+        video_compatible = (
+            source_info.video_codec == "h264"
+            and source_info.pixel_format in {"yuv420p", "yuvj420p"}
+        )
+        audio_compatible = source_info.audio_codec == "aac"
+        # ffprobe reports the same comma-separated demuxer aliases for MOV and
+        # MP4. A .mov source therefore needs at least a stream-copy remux; it
+        # must never be renamed to .mp4 solely because the alias list includes
+        # "mp4".
+        container_compatible = (
+            source.suffix.lower() == ".mp4"
+            and "mp4" in source_info.format_names
+        )
+        if video_compatible and audio_compatible and container_compatible:
+            os.replace(source, output)
+        else:
+            temporary = output.with_name(
+                f".{output.stem}.{uuid.uuid4().hex}.tmp.mp4"
+            )
+            arguments = [
+                "ffmpeg",
+                "-nostdin",
+                "-hide_banner",
+                "-loglevel",
+                "error",
+                "-n",
+                "-i",
+                str(source),
+                "-map",
+                "0:v:0",
+                "-map",
+                "0:a:0",
+                "-map_metadata",
+                "0",
+            ]
+            if video_compatible:
+                arguments.extend(("-c:v", "copy"))
+            else:
+                arguments.extend(
+                    (
+                        "-c:v",
+                        "libx264",
+                        "-preset",
+                        "medium",
+                        "-crf",
+                        "18",
+                        "-pix_fmt",
+                        "yuv420p",
+                    )
+                )
+            if audio_compatible:
+                arguments.extend(("-c:a", "copy"))
+            else:
+                arguments.extend(("-c:a", "aac", "-b:a", "192k"))
+            arguments.extend(("-movflags", "+faststart", str(temporary)))
+            try:
+                self._run_command(
+                    arguments,
+                    check=True,
+                    text=True,
+                    timeout=self.config.downloads.process_timeout_seconds,
+                )
+                converted = self._inspect_media(temporary)
+                self._validate_compatible_mp4(converted, source_info.height)
+                os.replace(temporary, output)
+            finally:
+                temporary.unlink(missing_ok=True)
+        self._validate_compatible_mp4(
+            self._inspect_media(output),
+            source_info.height,
+        )
+
+    def _validate_compatible_mp4(
+        self,
+        inspection: MediaInspection,
+        source_height: int,
+    ) -> None:
+        if (
+            "mp4" not in inspection.format_names
+            or inspection.video_codec != "h264"
+            or inspection.audio_codec != "aac"
+            or inspection.pixel_format not in {"yuv420p", "yuvj420p"}
+        ):
+            raise RuntimeError("MP4 compatibility verification failed")
+        if (
+            inspection.height != source_height
+            or inspection.height > self.config.downloads.export_max_height
+        ):
+            raise RuntimeError("MP4 export resolution verification failed")
+
+    @staticmethod
+    def _validate_audio_export(
+        inspection: MediaInspection,
+        mode: str,
+    ) -> None:
+        expected_codec = "aac" if mode == "m4a" else "mp3"
+        expected_container = "mp4" if mode == "m4a" else "mp3"
+        if inspection.video_codec is not None:
+            raise RuntimeError("audio export unexpectedly contains video")
+        if inspection.audio_codec != expected_codec:
+            raise RuntimeError(f"{mode.upper()} export codec verification failed")
+        if expected_container not in inspection.format_names:
+            raise RuntimeError(f"{mode.upper()} export container verification failed")
+
+    @staticmethod
+    def _fsync_file(path: Path) -> None:
+        descriptor = os.open(path, os.O_RDONLY)
+        try:
+            os.fsync(descriptor)
+        finally:
+            os.close(descriptor)
+
+    def _remove_export_work_directory(self, path: Path) -> None:
+        try:
+            parent = path.parent
+            if os.path.lexists(path):
+                if path.is_symlink() or path.is_file():
+                    path.unlink()
+                else:
+                    shutil.rmtree(path)
+                self._fsync_directory(parent)
+        except OSError:
+            LOG.exception("could not remove export work directory %s", path)
+
+    def reconcile_export_artifacts(self) -> None:
+        """Recover verified crash-published exports and discard incomplete staging."""
+
+        export_root = self.config.storage.media_base / self.config.downloads.export_directory
+        if not os.path.lexists(export_root):
+            return
+        try:
+            self._verify_existing_export_tree(export_root)
+        except OSError as exc:
+            LOG.error("could not inspect utility export tree: %s", exc)
+            return
+        except RuntimeError as exc:
+            LOG.error("utility export reconciliation refused unsafe storage: %s", exc)
+            return
+
+        jobs = {
+            job.job_id: job for job in self.store.list_reconcilable_export_jobs()
+        }
+        for job in jobs.values():
+            try:
+                self._reconcile_export_job(export_root, job.job_id, job.mode)
+            except BaseException:
+                LOG.exception("could not reconcile utility export %s", job.job_id)
+
+        work_root = export_root / ".work"
+        if not work_root.is_dir() or work_root.is_symlink():
+            return
+        for candidate in work_root.iterdir():
+            if not re.fullmatch(r"[0-9a-f]{32}", candidate.name):
+                LOG.warning("leaving unknown utility export staging entry %s", candidate)
+                continue
+            job = self.store.get_export_job(candidate.name)
+            if job is None:
+                LOG.warning("leaving untracked utility export staging entry %s", candidate)
+                continue
+            if job.status == "completed":
+                self._remove_export_work_directory(candidate)
+            elif job.job_id not in jobs:
+                LOG.warning(
+                    "leaving utility export staging for active job %s",
+                    candidate.name,
+                )
+
+    def _verify_existing_export_tree(self, export_root: Path) -> None:
+        current = self.config.storage.media_base
+        for part in self.config.downloads.export_directory.parts:
+            current = current / part
+            metadata = current.lstat()
+            if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
+                raise RuntimeError(f"unsafe export directory: {current}")
+        for child_name in (".work", "video", "audio"):
+            child = export_root / child_name
+            if not os.path.lexists(child):
+                continue
+            metadata = child.lstat()
+            if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
+                raise RuntimeError(f"unsafe export directory: {child}")
+
+    def _reconcile_export_job(
+        self,
+        export_root: Path,
+        job_id: str,
+        mode: str,
+    ) -> None:
+        work_directory = export_root / ".work" / job_id
+        final_parent = export_root / ("video" if mode == "mp4" else "audio")
+        final_directory = final_parent / job_id
+        if os.path.lexists(work_directory):
+            work_metadata = work_directory.lstat()
+            if stat.S_ISLNK(work_metadata.st_mode) or not stat.S_ISDIR(
+                work_metadata.st_mode
+            ):
+                raise RuntimeError("utility export staging directory is unsafe")
+        manifest = self._read_export_manifest(work_directory, job_id, mode)
+
+        if os.path.lexists(final_directory):
+            filenames = None if manifest is None else manifest
+            outputs = self._validate_export_output_directory(
+                final_directory,
+                mode,
+                expected_filenames=filenames,
+            )
+        elif manifest is not None:
+            ready_directory = work_directory / "ready"
+            outputs = self._validate_export_output_directory(
+                ready_directory,
+                mode,
+                expected_filenames=manifest,
+            )
+            self._ensure_real_directory(final_parent)
+            os.rename(ready_directory, final_directory)
+            self._fsync_directory(final_parent)
+        else:
+            if os.path.lexists(work_directory):
+                self._remove_export_work_directory(work_directory)
+            return
+
+        media_base = self.config.storage.media_base.resolve()
+        output_paths = tuple(
+            str((final_directory / output.name).resolve().relative_to(media_base))
+            for output in outputs
+        )
+        if self.store.recover_export_job(job_id, output_paths):
+            LOG.warning("recovered crash-published utility export %s", job_id)
+            if os.path.lexists(work_directory):
+                self._remove_export_work_directory(work_directory)
+
+    @staticmethod
+    def _read_export_manifest(
+        work_directory: Path,
+        job_id: str,
+        mode: str,
+    ) -> list[str] | None:
+        manifest = work_directory / EXPORT_MANIFEST_NAME
+        if not manifest.is_file() or manifest.is_symlink():
+            return None
+        if manifest.stat().st_size > 8 * 1024 * 1024:
+            raise RuntimeError("utility export manifest is too large")
+        try:
+            payload = json.loads(manifest.read_text(encoding="utf-8"))
+        except (OSError, UnicodeError, json.JSONDecodeError) as exc:
+            raise RuntimeError("utility export manifest is invalid") from exc
+        if (
+            not isinstance(payload, dict)
+            or payload.get("version") != 1
+            or payload.get("job_id") != job_id
+            or payload.get("mode") != mode
+        ):
+            raise RuntimeError("utility export manifest identity is invalid")
+        filenames = payload.get("filenames")
+        if not isinstance(filenames, list) or not filenames or len(filenames) > 10000:
+            raise RuntimeError("utility export manifest filenames are invalid")
+        expected_extension = ".mp4" if mode == "mp4" else EXPORT_AUDIO_EXTENSIONS[mode]
+        normalized: list[str] = []
+        for filename in filenames:
+            if (
+                not isinstance(filename, str)
+                or not filename
+                or Path(filename).name != filename
+                or Path(filename).suffix.lower() != expected_extension
+            ):
+                raise RuntimeError("utility export manifest contains an unsafe filename")
+            normalized.append(filename)
+        if len({value.casefold() for value in normalized}) != len(normalized):
+            raise RuntimeError("utility export manifest contains duplicate filenames")
+        return normalized
+
+    def _validate_export_output_directory(
+        self,
+        directory: Path,
+        mode: str,
+        *,
+        expected_filenames: list[str] | None,
+    ) -> list[Path]:
+        metadata = directory.lstat()
+        if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
+            raise RuntimeError("utility export output directory is unsafe")
+        outputs = sorted(directory.iterdir(), key=lambda path: path.name.casefold())
+        if not outputs:
+            raise RuntimeError("utility export output directory is empty")
+        if expected_filenames is not None and {
+            path.name for path in outputs
+        } != set(expected_filenames):
+            raise RuntimeError("utility export output does not match its manifest")
+        expected_extension = ".mp4" if mode == "mp4" else EXPORT_AUDIO_EXTENSIONS[mode]
+        for output in outputs:
+            if (
+                output.is_symlink()
+                or not output.is_file()
+                or output.suffix.lower() != expected_extension
+            ):
+                raise RuntimeError("utility export output contains an unsafe entry")
+            inspection = self._inspect_media(output)
+            if mode == "mp4":
+                if inspection.height is None:
+                    raise RuntimeError("MP4 export output has no verifiable height")
+                self._validate_compatible_mp4(inspection, inspection.height)
+            else:
+                self._validate_audio_export(inspection, mode)
+        return outputs
+
     @staticmethod
     def _failure_message(exc: BaseException) -> str:
+        command = getattr(exc, "cmd", None)
+        program = "media command"
+        if isinstance(command, (list, tuple)) and command:
+            program = Path(str(command[0])).name
         if isinstance(exc, subprocess.TimeoutExpired):
-            return f"yt-dlp timed out after {exc.timeout} seconds"
+            return f"{program} timed out after {exc.timeout} seconds"
         if isinstance(exc, subprocess.CalledProcessError):
-            return f"yt-dlp failed with exit code {exc.returncode}"
+            return f"{program} failed with exit code {exc.returncode}"
         return f"download failed: {exc}"
 
     def _run_command(
@@ -705,6 +1493,8 @@ class DownloadManager:
                 self._terminate_process(process, force=True)
         if self._owns_executor:
             self._executor.shutdown(wait=wait, cancel_futures=True)
+        if self._owns_export_executor:
+            self._export_executor.shutdown(wait=wait, cancel_futures=True)
 
     def active_future_count(self) -> int:
         with self._lock:

controller/models.py

@@ -10,6 +10,7 @@ MEDIA_STATES = frozenset(
 )
 LIVE_JOB_STATES = frozenset({"reserved", "running"})
 LIVE_GUARD_STATES = frozenset({"active", "review_required"})
+UTILITY_EXPORT_MODES = frozenset({"mp4", "m4a", "mp3"})
 
 
 @dataclass(frozen=True)
@@ -54,6 +55,27 @@ class DownloadJob:
     error: str | None
 
 
+@dataclass(frozen=True)
+class ExportReservation:
+    status: str
+    job_id: str | None = None
+
+
+@dataclass(frozen=True)
+class ExportJob:
+    job_id: str
+    mode: str
+    url: str
+    status: str
+    submitter: str | None
+    note: str | None
+    requested_at: float
+    started_at: float | None
+    finished_at: float | None
+    output_paths: tuple[str, ...]
+    error: str | None
+
+
 @dataclass(frozen=True)
 class DeleteToken:
     slot: int

controller/service.py

@@ -165,6 +165,11 @@ class ControllerService:
                 self._coordinator.start()
                 self.mqtt.start()
                 self._started = True
+                # A crash-published playlist may contain many files to probe.
+                # Queue that bounded work only after core readiness and keep it
+                # on the dedicated export executor, so MQTT reconnect and
+                # Player coordination never wait for recovery to finish.
+                self.downloads.schedule_export_reconciliation()
             except BaseException:
                 self._stop_event.set()
                 try:
@@ -252,6 +257,26 @@ class ControllerService:
             {"url": url, "submitter": submitter, "note": note},
         )
 
+    def enqueue_export(
+        self,
+        url: str,
+        mode: str,
+        submitter: str | None,
+        note: str | None,
+    ) -> dict[str, Any]:
+        return self._call(
+            "enqueue_export",
+            {
+                "url": url,
+                "mode": mode,
+                "submitter": submitter,
+                "note": note,
+            },
+        )
+
+    def list_export_jobs(self, *, limit: int = 10) -> list[dict[str, Any]]:
+        return self._call("list_export_jobs", {"limit": limit})
+
     def submit_control(self, payload: Mapping[str, Any]) -> dict[str, Any]:
         return self._call("control", dict(payload))
 
@@ -403,6 +428,35 @@ class ControllerService:
             self._handle_mqtt_event(event)
 
     def _handle_call(self, action: str, payload: Any) -> Any:
+        if action == "enqueue_export":
+            assert self.downloads is not None
+            reservation = self.downloads.enqueue_export(
+                payload["url"],
+                payload["mode"],
+                payload.get("submitter"),
+                payload.get("note"),
+            )
+            return {
+                "kind": "export",
+                "status": reservation.status,
+                "job_id": reservation.job_id,
+                "mode": payload["mode"],
+            }
+        if action == "list_export_jobs":
+            jobs = self.store.list_export_jobs(limit=payload["limit"])
+            return [
+                {
+                    "job_id": job.job_id,
+                    "mode": job.mode,
+                    "url": job.url,
+                    "status": job.status,
+                    "requested_at": job.requested_at,
+                    "finished_at": job.finished_at,
+                    "output_paths": list(job.output_paths),
+                    "error": job.error,
+                }
+                for job in jobs
+            ]
         if action == "enqueue_download":
             assert self.downloads is not None
             url = payload["url"]

controller/store.py

@@ -25,6 +25,8 @@ from .models import (
     DeleteToken,
     DownloadJob,
     DownloadReservation,
+    ExportJob,
+    ExportReservation,
     MigrationError,
     MigrationReport,
     PlayerObservation,
@@ -33,10 +35,11 @@ from .models import (
     SlotProtectedError,
     SlotRecord,
     StaleOperationError,
+    UTILITY_EXPORT_MODES,
 )
 
 
-SCHEMA_VERSION = 2
+SCHEMA_VERSION = 3
 LEGACY_MIGRATION_NAME = "legacy_playlists_yaml_v1"
 MAX_LEGACY_YAML_BYTES = 8 * 1024 * 1024
 
@@ -104,6 +107,24 @@ CREATE UNIQUE INDEX IF NOT EXISTS one_live_download_per_slot
 ON download_jobs(slot_number)
 WHERE status IN ('reserved','running');
 
+CREATE TABLE IF NOT EXISTS export_jobs (
+    job_id TEXT PRIMARY KEY,
+    mode TEXT NOT NULL CHECK (mode IN ('mp4','m4a','mp3')),
+    url TEXT NOT NULL,
+    status TEXT NOT NULL CHECK (status IN
+        ('queued','running','completed','failed','interrupted')),
+    submitter TEXT,
+    note TEXT,
+    requested_at REAL NOT NULL,
+    started_at REAL,
+    finished_at REAL,
+    output_paths_json TEXT NOT NULL DEFAULT '[]',
+    error TEXT
+);
+
+CREATE INDEX IF NOT EXISTS export_jobs_recent
+ON export_jobs(requested_at DESC, job_id DESC);
+
 CREATE TABLE IF NOT EXISTS slot_guards (
     guard_id TEXT PRIMARY KEY,
     slot_number INTEGER NOT NULL REFERENCES slots(slot_number),
@@ -401,6 +422,13 @@ class ControllerStore:
             """,
             (now,),
         )
+        connection.execute(
+            """
+            INSERT OR IGNORE INTO schema_migrations(version, name, applied_at)
+            VALUES (3, 'utility_export_jobs', ?)
+            """,
+            (now,),
+        )
         connection.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
 
     def _upgrade_database(self, connection: sqlite3.Connection) -> None:
@@ -409,12 +437,13 @@ class ControllerStore:
         version = connection.execute("PRAGMA user_version").fetchone()[0]
         if version == SCHEMA_VERSION:
             return
-        if version != 1:
+        if version not in {1, 2}:
             raise MigrationError(
-                f"unsupported Controller database schema {version}; expected 1 or {SCHEMA_VERSION}"
+                f"unsupported Controller database schema {version}; expected 1, 2, or {SCHEMA_VERSION}"
             )
         connection.execute("BEGIN IMMEDIATE")
         try:
+            if version < 2:
                 columns = {
                     row["name"]
                     for row in connection.execute(
@@ -432,6 +461,37 @@ class ControllerStore:
                     """,
                     (self._now(),),
                 )
+            connection.execute(
+                """
+                CREATE TABLE IF NOT EXISTS export_jobs (
+                    job_id TEXT PRIMARY KEY,
+                    mode TEXT NOT NULL CHECK (mode IN ('mp4','m4a','mp3')),
+                    url TEXT NOT NULL,
+                    status TEXT NOT NULL CHECK (status IN
+                        ('queued','running','completed','failed','interrupted')),
+                    submitter TEXT,
+                    note TEXT,
+                    requested_at REAL NOT NULL,
+                    started_at REAL,
+                    finished_at REAL,
+                    output_paths_json TEXT NOT NULL DEFAULT '[]',
+                    error TEXT
+                )
+                """
+            )
+            connection.execute(
+                """
+                CREATE INDEX IF NOT EXISTS export_jobs_recent
+                ON export_jobs(requested_at DESC, job_id DESC)
+                """
+            )
+            connection.execute(
+                """
+                INSERT OR IGNORE INTO schema_migrations(version, name, applied_at)
+                VALUES (3, 'utility_export_jobs', ?)
+                """,
+                (self._now(),),
+            )
             connection.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
         except BaseException:
             connection.rollback()
@@ -837,6 +897,15 @@ class ControllerStore:
                         row["generation"],
                     ),
                 )
+            connection.execute(
+                """
+                UPDATE export_jobs
+                SET status = 'interrupted', finished_at = ?,
+                    error = 'export interrupted by Controller restart'
+                WHERE status IN ('queued','running')
+                """,
+                (now,),
+            )
 
     def list_slots(self) -> list[SlotRecord]:
         with self._lock:
@@ -1136,6 +1205,189 @@ class ControllerStore:
                 error=row["error"],
             )
 
+    @staticmethod
+    def _row_to_export_job(row: sqlite3.Row) -> ExportJob:
+        try:
+            raw_paths = json.loads(row["output_paths_json"])
+        except (TypeError, ValueError, json.JSONDecodeError):
+            raw_paths = []
+        output_paths = tuple(
+            value for value in raw_paths if isinstance(value, str)
+        ) if isinstance(raw_paths, list) else ()
+        return ExportJob(
+            job_id=row["job_id"],
+            mode=row["mode"],
+            url=row["url"],
+            status=row["status"],
+            submitter=row["submitter"],
+            note=row["note"],
+            requested_at=row["requested_at"],
+            started_at=row["started_at"],
+            finished_at=row["finished_at"],
+            output_paths=output_paths,
+            error=row["error"],
+        )
+
+    def reserve_export_job(
+        self,
+        url: str,
+        mode: str,
+        submitter: str | None,
+        note: str | None,
+        *,
+        max_active: int,
+        job_id: str | None = None,
+    ) -> ExportReservation:
+        if not isinstance(url, str) or not url.strip():
+            raise ValueError("export URL must be non-empty")
+        if mode not in UTILITY_EXPORT_MODES:
+            raise ValueError("unsupported utility export mode")
+        if not isinstance(max_active, int) or isinstance(max_active, bool) or max_active < 1:
+            raise ValueError("max_active must be a positive integer")
+        job_id = job_id or uuid.uuid4().hex
+        with self._transaction() as connection:
+            active = connection.execute(
+                """
+                SELECT COUNT(*) FROM export_jobs
+                WHERE status IN ('queued','running')
+                """
+            ).fetchone()[0]
+            if active >= max_active:
+                return ExportReservation(status="queue_full")
+            connection.execute(
+                """
+                INSERT INTO export_jobs
+                    (job_id, mode, url, status, submitter, note, requested_at)
+                VALUES (?, ?, ?, 'queued', ?, ?, ?)
+                """,
+                (job_id, mode, url.strip(), submitter, note, self._now()),
+            )
+            return ExportReservation(status="queued", job_id=job_id)
+
+    def mark_export_running(self, job_id: str) -> bool:
+        with self._transaction() as connection:
+            return bool(
+                connection.execute(
+                    """
+                    UPDATE export_jobs SET status = 'running', started_at = ?
+                    WHERE job_id = ? AND status = 'queued'
+                    """,
+                    (self._now(), job_id),
+                ).rowcount
+            )
+
+    def complete_export_job(
+        self,
+        job_id: str,
+        output_paths: Sequence[str],
+    ) -> bool:
+        normalized = tuple(output_paths)
+        if not normalized or any(
+            not isinstance(path, str) or not path for path in normalized
+        ):
+            raise ValueError("completed export must contain output paths")
+        with self._transaction() as connection:
+            return bool(
+                connection.execute(
+                    """
+                    UPDATE export_jobs
+                    SET status = 'completed', finished_at = ?,
+                        output_paths_json = ?, error = NULL
+                    WHERE job_id = ? AND status = 'running'
+                    """,
+                    (self._now(), json.dumps(normalized), job_id),
+                ).rowcount
+            )
+
+    def fail_export_job(self, job_id: str, error: str) -> bool:
+        with self._transaction() as connection:
+            return bool(
+                connection.execute(
+                    """
+                    UPDATE export_jobs
+                    SET status = 'failed', finished_at = ?, error = ?
+                    WHERE job_id = ? AND status IN ('queued','running')
+                    """,
+                    (self._now(), error, job_id),
+                ).rowcount
+            )
+
+    def recover_export_job(
+        self,
+        job_id: str,
+        output_paths: Sequence[str],
+    ) -> bool:
+        """Record a fully verified export found during restart reconciliation."""
+
+        normalized = tuple(output_paths)
+        if not normalized or any(
+            not isinstance(path, str) or not path for path in normalized
+        ):
+            raise ValueError("recovered export must contain output paths")
+        with self._transaction() as connection:
+            now = self._now()
+            updated = connection.execute(
+                """
+                UPDATE export_jobs
+                SET status = 'completed', finished_at = ?,
+                    output_paths_json = ?, error = NULL
+                WHERE job_id = ?
+                  AND status IN ('queued','running','failed','interrupted')
+                """,
+                (now, json.dumps(normalized), job_id),
+            ).rowcount
+            if updated:
+                connection.execute(
+                    """
+                    INSERT INTO audit_events
+                        (occurred_at, event_type, actor, details_json)
+                    VALUES (?, 'utility_export_recovered', 'startup', ?)
+                    """,
+                    (
+                        now,
+                        json.dumps(
+                            {"job_id": job_id, "output_paths": normalized},
+                            sort_keys=True,
+                        ),
+                    ),
+                )
+            return bool(updated)
+
+    def get_export_job(self, job_id: str) -> ExportJob | None:
+        with self._lock:
+            row = self.connection.execute(
+                "SELECT * FROM export_jobs WHERE job_id = ?", (job_id,)
+            ).fetchone()
+            return None if row is None else self._row_to_export_job(row)
+
+    def list_export_jobs(self, *, limit: int = 20) -> list[ExportJob]:
+        if not isinstance(limit, int) or isinstance(limit, bool) or not 1 <= limit <= 100:
+            raise ValueError("export job limit must be between 1 and 100")
+        with self._lock:
+            rows = self.connection.execute(
+                """
+                SELECT * FROM export_jobs
+                ORDER BY
+                    CASE WHEN status IN ('queued','running') THEN 0 ELSE 1 END,
+                    requested_at DESC,
+                    job_id DESC
+                LIMIT ?
+                """,
+                (limit,),
+            ).fetchall()
+            return [self._row_to_export_job(row) for row in rows]
+
+    def list_reconcilable_export_jobs(self) -> list[ExportJob]:
+        with self._lock:
+            rows = self.connection.execute(
+                """
+                SELECT * FROM export_jobs
+                WHERE status IN ('failed','interrupted')
+                ORDER BY requested_at, job_id
+                """
+            ).fetchall()
+            return [self._row_to_export_job(row) for row in rows]
+
     def protect_slot(
         self,
         slot: int,

tests/test_app.py

@@ -14,7 +14,14 @@ from controller.service import ControllerService, ControllerUnavailableError
 
 
 class FakeControllerService:
-    def __init__(self, *, enqueue_result=None, enqueue_error=None, health=None):
+    def __init__(
+        self,
+        *,
+        enqueue_result=None,
+        enqueue_error=None,
+        health=None,
+        export_jobs=None,
+    ):
         self.enqueue_result = enqueue_result or {
             "status": "queued",
             "slot": 3,
@@ -29,7 +36,9 @@ class FakeControllerService:
             "router_state": "video",
             "degraded_reasons": [],
         }
+        self.export_jobs = list(export_jobs or [])
         self.enqueue_calls = []
+        self.export_calls = []
 
     def enqueue_download(self, url, submitter, note):
         self.enqueue_calls.append((url, submitter, note))
@@ -37,6 +46,20 @@ class FakeControllerService:
             raise self.enqueue_error
         return dict(self.enqueue_result)
 
+    def enqueue_export(self, url, mode, submitter, note):
+        self.export_calls.append((url, mode, submitter, note))
+        if self.enqueue_error is not None:
+            raise self.enqueue_error
+        return {
+            "kind": "export",
+            "status": "queued",
+            "job_id": "export-job",
+            "mode": mode,
+        }
+
+    def list_export_jobs(self, *, limit=10):
+        return [dict(job) for job in self.export_jobs[:limit]]
+
     def health(self):
         return dict(self.health_result)
 
@@ -71,11 +94,94 @@ class FlaskFactoryTests(unittest.TestCase):
         response = self._app(FakeControllerService()).test_client().get("/")
 
         self.assertEqual(response.status_code, 200)
-        self.assertIn(b"Submit YouTube Playlist", response.data)
+        self.assertIn(b"Submit YouTube Media", response.data)
         self.assertIn(b'action="/submit"', response.data)
         self.assertIn(b'name="url"', response.data)
         self.assertIn(b'name="submitter"', response.data)
         self.assertIn(b'name="note"', response.data)
+        self.assertIn(b'name="download_mode"', response.data)
+        self.assertIn(b'value="player_slot" checked', response.data)
+        self.assertIn(b"MP4 Export", response.data)
+        self.assertIn(b"M4A Audio", response.data)
+        self.assertIn(b"MP3 Audio", response.data)
+
+    def test_older_clients_omitting_mode_still_use_player_slot(self):
+        service = FakeControllerService()
+
+        response = self._post(service, submitter=" Alice ")
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(
+            service.enqueue_calls,
+            [("https://www.youtube.com/playlist?list=PL1234567890", "Alice", None)],
+        )
+        self.assertEqual(service.export_calls, [])
+
+    def test_each_utility_mode_uses_export_path_without_player_slot(self):
+        for mode, label in (
+            ("mp4", b"MP4 Export queued"),
+            ("m4a", b"M4A Audio queued"),
+            ("mp3", b"MP3 Audio queued"),
+        ):
+            with self.subTest(mode=mode):
+                service = FakeControllerService()
+                response = self._post(
+                    service,
+                    download_mode=mode,
+                    submitter="Operator",
+                    note="Utility",
+                )
+                self.assertEqual(response.status_code, 200)
+                self.assertIn(label, response.data)
+                self.assertEqual(service.enqueue_calls, [])
+                self.assertEqual(
+                    service.export_calls,
+                    [
+                        (
+                            "https://www.youtube.com/playlist?list=PL1234567890",
+                            mode,
+                            "Operator",
+                            "Utility",
+                        )
+                    ],
+                )
+
+    def test_invalid_download_mode_is_rejected_server_side(self):
+        service = FakeControllerService()
+
+        response = self._post(service, download_mode="../../slot1")
+
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(b"Invalid download type", response.data)
+        self.assertEqual(service.enqueue_calls, [])
+        self.assertEqual(service.export_calls, [])
+
+    def test_recent_export_status_shows_results_and_failures(self):
+        service = FakeControllerService(
+            export_jobs=[
+                {
+                    "job_id": "one",
+                    "mode": "mp4",
+                    "status": "completed",
+                    "output_paths": ["exports/video/one/Service.mp4"],
+                    "error": None,
+                },
+                {
+                    "job_id": "two",
+                    "mode": "mp3",
+                    "status": "failed",
+                    "output_paths": [],
+                    "error": "ffmpeg failed",
+                },
+            ]
+        )
+
+        response = self._app(service).test_client().get("/")
+
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(b"Recent utility exports", response.data)
+        self.assertIn(b"exports/video/one/Service.mp4", response.data)
+        self.assertIn(b"ffmpeg failed", response.data)
 
     def test_submit_rejects_non_youtube_and_malformed_urls(self):
         invalid_urls = (

tests/test_config_contracts.py

@@ -82,6 +82,10 @@ class ConfigurationTests(unittest.TestCase):
         self.assertEqual(config.storage.uncertain_review_after_seconds, 900.0)
         self.assertEqual(config.mqtt.publish_ack_timeout_seconds, 2.0)
         self.assertEqual(config.mqtt.router_echo_grace_seconds, 2.0)
+        self.assertEqual(config.downloads.export_directory, Path("exports"))
+        self.assertEqual(config.downloads.export_max_height, 1080)
+        self.assertEqual(config.downloads.export_max_workers, 1)
+        self.assertEqual(config.downloads.max_queued_exports, 20)
 
     def test_documented_environment_overrides_apply(self):
         env = {
@@ -175,6 +179,48 @@ class ConfigurationTests(unittest.TestCase):
                 with self.assertRaisesRegex(ConfigError, f"{field} must be an absolute path"):
                     load_config(path, environ={})
 
+    def test_export_directory_must_stay_relative_to_media_root(self):
+        invalid = ("/tmp/exports", "../exports", "nested/../exports", r"nested\\exports")
+        for value in invalid:
+            with self.subTest(value=value), tempfile.TemporaryDirectory() as directory:
+                path = Path(directory) / "config.yaml"
+                path.write_text(
+                    f"downloads:\n  export_directory: {value!r}\n",
+                    encoding="utf-8",
+                )
+                with self.assertRaisesRegex(ConfigError, "safe relative directory"):
+                    load_config(path, environ={})
+
+    def test_export_directory_cannot_overlap_player_slot_storage(self):
+        invalid = ("slot1", "SLOT10/exports", ".controller-trash/exports")
+        for value in invalid:
+            with self.subTest(value=value), tempfile.TemporaryDirectory() as directory:
+                path = Path(directory) / "config.yaml"
+                path.write_text(
+                    f"downloads:\n  export_directory: {value!r}\n",
+                    encoding="utf-8",
+                )
+                with self.assertRaisesRegex(ConfigError, "slot storage"):
+                    load_config(path, environ={})
+
+    def test_export_limits_are_bounded_positive_integers(self):
+        cases = (
+            ("export_max_height", 0),
+            ("export_max_workers", 0),
+            ("export_max_workers", 11),
+            ("max_queued_exports", 0),
+            ("max_queued_exports", 101),
+        )
+        for field, value in cases:
+            with self.subTest(field=field), tempfile.TemporaryDirectory() as directory:
+                path = Path(directory) / "config.yaml"
+                path.write_text(
+                    f"downloads:\n  {field}: {value}\n",
+                    encoding="utf-8",
+                )
+                with self.assertRaisesRegex(ConfigError, field):
+                    load_config(path, environ={})
+
 
 class ImportLifecycleTests(unittest.TestCase):
     def test_controller_core_import_does_not_connect_or_create_runtime_paths(self):

tests/test_downloads.py

@@ -8,8 +8,14 @@ import unittest
 from concurrent.futures import Future
 from pathlib import Path
 
+from yt_dlp import YoutubeDL
+
 from controller.config import ControllerConfig, DownloadConfig, StorageConfig
-from controller.downloads import DownloadManager, canonical_playlist_path
+from controller.downloads import (
+    DownloadManager,
+    canonical_playlist_path,
+    player_format_selector,
+)
 from controller.models import SlotProtectedError
 from controller.store import ControllerStore
 
@@ -169,7 +175,120 @@ class DownloadManagerTests(unittest.TestCase):
         self.assertTrue(all(not path.exists() for path in working_paths))
         self.assertEqual(self.cookie.read_text(encoding="utf-8"), "cookie data")
         download_arguments = runner.calls[-1][0]
-        self.assertIn("bv*[height<=720]+ba/b[height<=720]/b", download_arguments)
+        self.assertIn("bv*[height<=720]+ba/b[height<=720]", download_arguments)
+
+    @staticmethod
+    def _select_player_format(max_height, formats):
+        with YoutubeDL({"quiet": True}) as ydl:
+            selector = ydl.build_format_selector(
+                player_format_selector(max_height)
+            )
+            selected = list(ydl._select_formats(formats, selector))
+        return selected
+
+    def test_player_selector_chooses_separate_true_1080_video_and_audio(self):
+        formats = [
+            {
+                "format_id": "22",
+                "url": "https://example.test/720.mp4",
+                "ext": "mp4",
+                "height": 720,
+                "width": 1280,
+                "vcodec": "avc1.64001f",
+                "acodec": "mp4a.40.2",
+                "tbr": 2500,
+                "filesize": 1,
+            },
+            {
+                "format_id": "399",
+                "url": "https://example.test/1080.mp4",
+                "ext": "mp4",
+                "height": 1080,
+                "width": 1920,
+                "vcodec": "av01.0.08M.08",
+                "acodec": "none",
+                "tbr": 4500,
+                "filesize": 1,
+            },
+            {
+                "format_id": "251",
+                "url": "https://example.test/audio.webm",
+                "ext": "webm",
+                "vcodec": "none",
+                "acodec": "opus",
+                "tbr": 160,
+                "filesize": 1,
+            },
+        ]
+
+        selected = self._select_player_format(1080, formats)
+
+        self.assertEqual(len(selected), 1)
+        self.assertEqual(selected[0]["height"], 1080)
+        self.assertEqual(
+            [item["format_id"] for item in selected[0]["requested_formats"]],
+            ["399", "251"],
+        )
+        self.assertEqual(selected[0]["vcodec"], "av01.0.08M.08")
+        self.assertEqual(selected[0]["acodec"], "opus")
+
+    def test_player_selector_preserves_lower_source_resolution(self):
+        formats = [
+            {
+                "format_id": "18",
+                "url": "https://example.test/360.mp4",
+                "ext": "mp4",
+                "height": 360,
+                "width": 640,
+                "vcodec": "avc1.42001e",
+                "acodec": "mp4a.40.2",
+                "tbr": 700,
+                "filesize": 1,
+            },
+            {
+                "format_id": "22",
+                "url": "https://example.test/720.mp4",
+                "ext": "mp4",
+                "height": 720,
+                "width": 1280,
+                "vcodec": "avc1.64001f",
+                "acodec": "mp4a.40.2",
+                "tbr": 2500,
+                "filesize": 1,
+            },
+        ]
+
+        selected = self._select_player_format(1080, formats)
+
+        self.assertEqual(len(selected), 1)
+        self.assertEqual(selected[0]["format_id"], "22")
+        self.assertEqual(selected[0]["height"], 720)
+
+    def test_player_selector_never_uses_unfiltered_over_height_fallback(self):
+        formats = [
+            {
+                "format_id": "401",
+                "url": "https://example.test/2160.mp4",
+                "ext": "mp4",
+                "height": 2160,
+                "width": 3840,
+                "vcodec": "av01.0.12M.08",
+                "acodec": "none",
+                "tbr": 12000,
+                "filesize": 1,
+            },
+            {
+                "format_id": "251",
+                "url": "https://example.test/audio.webm",
+                "ext": "webm",
+                "vcodec": "none",
+                "acodec": "opus",
+                "tbr": 160,
+                "filesize": 1,
+            },
+        ]
+
+        self.assertEqual(self._select_player_format(1080, formats), [])
 
     def test_worker_failure_is_contained_and_persisted(self):
         events = []

tests/test_exports.py

@@ -0,0 +1,655 @@
+import json
+import subprocess
+import tempfile
+import unittest
+from concurrent.futures import Future
+from pathlib import Path
+
+from yt_dlp import YoutubeDL
+
+from controller.config import ControllerConfig, DownloadConfig, StorageConfig
+from controller.downloads import (
+    DownloadManager,
+    mp4_export_format_selector,
+    safe_export_filename,
+)
+from controller.store import ControllerStore
+
+
+class ImmediateExecutor:
+    def submit(self, function, *args, **kwargs):
+        future = Future()
+        try:
+            future.set_result(function(*args, **kwargs))
+        except BaseException as exc:
+            future.set_exception(exc)
+        return future
+
+
+class QueuedExecutor:
+    def __init__(self):
+        self.calls = []
+
+    def submit(self, function, *args, **kwargs):
+        future = Future()
+        self.calls.append((future, function, args, kwargs))
+        return future
+
+
+class FakeExportRunner:
+    def __init__(
+        self,
+        *,
+        video_codec="h264",
+        audio_codec="aac",
+        pixel_format="yuv420p",
+        container="mp4",
+        height=1080,
+        playlist_items=1,
+        fail_download=False,
+    ):
+        self.video_codec = video_codec
+        self.audio_codec = audio_codec
+        self.pixel_format = pixel_format
+        self.container = container
+        self.height = height
+        self.playlist_items = playlist_items
+        self.fail_download = fail_download
+        self.calls = []
+
+    def __call__(self, arguments, **kwargs):
+        arguments = list(arguments)
+        self.calls.append((arguments, dict(kwargs)))
+        program = Path(arguments[0]).name
+        if program == "yt-dlp-test":
+            if self.fail_download:
+                raise subprocess.CalledProcessError(9, arguments)
+            template = Path(arguments[arguments.index("-o") + 1])
+            template.parent.mkdir(parents=True, exist_ok=True)
+            if "--audio-format" in arguments:
+                extension = "." + arguments[arguments.index("--audio-format") + 1]
+            else:
+                extension = {
+                    "mp4": ".mp4",
+                    "matroska": ".mkv",
+                    "webm": ".webm",
+                    "mov": ".mov",
+                }[self.container]
+            for index in range(1, self.playlist_items + 1):
+                (template.parent / f"{index:03d} - Tést Item {index} [id{index}]{extension}").write_bytes(
+                    f"source-{index}".encode()
+                )
+            return subprocess.CompletedProcess(arguments, 0, stdout="", stderr="")
+        if program == "ffmpeg":
+            Path(arguments[-1]).write_bytes(b"converted")
+            return subprocess.CompletedProcess(arguments, 0, stdout="", stderr="")
+        if program == "ffprobe":
+            path = Path(arguments[-1])
+            converted = path.is_file() and path.read_bytes() == b"converted"
+            if path.suffix == ".m4a":
+                payload = {
+                    "streams": [{"codec_type": "audio", "codec_name": "aac"}],
+                    "format": {"format_name": "mov,mp4,m4a,3gp,3g2,mj2"},
+                }
+            elif path.suffix == ".mp3":
+                payload = {
+                    "streams": [{"codec_type": "audio", "codec_name": "mp3"}],
+                    "format": {"format_name": "mp3"},
+                }
+            else:
+                payload = {
+                    "streams": [
+                        {
+                            "codec_type": "video",
+                            "codec_name": "h264" if converted else self.video_codec,
+                            "pix_fmt": "yuv420p" if converted else self.pixel_format,
+                            "width": 1920 if self.height == 1080 else 1280,
+                            "height": self.height,
+                        },
+                        {
+                            "codec_type": "audio",
+                            "codec_name": "aac" if converted else self.audio_codec,
+                        },
+                    ],
+                    "format": {
+                        "format_name": "mov,mp4,m4a,3gp,3g2,mj2"
+                        if converted or self.container in {"mp4", "mov"}
+                        else self.container
+                    },
+                }
+            return subprocess.CompletedProcess(
+                arguments,
+                0,
+                stdout=json.dumps(payload),
+                stderr="",
+            )
+        raise AssertionError(f"unexpected command: {arguments!r}")
+
+    @property
+    def yt_dlp_calls(self):
+        return [arguments for arguments, _ in self.calls if Path(arguments[0]).name == "yt-dlp-test"]
+
+    @property
+    def ffmpeg_calls(self):
+        return [arguments for arguments, _ in self.calls if Path(arguments[0]).name == "ffmpeg"]
+
+
+class UtilityExportTests(unittest.TestCase):
+    def setUp(self):
+        self.tempdir = tempfile.TemporaryDirectory()
+        base = Path(self.tempdir.name)
+        self.media = base / "media"
+        self.data = base / "data"
+        self.config = ControllerConfig(
+            storage=StorageConfig(media_base=self.media, data_base=self.data),
+            downloads=DownloadConfig(
+                yt_dlp_binary="yt-dlp-test",
+                cookie_path=None,
+                max_workers=1,
+                process_timeout_seconds=30,
+                max_height=1080,
+                export_directory=Path("exports"),
+                export_max_height=1080,
+                max_queued_exports=5,
+            ),
+        )
+        self.store = ControllerStore(self.config.storage).open()
+
+    def tearDown(self):
+        self.store.close()
+        self.tempdir.cleanup()
+
+    def manager(self, runner, events=None):
+        return DownloadManager(
+            self.config,
+            self.store,
+            runner=runner,
+            executor=ImmediateExecutor(),
+            export_executor=ImmediateExecutor(),
+            on_event=(events if events is not None else []).append,
+        )
+
+    def assert_player_slots_empty(self):
+        self.assertTrue(
+            all(slot.media_state == "empty" for slot in self.store.list_slots())
+        )
+
+    def test_compatible_mp4_exports_without_slot_or_transcode(self):
+        events = []
+        runner = FakeExportRunner()
+        result = self.manager(runner, events).enqueue_export(
+            "https://www.youtube.com/watch?v=one", "mp4", "Alice", "Slides"
+        )
+
+        self.assertEqual(result.status, "queued")
+        job = self.store.get_export_job(result.job_id)
+        self.assertEqual(job.status, "completed")
+        self.assertEqual(len(job.output_paths), 1)
+        self.assertTrue(job.output_paths[0].startswith(f"exports/video/{result.job_id}/"))
+        self.assertTrue(job.output_paths[0].endswith(".mp4"))
+        self.assertTrue((self.media / job.output_paths[0]).is_file())
+        self.assertEqual(events, [])
+        self.assertEqual(runner.ffmpeg_calls, [])
+        self.assert_player_slots_empty()
+
+        arguments = runner.yt_dlp_calls[0]
+        self.assertIn("--yes-playlist", arguments)
+        self.assertIn("--windows-filenames", arguments)
+        self.assertIn("--no-overwrites", arguments)
+        self.assertEqual(
+            arguments[arguments.index("-f") + 1],
+            "bv*[height<=1080]+ba/b[height<=1080]",
+        )
+        self.assertEqual(
+            arguments[arguments.index("-S") + 1],
+            "res:1080,vcodec:h264,acodec:aac",
+        )
+
+    def test_incompatible_mp4_transcodes_only_incompatible_codecs(self):
+        runner = FakeExportRunner(
+            video_codec="vp9",
+            audio_codec="opus",
+            pixel_format="yuv420p",
+            container="webm",
+        )
+        result = self.manager(runner).enqueue_export(
+            "https://www.youtube.com/watch?v=vp9", "mp4", None, None
+        )
+
+        self.assertEqual(self.store.get_export_job(result.job_id).status, "completed")
+        self.assertEqual(len(runner.ffmpeg_calls), 1)
+        arguments = runner.ffmpeg_calls[0]
+        self.assertIn("libx264", arguments)
+        self.assertIn("-crf", arguments)
+        self.assertEqual(arguments[arguments.index("-crf") + 1], "18")
+        self.assertIn("yuv420p", arguments)
+        self.assertIn("aac", arguments)
+        self.assertIn("192k", arguments)
+        self.assertNotIn("-vf", arguments)
+        self.assertNotIn("scale", " ".join(arguments))
+        self.assert_player_slots_empty()
+
+    def test_compatible_h264_aac_in_other_container_is_remuxed_only(self):
+        runner = FakeExportRunner(container="matroska")
+        result = self.manager(runner).enqueue_export(
+            "https://www.youtube.com/watch?v=mkv", "mp4", None, None
+        )
+
+        self.assertEqual(self.store.get_export_job(result.job_id).status, "completed")
+        arguments = runner.ffmpeg_calls[0]
+        self.assertEqual(arguments[arguments.index("-c:v") + 1], "copy")
+        self.assertEqual(arguments[arguments.index("-c:a") + 1], "copy")
+        self.assertNotIn("libx264", arguments)
+
+    def test_quicktime_alias_is_remuxed_instead_of_renamed_to_mp4(self):
+        runner = FakeExportRunner(container="mov")
+        result = self.manager(runner).enqueue_export(
+            "https://www.youtube.com/watch?v=mov", "mp4", None, None
+        )
+
+        self.assertEqual(self.store.get_export_job(result.job_id).status, "completed")
+        arguments = runner.ffmpeg_calls[0]
+        self.assertEqual(arguments[arguments.index("-c:v") + 1], "copy")
+        self.assertEqual(arguments[arguments.index("-c:a") + 1], "copy")
+        self.assertNotIn("libx264", arguments)
+
+    def test_export_reconciliation_is_submitted_to_dedicated_executor(self):
+        export_executor = QueuedExecutor()
+        manager = DownloadManager(
+            self.config,
+            self.store,
+            runner=FakeExportRunner(),
+            executor=ImmediateExecutor(),
+            export_executor=export_executor,
+        )
+
+        self.assertTrue(manager.schedule_export_reconciliation())
+        self.assertEqual(len(export_executor.calls), 1)
+        _, function, arguments, keyword_arguments = export_executor.calls[0]
+        self.assertEqual(function, manager.reconcile_export_artifacts)
+        self.assertEqual(arguments, ())
+        self.assertEqual(keyword_arguments, {})
+
+    def test_lower_resolution_mp4_is_preserved_without_scaling(self):
+        runner = FakeExportRunner(height=720)
+        result = self.manager(runner).enqueue_export(
+            "https://www.youtube.com/watch?v=720", "mp4", None, None
+        )
+
+        self.assertEqual(self.store.get_export_job(result.job_id).status, "completed")
+        self.assertEqual(runner.ffmpeg_calls, [])
+
+    def test_over_height_mp4_is_rejected_even_if_downloader_misbehaves(self):
+        runner = FakeExportRunner(height=2160)
+        result = self.manager(runner).enqueue_export(
+            "https://www.youtube.com/watch?v=2160", "mp4", None, None
+        )
+
+        job = self.store.get_export_job(result.job_id)
+        self.assertEqual(job.status, "failed")
+        self.assertIn("maximum height", job.error)
+        self.assertFalse((self.media / "exports" / "video" / result.job_id).exists())
+        self.assert_player_slots_empty()
+
+    def test_audio_exports_use_intentional_audio_only_strategies(self):
+        for mode, expected_selector in (
+            ("m4a", "ba/b"),
+            ("mp3", "ba/b"),
+        ):
+            with self.subTest(mode=mode):
+                events = []
+                runner = FakeExportRunner()
+                result = self.manager(runner, events).enqueue_export(
+                    f"https://www.youtube.com/watch?v={mode}", mode, None, None
+                )
+                job = self.store.get_export_job(result.job_id)
+                self.assertEqual(job.status, "completed")
+                self.assertEqual(len(job.output_paths), 1)
+                self.assertTrue(job.output_paths[0].endswith(f".{mode}"))
+                arguments = runner.yt_dlp_calls[0]
+                self.assertEqual(arguments[arguments.index("-f") + 1], expected_selector)
+                self.assertIn("-x", arguments)
+                self.assertEqual(
+                    arguments[arguments.index("--audio-format") + 1], mode
+                )
+                self.assertEqual(
+                    arguments[arguments.index("--audio-quality") + 1], "0"
+                )
+                if mode == "m4a":
+                    self.assertEqual(
+                        arguments[arguments.index("-S") + 1],
+                        "abr,acodec:aac",
+                    )
+                self.assertEqual(runner.ffmpeg_calls, [])
+                self.assertEqual(events, [])
+        self.assert_player_slots_empty()
+
+    def test_playlist_items_publish_as_individual_collision_safe_files(self):
+        runner = FakeExportRunner(playlist_items=3)
+        result = self.manager(runner).enqueue_export(
+            "https://www.youtube.com/playlist?list=three", "m4a", None, None
+        )
+
+        job = self.store.get_export_job(result.job_id)
+        self.assertEqual(job.status, "completed")
+        self.assertEqual(len(job.output_paths), 3)
+        self.assertEqual(len(set(job.output_paths)), 3)
+        self.assertTrue(all((self.media / path).is_file() for path in job.output_paths))
+
+    def test_export_failure_and_expiry_never_mutate_player_or_export_files(self):
+        slot_one = self.media / "slot1"
+        slot_one.mkdir(parents=True)
+        protected_file = slot_one / "active.mp4"
+        protected_file.write_bytes(b"active")
+        failed = self.manager(FakeExportRunner(fail_download=True)).enqueue_export(
+            "https://www.youtube.com/watch?v=fail", "mp3", None, None
+        )
+        self.assertEqual(self.store.get_export_job(failed.job_id).status, "failed")
+        self.assertEqual(protected_file.read_bytes(), b"active")
+        self.assert_player_slots_empty()
+
+        completed = self.manager(FakeExportRunner()).enqueue_export(
+            "https://www.youtube.com/watch?v=keep", "m4a", None, None
+        )
+        export_path = self.media / self.store.get_export_job(completed.job_id).output_paths[0]
+        self.manager(FakeExportRunner()).expire_due_slots()
+        self.assertTrue(export_path.is_file())
+        self.assertEqual(protected_file.read_bytes(), b"active")
+
+    def test_symlinked_export_root_cannot_redirect_into_player_slot(self):
+        slot_one = self.media / "slot1"
+        slot_one.mkdir(parents=True)
+        protected_file = slot_one / "active.mp4"
+        protected_file.write_bytes(b"active")
+        (self.media / "exports").symlink_to(slot_one, target_is_directory=True)
+
+        result = self.manager(FakeExportRunner()).enqueue_export(
+            "https://www.youtube.com/watch?v=escape", "m4a", None, None
+        )
+
+        self.assertEqual(self.store.get_export_job(result.job_id).status, "failed")
+        self.assertIn("unsafe export directory", self.store.get_export_job(result.job_id).error)
+        self.assertEqual(protected_file.read_bytes(), b"active")
+
+    def test_export_work_uses_dedicated_executor_capacity(self):
+        player_executor = QueuedExecutor()
+        export_executor = QueuedExecutor()
+        manager = DownloadManager(
+            self.config,
+            self.store,
+            runner=FakeExportRunner(),
+            executor=player_executor,
+            export_executor=export_executor,
+        )
+
+        manager.enqueue_export(
+            "https://www.youtube.com/watch?v=export", "mp4", None, None
+        )
+        manager.enqueue(
+            "https://www.youtube.com/watch?v=player", None, None
+        )
+
+        self.assertEqual(len(export_executor.calls), 1)
+        self.assertEqual(len(player_executor.calls), 1)
+
+    def test_restart_reconciles_published_export_and_cleans_manifest(self):
+        runner = FakeExportRunner()
+        manager = self.manager(runner)
+        job_id = "a" * 32
+        self.store.reserve_export_job(
+            "https://www.youtube.com/watch?v=recover",
+            "mp4",
+            None,
+            None,
+            max_active=5,
+            job_id=job_id,
+        )
+        self.store.mark_export_running(job_id)
+        self.store.fail_export_job(job_id, "database completion failed")
+        work = self.media / "exports" / ".work" / job_id
+        published = self.media / "exports" / "video" / job_id
+        work.mkdir(parents=True)
+        published.mkdir(parents=True)
+        filename = "Recovered.mp4"
+        (published / filename).write_bytes(b"compatible")
+        manager._write_export_manifest(
+            work,
+            job_id=job_id,
+            mode="mp4",
+            filenames=[filename],
+        )
+
+        manager.reconcile_export_artifacts()
+
+        job = self.store.get_export_job(job_id)
+        self.assertEqual(job.status, "completed")
+        self.assertEqual(
+            job.output_paths,
+            (f"exports/video/{job_id}/{filename}",),
+        )
+        self.assertFalse(work.exists())
+        self.assertTrue((published / filename).is_file())
+
+    def test_restart_atomically_publishes_manifested_ready_export(self):
+        runner = FakeExportRunner()
+        manager = self.manager(runner)
+        job_id = "b" * 32
+        self.store.reserve_export_job(
+            "https://www.youtube.com/watch?v=staged",
+            "m4a",
+            None,
+            None,
+            max_active=5,
+            job_id=job_id,
+        )
+        self.store.mark_export_running(job_id)
+        self.store.fail_export_job(job_id, "interrupted after conversion")
+        work = self.media / "exports" / ".work" / job_id
+        ready = work / "ready"
+        ready.mkdir(parents=True)
+        filename = "Recovered.m4a"
+        (ready / filename).write_bytes(b"aac")
+        manager._write_export_manifest(
+            work,
+            job_id=job_id,
+            mode="m4a",
+            filenames=[filename],
+        )
+
+        manager.reconcile_export_artifacts()
+
+        job = self.store.get_export_job(job_id)
+        output = self.media / job.output_paths[0]
+        self.assertEqual(job.status, "completed")
+        self.assertTrue(output.is_file())
+        self.assertFalse(work.exists())
+
+    def test_restart_discards_only_known_incomplete_export_staging(self):
+        manager = self.manager(FakeExportRunner())
+        job_id = "c" * 32
+        self.store.reserve_export_job(
+            "https://www.youtube.com/watch?v=incomplete",
+            "mp3",
+            None,
+            None,
+            max_active=5,
+            job_id=job_id,
+        )
+        self.store.fail_export_job(job_id, "interrupted download")
+        work = self.media / "exports" / ".work" / job_id
+        work.mkdir(parents=True)
+        (work / "partial.part").write_bytes(b"partial")
+        unknown = self.media / "exports" / ".work" / ("d" * 32)
+        unknown.mkdir()
+        (unknown / "operator-file").write_bytes(b"keep")
+
+        manager.reconcile_export_artifacts()
+
+        self.assertFalse(work.exists())
+        self.assertTrue((unknown / "operator-file").is_file())
+
+    def test_safe_filename_blocks_traversal_preserves_unicode_and_deduplicates(self):
+        used = set()
+        first = safe_export_filename(
+            "../../slot1/CON:Tést\x00?", ".mp4", used_casefolded=used
+        )
+        second = safe_export_filename(
+            "../../slot1/CON:Tést\x00?", ".mp4", used_casefolded=used
+        )
+        long_name = safe_export_filename("🎬" * 500, ".mp3")
+
+        self.assertNotIn("/", first)
+        self.assertNotIn("\\", first)
+        self.assertNotIn("..", Path(first).parts)
+        self.assertIn("Tést", first)
+        self.assertNotEqual(first.casefold(), second.casefold())
+        self.assertTrue(second.endswith(" (2).mp4"))
+        self.assertLessEqual(len(Path(long_name).stem.encode("utf-8")), 180)
+
+    def test_mp4_sorting_keeps_resolution_ahead_of_codec_preference(self):
+        formats = [
+            {
+                "format_id": "av1-1080",
+                "url": "https://example.test/av1",
+                "ext": "webm",
+                "height": 1080,
+                "width": 1920,
+                "vcodec": "av01.0.08M.08",
+                "acodec": "none",
+                "tbr": 4000,
+                "filesize": 1,
+            },
+            {
+                "format_id": "h264-720",
+                "url": "https://example.test/h264",
+                "ext": "mp4",
+                "height": 720,
+                "width": 1280,
+                "vcodec": "avc1.64001f",
+                "acodec": "none",
+                "tbr": 3000,
+                "filesize": 1,
+            },
+            {
+                "format_id": "aac",
+                "url": "https://example.test/aac",
+                "ext": "m4a",
+                "vcodec": "none",
+                "acodec": "mp4a.40.2",
+                "tbr": 128,
+                "filesize": 1,
+            },
+        ]
+        with YoutubeDL(
+            {
+                "quiet": True,
+                "format_sort": ["res:1080", "vcodec:h264", "acodec:aac"],
+            }
+        ) as ydl:
+            info = {"formats": formats}
+            ydl.sort_formats(info)
+            selector = ydl.build_format_selector(mp4_export_format_selector(1080))
+            selected = list(ydl._select_formats(info["formats"], selector))
+
+        self.assertEqual(selected[0]["height"], 1080)
+        self.assertEqual(selected[0]["requested_formats"][0]["format_id"], "av1-1080")
+
+    def test_mp4_sorting_prefers_native_h264_aac_at_equal_quality(self):
+        formats = [
+            {
+                "format_id": "vp9-1080",
+                "url": "https://example.test/vp9",
+                "ext": "webm",
+                "height": 1080,
+                "width": 1920,
+                "vcodec": "vp9",
+                "acodec": "none",
+                "tbr": 4200,
+                "filesize": 1,
+            },
+            {
+                "format_id": "h264-1080",
+                "url": "https://example.test/h264",
+                "ext": "mp4",
+                "height": 1080,
+                "width": 1920,
+                "vcodec": "avc1.640028",
+                "acodec": "none",
+                "tbr": 4000,
+                "filesize": 1,
+            },
+            {
+                "format_id": "opus",
+                "url": "https://example.test/opus",
+                "ext": "webm",
+                "vcodec": "none",
+                "acodec": "opus",
+                "tbr": 160,
+                "filesize": 1,
+            },
+            {
+                "format_id": "aac",
+                "url": "https://example.test/aac",
+                "ext": "m4a",
+                "vcodec": "none",
+                "acodec": "mp4a.40.2",
+                "tbr": 128,
+                "filesize": 1,
+            },
+        ]
+        with YoutubeDL(
+            {
+                "quiet": True,
+                "format_sort": ["res:1080", "vcodec:h264", "acodec:aac"],
+            }
+        ) as ydl:
+            info = {"formats": formats}
+            ydl.sort_formats(info)
+            selector = ydl.build_format_selector(mp4_export_format_selector(1080))
+            selected = list(ydl._select_formats(info["formats"], selector))
+
+        self.assertEqual(
+            [item["format_id"] for item in selected[0]["requested_formats"]],
+            ["h264-1080", "aac"],
+        )
+
+    def test_m4a_sorting_keeps_audio_quality_ahead_of_native_aac_preference(self):
+        formats = [
+            {
+                "format_id": "aac-low",
+                "url": "https://example.test/aac-low",
+                "ext": "m4a",
+                "vcodec": "none",
+                "acodec": "mp4a.40.2",
+                "abr": 48,
+                "tbr": 48,
+                "filesize": 1,
+            },
+            {
+                "format_id": "opus-high",
+                "url": "https://example.test/opus-high",
+                "ext": "webm",
+                "vcodec": "none",
+                "acodec": "opus",
+                "abr": 160,
+                "tbr": 160,
+                "filesize": 1,
+            },
+        ]
+        with YoutubeDL(
+            {"quiet": True, "format_sort": ["abr", "acodec:aac"]}
+        ) as ydl:
+            info = {"formats": formats}
+            ydl.sort_formats(info)
+            selected = list(
+                ydl._select_formats(
+                    info["formats"],
+                    ydl.build_format_selector("ba/b"),
+                )
+            )
+
+        self.assertEqual(selected[0]["format_id"], "opus-high")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_service.py

@@ -89,11 +89,27 @@ class FakeDownloadManager:
         self.stopped = False
         self.delete_requests = []
         self.enqueue_calls = []
+        self.export_calls = []
+        self.export_reconciled = False
 
     def enqueue(self, url, submitter, note):
         self.enqueue_calls.append((url, submitter, note))
         return self.store.reserve_download(url, submitter, note)
 
+    def enqueue_export(self, url, mode, submitter, note):
+        self.export_calls.append((url, mode, submitter, note))
+        return self.store.reserve_export_job(
+            url,
+            mode,
+            submitter,
+            note,
+            max_active=20,
+        )
+
+    def schedule_export_reconciliation(self):
+        self.export_reconciled = True
+        return True
+
     def recover_pending_deletions(self):
         self.recovered = True
 
@@ -171,6 +187,7 @@ class ControllerServiceTests(unittest.TestCase):
         )
         self.service.start()
         self.assertFalse(self.downloads.recovered)
+        self.assertTrue(self.downloads.export_reconciled)
         return self.service
 
     def ready_slot(self, slot=1):
@@ -497,6 +514,62 @@ class ControllerServiceTests(unittest.TestCase):
         self.assertIn("mqtt_disconnected", health["degraded_reasons"])
         self.assertIn("awaiting_player_evidence", health["degraded_reasons"])
 
+    def test_utility_export_is_slot_free_and_not_gated_by_player_evidence(self):
+        self.start_service()
+        self.assertIn(
+            "awaiting_player_evidence",
+            self.service.health()["degraded_reasons"],
+        )
+
+        result = self.service.enqueue_export(
+            "https://www.youtube.com/playlist?list=utility",
+            "mp4",
+            "Alice",
+            "ProPresenter",
+        )
+
+        self.assertEqual(result["kind"], "export")
+        self.assertEqual(result["status"], "queued")
+        self.assertEqual(result["mode"], "mp4")
+        self.assertIsNotNone(result["job_id"])
+        self.assertEqual(
+            self.downloads.export_calls,
+            [
+                (
+                    "https://www.youtube.com/playlist?list=utility",
+                    "mp4",
+                    "Alice",
+                    "ProPresenter",
+                )
+            ],
+        )
+        self.assertTrue(
+            all(slot.media_state == "empty" for slot in self.store.list_slots())
+        )
+        self.assertEqual(self.mqtt.json_messages, [])
+        self.assertEqual(self.mqtt.legacy_messages, [])
+        jobs = self.service.list_export_jobs(limit=10)
+        self.assertEqual(len(jobs), 1)
+        self.assertEqual(jobs[0]["job_id"], result["job_id"])
+        self.assertEqual(jobs[0]["output_paths"], [])
+
+    def test_invalid_utility_mode_is_rejected_without_player_side_effects(self):
+        self.start_service()
+
+        with self.assertRaisesRegex(ValueError, "unsupported utility export mode"):
+            self.service.enqueue_export(
+                "https://www.youtube.com/watch?v=one",
+                "slot1",
+                None,
+                None,
+            )
+
+        self.assertTrue(
+            all(slot.media_state == "empty" for slot in self.store.list_slots())
+        )
+        self.assertEqual(self.mqtt.json_messages, [])
+        self.assertEqual(self.mqtt.legacy_messages, [])
+
     def test_cleanup_waits_for_post_start_player_evidence_but_new_ingest_does_not(self):
         self.start_service()
         self.service._next_expiry = self.monotonic()

tests/test_store.py

@@ -50,7 +50,7 @@ class FreshStoreTests(StoreTestCase):
             self.assertTrue(all(slot.media_state == "empty" for slot in slots))
             self.assertEqual(store.player_observation().phase, "unknown")
 
-    def test_schema_one_database_upgrades_external_reconciliation_atomically(self):
+    def test_schema_one_database_upgrades_reconciliation_and_exports_atomically(self):
         with self.open_store() as store:
             database_path = store.path
         connection = sqlite3.connect(database_path)
@@ -67,7 +67,7 @@ class FreshStoreTests(StoreTestCase):
         with self.open_store() as upgraded:
             self.assertEqual(
                 upgraded.connection.execute("PRAGMA user_version").fetchone()[0],
-                2,
+                3,
             )
             columns = {
                 row["name"]
@@ -80,6 +80,164 @@ class FreshStoreTests(StoreTestCase):
                 "SELECT name FROM schema_migrations WHERE version = 2"
             ).fetchone()
             self.assertEqual(migration["name"], "external_playback_reconciliation")
+            export_migration = upgraded.connection.execute(
+                "SELECT name FROM schema_migrations WHERE version = 3"
+            ).fetchone()
+            self.assertEqual(export_migration["name"], "utility_export_jobs")
+            tables = {
+                row["name"]
+                for row in upgraded.connection.execute(
+                    "SELECT name FROM sqlite_master WHERE type = 'table'"
+                ).fetchall()
+            }
+            self.assertIn("export_jobs", tables)
+
+    def test_schema_two_database_adds_only_utility_export_jobs(self):
+        with self.open_store() as store:
+            database_path = store.path
+        connection = sqlite3.connect(database_path)
+        try:
+            connection.execute("DROP TABLE export_jobs")
+            connection.execute("DELETE FROM schema_migrations WHERE version = 3")
+            connection.execute("PRAGMA user_version = 2")
+            connection.commit()
+        finally:
+            connection.close()
+
+        with self.open_store() as upgraded:
+            self.assertEqual(
+                upgraded.connection.execute("PRAGMA user_version").fetchone()[0],
+                3,
+            )
+            self.assertEqual(upgraded.list_export_jobs(), [])
+
+    def test_export_jobs_are_slot_free_bounded_and_durable(self):
+        with self.open_store() as store:
+            first = store.reserve_export_job(
+                "https://www.youtube.com/watch?v=one",
+                "mp4",
+                "Alice",
+                "Sunday",
+                max_active=1,
+                job_id="export-one",
+            )
+            self.assertEqual(first.status, "queued")
+            self.assertEqual(first.job_id, "export-one")
+            self.assertEqual(
+                store.reserve_export_job(
+                    "https://www.youtube.com/watch?v=two",
+                    "mp3",
+                    None,
+                    None,
+                    max_active=1,
+                ).status,
+                "queue_full",
+            )
+            self.assertTrue(store.mark_export_running("export-one"))
+            self.assertTrue(
+                store.complete_export_job(
+                    "export-one",
+                    ("exports/video/export-one/Service.mp4",),
+                )
+            )
+            job = store.get_export_job("export-one")
+            self.assertEqual(job.mode, "mp4")
+            self.assertEqual(job.status, "completed")
+            self.assertEqual(job.submitter, "Alice")
+            self.assertEqual(
+                job.output_paths,
+                ("exports/video/export-one/Service.mp4",),
+            )
+            self.assertTrue(all(slot.media_state == "empty" for slot in store.list_slots()))
+
+    def test_export_listing_keeps_live_jobs_visible_before_recent_history(self):
+        with self.open_store() as store:
+            store.reserve_export_job(
+                "https://www.youtube.com/watch?v=active",
+                "mp4",
+                None,
+                None,
+                max_active=5,
+                job_id="active-old",
+            )
+            self.clock.value += 1
+            for index in range(3):
+                job_id = f"completed-{index}"
+                store.reserve_export_job(
+                    f"https://www.youtube.com/watch?v={index}",
+                    "mp3",
+                    None,
+                    None,
+                    max_active=5,
+                    job_id=job_id,
+                )
+                store.mark_export_running(job_id)
+                store.complete_export_job(
+                    job_id,
+                    (f"exports/audio/{job_id}/item.mp3",),
+                )
+                self.clock.value += 1
+
+            listed = store.list_export_jobs(limit=2)
+            self.assertEqual(listed[0].job_id, "active-old")
+            self.assertEqual(listed[0].status, "queued")
+            self.assertEqual(listed[1].job_id, "completed-2")
+
+    def test_restart_marks_live_exports_interrupted_without_touching_slots(self):
+        store = self.open_store()
+        store.reserve_export_job(
+            "https://www.youtube.com/playlist?list=one",
+            "m4a",
+            None,
+            None,
+            max_active=2,
+            job_id="queued-export",
+        )
+        store.reserve_export_job(
+            "https://www.youtube.com/watch?v=two",
+            "mp3",
+            None,
+            None,
+            max_active=2,
+            job_id="running-export",
+        )
+        store.mark_export_running("running-export")
+        store.close()
+
+        with self.open_store() as reopened:
+            jobs = {job.job_id: job for job in reopened.list_export_jobs()}
+            self.assertEqual(jobs["queued-export"].status, "interrupted")
+            self.assertEqual(jobs["running-export"].status, "interrupted")
+            self.assertTrue(
+                all("Controller restart" in job.error for job in jobs.values())
+            )
+            self.assertTrue(
+                all(slot.media_state == "empty" for slot in reopened.list_slots())
+            )
+
+    def test_export_failures_do_not_corrupt_player_inventory(self):
+        with self.open_store() as store:
+            slot = store.reserve_download(
+                "https://www.youtube.com/watch?v=managed",
+                None,
+                None,
+                job_id="managed",
+            )
+            store.fail_download(
+                "managed", slot.slot, slot.generation, "managed failure"
+            )
+            before = store.get_slot(slot.slot)
+            export = store.reserve_export_job(
+                "https://www.youtube.com/watch?v=export",
+                "mp3",
+                None,
+                None,
+                max_active=5,
+                job_id="failed-export",
+            )
+            self.assertEqual(export.status, "queued")
+            self.assertTrue(store.fail_export_job("failed-export", "ffmpeg failed"))
+            self.assertEqual(store.get_slot(slot.slot), before)
 
     def test_slot_reservation_is_committed_before_worker_transition(self):
         with self.open_store() as store: