fix: harden utility export recovery

publish_router_mode registered its self-echo correlation marker before
publishing but removed it in a finally block the instant the QoS 1 publish()
call observed its PUBACK. A broker may legally deliver a Controller's own
subscription copy of that same publish after the PUBACK, since MQTT does not
order the two; when that happened, the Controller mistook its own late echo
for live operator intent and relinquished automatic restoration ownership.

Keep the marker claimable for a short, bounded mqtt.router_echo_grace_seconds
window (default 2s) after the PUBACK instead of clearing it immediately.
Genuine external/operator commands are unaffected: once the grace window
elapses, a matching payload is still correctly treated as operator intent
and ownership is relinquished as before.

Verified with deterministic fake-clock unit tests plus a real Mosquitto
broker integration test (tests/test_router_echo_mosquitto_integration.py,
skipped automatically when Docker is unavailable).

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
                                                             |
                               +-----------------------------+------------------+
@@ -345,8 +346,12 @@ when all of these remain true:
 - No live, non-self mode command and no conflicting state change indicated an
   operator override.
 
-Controller subscribes to both Router mode set and state topics. Its exact
+Controller subscribes to both Router mode set and state topics. Its own
-in-flight echo is recognized; every other live mode-set publication is treated
+mode-set publication is correlated by exact payload and stays recognized as
+a self-echo both while the publish is in flight and, because a broker may
+legally deliver a subscription echo after the PUBACK that acknowledged the
+same publish, for a short `mqtt.router_echo_grace_seconds` window afterward
+(default 2 seconds). Every other live mode-set publication is treated
 conservatively as operator intent, even if it repeats `video`. A retained set
 message is ignored as stale intent. Controller does not publish Router state;
 it accepts a received live or retained state message as the Router's current
@@ -407,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
@@ -548,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 |
@@ -611,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
+record makes the migration idempotent. Existing schema-v1 or schema-v2
-database is upgraded transactionally rather than reimporting YAML.
+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:
 
@@ -682,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
+- Downloads: two Player-slot workers, one dedicated utility-export worker,
-  optional cookies.
+  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.
@@ -714,8 +779,10 @@ permissions, and rotate it as a credential.
 
 The HTTP surface is intentionally small:
 
-- `GET /` renders the playlist submission form.
+- `GET /` renders the submission form and recent utility-export status.
-- `POST /submit` accepts an HTTP(S) URL plus optional submitter and note.
+- `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`,
@@ -748,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,
@@ -846,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.
@@ -877,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
@@ -900,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:
 
@@ -927,8 +1018,11 @@ PYTHONDONTWRITEBYTECODE=1 .venv/bin/python -m unittest discover -s tests -v
 
 - The web form is unauthenticated and listens on `0.0.0.0` by default. Restrict
   the published port with host firewall/network ACLs or place it behind an
-  authenticated reverse proxy. URL validation checks only the `http`/`https`
+  authenticated reverse proxy. Submitted URLs must use `http`/`https`, carry no
-  shape; yt-dlp performs the outbound request.
+  embedded userinfo, and resolve (by exact, normalized hostname match) to an
+  official YouTube host: `youtube.com`, `www.youtube.com`, `m.youtube.com`,
+  `music.youtube.com`, or `youtu.be`. URL, submitter, note, and total request
+  size are also server-side bounded. yt-dlp performs the outbound request.
 - MQTT username/password support is available, but application-level TLS is not
   configured here. Keep this connection on a trusted/protected network or add a
   separately managed secure tunnel/proxy, and grant the Controller only the
@@ -958,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,28 +259,107 @@ 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>
 </html>
 """
 
+ALLOWED_YOUTUBE_HOSTS = frozenset(
+    {
+        "youtube.com",
+        "www.youtube.com",
+        "m.youtube.com",
+        "music.youtube.com",
+        "youtu.be",
+    }
+)
+
+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:
+    if not value or len(value) > MAX_URL_LENGTH:
+        return False
     try:
         parsed = urlsplit(value)
+        hostname = parsed.hostname
+        has_userinfo = parsed.username is not None or parsed.password is not None
     except ValueError:
         return False
-    return parsed.scheme.lower() in {"http", "https"} and bool(parsed.netloc)
+    if parsed.scheme.lower() not in {"http", "https"}:
+        return False
+    if has_userinfo or not hostname:
+        return False
+    return hostname.lower() in ALLOWED_YOUTUBE_HOSTS
 
 
 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)
@@ -277,26 +403,66 @@ def create_app(
     web_app.config.update(
         TESTING=testing,
         SECRET_KEY=secret_key or "controller-test-only-secret",
+        MAX_CONTENT_LENGTH=MAX_REQUEST_CONTENT_LENGTH,
     )
     web_app.extensions["controller_service"] = service
     web_app.extensions["controller_config"] = config
 
     @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):
+        flash("The submission was too large.")
+        return redirect(url_for("index"))
 
     @web_app.post("/submit")
     def submit():
         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("Invalid URL. Use a complete http:// or https:// address.")
+            flash(
+                "Invalid URL. Only YouTube links are accepted "
+                "(e.g. https://www.youtube.com/... or https://youtu.be/...)."
+            )
+            return redirect(url_for("index"))
+        if submitter is not None and len(submitter) > MAX_SUBMITTER_LENGTH:
+            flash(f"Your name must be {MAX_SUBMITTER_LENGTH} characters or fewer.")
+            return redirect(url_for("index"))
+        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

@@ -10,6 +10,11 @@ mqtt:
   command_queue_size: 512
   # QoS 1 is accepted only after the broker PUBACKs within this bound.
   publish_ack_timeout_seconds: 2
+  # Audio Router self-echo correlation stays claimable for this long after
+  # a Controller-originated mode-set publish is PUBACK'd, so a subscription
+  # echo that is delivered after the PUBACK is still recognized as ours
+  # instead of being mistaken for operator intent.
+  router_echo_grace_seconds: 2
 
 topics:
   player_command: youtube/player/cmd
@@ -43,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

@@ -33,6 +33,7 @@ class MqttConfig:
     reconnect_max_seconds: int = 30
     command_queue_size: int = 512
     publish_ack_timeout_seconds: float = 2.0
+    router_echo_grace_seconds: float = 2.0
 
 
 @dataclass(frozen=True)
@@ -62,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)
@@ -171,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
 
@@ -219,6 +241,11 @@ def load_config(
             "mqtt.publish_ack_timeout_seconds",
             minimum=0.1,
         ),
+        router_echo_grace_seconds=_number(
+            mqtt_raw.get("router_echo_grace_seconds", 2.0),
+            "mqtt.router_echo_grace_seconds",
+            minimum=0.0,
+        ),
     )
     if mqtt.reconnect_max_seconds < mqtt.reconnect_min_seconds:
         raise ConfigError("mqtt.reconnect_max_seconds must be >= reconnect_min_seconds")
@@ -271,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",
-                    (
+                    player_format_selector(self.config.downloads.max_height),
-                        f"bv*[height<={self.config.downloads.max_height}]+ba/"
-                        f"b[height<={self.config.downloads.max_height}]/b"
-                    ),
                     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/mqtt_adapter.py

@@ -6,6 +6,7 @@ import json
 import logging
 import queue
 import threading
+import time
 import uuid
 from dataclasses import dataclass
 from typing import Any, Mapping
@@ -58,6 +59,7 @@ class MqttAdapter:
         *,
         client: Any | None = None,
         events: "queue.Queue[MqttEvent] | None" = None,
+        monotonic=time.monotonic,
     ):
         self.config = config
         self.events: "queue.Queue[MqttEvent]" = events or queue.Queue(
@@ -73,8 +75,16 @@ class MqttAdapter:
         self._started = False
         self._loop_started = False
         self._bootstrap_thread: threading.Thread | None = None
+        self._monotonic = monotonic
         self._router_echo_lock = threading.Lock()
-        self._pending_router_echoes: dict[str, bytes] = {}
+        # token -> (exact payload, grace deadline). The deadline is ``None``
+        # while the correlated publish() call has not yet observed its
+        # PUBACK; such an entry never expires on its own, bounded instead by
+        # the publish ack timeout. Once PUBACK is observed, the entry keeps
+        # a short grace window so a subscription echo delivered just after
+        # the PUBACK is still recognized as ours rather than mistaken for
+        # operator intent.
+        self._pending_router_echoes: dict[str, tuple[bytes, float | None]] = {}
         self.last_error: str | None = None
         self._configure_client()
 
@@ -466,15 +476,23 @@ class MqttAdapter:
     def publish_router_mode(self, mode: str) -> PublishReceipt:
         # A broker normally sends our publication back through the matching
         # subscription. Register the exact payload before publishing because
-        # Paho's network thread may receive that echo before publish() returns.
+        # Paho's network thread may receive that echo before publish()
-        # The marker is removed as soon as the QoS 1 publish call completes;
+        # returns. A PUBACK only proves broker receipt of our publish, not
-        # this intentionally prefers a false ownership loss over mistaking a
+        # that the broker has already forwarded (or even scheduled) our own
-        # later same-value operator command for a Controller echo.
+        # subscription copy of it; that copy can legally arrive after the
+        # PUBACK. So the marker is not removed the instant publish()
+        # completes: it stays claimable for a short bounded grace window
+        # afterward, and is removed immediately once actually claimed or
+        # once the grace window elapses. This closes the specific race where
+        # the Controller's own late echo would otherwise be mistaken for
+        # live operator intent and cause automatic restoration ownership to
+        # be relinquished.
         echo_token = uuid.uuid4().hex
+        payload = mode.encode("utf-8")
         with self._router_echo_lock:
-            self._pending_router_echoes[echo_token] = mode.encode("utf-8")
+            self._prune_router_echoes_locked()
-        try:
+            self._pending_router_echoes[echo_token] = (payload, None)
-            return self._publish(
+        receipt = self._publish(
             self.config.topics.router_mode_set,
             mode,
             qos=ROUTER_COMMAND_QOS,
@@ -482,20 +500,39 @@ class MqttAdapter:
             require_connection=True,
             invalidate_transport_on_uncertain=True,
         )
-        finally:
+        grace_deadline = (
+            self._monotonic() + self.config.mqtt.router_echo_grace_seconds
+        )
         with self._router_echo_lock:
-                self._pending_router_echoes.pop(echo_token, None)
+            if echo_token in self._pending_router_echoes:
+                self._pending_router_echoes[echo_token] = (payload, grace_deadline)
+        return receipt
 
     def _claim_router_echo(self, payload: bytes) -> bool:
-        """Claim one exact mode-set echo from an in-flight local publish."""
+        """Claim one exact mode-set echo from an in-flight or recent local publish."""
 
         with self._router_echo_lock:
-            for token, expected in tuple(self._pending_router_echoes.items()):
+            self._prune_router_echoes_locked()
+            for token, (expected, _deadline) in tuple(
+                self._pending_router_echoes.items()
+            ):
                 if expected == payload:
                     del self._pending_router_echoes[token]
                     return True
         return False
 
+    def _prune_router_echoes_locked(self) -> None:
+        """Drop expired grace-window entries. Caller holds ``_router_echo_lock``."""
+
+        now = self._monotonic()
+        expired = [
+            token
+            for token, (_, deadline) in self._pending_router_echoes.items()
+            if deadline is not None and deadline <= now
+        ]
+        for token in expired:
+            del self._pending_router_echoes[token]
+
     def publish_legacy(self, topic: str, payload: str, *, retain: bool = True) -> PublishReceipt:
         return self._publish(
             topic,

controller/service.py

@@ -81,6 +81,7 @@ class ControllerService:
         self.mqtt = mqtt_adapter or MqttAdapter(
             config,
             events=self.events,
+            monotonic=monotonic,
         )
         self.downloads = download_manager
         self.playback = playback
@@ -164,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:
@@ -251,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))
 
@@ -402,15 +428,46 @@ 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
-            if not self._cleanup_allowed():
+            url = payload["url"]
+            if not self._cleanup_allowed() and self._ingest_requires_evidence(url):
                 raise ControllerUnavailableError(
-                    "downloads are deferred until fresh Player evidence or "
+                    "refreshing or replacing existing playlist media is deferred "
-                    "administrative playback reconciliation"
+                    "until fresh Player evidence or administrative playback "
+                    "reconciliation"
                 )
             reservation = self.downloads.enqueue(
-                payload["url"], payload.get("submitter"), payload.get("note")
+                url, payload.get("submitter"), payload.get("note")
             )
             result: dict[str, Any] = {
                 "status": "queued" if reservation.status == "reserved" else reservation.status,
@@ -1261,6 +1318,20 @@ class ControllerService:
     def _cleanup_allowed(self) -> bool:
         return self._player_evidence_ready or self._operator_cleanup_override
 
+    def _ingest_requires_evidence(self, url: Any) -> bool:
+        """True when ingesting ``url`` could reuse/refresh occupied media.
+
+        A URL that does not already occupy a slot can only land in a slot
+        that ``reserve_download`` independently proves is empty and
+        unguarded, so it cannot mutate, delete, or replace media the Player
+        may currently be using. Only the reuse/refresh path needs the same
+        conservative evidence gate as deletion and expiry.
+        """
+
+        if not isinstance(url, str):
+            return False
+        return self.store.find_occupied_slot_for_url(url) is not None
+
     def _schedule_deletion_recovery(self) -> None:
         if not self._cleanup_allowed():
             return

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:
@@ -962,6 +1031,31 @@ class ControllerStore:
             (slot,),
         ).fetchone() is not None
 
+    def find_occupied_slot_for_url(self, url: str) -> int | None:
+        """Return the slot already holding ``url``, if any.
+
+        A non-``None`` result means ingest for this exact URL would reuse or
+        refresh a slot that already holds media, rather than land in a slot
+        that is genuinely empty. Callers use this to decide whether an
+        ingest request needs fresh Player evidence before proceeding.
+        """
+
+        if not isinstance(url, str):
+            return None
+        url = url.strip()
+        if not url:
+            return None
+        with self._lock:
+            row = self.connection.execute(
+                """
+                SELECT slot_number FROM slots
+                WHERE url = ? AND media_state <> 'empty'
+                ORDER BY slot_number LIMIT 1
+                """,
+                (url,),
+            ).fetchone()
+            return None if row is None else int(row["slot_number"])
+
     def mark_download_running(self, job_id: str, slot: int, generation: int) -> bool:
         with self._transaction() as connection:
             now = self._now()
@@ -1111,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)
 
@@ -53,7 +76,13 @@ class FlaskFactoryTests(unittest.TestCase):
     def _app(self, service):
         return self._factory()(self.config, service, testing=True)
 
-    def _post(self, service, *, url="https://example.com/playlist", **fields):
+    def _post(
+        self,
+        service,
+        *,
+        url="https://www.youtube.com/playlist?list=PL1234567890",
+        **fields,
+    ):
         app = self._app(service)
         return app.test_client().post(
             "/submit",
@@ -65,21 +94,118 @@ 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_submit_accepts_only_http_or_https_urls_with_a_network_location(self):
+    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 = (
             "",
             "http",
             "https:///missing-host",
-            "httpfoo://example.com/video",
+            "httpfoo://youtube.com/video",
-            "ftp://example.com/video",
+            "ftp://youtube.com/video",
             "javascript:alert(1)",
             "/relative/path",
+            # Arbitrary external hosts, including lookalikes and substring
+            # tricks, must be rejected by exact normalized hostname match.
+            "https://example.com/video",
+            "https://notyoutube.com/video",
+            "https://youtube.com.evil.com/video",
+            "https://evil.com/?u=youtube.com",
+            "https://xyoutube.com/watch?v=abc",
+            "https://youtube.com.attacker.net/watch?v=abc",
+            # Embedded userinfo must be rejected even when the host matches.
+            "https://user:[email protected]/watch?v=abc",
+            "https://[email protected]/watch?v=abc",
+            "https://[email protected]/watch?v=abc",
+            # Oversized URL.
+            "https://www.youtube.com/watch?v=" + ("a" * 3000),
         )
         for url in invalid_urls:
             with self.subTest(url=url):
@@ -89,7 +215,18 @@ class FlaskFactoryTests(unittest.TestCase):
                 self.assertIn(b"Invalid URL", response.data)
                 self.assertEqual(service.enqueue_calls, [])
 
-        for url in ("http://example.com/video", "https://example.com/video"):
+    def test_submit_accepts_only_explicit_official_youtube_hosts(self):
+        valid_urls = (
+            "https://youtube.com/watch?v=abc",
+            "https://www.youtube.com/watch?v=abc",
+            "http://www.youtube.com/watch?v=abc",
+            "https://m.youtube.com/watch?v=abc",
+            "https://music.youtube.com/watch?v=abc",
+            "https://youtu.be/abc",
+            "https://YouTube.com/watch?v=abc",
+            "https://WWW.YOUTUBE.COM/watch?v=abc",
+        )
+        for url in valid_urls:
             with self.subTest(url=url):
                 service = FakeControllerService()
                 response = self._post(service, url=url)
@@ -103,7 +240,7 @@ class FlaskFactoryTests(unittest.TestCase):
 
         response = self._post(
             service,
-            url="https://example.com/playlist",
+            url="https://www.youtube.com/playlist?list=PL1234567890",
             submitter="  Alice  ",
             note="  Before service  ",
         )
@@ -113,8 +250,49 @@ class FlaskFactoryTests(unittest.TestCase):
         self.assertIn(b"start downloading shortly", response.data)
         self.assertEqual(
             service.enqueue_calls,
-            [("https://example.com/playlist", "Alice", "Before service")],
+            [
+                (
+                    "https://www.youtube.com/playlist?list=PL1234567890",
+                    "Alice",
+                    "Before service",
                 )
+            ],
+        )
+
+    def test_submit_rejects_oversized_submitter_and_note(self):
+        module = importlib.import_module("app")
+        service = FakeControllerService()
+        response = self._post(
+            service,
+            submitter="x" * (module.MAX_SUBMITTER_LENGTH + 1),
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(b"must be", response.data)
+        self.assertEqual(service.enqueue_calls, [])
+
+        service = FakeControllerService()
+        response = self._post(
+            service,
+            note="x" * (module.MAX_NOTE_LENGTH + 1),
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(b"must be", response.data)
+        self.assertEqual(service.enqueue_calls, [])
+
+    def test_submit_rejects_oversized_request_body(self):
+        service = FakeControllerService()
+        app = self._app(service)
+        response = app.test_client().post(
+            "/submit",
+            data={
+                "url": "https://www.youtube.com/watch?v=abc",
+                "note": "x" * 64_000,
+            },
+            follow_redirects=True,
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(b"too large", response.data.lower())
+        self.assertEqual(service.enqueue_calls, [])
 
     def test_reused_submission_preserves_existing_slot_message(self):
         service = FakeControllerService(

tests/test_config_contracts.py

@@ -81,6 +81,11 @@ class ConfigurationTests(unittest.TestCase):
         self.assertIsNone(config.router.fallback_restore_mode)
         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 = {
@@ -152,6 +157,15 @@ class ConfigurationTests(unittest.TestCase):
                 with self.assertRaisesRegex(ConfigError, "must be finite"):
                     load_config(path, environ={})
 
+    def test_negative_router_echo_grace_is_rejected(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "config.yaml"
+            path.write_text(
+                "mqtt:\n  router_echo_grace_seconds: -1\n", encoding="utf-8"
+            )
+            with self.assertRaisesRegex(ConfigError, "router_echo_grace_seconds"):
+                load_config(path, environ={})
+
     def test_runtime_paths_must_be_absolute(self):
         cases = (
             ("storage:\n  media_base: relative/media\n", "storage.media_base"),
@@ -165,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_mqtt_adapter.py

@@ -10,6 +10,14 @@ from controller.config import ControllerConfig, MqttConfig
 from controller.mqtt_adapter import MqttAdapter
 
 
+class MutableClock:
+    def __init__(self, value=0.0):
+        self.value = value
+
+    def __call__(self):
+        return self.value
+
+
 class FakeMessageInfo:
     def __init__(self, mid, published):
         self.rc = mqtt.MQTT_ERR_SUCCESS
@@ -245,7 +253,7 @@ class MqttAdapterTests(unittest.TestCase):
             ("audio/player/mode/set", "video", 1, False),
         )
 
-    def test_router_self_echo_is_marked_only_while_publish_is_in_flight(self):
+    def test_router_self_echo_is_marked_while_publish_is_in_flight(self):
         self.adapter.start()
         self.adapter._on_connect(self.client, None, None, 0)
         self.events.get_nowait()
@@ -268,9 +276,28 @@ class MqttAdapterTests(unittest.TestCase):
         self.assertTrue(self.adapter.publish_router_mode("video").accepted)
         self.assertTrue(self.events.get_nowait().controller_echo)
 
-        self.client.publish_hook = None
+    def test_router_self_echo_delivered_after_puback_is_still_claimed_within_grace(
-        self.adapter._on_message(
+        self,
-            self.client,
+    ):
+        # Regression: a broker may legally deliver our own subscription echo
+        # after the PUBACK that already acknowledged the same publish. If the
+        # echo marker were cleared the instant publish() observes that
+        # PUBACK, this late echo would be mistaken for live operator intent
+        # and cause automatic Router restoration ownership to be relinquished.
+        clock = MutableClock(1000.0)
+        client = FakeClient()
+        events = queue.Queue()
+        adapter = MqttAdapter(self.config, client=client, events=events, monotonic=clock)
+        try:
+            adapter.start()
+            adapter._on_connect(client, None, None, 0)
+            events.get_nowait()
+
+            self.assertTrue(adapter.publish_router_mode("video").accepted)
+            clock.value += self.config.mqtt.router_echo_grace_seconds - 0.5
+
+            adapter._on_message(
+                client,
                 None,
                 SimpleNamespace(
                     topic="audio/player/mode/set",
@@ -281,7 +308,41 @@ class MqttAdapterTests(unittest.TestCase):
                     mid=32,
                 ),
             )
-        self.assertFalse(self.events.get_nowait().controller_echo)
+            self.assertTrue(events.get_nowait().controller_echo)
+        finally:
+            adapter.stop()
+
+    def test_router_self_echo_marker_expires_after_grace_window(self):
+        # A genuine external/operator command that happens to repeat the same
+        # value must still be recognized as operator intent once the bounded
+        # grace window has elapsed; the marker must not linger forever.
+        clock = MutableClock(1000.0)
+        client = FakeClient()
+        events = queue.Queue()
+        adapter = MqttAdapter(self.config, client=client, events=events, monotonic=clock)
+        try:
+            adapter.start()
+            adapter._on_connect(client, None, None, 0)
+            events.get_nowait()
+
+            self.assertTrue(adapter.publish_router_mode("video").accepted)
+            clock.value += self.config.mqtt.router_echo_grace_seconds + 0.5
+
+            adapter._on_message(
+                client,
+                None,
+                SimpleNamespace(
+                    topic="audio/player/mode/set",
+                    payload=b"video",
+                    qos=1,
+                    retain=False,
+                    dup=False,
+                    mid=32,
+                ),
+            )
+            self.assertFalse(events.get_nowait().controller_echo)
+        finally:
+            adapter.stop()
 
     def test_retained_router_set_never_consumes_controller_echo_marker(self):
         self.adapter.start()

tests/test_router_echo_mosquitto_integration.py

@@ -0,0 +1,206 @@
+"""Real-broker smoke test for Audio Router self-echo correlation.
+
+The fake MQTT client used by ``tests/test_mqtt_adapter.py`` delivers a
+publish's own echo synchronously, inside the ``publish()`` call itself, so it
+cannot exercise the actual ordering risk: a real broker may legally deliver a
+Controller-originated Router mode-set message back through the Controller's
+own subscription either before or after the PUBACK that already acknowledged
+the same publish. This test runs the real ``MqttAdapter`` against a real
+Mosquitto broker in Docker to prove that ordering no longer matters.
+
+Skipped automatically when Docker (or a Mosquitto broker container) is not
+available, so it never blocks the primary regression suite.
+"""
+
+from __future__ import annotations
+
+import queue
+import shutil
+import socket
+import subprocess
+import tempfile
+import time
+import unittest
+import uuid
+from pathlib import Path
+
+import paho.mqtt.client as mqtt
+
+from controller.config import ControllerConfig, MqttConfig
+from controller.mqtt_adapter import MqttAdapter
+
+
+def _docker_available() -> bool:
+    return shutil.which("docker") is not None
+
+
+def _free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+@unittest.skipUnless(
+    _docker_available(), "docker is not available for a real Mosquitto broker"
+)
+class RouterSelfEchoMosquittoIntegrationTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.port = _free_port()
+        cls.container_name = f"controller-echo-test-{uuid.uuid4().hex[:10]}"
+        cls.tempdir = tempfile.TemporaryDirectory()
+        conf_path = Path(cls.tempdir.name) / "mosquitto.conf"
+        conf_path.write_text("listener 1883\nallow_anonymous true\n", encoding="utf-8")
+        try:
+            subprocess.run(
+                [
+                    "docker",
+                    "run",
+                    "-d",
+                    "--rm",
+                    "--name",
+                    cls.container_name,
+                    "-p",
+                    f"127.0.0.1:{cls.port}:1883",
+                    "-v",
+                    f"{conf_path}:/mosquitto/config/mosquitto.conf:ro",
+                    "eclipse-mosquitto",
+                ],
+                check=True,
+                capture_output=True,
+                text=True,
+                timeout=30,
+            )
+        except (OSError, subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc:
+            cls.tempdir.cleanup()
+            raise unittest.SkipTest(
+                f"could not start a real Mosquitto broker container: {exc}"
+            )
+        try:
+            cls._wait_for_broker(cls.port)
+        except AssertionError as exc:
+            subprocess.run(
+                ["docker", "rm", "-f", cls.container_name],
+                capture_output=True,
+                timeout=15,
+            )
+            cls.tempdir.cleanup()
+            raise unittest.SkipTest(str(exc))
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        subprocess.run(
+            ["docker", "rm", "-f", cls.container_name],
+            capture_output=True,
+            timeout=15,
+        )
+        cls.tempdir.cleanup()
+
+    @staticmethod
+    def _wait_for_broker(port: int, timeout: float = 15.0) -> None:
+        deadline = time.monotonic() + timeout
+        last_exc: OSError | None = None
+        while time.monotonic() < deadline:
+            try:
+                with socket.create_connection(("127.0.0.1", port), timeout=0.5):
+                    return
+            except OSError as exc:
+                last_exc = exc
+                time.sleep(0.2)
+        raise AssertionError(f"Mosquitto broker never became reachable: {last_exc}")
+
+    def _config(self) -> ControllerConfig:
+        return ControllerConfig(
+            mqtt=MqttConfig(
+                host="127.0.0.1",
+                port=self.port,
+                client_id=f"controller-echo-{uuid.uuid4().hex[:10]}",
+                reconnect_min_seconds=1,
+                reconnect_max_seconds=1,
+                command_queue_size=32,
+                publish_ack_timeout_seconds=5.0,
+                router_echo_grace_seconds=2.0,
+            )
+        )
+
+    def test_controller_recognizes_its_own_router_echo_over_a_real_broker(self) -> None:
+        config = self._config()
+        adapter = MqttAdapter(config)
+        adapter.start()
+        try:
+            self._wait_until(
+                lambda: adapter.connected,
+                "Controller MQTT adapter never connected to the real broker",
+            )
+            receipt = adapter.publish_router_mode("video")
+            self.assertTrue(receipt.accepted, receipt.reason)
+
+            event = self._next_message_event(adapter, timeout=5.0)
+            self.assertEqual(event.topic, config.topics.router_mode_set)
+            self.assertEqual(event.payload, b"video")
+            self.assertTrue(
+                event.controller_echo,
+                "Controller did not recognize its own Router mode-set echo "
+                "delivered by a real broker; the PUBACK-vs-subscription-echo "
+                "race was not closed",
+            )
+        finally:
+            adapter.stop()
+
+    def test_genuine_external_router_command_is_not_treated_as_self_echo(self) -> None:
+        config = self._config()
+        adapter = MqttAdapter(config)
+        adapter.start()
+        operator = mqtt.Client(
+            mqtt.CallbackAPIVersion.VERSION2,
+            client_id=f"operator-{uuid.uuid4().hex[:10]}",
+        )
+        try:
+            self._wait_until(
+                lambda: adapter.connected,
+                "Controller MQTT adapter never connected to the real broker",
+            )
+            operator.connect(config.mqtt.host, config.mqtt.port, keepalive=30)
+            operator.loop_start()
+            info = operator.publish(
+                config.topics.router_mode_set, "service", qos=1, retain=False
+            )
+            info.wait_for_publish(timeout=5.0)
+
+            event = self._next_message_event(adapter, timeout=5.0)
+            self.assertEqual(event.payload, b"service")
+            self.assertFalse(
+                event.controller_echo,
+                "A genuine external Router command was mistaken for a "
+                "Controller echo; automatic restoration ownership would be "
+                "incorrectly preserved instead of relinquished",
+            )
+        finally:
+            operator.loop_stop()
+            operator.disconnect()
+            adapter.stop()
+
+    @staticmethod
+    def _wait_until(predicate, message: str, timeout: float = 10.0) -> None:
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            if predicate():
+                return
+            time.sleep(0.05)
+        raise AssertionError(message)
+
+    @staticmethod
+    def _next_message_event(adapter: MqttAdapter, *, timeout: float):
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            try:
+                event = adapter.events.get(timeout=0.2)
+            except queue.Empty:
+                continue
+            if event.kind == "message":
+                return event
+        raise AssertionError("no MQTT message event was observed within the timeout")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_service.py

@@ -88,6 +88,27 @@ class FakeDownloadManager:
         self.expiry_calls = 0
         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
@@ -166,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):
@@ -492,22 +514,80 @@ class ControllerServiceTests(unittest.TestCase):
         self.assertIn("mqtt_disconnected", health["degraded_reasons"])
         self.assertIn("awaiting_player_evidence", health["degraded_reasons"])
 
-    def test_cleanup_and_downloads_wait_for_post_start_player_evidence(self):
+    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()
         time.sleep(0.2)
 
         self.assertEqual(self.downloads.expiry_calls, 0)
         self.assertFalse(self.downloads.recovered)
-        with self.assertRaisesRegex(
+
-            ControllerUnavailableError,
+        # A genuinely new playlist cannot mutate, delete, or replace any
-            "deferred until fresh Player evidence",
+        # occupied slot, so it is not blocked by the missing evidence that
-        ):
+        # still defers cleanup/expiry below.
-            self.service.enqueue_download(
+        result = self.service.enqueue_download(
-                "https://example.test/deferred",
+            "https://example.test/brand-new",
             None,
             None,
         )
+        self.assertEqual(result["status"], "queued")
+        self.assertFalse(result["reused_slot"])
 
         self.send_message(
             self.config.topics.player_state,
@@ -540,6 +620,106 @@ class ControllerServiceTests(unittest.TestCase):
             self.service.health()["degraded_reasons"],
         )
 
+    def test_ingest_matching_an_occupied_slot_url_remains_gated_without_evidence(self):
+        self.start_service()
+        self.ready_slot(1)
+
+        with self.assertRaisesRegex(
+            ControllerUnavailableError,
+            "fresh Player evidence",
+        ):
+            self.service.enqueue_download("https://example.test/1", None, None)
+
+    def test_controller_only_restart_with_idle_connected_player_allows_new_ingest(self):
+        """Regression for the too-broad evidence gate deadlock.
+
+        A Controller-only restart with the Player still connected and idle
+        never produces fresh detailed evidence, because the Player has no
+        reason to reconnect or emit unsolicited status. New ingest into an
+        empty slot must still work; refresh/reuse of already-occupied media,
+        expiry, cleanup recovery, and manual deletion must all stay gated
+        until fresh evidence or an explicit administrative override arrives.
+        """
+
+        self.start_service()
+        self.ready_slot(1)
+        self.send_message(
+            self.config.topics.player_status,
+            {"state": "idle", "message": "player online"},
+            qos=1,
+        )
+        self.wait_for(
+            lambda: "awaiting_player_evidence"
+            not in self.service.health()["degraded_reasons"],
+            "evidence did not clear before the simulated restart",
+        )
+        self.service.stop()
+
+        # Simulate a Controller-only restart against the same durable store:
+        # the Player process itself never restarted, stays connected and
+        # idle, and therefore never sends another fresh status.
+        self.events = queue.Queue()
+        restarted_store = ControllerStore(self.config.storage, now=self.wall)
+        restarted_mqtt = FakeMqttAdapter()
+        restarted_downloads = FakeDownloadManager(restarted_store)
+        restarted_service = ControllerService(
+            self.config,
+            store=restarted_store,
+            mqtt_adapter=restarted_mqtt,
+            event_queue=self.events,
+            download_manager=restarted_downloads,
+            monotonic=self.monotonic,
+            wall_clock=self.wall,
+        )
+        restarted_service.start()
+        self.addCleanup(
+            lambda: restarted_service.stop() if restarted_service.started else None
+        )
+        self.assertIn(
+            "awaiting_player_evidence",
+            restarted_service.health()["degraded_reasons"],
+        )
+
+        # A genuinely new playlist still ingests safely with no evidence.
+        result = restarted_service.enqueue_download(
+            "https://example.test/after-restart", None, None
+        )
+        self.assertEqual(result["status"], "queued")
+        self.assertFalse(result["reused_slot"])
+        self.assertNotEqual(result["slot"], 1)
+
+        # Refreshing the URL already occupying slot 1 remains gated.
+        with self.assertRaisesRegex(
+            ControllerUnavailableError,
+            "fresh Player evidence",
+        ):
+            restarted_service.enqueue_download("https://example.test/1", None, None)
+
+        # Expiry and cleanup recovery remain deferred.
+        restarted_service._next_expiry = self.monotonic()
+        time.sleep(0.2)
+        self.assertEqual(restarted_downloads.expiry_calls, 0)
+        self.assertFalse(restarted_downloads.recovered)
+
+        # Manual deletion of occupied slot 1 still requires the explicit
+        # confirm_player_unknown override; it is not silently unlocked.
+        with self.assertRaisesRegex(
+            SlotProtectedError,
+            "confirm_player_unknown=true",
+        ):
+            restarted_service.submit_control({"action": "delete_slot", "slot": 1})
+
+        # Fresh Player evidence still unlocks cleanup normally afterward.
+        self.send_message(
+            self.config.topics.player_status,
+            {"state": "idle", "message": "player online"},
+            qos=1,
+        )
+        self.wait_for(
+            lambda: restarted_downloads.recovered,
+            "fresh evidence did not unlock deferred recovery after restart",
+        )
+
     def test_administrator_can_explicitly_unlock_cleanup_without_player_evidence(self):
         self.start_service()
         review = self.service.health()["cleanup_reconciliation"]

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:
@@ -98,6 +256,20 @@ class FreshStoreTests(StoreTestCase):
             self.assertTrue(store.mark_download_running("job-one", 1, 1))
             self.assertEqual(store.get_slot(1).media_state, "downloading")
 
+    def test_find_occupied_slot_for_url_only_matches_non_empty_slots(self):
+        with self.open_store() as store:
+            url = "https://www.youtube.com/playlist?list=one"
+            self.assertIsNone(store.find_occupied_slot_for_url(url))
+            self.assertIsNone(store.find_occupied_slot_for_url("  "))
+            self.assertIsNone(store.find_occupied_slot_for_url(None))
+
+            store.reserve_download(url, None, None, job_id="job-one")
+            self.assertEqual(store.find_occupied_slot_for_url(url), 1)
+            self.assertEqual(store.find_occupied_slot_for_url(f"  {url}  "), 1)
+            self.assertIsNone(
+                store.find_occupied_slot_for_url("https://example.test/other")
+            )
+
     def test_concurrent_requests_get_only_one_remaining_slot(self):
         with self.open_store() as store:
             for index in range(1, 10):