diff --git a/CHANGELOG.md b/CHANGELOG.md
index 38c590bd..8e6e9b51 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,8 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Security
+
+- **Authorization on DVR recording playback endpoints**. `RecordingViewSet.file` and `RecordingViewSet.hls` now require an authenticated session and enforce a per-user channel-access check before serving any bytes. Admins (`user_level >= 10`) are always allowed; standard users are allowed only when the recording's source channel is visible under their channel-profile assignments and within their `user_level`, mirroring the same logic used by `stream_xc` for live channels. Unauthenticated requests now receive `403 Forbidden` instead of being served. The pre-existing `network_access_allowed(request, "STREAMS")` perimeter check is retained as a separate, prior gate so external IPs can be blocked from streaming entirely even with a valid token.
+
 ### Fixed
 
+- **Empty show/season folders left behind after deleting a recording**. Deleting a recording removes the MKV (or HLS working directory if still in progress) but previously left the parent show / season directories on disk even when they no longer contained any other recordings. After file cleanup, `RecordingViewSet.destroy` now walks up from the deleted file's parent directory and removes any now-empty directories, stopping at the `/data/recordings` library root so the root itself is never touched.
+- **Premature DVR concat on graceful Celery worker shutdown**. A `worker_shutting_down` signal handler in `apps/channels/tasks.py` now sets a module-level `_DVR_SHUTTING_DOWN` flag. The `run_recording` loop checks this flag after FFmpeg exits and, if the recording's `end_time` has not yet passed, skips concatenation and persists `status="interrupted"` with `interrupted_reason="server_shutdown"`. The HLS working directory is preserved across the restart so the recovery path on the next worker startup can resume segment numbering from where it left off rather than truncating the show.
 - **Channel start/stop notifications missing name**: the "channel started" toast never showed up and the "channel stopped" toast showed the raw UUID instead of the channel name. `channel_name` is now written into the Redis metadata hash at init time and included in every stats payload pushed over WebSocket; the frontend notification functions read `ch.channel_name` from the stats payload directly, falling back to the store lookup and then a formatted placeholder. Both notifications now display the correct channel name.
 - **Channel form reset on group creation**: creating a new channel group from within the channel create/edit form no longer wipes all filled-in form fields. The newly created group is also automatically selected in the channel group field after it is saved. (Fixes #545)
 - **EPG channel name truncation**: EPG sources that include long `<display-name>` values (e.g. event-based channels with descriptions appended to the name) would crash the channel-parse task with a `value too long for type character varying(255)` PostgreSQL error and silently discard the entire batch. The `EPGData.name`, `Stream.name`, and `Channel.name` fields have been widened to 512 characters, and names exceeding this limit are now truncated with a warning log rather than aborting the import. (Fixes #1134)
@@ -16,6 +22,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **In-progress recording playback from the DVR page**. The Watch button on a recording card is now enabled while the recording is still in progress, and the in-app floating player can play the live HLS playlist with full timeshift / scrub-back to the start of the recording.
+  - The frontend now bundles `hls.js` and routes any `.m3u8` URL through it (Chrome, Edge, Firefox, and other Chromium-based browsers). Native HLS via `<video src=>` is reserved for Safari, where it Just Works for the same playlist.
+  - hls.js requests are authenticated via `xhrSetup`, attaching the same `Authorization: Bearer <accessToken>` header the live mpegts.js player already uses, so the new per-user authorization check on the recording endpoints (see Security) is satisfied for every playlist refresh and segment fetch.
+- **Watch DVR recordings live while they are still recording**: `run_recording` has been refactored from a single TS file capture to an FFmpeg HLS segmentation pipeline. While recording is in progress, the `file_url` exposed on the recording points to a new `/api/channels/recordings/{id}/hls/index.m3u8` endpoint instead of the eventual MKV path, so any HLS-capable client (the built-in web player, VLC, infuse, channels DVR clients, etc.) can join the stream at any time and watch from the live edge. When the recording ends, the segments are concatenated into the final MKV, `file_url` is updated to point to the existing `/file/` endpoint, and the HLS working directory is cleaned up. Hitting `/file/` while a recording is still in progress now redirects to the HLS playlist rather than 404, and hitting `/hls/index.m3u8` after the recording has finalized redirects to `/file/`, so URLs cached by clients in either form continue to work across the transition.
+- New HLS playback endpoint (`RecordingViewSet.hls`) serves `.m3u8` and `.ts` files out of the recording's working directory with path traversal protection (`os.path.realpath` containment check). Playlist files are rewritten on the fly so each segment line becomes an absolute URL routed back through this endpoint, preserving authentication and path isolation. Both this and the existing `/file/` endpoint are gated by the `STREAMS` network policy and a per-user authorization check (see Security).
+- Explicit `path('recordings/<int:pk>/hls/<path:seg_path>', ...)` route registered in `apps/channels/api_urls.py` _before_ `router.urls`. DRF's `DefaultRouter` appends a mandatory trailing slash to every URL it generates, but HLS players request segments by their natural filenames (`seg_00001.ts`) without a trailing slash, so the explicit route is required for the player to ever reach the view.
+- `DISPATCHARR_WEB_HOST` environment variable for modular deployments. The Celery container needs to reach the uWSGI / web container to fetch HLS segments while recording (FFmpeg pulls from the public stream URL, the same way any other client does). `get_dvr_stream_base_url()` resolves the base URL deterministically: AIO / dev / debug containers reach uWSGI on `127.0.0.1:$DISPATCHARR_PORT` since Celery and uWSGI share the container, while modular deployments use `http://$DISPATCHARR_WEB_HOST:$DISPATCHARR_PORT` and default `DISPATCHARR_WEB_HOST` to `web` (the compose service name). Documented as a commented-out override in `docker/docker-compose.yml`.
+- HLS viewer heartbeat. Every `.ts` request refreshes a Redis key `dvr:hls_viewer:{id}` with a 20 second TTL. After concat succeeds, the recording task waits for that key to expire naturally before deleting the HLS working directory, so an actively-watching client is never cut off mid-segment when the recording ends. Cleanup happens within 20 seconds of the last client stopping, with a 4 hour safety cap as a guard against a stuck Redis state.
 - **VOD start/stop notifications**: the frontend now shows a toast notification when a VOD stream starts or stops. `vod_started` and `vod_stopped` WebSocket events are fired from the backend when a new provider connection is opened or the last active stream on a session ends. The 1-second delayed-cleanup window is used as a settle period before firing `vod_stopped`, so seek reconnects that re-establish `active_streams` within that window suppress the notification. A `vod_stats` WebSocket push is sent alongside every event to keep the Stats page connection table in sync in real time. `vod_start` and `vod_stop` system events are also written to the system event log for each transition, and are now selectable as integration trigger events (webhook/script) in the Connect page.
 - **Channel Profiles column in Users table**: users can now see all channel profiles assigned to each user directly in the Users table. Added tooltips for profile names to handle overflow, adjusted column sizing to prevent overflow between columns, and improved the table layout for better readability. (Closes #819) — Thanks [@damien-alt-sudo](https://github.com/damien-alt-sudo)
 - **Plugin warning & disclaimer components**: extracted shared plugin warning UI into a new `PluginWarnings.jsx` component and normalized warning/disclaimer usage across all plugin action modals. New reusable components: `PluginSecurityWarning` (untrusted code), `PluginSupportDisclaimer` (community support scope), `PluginDowngradeWarning` (version downgrade), `PluginInfoNote` (informational), and `PluginRestartWarning` (backend restart on import). Updated `Plugins.jsx`, `AvailablePluginCard.jsx`, and `PluginCard.jsx` to use the shared components. — Thanks [@sethwv](https://github.com/sethwv)
@@ -42,9 +56,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Eliminated per-tick DB queries from the channel stats system. Previously `get_basic_channel_info()` in `channel_status.py` issued a `Stream.objects.filter()` and an `M3UAccountProfile.objects.filter()` on every stats tick for each active channel to resolve display names. Channel name and stream name are now written into the Redis metadata hash at channel-init time (via `initialize_channel`) and read back directly during stats collection, zero DB queries per tick. `m3u_profile_name` is now resolved on the frontend from the already-loaded playlists store rather than being pushed from the backend.
 - Eliminated a redundant `Channel` DB lookup inside `ChannelService.initialize_channel()`. `views.py` already fetches the `Channel` object via `get_stream_object()` before calling `initialize_channel()`; `channel.name` is now passed as an optional `channel_name` parameter, so the service uses the caller-supplied value and only falls back to a DB query when the name is not provided (e.g. stream-preview paths). A matching `stream_name` parameter was added for the same reason; the `stream_name` DB query is skipped entirely on normal channel start since a channel name is always available.
 - Eliminated a redundant `Stream` DB lookup on stream switches. `get_stream_info_for_switch()` already fetches the `Stream` object to build the URL; it now includes `stream_name` in its return dict. `change_stream_url()` captures that value and passes it through to `_update_channel_metadata()`, which skips its own `Stream.objects.filter()` when the name is already known.
+- **Dedicated thread-pool Celery worker for DVR recordings**. `run_recording` is long-running and almost entirely I/O-bound: it loops in short ticks polling FFmpeg, the DB, and Redis for the full duration of the recording. Running it on the default prefork worker pool (`--autoscale=6,1`) meant at most 6 concurrent recordings, and only if no other background work was running. M3U refreshes, EPG parsing, channel matching, comskip, etc. all competed for the same 6 slots, so if every worker was busy when a recording's start time arrived, the task would queue in Redis and FFmpeg would not start until a worker freed up, causing the user to miss the beginning of the show.
+  - A second Celery worker is now started alongside the default one, configured with `--pool=threads --concurrency=20` and bound to a dedicated `dvr` queue. Threads fit this workload because every blocking call in `run_recording` releases the GIL (sleep, subprocess wait, file / DB / Redis I/O) and FFmpeg itself runs in a separate OS process. A `task_routes` entry in `dispatcharr/celery.py` routes `apps.channels.tasks.run_recording` to the `dvr` queue regardless of how it is dispatched (Celery Beat `PeriodicTask`, `.delay()`, etc.). The default prefork worker now subscribes only to the `celery` queue (`-Q celery -n default@%h --autoscale=6,1`), so recordings can no longer starve background tasks and background tasks can no longer delay recordings.
+
+  - Net result: up to 20 concurrent recordings, zero startup delay regardless of EPG / M3U refresh activity, and a memory cost of roughly 80 to 120 MB for the second always-on worker process. Each additional concurrent recording costs nearly nothing on top because all 20 threads share that single process. Applied to AIO (`docker/uwsgi.ini`), dev (`docker/uwsgi.dev.ini`), debug (`docker/uwsgi.debug.ini`), and the modular celery container (`docker/entrypoint.celery.sh`).
 
 ### Changed
 
+- DVR recording pipeline rewritten around FFmpeg + HLS. The previous implementation wrote a single growing `.ts` file via the proxy and remuxed it to `.mkv` at the end; the new implementation runs FFmpeg with `-f hls`, regenerates monotonic PTS to handle erratic IPTV source timestamps, shifts output timestamps so they start at 0 (fixing negative-PTS streams that prevented HLS segment boundary detection), and concatenates the resulting segments into the final MKV at the end of the recording. Server-restart resume now continues segment numbering from the previous session rather than overwriting earlier segments.
+- Stall detection added to the recording loop. Once the first segment confirms the stream is flowing, the loop watches both segment count and segment file mtime. If neither advances for the configured stall window (e.g. when an upstream proxy ghost-kills the client), FFmpeg is signaled and the recording ends cleanly rather than hanging until `end_time`. Recording duration is honored via SIGINT so FFmpeg writes `#EXT-X-ENDLIST` cleanly into the final playlist.
+- `end_time` extension is now picked up by the running recording without restarting FFmpeg. The DB poll loop simply updates the in-memory deadline.
+- Recording cancellation cleanup updated for the new layout. `RecordingViewSet.destroy()` now removes the HLS working directory via a `_safe_rmtree` helper (with the same `allowed_roots` containment check used for the existing `_safe_remove`) instead of trying to delete the obsolete `_temp_file_path`.
+- HLS working directory lifecycle in `custom_properties` is now two-phase. After concat succeeds, only `file_url` and `output_file_url` are updated so new client requests are redirected to the final MKV via `/file/`; `_hls_dir` is intentionally preserved through the viewer-wait grace period so in-flight `.ts` requests keep resolving. `_hls_dir` is cleared from `custom_properties` only after `shutil.rmtree` actually removes the directory.
 - **Column resizing in `CustomTable`**: column widths are now propagated to body cells via CSS custom properties (`--header-{id}-size`) injected on the table wrapper, rather than reading `column.getSize()` directly in each cell's React style. This decouples body-cell widths from React renders so that memoized rows (which skip re-renders for performance) still reflect resize changes instantly via CSS cascade.
 - Decoupled row expansion from row selection in `CustomTable`, expanding a channel row no longer also selects it. Added an `onRowExpansionChange` callback to `useTable` so callers can react to expansion changes independently of selection state.
 - Fixed a stale-closure bug in memoized channel row checkboxes, `selectedTableIdsRef` now ensures the checkbox `onChange` handler and `handleShiftSelect` always read the current selection set rather than the stale set captured at render time. Without this, clicking any checkbox after the first would silently deselect previously checked rows.
diff --git a/apps/channels/api_urls.py b/apps/channels/api_urls.py
index bd53ae45..b2af3167 100644
--- a/apps/channels/api_urls.py
+++ b/apps/channels/api_urls.py
@@ -49,6 +49,11 @@ urlpatterns = [
     path('series-rules/bulk-remove/', BulkRemoveSeriesRecordingsAPIView.as_view(), name='bulk_remove_series_recordings'),
     path('series-rules/<path:tvg_id>/', DeleteSeriesRuleAPIView.as_view(), name='delete_series_rule'),
     path('recordings/bulk-delete-upcoming/', BulkDeleteUpcomingRecordingsAPIView.as_view(), name='bulk_delete_upcoming_recordings'),
+    path(
+        'recordings/<int:pk>/hls/<path:seg_path>',
+        RecordingViewSet.as_view({'get': 'hls'}),
+        name='recording-hls',
+    ),
     path('dvr/comskip-config/', ComskipConfigAPIView.as_view(), name='comskip_config'),
 ]
 
diff --git a/apps/channels/api_views.py b/apps/channels/api_views.py
index 3e786ff1..d3c76f88 100644
--- a/apps/channels/api_views.py
+++ b/apps/channels/api_views.py
@@ -62,13 +62,14 @@ from rest_framework.filters import SearchFilter, OrderingFilter
 from apps.epg.models import EPGData
 from apps.vod.models import Movie, Series
 from django.db.models import Q
-from django.http import HttpResponse, StreamingHttpResponse, FileResponse, Http404
+from django.http import HttpResponse, StreamingHttpResponse, FileResponse, Http404, JsonResponse, HttpResponseRedirect
 from django.utils import timezone
 import mimetypes
 from django.conf import settings
 
 from rest_framework.pagination import PageNumberPagination
 
+from dispatcharr.utils import network_access_allowed
 
 
 logger = logging.getLogger(__name__)
@@ -2444,13 +2445,58 @@ class RecordingViewSet(viewsets.ModelViewSet):
 
     def get_permissions(self):
         # Allow unauthenticated playback of recording files (like other streaming endpoints)
-        if self.action == 'file':
+        if self.action in ('file', 'hls'):
             return [AllowAny()]
         try:
             return [perm() for perm in permission_classes_by_action[self.action]]
         except KeyError:
             return [Authenticated()]
 
+    def _user_can_play_recording(self, request, recording):
+        """Authorization gate for recording playback (file/hls actions).
+
+        Mirrors how live stream endpoints authorize non-admin users, but
+        unlike the XC-style endpoints these URLs carry no credentials of
+        their own, so we require an authenticated session/JWT:
+          * Unauthenticated requests → denied.
+          * Admins (user_level >= 10) → allowed.
+          * Authenticated non-admins → allowed only if the recording's
+            source channel is visible under their channel-profile
+            assignments and within their user_level.
+
+        The network_access_allowed(request, "STREAMS") check applied
+        before this is a network-perimeter gate (e.g. block external IPs
+        from streaming at all); it is not a substitute for per-user
+        authorization.
+        """
+        user = getattr(request, "user", None)
+        if not user or not getattr(user, "is_authenticated", False):
+            return False
+        if getattr(user, "user_level", 0) >= 10:
+            return True
+
+        channel = getattr(recording, "channel", None)
+        if channel is None:
+            # Recording with no source channel, only admins can play.
+            return False
+
+        try:
+            user_profile_count = user.channel_profiles.count()
+        except Exception:
+            user_profile_count = 0
+
+        filters = {
+            "id": channel.id,
+            "user_level__lte": user.user_level,
+        }
+        if user_profile_count > 0:
+            filters["channelprofilemembership__enabled"] = True
+            filters["channelprofilemembership__channel_profile__in"] = (
+                user.channel_profiles.all()
+            )
+            return Channel.objects.filter(**filters).distinct().exists()
+        return Channel.objects.filter(**filters).exists()
+
     @action(detail=True, methods=["post"], url_path="comskip")
     def comskip(self, request, pk=None):
         """Trigger comskip processing for this recording."""
@@ -2464,14 +2510,32 @@ class RecordingViewSet(viewsets.ModelViewSet):
 
     @action(detail=True, methods=["get"], url_path="file")
     def file(self, request, pk=None):
-        """Stream a recorded file with HTTP Range support for seeking."""
+        """Stream a completed recording file with HTTP Range support for seeking.
+
+        For in-progress recordings, file_url in custom_properties points to
+        /hls/index.m3u8.  If a client hits this endpoint while the recording
+        is still running (or the MKV is not yet produced), it is redirected to
+        the HLS playlist endpoint.
+        """
+        if not network_access_allowed(request, "STREAMS"):
+            return JsonResponse({"error": "Forbidden"}, status=403)
         recording = get_object_or_404(Recording, pk=pk)
+        if not self._user_can_play_recording(request, recording):
+            return JsonResponse({"error": "Forbidden"}, status=403)
         cp = recording.custom_properties or {}
         file_path = cp.get("file_path")
         file_name = cp.get("file_name") or "recording"
 
-        if not file_path or not os.path.exists(file_path):
-            raise Http404("Recording file not found")
+        if not file_path or not os.path.exists(file_path) or os.path.getsize(file_path) == 0:
+            # Redirect to HLS if recording is still in progress
+            hls_dir = cp.get("_hls_dir")
+            if hls_dir and os.path.isdir(hls_dir):
+                hls_url = request.build_absolute_uri(
+                    f"/api/channels/recordings/{pk}/hls/index.m3u8"
+                )
+                return HttpResponseRedirect(hls_url)
+            if not file_path or not os.path.exists(file_path):
+                raise Http404("Recording file not found")
 
         # Guess content type
         ext = os.path.splitext(file_path)[1].lower()
@@ -2532,6 +2596,74 @@ class RecordingViewSet(viewsets.ModelViewSet):
         response["Content-Disposition"] = f"inline; filename=\"{file_name}\""
         return response
 
+    @action(detail=True, methods=["get"], url_path="hls/(?P<seg_path>.+)")
+    def hls(self, request, pk=None, seg_path=None):
+        """Serve HLS playlist and segment files for an in-progress (or completed) recording.
+
+        Clients connecting during recording should use the m3u8 URL returned in
+        custom_properties.file_url.  Segment URLs inside the playlist are rewritten
+        to route through this endpoint so authentication and path isolation are
+        preserved.
+        """
+        if not network_access_allowed(request, "STREAMS"):
+            return JsonResponse({"error": "Forbidden"}, status=403)
+        recording = get_object_or_404(Recording, pk=pk)
+        if not self._user_can_play_recording(request, recording):
+            return JsonResponse({"error": "Forbidden"}, status=403)
+        cp = recording.custom_properties or {}
+        hls_dir = cp.get("_hls_dir")
+
+        if not hls_dir or not os.path.isdir(hls_dir):
+            # HLS dir is gone, recording is likely complete.  Redirect to the
+            # permanent MKV endpoint for .m3u8 requests so clients that still
+            # have the HLS URL bookmarked get a useful response.
+            cp = recording.custom_properties or {}
+            file_path = cp.get("file_path")
+            if seg_path.endswith(".m3u8") and file_path and os.path.exists(file_path) and os.path.getsize(file_path) > 0:
+                return HttpResponseRedirect(
+                    request.build_absolute_uri(f"/api/channels/recordings/{pk}/file/")
+                )
+            raise Http404("HLS content not available for this recording")
+
+        # Security: prevent path traversal outside the HLS directory
+        safe_dir = os.path.realpath(hls_dir)
+        requested = os.path.realpath(os.path.join(hls_dir, seg_path))
+        if not requested.startswith(safe_dir + os.sep) and requested != safe_dir:
+            return Response({"error": "Forbidden"}, status=403)
+
+        if not os.path.isfile(requested):
+            raise Http404(f"HLS file not found: {seg_path}")
+
+        if seg_path.endswith(".m3u8"):
+            # Rewrite relative segment lines to absolute URLs through this API
+            base_url = request.build_absolute_uri(
+                f"/api/channels/recordings/{pk}/hls/"
+            )
+            lines = []
+            with open(requested) as _f:
+                for line in _f:
+                    stripped = line.strip()
+                    if stripped and not stripped.startswith("#"):
+                        lines.append(f"{base_url}{stripped}\n")
+                    else:
+                        lines.append(line)
+            return HttpResponse("".join(lines), content_type="application/x-mpegURL")
+
+        if seg_path.endswith(".ts"):
+            # Refresh the viewer heartbeat in Redis so the Celery task knows an
+            # active client is still fetching segments.  TTL is 20 s, enough for
+            # three 4-second segments plus network margin.
+            try:
+                from core.utils import RedisClient
+                _rv = RedisClient.get_client(max_retries=1, retry_interval=0)
+                if _rv:
+                    _rv.set(f"dvr:hls_viewer:{pk}", "1", ex=20)
+            except Exception:
+                pass
+            return FileResponse(open(requested, "rb"), content_type="video/mp2t")
+
+        raise Http404("Unsupported HLS file type")
+
     @action(detail=True, methods=["post"], url_path="stop")
     def stop(self, request, pk=None):
         """Stop a recording early while retaining the partial content for playback."""
@@ -2838,7 +2970,7 @@ class RecordingViewSet(viewsets.ModelViewSet):
         cp = instance.custom_properties or {}
         rec_status = cp.get("status", "")
         file_path = cp.get("file_path")
-        temp_ts_path = cp.get("_temp_file_path")
+        hls_dir = cp.get("_hls_dir")
         channel_uuid = str(instance.channel.uuid)
 
         # 1. Delete the DB record (also fires post_delete → revoke_task_on_delete)
@@ -2871,6 +3003,41 @@ class RecordingViewSet(viewsets.ModelViewSet):
             except Exception as ex:
                 logger.warning(f"Failed to delete recording artifact {path}: {ex}")
 
+        def _safe_rmtree(path: str):
+            if not path or not isinstance(path, str):
+                return
+            try:
+                import shutil as _shutil
+                if any(path.startswith(root) for root in allowed_roots) and os.path.isdir(path):
+                    _shutil.rmtree(path)
+                    logger.info(f"Deleted recording HLS directory: {path}")
+            except Exception as ex:
+                logger.warning(f"Failed to delete HLS directory {path}: {ex}")
+
+        # Clean up empty parent directories up to the recordings root to prevent orphaned folders from accumulating over time.
+        recordings_root = os.path.normpath('/data/recordings')
+
+        def _prune_empty_parents(path: str):
+            if not path or not isinstance(path, str):
+                return
+            try:
+                parent = os.path.dirname(os.path.normpath(path))
+                while (
+                    parent
+                    and parent != recordings_root
+                    and parent.startswith(recordings_root + os.sep)
+                    and os.path.isdir(parent)
+                    and not os.listdir(parent)
+                ):
+                    try:
+                        os.rmdir(parent)
+                        logger.info(f"Removed empty recording directory: {parent}")
+                    except OSError:
+                        break
+                    parent = os.path.dirname(parent)
+            except Exception as ex:
+                logger.debug(f"Unable to prune empty parents for {path}: {ex}")
+
         def _background_cancel():
             # Only stop the DVR client if the recording was actively streaming.
             # Stopping for completed/upcoming recordings would kill an unrelated
@@ -2888,7 +3055,13 @@ class RecordingViewSet(viewsets.ModelViewSet):
             # Best-effort file cleanup in case run_recording already exited
             # before the DB delete.
             _safe_remove(file_path)
-            _safe_remove(temp_ts_path)
+            _safe_rmtree(hls_dir)
+
+            # If removing the file/HLS dir leaves the show/season folder
+            # empty, clean those up too.  Both paths share the same parent
+            # in normal layouts, but run the prune for each just in case.
+            _prune_empty_parents(file_path)
+            _prune_empty_parents(hls_dir)
 
             try:
                 from django.db import connection as _conn
diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py
index 59122d9f..ee38c1f3 100755
--- a/apps/channels/tasks.py
+++ b/apps/channels/tasks.py
@@ -6,13 +6,17 @@ import re
 import requests
 import time
 import json
+import shutil
 import subprocess
 import signal
+import threading
+from collections import deque
 from zoneinfo import ZoneInfo
 from datetime import datetime, timedelta
 import gc
 
 from celery import shared_task
+from celery.signals import worker_shutting_down
 from django.utils.text import slugify
 from rapidfuzz import fuzz
 
@@ -30,6 +34,21 @@ from urllib.parse import quote
 logger = logging.getLogger(__name__)
 
 
+# Flag set by the Celery `worker_shutting_down` signal so in-flight
+# `run_recording` tasks can distinguish a graceful worker shutdown (e.g.
+# `docker stop`, container update) from the natural end of a recording.
+# When set, the post-FFmpeg path leaves the HLS working directory intact
+# and marks the recording "interrupted" so `recover_recordings_on_startup`
+# can resume it on the next boot via the existing path-reuse logic.
+_DVR_SHUTTING_DOWN = False
+
+
+@worker_shutting_down.connect
+def _dvr_mark_shutting_down(**_kwargs):
+    global _DVR_SHUTTING_DOWN
+    _DVR_SHUTTING_DOWN = True
+
+
 _url_validation_cache = {}
 _URL_CACHE_TTL = 300  # seconds
 
@@ -1584,9 +1603,14 @@ def _parse_epg_tv_movie_info(program):
     return is_movie, season, episode, year, sub_title
 
 
-def _build_output_paths(channel, program, start_time, end_time):
+def _build_output_paths(channel, program, start_time, end_time, recording_id):
     """
-    Build (final_path, temp_ts_path, final_filename) using DVR templates.
+    Build (final_path, hls_dir, final_filename) using DVR templates.
+
+    The HLS working directory is hidden and tagged with the recording id
+    (e.g. ``.dvr_71_hls``) so it cannot collide with another recording's
+    files and so orphan-cleanup utilities can identify internal scratch
+    directories unambiguously.
     """
     from core.models import CoreSettings
     # Root for DVR recordings: fixed to /data/recordings inside the container
@@ -1653,23 +1677,17 @@ def _build_output_paths(channel, program, start_time, end_time):
     final_path = rel_path if rel_path.startswith('/') else os.path.join(library_root, rel_path)
     final_path = os.path.normpath(final_path)
 
-    # Avoid overwriting an existing file from a different recording.
-    # Check BOTH .mkv and .ts — a pre-restart TS segment may exist at
-    # the same base name even when the MKV is a 0-byte placeholder.
+    # Avoid overwriting an existing MKV from a different recording.  The HLS
+    # working directory is keyed by recording id, so it cannot collide and is
+    # not part of this check.
     base, ext = os.path.splitext(final_path)
     counter = 1
     while True:
-        candidate_base = final_path[:-len(ext)]  # strip extension
-        ts_candidate = candidate_base + '.ts'
         try:
             mkv_occupied = os.stat(final_path).st_size > 0
         except OSError:
             mkv_occupied = False
-        try:
-            ts_occupied = os.stat(ts_candidate).st_size > 0
-        except OSError:
-            ts_occupied = False
-        if not mkv_occupied and not ts_occupied:
+        if not mkv_occupied:
             break
         counter += 1
         final_path = f"{base}_{counter}{ext}"
@@ -1677,37 +1695,35 @@ def _build_output_paths(channel, program, start_time, end_time):
     # Ensure directory exists
     os.makedirs(os.path.dirname(final_path), exist_ok=True)
 
-    # Derive temp TS path in same directory
-    base_no_ext = os.path.splitext(os.path.basename(final_path))[0]
-    temp_ts_path = os.path.join(os.path.dirname(final_path), f"{base_no_ext}.ts")
-    return final_path, temp_ts_path, os.path.basename(final_path)
+    # HLS working directory: hidden and id-tagged alongside the final MKV.
+    hls_dir = os.path.join(os.path.dirname(final_path), f".dvr_{recording_id}_hls")
+    return final_path, hls_dir, os.path.basename(final_path)
 
 
-def build_dvr_candidates():
-    """Build ordered list of candidate base URLs for DVR TS streaming.
+def get_dvr_stream_base_url():
+    """Return the single correct base URL for DVR to reach the TS stream proxy.
 
-    Reads environment variables to determine which URLs to try:
-    - DISPATCHARR_INTERNAL_TS_BASE_URL: explicit override (first priority)
-    - DISPATCHARR_PORT: the external port (default 9191)
-    - DISPATCHARR_ENV/DISPATCHARR_DEBUG/REDIS_HOST: dev-mode detection
-    - DISPATCHARR_INTERNAL_API_BASE: override for the docker service URL
+    Priority:
+    1. DISPATCHARR_INTERNAL_TS_BASE_URL — explicit override, always wins.
+    2. Modular mode (DISPATCHARR_ENV=modular) — celery runs in a separate container
+       and must reach the web container by its Docker service name on DISPATCHARR_PORT.
+       Override the host with DISPATCHARR_WEB_HOST for non-standard compose setups.
+    3. AIO / dev / debug — celery shares the container with uwsgi which binds on
+       port 5656; use 127.0.0.1 to avoid any nginx layer.
     """
     explicit = os.environ.get('DISPATCHARR_INTERNAL_TS_BASE_URL')
-    dispatcharr_port = os.environ.get('DISPATCHARR_PORT', '9191')
-    is_dev = (os.environ.get('DISPATCHARR_ENV', '').lower() == 'dev') or \
-             (os.environ.get('DISPATCHARR_DEBUG', '').lower() == 'true') or \
-             (os.environ.get('REDIS_HOST', 'redis') in ('localhost', '127.0.0.1'))
-    candidates = []
     if explicit:
-        candidates.append(explicit)
-    if is_dev:
-        # Debug container typically exposes API on 5656 (uwsgi internal port)
-        candidates.extend(['http://127.0.0.1:5656', f'http://127.0.0.1:{dispatcharr_port}'])
-    # Docker service name fallback — use DISPATCHARR_PORT so modular mode works with custom ports
-    candidates.append(os.environ.get('DISPATCHARR_INTERNAL_API_BASE', f'http://web:{dispatcharr_port}'))
-    # Last-resort localhost ports
-    candidates.extend(['http://localhost:5656', f'http://localhost:{dispatcharr_port}'])
-    return candidates
+        return explicit.rstrip('/')
+
+    dispatcharr_env = os.environ.get('DISPATCHARR_ENV', 'aio').lower()
+
+    if dispatcharr_env == 'modular':
+        host = os.environ.get('DISPATCHARR_WEB_HOST', 'web')
+        port = os.environ.get('DISPATCHARR_PORT', '9191')
+        return f'http://{host}:{port}'
+
+    # AIO, dev, debug: celery and uwsgi share the container, reach uwsgi directly
+    return 'http://127.0.0.1:5656'
 
 
 @shared_task
@@ -1759,11 +1775,10 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str):
     start_time = datetime.fromisoformat(start_time_str)
     end_time = datetime.fromisoformat(end_time_str)
 
-    duration_seconds = int((end_time - start_time).total_seconds())
     # Build output paths from templates (refined after loading Recording cp below)
     filename = None
     final_path = None
-    temp_ts_path = None
+    hls_dir = None
 
     channel_layer = get_channel_layer()
 
@@ -1810,7 +1825,7 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str):
             "started_at": str(datetime.now()),
         })
         # Provide a predictable playback URL for the frontend
-        cp["file_url"] = f"/api/channels/recordings/{recording_id}/file/"
+        cp["file_url"] = f"/api/channels/recordings/{recording_id}/hls/index.m3u8"
         cp["output_file_url"] = cp["file_url"]
 
         # Determine program info (may include id for deeper details)
@@ -1825,10 +1840,40 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str):
                 program.update(epg_match)
                 cp["program"] = program
 
-        final_path, temp_ts_path, filename = _build_output_paths(channel, program, start_time, end_time)
+        # Resume into the existing working set if this task is a recovery run.
+        # `recover_recordings_on_startup` re-dispatches `run_recording` for an
+        # interrupted recording with `_hls_dir` and `file_path` already set in
+        # custom_properties.  Re-running `_build_output_paths` would allocate a
+        # fresh `Foo_2.mkv` / `.dvr_{id}_hls` pair and orphan the original
+        # segments, producing two MKVs at the end.  Reuse the original paths
+        # whenever the HLS directory is still on disk.
+        existing_hls = cp.get("_hls_dir")
+        existing_file = cp.get("file_path")
+        if (
+            existing_hls
+            and existing_file
+            and os.path.isdir(existing_hls)
+        ):
+            final_path = existing_file
+            hls_dir = existing_hls
+            filename = os.path.basename(final_path)
+            try:
+                _seg_count = sum(
+                    1 for f in os.listdir(hls_dir) if f.endswith(".ts")
+                )
+            except OSError:
+                _seg_count = 0
+            logger.info(
+                f"run_recording {recording_id}: resuming into existing HLS dir "
+                f"{hls_dir} ({_seg_count} segment(s)), final={final_path}"
+            )
+        else:
+            final_path, hls_dir, filename = _build_output_paths(
+                channel, program, start_time, end_time, recording_id
+            )
         cp["file_name"] = filename
         cp["file_path"] = final_path
-        cp["_temp_file_path"] = temp_ts_path
+        cp["_hls_dir"] = hls_dir
 
         # Resolve poster art via the shared pipeline (EPG → VOD → TMDB/OMDb →
         # TVMaze/iTunes → direct program fields → Logo table → channel logo).
@@ -1840,11 +1885,16 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str):
         if poster_url and "poster_url" not in cp:
             cp["poster_url"] = poster_url
 
-        # Ensure destination exists so it's visible immediately
+        # Ensure destination directory exists.  No placeholder MKV is created
+        # here on purpose: with the HLS pipeline, the final MKV is materialized
+        # by the post-recording concat step, and `/file/` redirects to `/hls/`
+        # while no MKV exists, so a 0-byte placeholder serves no functional
+        # purpose.  Writing one previously caused an orphan-file bug on server
+        # restart in cases where the resume path could not adopt the original
+        # final_path, since the second run would generate a fresh placeholder
+        # at a new path and leave the original behind.
         try:
             os.makedirs(os.path.dirname(final_path), exist_ok=True)
-            if not os.path.exists(final_path):
-                open(final_path, 'ab').close()
         except Exception:
             pass
 
@@ -1863,7 +1913,7 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str):
 
         # Merge only the keys explicitly set into the fresh copy
         for key in ("status", "started_at", "file_url", "output_file_url",
-                     "file_name", "file_path", "_temp_file_path",
+                     "file_name", "file_path", "_hls_dir",
                      "program", "poster_logo_id", "poster_url"):
             if key in cp:
                 fresh_cp[key] = cp[key]
@@ -1886,258 +1936,374 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str):
     interrupted_reason = None
     bytes_written = 0
 
-    from requests.exceptions import ReadTimeout, ConnectionError as ReqConnectionError, ChunkedEncodingError
+    base_url = get_dvr_stream_base_url()
+    logger.info(f"DVR recording {recording_id}: using stream base URL {base_url}")
 
-    candidates = build_dvr_candidates()
-
-    chosen_base = None
-    last_error = None
-    bytes_written = 0
-    interrupted = False
-    interrupted_reason = None
-
-    def _check_recording_cancelled(rid):
-        """Check if a recording was stopped by user or deleted.
-
-        Returns (should_exit, is_interrupted, reason) where should_exit
-        indicates the stream loop must terminate.
-        """
-        try:
-            rec = Recording.objects.filter(id=rid).only("custom_properties").first()
-            if rec is None:
-                return True, True, "recording_deleted"
-            if (rec.custom_properties or {}).get("status") == "stopped":
-                return True, False, "stopped_by_user"
-        except Exception:
-            pass
-        return False, False, None
-
-    # --- Retry / reconnection constants ---
-    # Stream reconnection: retry the same TS proxy base on transient
-    # connectivity loss.  Counter resets when data resumes.
-    _dvr_max_reconnects = 5
-    _dvr_reconnect_delay = 2.0  # seconds
-    # DB save retry: exponential backoff (1s, 2s, 4s) for transient errors.
+    # --- DB retry constants ---
     _dvr_db_max_retries = 3
     _dvr_db_retry_interval = 1  # seconds (base for exponential backoff)
-    # FFmpeg remux retry: covers transient I/O errors.
-    _dvr_remux_max_retries = 2
-    _dvr_remux_retry_interval = 2  # seconds (base for exponential backoff)
 
-    for base in candidates:
-        test_url = f"{base.rstrip('/')}/proxy/ts/stream/{channel.uuid}"
-        logger.info(f"DVR recording {recording_id}: trying TS base {base}")
+    # Redis key used to signal that an HLS client is actively fetching segments.
+    # The hls endpoint refreshes this on every .ts request; we check it before
+    # deleting the HLS directory so in-flight downloads are not cut short.
+    _hls_viewer_key = f"dvr:hls_viewer:{recording_id}"
 
-        _reconnects = 0
-        _file_mode = 'wb'
-        _stream_started_at = None
-        _done = False
+    # --- HLS recording pipeline ---
+    # Launch FFmpeg with the deterministic base URL resolved above.  Wait up to
+    # _first_segment_timeout seconds for the first segment to appear before
+    # treating the stream as unavailable.
+    from django.utils import timezone as _tz
 
-        while True:  # Reconnection loop for this base
+    last_error = None
+    ffmpeg_proc = None
+    hls_m3u8 = None
+    hls_seg_pattern = None
+    _stream_confirmed = False
+
+    if hls_dir:
+        os.makedirs(hls_dir, exist_ok=True)
+        hls_m3u8 = os.path.join(hls_dir, "index.m3u8")
+        hls_seg_pattern = os.path.join(hls_dir, "seg_%05d.ts")
+
+    base = base_url
+
+    if not interrupted:
+        # Check for stop/delete before starting
+        try:
+            _pre = Recording.objects.filter(id=recording_id).only("custom_properties").first()
+            if _pre is None:
+                interrupted = True
+                interrupted_reason = "recording_deleted"
+            elif (_pre.custom_properties or {}).get("status") == "stopped":
+                interrupted = False
+                interrupted_reason = "stopped_by_user"
+        except Exception:
+            pass
+
+    if not interrupted:
+        stream_url = f"{base}/proxy/ts/stream/{channel.uuid}"
+        logger.info(f"DVR recording {recording_id}: stream URL: {stream_url}")
+
+    if not interrupted:
+
+        # Continue segment numbering from any previous session (server-restart resume)
+        existing_segs = sorted(
+            f for f in os.listdir(hls_dir) if f.startswith("seg_") and f.endswith(".ts")
+        ) if hls_dir else []
+        hls_start_number = len(existing_segs)
+
+        ffmpeg_cmd = [
+            "ffmpeg", "-y",
+            "-reconnect", "1",
+            "-reconnect_streamed", "1",
+            "-reconnect_delay_max", "5",
+            "-user_agent", f"Dispatcharr-DVR/recording-{recording_id}",
+            # Regenerate monotonic PTS to handle erratic/discontinuous timestamps
+            # from IPTV sources.
+            "-fflags", "+genpts",
+            "-i", stream_url,
+            "-c", "copy",
+            # Shift output timestamps so they start from 0, fixing negative PTS
+            # values that can prevent segment boundary detection in the HLS muxer.
+            "-avoid_negative_ts", "make_zero",
+            "-f", "hls",
+            "-hls_time", "4",
+            "-hls_list_size", "0",
+            "-hls_flags", "append_list+omit_endlist+independent_segments",
+            "-start_number", str(hls_start_number),
+            "-hls_segment_filename", hls_seg_pattern,
+            hls_m3u8,
+        ]
+
+        logger.info(f"DVR recording {recording_id}: starting FFmpeg — stream URL: {stream_url}")
+        logger.info(f"DVR recording {recording_id}: HLS output dir: {hls_dir}")
+        logger.debug(f"DVR recording {recording_id}: FFmpeg command: {' '.join(str(a) for a in ffmpeg_cmd)}")
+        # Rolling tail of FFmpeg stderr lines for post-mortem diagnostics
+        _ffmpeg_stderr_tail = deque(maxlen=200)
+        try:
+            ffmpeg_proc = subprocess.Popen(
+                ffmpeg_cmd,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.PIPE,
+            )
+        except Exception as _fe:
+            last_error = str(_fe)
+            logger.warning(f"DVR recording {recording_id}: failed to launch FFmpeg: {_fe}")
+            ffmpeg_proc = None
+
+        # Drain FFmpeg stderr in a background thread to prevent the OS pipe
+        # buffer (~64 KB on Linux) from filling up, which would block FFmpeg's
+        # writes and cause it to silently stop demuxing/segmenting after a few
+        # minutes of normal progress output. Lines are logged to the Dispatcharr
+        # log and tail-buffered for diagnostics.
+        def _drain_ffmpeg_stderr(proc, rec_id, tail):
+            # FFmpeg emits progress lines terminated with \r (carriage return)
+            # so it can rewrite them in place; only non-progress messages end
+            # with \n. Split on either to capture both kinds.
             try:
-                with requests.get(
-                    test_url,
-                    headers={
-                        'User-Agent': f'Dispatcharr-DVR/recording-{recording_id}',
-                    },
-                    stream=True,
-                    timeout=(10, 15),
-                ) as response:
-                    response.raise_for_status()
-
-                    _test_window = 3.0
-                    _window_start = time.time()
-                    _stop_poll_interval = 2.0
-                    _last_stop_poll = time.time()
-
-                    with open(temp_ts_path, _file_mode) as file:
-                        if _stream_started_at is None:
-                            _stream_started_at = time.time()
-
-                        for chunk in response.iter_content(chunk_size=8192):
-                            if not chunk:
-                                if not chosen_base and (time.time() - _window_start) > _test_window:
-                                    break
-                                continue
-
-                            if not chosen_base:
-                                chosen_base = base
-
-                            # Data received after reconnect — connection restored
-                            if _reconnects > 0:
-                                logger.info(
-                                    f"DVR recording {recording_id}: "
-                                    f"stream resumed after reconnect"
-                                )
-                                _reconnects = 0
-
-                            file.write(chunk)
-                            bytes_written += len(chunk)
-
-                            elapsed = time.time() - _stream_started_at
-                            if elapsed > duration_seconds:
-                                break
-
-                            # Periodic DB poll: stop, delete, end_time extension
-                            _now = time.time()
-                            if _now - _last_stop_poll >= _stop_poll_interval:
-                                _last_stop_poll = _now
-                                try:
-                                    _sc = Recording.objects.filter(
-                                        id=recording_id
-                                    ).only("custom_properties", "end_time").first()
-                                    if _sc is None:
-                                        logger.info(
-                                            f"DVR recording {recording_id}: "
-                                            f"deleted — exiting stream loop"
-                                        )
-                                        interrupted = False
-                                        break
-                                    if (_sc.custom_properties or {}).get("status") == "stopped":
-                                        logger.info(
-                                            f"DVR recording {recording_id}: "
-                                            f"stop requested — exiting stream loop"
-                                        )
-                                        break
-                                    try:
-                                        new_end = _sc.end_time
-                                        if new_end is not None:
-                                            from django.utils import timezone as _tz
-                                            if _tz.is_naive(new_end):
-                                                new_end = _tz.make_aware(new_end)
-                                            _ref = start_time
-                                            if _tz.is_naive(_ref):
-                                                _ref = _tz.make_aware(_ref)
-                                            new_duration = int(
-                                                (new_end - _ref).total_seconds()
-                                            )
-                                            if new_duration > duration_seconds:
-                                                logger.info(
-                                                    f"DVR recording {recording_id}: "
-                                                    f"end_time extended to {new_end}, "
-                                                    f"new duration {new_duration}s"
-                                                )
-                                                duration_seconds = new_duration
-                                    except Exception:
-                                        pass
-                                except Exception:
-                                    pass
-
-                    # iter_content exhausted or loop exited normally
-                    if bytes_written > 0:
-                        logger.info(
-                            f"DVR recording {recording_id}: "
-                            f"stream complete, {bytes_written} bytes written"
-                        )
-                        _done = True
-                    else:
-                        last_error = f"no_data_from_{base}"
-                        logger.warning(
-                            f"DVR recording {recording_id}: no data from "
-                            f"{base} within {_test_window}s, trying next base"
-                        )
-                        try:
-                            if os.path.exists(temp_ts_path) and os.path.getsize(temp_ts_path) == 0:
-                                os.remove(temp_ts_path)
-                        except FileNotFoundError:
-                            pass
-                    break  # Exit reconnection loop
-
-            except (ReadTimeout, ReqConnectionError, ChunkedEncodingError) as e:
-                if bytes_written > 0:
-                    # Active stream lost — check cancellation before reconnecting
-                    should_exit, is_int, reason = _check_recording_cancelled(recording_id)
-                    if should_exit:
-                        interrupted = is_int
-                        interrupted_reason = reason
-                        if reason == "stopped_by_user":
-                            logger.info(
-                                f"DVR recording {recording_id}: "
-                                f"stopped by user — ending stream"
-                            )
-                        _done = True
+                buf = bytearray()
+                stream = proc.stderr
+                while True:
+                    byte = stream.read(1)
+                    if not byte:
                         break
-
-                    _reconnects += 1
-                    if _reconnects <= _dvr_max_reconnects:
-                        logger.warning(
-                            f"DVR recording {recording_id}: connection lost "
-                            f"({type(e).__name__}), reconnecting "
-                            f"({_reconnects}/{_dvr_max_reconnects}) "
-                            f"in {_dvr_reconnect_delay}s..."
-                        )
-                        time.sleep(_dvr_reconnect_delay)
-                        _file_mode = 'ab'
+                    if byte in (b'\r', b'\n'):
+                        if buf:
+                            line = buf.decode('utf-8', errors='replace').strip()
+                            buf.clear()
+                            if line:
+                                tail.append(line)
+                                low = line.lower()
+                                if 'error' in low or 'failed' in low or 'invalid' in low:
+                                    logger.warning(f"DVR recording {rec_id} ffmpeg: {line}")
+                                else:
+                                    logger.debug(f"DVR recording {rec_id} ffmpeg: {line}")
                         continue
+                    buf.append(byte[0])
+                    # Safety net: flush absurdly long lines so a malformed
+                    # stream can't grow the buffer without bound.
+                    if len(buf) > 4096:
+                        line = buf.decode('utf-8', errors='replace').strip()
+                        buf.clear()
+                        if line:
+                            tail.append(line)
+                            logger.debug(f"DVR recording {rec_id} ffmpeg: {line}")
+                # Flush any trailing content
+                if buf:
+                    line = buf.decode('utf-8', errors='replace').strip()
+                    if line:
+                        tail.append(line)
+                        logger.debug(f"DVR recording {rec_id} ffmpeg: {line}")
+            except Exception as _de:
+                logger.debug(f"DVR recording {rec_id}: stderr drain ended: {_de}")
 
-                    logger.error(
-                        f"DVR recording {recording_id}: max reconnects "
-                        f"({_dvr_max_reconnects}) exceeded — ending recording"
-                    )
-                    interrupted = True
-                    interrupted_reason = (
-                        f"stream_interrupted: max reconnects exceeded ({e})"
-                    )
-                    _done = True
-                    break
+        if ffmpeg_proc is not None and ffmpeg_proc.stderr is not None:
+            _stderr_thread = threading.Thread(
+                target=_drain_ffmpeg_stderr,
+                args=(ffmpeg_proc, recording_id, _ffmpeg_stderr_tail),
+                daemon=True,
+                name=f"dvr-ffmpeg-stderr-{recording_id}",
+            )
+            _stderr_thread.start()
 
-                # No data received yet — retry same base before moving on
-                should_exit, is_int, reason = _check_recording_cancelled(recording_id)
-                if should_exit:
-                    interrupted = is_int
-                    interrupted_reason = reason
-                    _done = True
-                    break
-                _reconnects += 1
-                if _reconnects <= _dvr_max_reconnects:
+        end_timestamp = end_time.timestamp()
+        _stop_poll_interval = 2.0
+        _last_stop_poll = time.time()
+        _ffmpeg_start = time.time()
+        _first_segment_timeout = 15.0
+        _stall_timeout = 60.0  # seconds without new segments → source stream gone
+        _stream_confirmed = False
+        _last_seg_count = hls_start_number
+        _last_new_seg_time = time.time()
+
+        while ffmpeg_proc.poll() is None:
+            time.sleep(0.5)
+            now = time.time()
+
+            segs_now = [
+                f for f in os.listdir(hls_dir)
+                if f.startswith("seg_") and f.endswith(".ts")
+            ] if hls_dir else []
+
+            # Wait for the first segment to confirm data is flowing
+            if not _stream_confirmed:
+                if segs_now:
+                    _stream_confirmed = True
+                    _last_seg_count = len(segs_now)
+                    _last_new_seg_time = now
+                    logger.info(
+                        f"DVR recording {recording_id}: first HLS segment written, stream confirmed"
+                    )
+                elif now - _ffmpeg_start > _first_segment_timeout:
                     logger.warning(
-                        f"DVR recording {recording_id}: initial connection "
-                        f"to {base} failed ({type(e).__name__}), retrying "
-                        f"({_reconnects}/{_dvr_max_reconnects}) "
-                        f"in {_dvr_reconnect_delay}s..."
+                        f"DVR recording {recording_id}: no HLS segments produced after "
+                        f"{_first_segment_timeout}s from {base} — stream unavailable"
                     )
-                    time.sleep(_dvr_reconnect_delay)
-                    continue
-                last_error = str(e)
-                logger.warning(
-                    f"DVR recording {recording_id}: base {base} exhausted "
-                    f"retries ({_dvr_max_reconnects}): {e}"
-                )
-                break
-
-            except Exception as e:
-                last_error = str(e)
-                logger.warning(f"DVR recording {recording_id}: base {base} failed: {e}")
-                if bytes_written > 0:
-                    should_exit, is_int, reason = _check_recording_cancelled(recording_id)
-                    if should_exit and reason == "stopped_by_user":
-                        interrupted = False
-                        logger.info(
-                            f"DVR recording {recording_id}: "
-                            f"stopped by user — ending stream"
-                        )
-                    else:
-                        interrupted = True
-                        interrupted_reason = f"stream_interrupted: {e}"
-                    _done = True
+                    ffmpeg_proc.kill()
+                    try:
+                        ffmpeg_proc.wait(timeout=5)
+                    except subprocess.TimeoutExpired:
+                        pass
+                    last_error = f"no_segments_in_{_first_segment_timeout}s_from_{base}"
+                    ffmpeg_proc = None
                     break
-                should_exit, is_int, reason = _check_recording_cancelled(recording_id)
-                if should_exit:
-                    interrupted = is_int
-                    interrupted_reason = reason
-                    _done = True
+            else:
+                # Stream was confirmed, detect source stalls (e.g. proxy ghost-kills the client)
+                if len(segs_now) > _last_seg_count:
+                    _last_seg_count = len(segs_now)
+                    _last_new_seg_time = now
+                else:
+                    # Also treat a recently-modified segment file as activity.
+                    # With erratic source timestamps, FFmpeg may buffer data inside
+                    # a partially-written segment for longer than hls_time, so the
+                    # segment count won't increase even though data is flowing.
+                    # Only stat the most recent segment (highest filename) to keep
+                    # this O(1) per tick instead of O(N) over a long recording.
+                    try:
+                        if segs_now and hls_dir:
+                            _newest = max(segs_now)
+                            _newest_mtime = os.path.getmtime(
+                                os.path.join(hls_dir, _newest)
+                            )
+                            if _newest_mtime > _last_new_seg_time:
+                                _last_new_seg_time = _newest_mtime
+                    except Exception:
+                        pass
+                if now - _last_new_seg_time > _stall_timeout:
+                    logger.warning(
+                        f"DVR recording {recording_id}: no new HLS segments for "
+                        f"{_stall_timeout:.0f}s — source stream stalled, stopping FFmpeg"
+                    )
+                    ffmpeg_proc.send_signal(signal.SIGINT)
+                    try:
+                        ffmpeg_proc.wait(timeout=10)
+                    except subprocess.TimeoutExpired:
+                        ffmpeg_proc.kill()
+                    break
+
+            # Duration check — SIGINT lets FFmpeg write #EXT-X-ENDLIST cleanly
+            if now >= end_timestamp:
+                logger.info(
+                    f"DVR recording {recording_id}: scheduled end time reached, stopping FFmpeg"
+                )
+                ffmpeg_proc.send_signal(signal.SIGINT)
+                try:
+                    ffmpeg_proc.wait(timeout=10)
+                except subprocess.TimeoutExpired:
+                    ffmpeg_proc.kill()
                 break
 
-        if _done:
-            break
+            # Periodic DB poll: stop, delete, end_time extension
+            if now - _last_stop_poll >= _stop_poll_interval:
+                _last_stop_poll = now
+                try:
+                    _sc = Recording.objects.filter(
+                        id=recording_id
+                    ).only("custom_properties", "end_time").first()
+                    if _sc is None:
+                        logger.info(
+                            f"DVR recording {recording_id}: deleted — stopping FFmpeg"
+                        )
+                        ffmpeg_proc.send_signal(signal.SIGINT)
+                        try:
+                            ffmpeg_proc.wait(timeout=10)
+                        except subprocess.TimeoutExpired:
+                            ffmpeg_proc.kill()
+                        interrupted = False
+                        break
+                    if (_sc.custom_properties or {}).get("status") == "stopped":
+                        logger.info(
+                            f"DVR recording {recording_id}: stop requested — stopping FFmpeg"
+                        )
+                        ffmpeg_proc.send_signal(signal.SIGINT)
+                        try:
+                            ffmpeg_proc.wait(timeout=10)
+                        except subprocess.TimeoutExpired:
+                            ffmpeg_proc.kill()
+                        break
+                    # Handle end_time extension — just update the deadline, no restart needed
+                    try:
+                        new_end = _sc.end_time
+                        if new_end is not None:
+                            if _tz.is_naive(new_end):
+                                new_end = _tz.make_aware(new_end)
+                            new_ts = new_end.timestamp()
+                            if new_ts > end_timestamp:
+                                logger.info(
+                                    f"DVR recording {recording_id}: end_time extended to {new_end}"
+                                )
+                                end_timestamp = new_ts
+                    except Exception:
+                        pass
+                except Exception:
+                    pass
 
-    if chosen_base is None and bytes_written == 0:
+        # If FFmpeg exited after confirming the stream, log the exit code so we know why it stopped.
+        if _stream_confirmed and ffmpeg_proc is not None:
+            _exit_code = ffmpeg_proc.poll()
+            if _exit_code is not None and _exit_code != 0:
+                logger.warning(
+                    f"DVR recording {recording_id}: FFmpeg exited unexpectedly (rc={_exit_code}) "
+                    f"after stream was confirmed — source stream likely disconnected"
+                )
+
+        # If FFmpeg exited without the stream being confirmed, log the tail of
+        # captured stderr for diagnosis. Lines were already logged live by the
+        # drain thread; this surfaces the recent context in a single message.
+        elif not _stream_confirmed and ffmpeg_proc is not None:
+            try:
+                _exit_code = ffmpeg_proc.poll()
+                last_error = f"rc={_exit_code} from {base}"
+                _tail_text = "\n".join(_ffmpeg_stderr_tail)
+                if _tail_text:
+                    logger.warning(
+                        f"DVR recording {recording_id}: FFmpeg exited (rc={_exit_code}) "
+                        f"for {base} without producing segments.\nFFmpeg stderr tail:\n{_tail_text[-1000:]}"
+                    )
+                else:
+                    logger.warning(
+                        f"DVR recording {recording_id}: FFmpeg exited (rc={_exit_code}) "
+                        f"for {base} without producing segments (no stderr output)"
+                    )
+            except Exception:
+                pass
+
+    # Ensure FFmpeg has fully exited (covers cases where the loop broke early)
+    if ffmpeg_proc is not None and ffmpeg_proc.poll() is None:
+        try:
+            ffmpeg_proc.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            ffmpeg_proc.kill()
+
+    # If the loop broke because the Celery worker is shutting down (e.g.
+    # docker stop, container update) and the recording window is still open,
+    # leave the HLS working directory intact and mark the recording as
+    # interrupted.  `recover_recordings_on_startup` will re-dispatch this task
+    # on the next boot, and the resume-path logic in the prep block will
+    # adopt the existing `_hls_dir` and `file_path` so FFmpeg appends to the
+    # same playlist and the eventual concat produces a single MKV.
+    #
+    # Note: `_DVR_SHUTTING_DOWN` is per-process. The DVR queue currently uses
+    # a threads pool worker (one process, all threads see the flag). If the
+    # DVR queue is ever switched back to prefork, the parent's flag will not
+    # propagate to in-flight child workers.
+    if _DVR_SHUTTING_DOWN and time.time() < end_timestamp:
+        try:
+            _shutdown_rec = Recording.objects.filter(id=recording_id).first()
+            if _shutdown_rec:
+                _shutdown_cp = _shutdown_rec.custom_properties or {}
+                # Preserve _hls_dir / file_path / file_url exactly as they are
+                # so the HLS endpoint keeps serving until the worker dies and
+                # so recovery can find the working directory on next boot.
+                _shutdown_cp["status"] = "interrupted"
+                _shutdown_cp["interrupted_reason"] = "server_shutdown"
+                _shutdown_cp["ended_at"] = str(datetime.now())
+                _shutdown_rec.custom_properties = _shutdown_cp
+                _shutdown_rec.save(update_fields=["custom_properties"])
+        except Exception as _se:
+            logger.warning(
+                f"DVR recording {recording_id}: failed to flag interrupted on shutdown: {_se}"
+            )
+        logger.info(
+            f"DVR recording {recording_id}: worker shutting down with "
+            f"{int(end_timestamp - time.time())}s of recording window remaining; "
+            f"leaving HLS dir intact for resume on next boot, skipping concat."
+        )
+        return
+
+    if not _stream_confirmed and not interrupted:
         interrupted = True
-        interrupted_reason = f"no_stream_data: {last_error or 'all_bases_failed'}"
+        interrupted_reason = f"no_stream_data: {last_error or 'ffmpeg_failed'}"
 
-    # If no bytes were written at all, check whether this was a deliberate stop or a
-    # genuine failure.  The exception handler above already sets interrupted=False when
-    # it detects "stopped" status, but do not override that decision here.
+    # Measure bytes from HLS segment files
+    try:
+        bytes_written = sum(
+            os.path.getsize(os.path.join(hls_dir, f))
+            for f in os.listdir(hls_dir)
+            if f.startswith("seg_") and f.endswith(".ts")
+        ) if hls_dir and os.path.isdir(hls_dir) else 0
+    except Exception:
+        bytes_written = 0
     if bytes_written == 0 and not interrupted:
         _deliberately_stopped = False
         try:
@@ -2178,19 +2344,19 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str):
             # After the loop, the file and response are closed automatically.
             logger.info(f"Finished recording for channel {channel.name}")
 
-        # Log system event for recording end
-        try:
-            from core.utils import log_system_event
-            log_system_event(
-                'recording_end',
-                channel_id=channel.uuid,
-                channel_name=channel.name,
-                recording_id=recording_id,
-                interrupted=interrupted,
-                bytes_written=bytes_written
-            )
-        except Exception as e:
-            logger.error(f"Could not log recording end event: {e}")
+    # Log system event for recording end
+    try:
+        from core.utils import log_system_event
+        log_system_event(
+            'recording_end',
+            channel_id=channel.uuid,
+            channel_name=channel.name,
+            recording_id=recording_id,
+            interrupted=interrupted,
+            bytes_written=bytes_written
+        )
+    except Exception as e:
+        logger.error(f"Could not log recording end event: {e}")
 
     # If the Recording was deleted (cancelled by user), skip post-processing
     recording_cancelled = not Recording.objects.filter(id=recording_id).exists()
@@ -2199,261 +2365,248 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str):
         # Clean up all artifacts for the cancelled recording,
         # including any pre-restart .ts segments from server recovery.
         # Use the in-memory recording_obj since the DB row is already deleted.
-        _cancel_cleanup = [temp_ts_path, final_path]
-        _cancel_cp = (recording_obj.custom_properties or {}) if recording_obj else {}
-        _cancel_cleanup.extend(_cancel_cp.get("_pre_restart_ts_paths", []))
-        for _cleanup_path in _cancel_cleanup:
-            if not _cleanup_path:
-                continue
+        if hls_dir and os.path.isdir(hls_dir):
             try:
-                os.remove(_cleanup_path)
-                logger.info(f"Cleaned up cancelled recording artifact: {_cleanup_path}")
-            except FileNotFoundError:
-                pass
+                shutil.rmtree(hls_dir)
+                logger.info(f"Cleaned up cancelled recording HLS directory: {hls_dir}")
+            except Exception as _e:
+                logger.warning(f"Failed to remove HLS directory {hls_dir}: {_e}")
+        if final_path:
+            try:
+                if os.path.exists(final_path):
+                    os.remove(final_path)
+                    logger.info(f"Cleaned up cancelled recording MKV placeholder: {final_path}")
             except Exception:
                 pass
         return
 
-    # Concatenate pre-restart .ts segments with the current segment.
-    # Instead of creating an intermediate combined.ts and then remuxing to
-    # MKV (which loses timestamp boundary info and causes playback freezes
-    # at the splice point), go directly from the concat list → MKV.
-    # This lets ffmpeg's MKV muxer see each segment boundary and write
-    # correct cue points / clusters for seamless seeking.
-    _concat_did_remux = False
-    try:
-        _rec_obj_for_concat = Recording.objects.filter(id=recording_id).only("custom_properties").first()
-        _concat_cp = (_rec_obj_for_concat.custom_properties or {}) if _rec_obj_for_concat else {}
-        pre_restart_segments = _concat_cp.get("_pre_restart_ts_paths", [])
-        # Filter to segments that still exist on disk and have data
-        def _has_data(p):
-            try:
-                return os.stat(p).st_size > 0
-            except OSError:
-                return False
-        pre_restart_segments = [p for p in pre_restart_segments if p and _has_data(p)]
-        if pre_restart_segments and temp_ts_path and os.path.exists(temp_ts_path):
-            all_segments = pre_restart_segments + [temp_ts_path]
-            concat_list_path = temp_ts_path + ".concat.txt"
-            try:
-                with open(concat_list_path, "w") as cl:
-                    for seg in all_segments:
-                        cl.write(f"file '{seg}'\n")
+    # --- Post-processing: concat HLS segments → final MKV ---
+    remux_success = False
+    hls_m3u8 = os.path.join(hls_dir, "index.m3u8") if hls_dir else None
 
-                # Direct concat → MKV in a single pass.
-                # -reset_timestamps 1 tells the concat demuxer to reset
-                # timestamps at each segment boundary, eliminating the
-                # discontinuity that causes playback to freeze at the
-                # splice point.
-                concat_result = subprocess.run(
+    def _get_hls_segments(m3u8_path, seg_dir):
+        """Return ordered segment paths from an HLS m3u8 playlist."""
+        segs = []
+        try:
+            with open(m3u8_path) as _f:
+                for _line in _f:
+                    _line = _line.strip()
+                    if _line and not _line.startswith('#'):
+                        sp = os.path.join(seg_dir, _line) if not os.path.isabs(_line) else _line
+                        if os.path.exists(sp):
+                            segs.append(sp)
+        except Exception as _e:
+            logger.warning(f"DVR recording {recording_id}: failed to parse m3u8: {_e}")
+        return segs
+
+    segments = (
+        _get_hls_segments(hls_m3u8, hls_dir)
+        if (hls_m3u8 and os.path.exists(hls_m3u8))
+        else []
+    )
+    if not segments and hls_dir and os.path.isdir(hls_dir):
+        # Fallback: sort all segment files by name if m3u8 is missing or empty
+        try:
+            segments = sorted(
+                os.path.join(hls_dir, f)
+                for f in os.listdir(hls_dir)
+                if f.startswith("seg_") and f.endswith(".ts")
+            )
+        except Exception:
+            segments = []
+
+    if segments:
+        concat_list_path = os.path.join(hls_dir, "concat.txt")
+        try:
+            with open(concat_list_path, "w") as _cl:
+                for seg in segments:
+                    _escaped = seg.replace("'", "'\\''")
+                    _cl.write(f"file '{_escaped}'\n")
+
+            concat_result = subprocess.run(
+                [
+                    "ffmpeg", "-y",
+                    "-f", "concat", "-safe", "0",
+                    "-i", concat_list_path,
+                    "-c", "copy",
+                    final_path,
+                ],
+                stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
+            )
+            _ok = (
+                concat_result.returncode == 0
+                and os.path.exists(final_path)
+                and os.path.getsize(final_path) > 0
+            )
+            _fallback_used = False
+            # MP4-intermediate fallback: some MPEG-TS streams (parameter set
+            # changes mid-stream, weird PMT updates, audio discontinuities)
+            # fail to mux directly into Matroska but go through an MP4
+            # container cleanly, which can then be remuxed losslessly to
+            # MKV.  Try this before declaring the recording lost.
+            if not _ok:
+                logger.warning(
+                    f"DVR recording {recording_id}: direct HLS\u2192MKV concat failed "
+                    f"(rc={concat_result.returncode}); attempting MP4-intermediate "
+                    f"fallback. stderr: {(concat_result.stderr or '')[:300]}"
+                )
+                try:
+                    if os.path.exists(final_path) and os.path.getsize(final_path) == 0:
+                        os.remove(final_path)
+                except OSError:
+                    pass
+                _intermediate_mp4 = os.path.join(
+                    hls_dir, f".dvr_{recording_id}_intermediate.mp4"
+                )
+                try:
+                    if os.path.exists(_intermediate_mp4):
+                        os.remove(_intermediate_mp4)
+                except OSError:
+                    pass
+                _mp4_concat = subprocess.run(
                     [
                         "ffmpeg", "-y",
-                        "-fflags", "+genpts+igndts+discardcorrupt",
-                        "-err_detect", "ignore_err",
                         "-f", "concat", "-safe", "0",
-                        "-segment_time_metadata", "1",
                         "-i", concat_list_path,
-                        "-reset_timestamps", "1",
-                        "-map", "0",
                         "-c", "copy",
-                        final_path,
+                        "-bsf:a", "aac_adtstoasc",
+                        _intermediate_mp4,
                     ],
                     stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
                 )
-                if concat_result.returncode == 0 and os.path.exists(final_path) and os.path.getsize(final_path) > 0:
-                    _concat_did_remux = True
-                    # Clean up individual TS segments (including current)
-                    for seg in all_segments:
-                        try:
-                            os.remove(seg)
-                        except OSError:
-                            pass
+                if (
+                    _mp4_concat.returncode == 0
+                    and os.path.exists(_intermediate_mp4)
+                    and os.path.getsize(_intermediate_mp4) > 0
+                ):
+                    _mp4_to_mkv = subprocess.run(
+                        [
+                            "ffmpeg", "-y",
+                            "-i", _intermediate_mp4,
+                            "-c", "copy",
+                            final_path,
+                        ],
+                        stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
+                    )
+                    if (
+                        _mp4_to_mkv.returncode == 0
+                        and os.path.exists(final_path)
+                        and os.path.getsize(final_path) > 0
+                    ):
+                        _ok = True
+                        _fallback_used = True
+                    else:
+                        logger.error(
+                            f"DVR recording {recording_id}: MP4\u2192MKV remux step "
+                            f"failed (rc={_mp4_to_mkv.returncode}). stderr: "
+                            f"{(_mp4_to_mkv.stderr or '')[:300]}"
+                        )
+                else:
+                    logger.error(
+                        f"DVR recording {recording_id}: HLS\u2192MP4 concat fallback "
+                        f"failed (rc={_mp4_concat.returncode}). stderr: "
+                        f"{(_mp4_concat.stderr or '')[:300]}"
+                    )
+                try:
+                    if os.path.exists(_intermediate_mp4):
+                        os.remove(_intermediate_mp4)
+                except OSError:
+                    pass
+
+            if _ok:
+                remux_success = True
+                if _fallback_used:
                     logger.info(
-                        f"DVR recording {recording_id}: concat→MKV succeeded — "
-                        f"{len(all_segments)} segments → {os.path.basename(final_path)} "
+                        f"DVR recording {recording_id}: HLS\u2192MP4\u2192MKV fallback "
+                        f"concat succeeded \u2014 {len(segments)} segments \u2192 "
+                        f"{os.path.basename(final_path)} "
                         f"({os.path.getsize(final_path):,} bytes)"
                     )
                 else:
-                    logger.warning(
-                        f"DVR recording {recording_id}: direct concat→MKV failed "
-                        f"(rc={concat_result.returncode}), falling back to "
-                        f"normal remux with current segment only. "
-                        f"stderr: {(concat_result.stderr or '')[:500]}"
+                    logger.info(
+                        f"DVR recording {recording_id}: HLS\u2192MKV concat succeeded \u2014 "
+                        f"{len(segments)} segments \u2192 {os.path.basename(final_path)} "
+                        f"({os.path.getsize(final_path):,} bytes)"
                     )
-            finally:
+                # Update DB so *new* client requests go to /file/ (the final MKV)
+                # rather than the soon-to-be-removed /hls/ endpoint.  Important:
+                # we do NOT clear _hls_dir here, active viewers still in the
+                # 20s heartbeat window need it to keep fetching .ts segments
+                # during the viewer-wait grace period below.  _hls_dir is
+                # cleared only after the directory is actually removed.
                 try:
-                    os.remove(concat_list_path)
-                except OSError:
-                    pass
-            # Clear the pre-restart paths from custom_properties
-            if _rec_obj_for_concat:
-                _ccp = _rec_obj_for_concat.custom_properties or {}
-                _ccp.pop("_pre_restart_ts_paths", None)
-                _ccp.pop("interrupted_reason", None)
-                _rec_obj_for_concat.custom_properties = _ccp
-                _rec_obj_for_concat.save(update_fields=["custom_properties"])
-    except Exception as e:
-        logger.warning(
-            f"DVR recording {recording_id}: segment concatenation error "
-            f"({type(e).__name__}: {e}), proceeding with current segment only."
-        )
-
-    # Remux TS to MKV container with retry for transient I/O errors
-    # (Skip if concat already produced the final MKV directly.)
-    remux_success = _concat_did_remux
-    existing_mkv_size = 0
-    try:
-        if final_path and os.path.exists(final_path):
-            existing_mkv_size = os.path.getsize(final_path)
-    except OSError:
-        pass
-    for _remux_attempt in range(_dvr_remux_max_retries):
-        if remux_success:
-            break
-        try:
-            if temp_ts_path and os.path.exists(temp_ts_path):
-                # First attempt: Direct TS to MKV remux
-                result = subprocess.run([
-                    "ffmpeg", "-y",
-                    "-fflags", "+genpts+igndts+discardcorrupt",  # Regenerate timestamps, ignore DTS
-                    "-err_detect", "ignore_err",   # Ignore minor stream errors
-                    "-i", temp_ts_path,
-                    "-map", "0",  # Map all streams
-                    "-c", "copy",
-                    final_path
-                ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
-
-                # Check if FFmpeg succeeded (return code 0) and output file is valid
-                if result.returncode == 0 and os.path.exists(final_path) and os.path.getsize(final_path) > 0:
-                    remux_success = True
-                    logger.info(f"Direct TS→MKV remux succeeded for {os.path.basename(final_path)}")
-                else:
-                    # Direct remux failed - try fallback: TS → MP4 → MKV to fix timestamps
-                    logger.warning(f"Direct TS→MKV remux failed (return code: {result.returncode}), trying fallback TS→MP4→MKV")
-
-                    # Clean up partial/failed MKV
-                    try:
-                        if os.path.exists(final_path):
-                            os.remove(final_path)
-                    except Exception:
-                        pass
-
-                    # Step 1: TS → MP4 (MP4 container handles broken timestamps better)
-                    temp_mp4_path = os.path.splitext(temp_ts_path)[0] + ".mp4"
-                    result_mp4 = subprocess.run([
-                        "ffmpeg", "-y",
-                        "-fflags", "+genpts+igndts+discardcorrupt",
-                        "-err_detect", "ignore_err",
-                        "-i", temp_ts_path,
-                        "-map", "0",
-                        "-c", "copy",
-                        temp_mp4_path
-                    ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
-
-                    if result_mp4.returncode == 0 and os.path.exists(temp_mp4_path) and os.path.getsize(temp_mp4_path) > 0:
-                        logger.info(f"TS→MP4 conversion succeeded, now converting MP4→MKV")
-
-                        # Step 2: MP4 → MKV (clean timestamps from MP4)
-                        result_mkv = subprocess.run([
-                            "ffmpeg", "-y",
-                            "-i", temp_mp4_path,
-                            "-map", "0",
-                            "-c", "copy",
-                            final_path
-                        ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
-
-                        if result_mkv.returncode == 0 and os.path.exists(final_path) and os.path.getsize(final_path) > 0:
-                            remux_success = True
-                            logger.info(f"Fallback TS→MP4→MKV remux succeeded for {os.path.basename(final_path)}")
-                        else:
-                            logger.error(f"MP4→MKV conversion failed (return code: {result_mkv.returncode})")
-
-                        # Clean up temp MP4
-                        try:
-                            if os.path.exists(temp_mp4_path):
-                                os.remove(temp_mp4_path)
-                        except Exception:
-                            pass
-                    else:
-                        logger.error(f"TS→MP4 conversion failed (return code: {result_mp4.returncode})")
-
-                # Sanity-check the remuxed file.  Two checks:
-                # 1. If a pre-existing MKV was overwritten, reject a
-                #    file that is drastically smaller (duplicate-task
-                #    overwrite protection).
-                # 2. If the MKV is smaller than the .ts source, the
-                #    remux likely produced a corrupt or truncated file.
-                if remux_success:
-                    try:
-                        new_size = os.path.getsize(final_path)
-                        ts_size = os.path.getsize(temp_ts_path) if temp_ts_path and os.path.exists(temp_ts_path) else 0
-                        reject = False
-                        if existing_mkv_size > 0 and new_size < existing_mkv_size * 0.5:
-                            logger.error(
-                                f"DVR recording {recording_id}: new MKV "
-                                f"({new_size:,} bytes) is less than 50%% of "
-                                f"the previous MKV ({existing_mkv_size:,} bytes) "
-                                f"— refusing to overwrite. Keeping .ts for "
-                                f"manual recovery."
-                            )
-                            reject = True
-                        elif ts_size > 0 and new_size < ts_size * 0.1:
-                            logger.error(
-                                f"DVR recording {recording_id}: remuxed MKV "
-                                f"({new_size:,} bytes) is less than 10%% of "
-                                f"the source TS ({ts_size:,} bytes) — likely "
-                                f"corrupt. Keeping .ts for manual recovery."
-                            )
-                            reject = True
-                        if reject:
-                            remux_success = False
-                            try:
-                                os.remove(final_path)
-                            except OSError:
-                                pass
-                    except OSError:
-                        pass
-
-                # Clean up temp TS file only on successful remux
-                if remux_success:
-                    try:
-                        os.remove(temp_ts_path)
-                        logger.debug(f"Cleaned up temp TS file: {temp_ts_path}")
-                    except Exception as e:
-                        logger.warning(f"Failed to remove temp TS file: {e}")
-                else:
-                    # Keep TS file for debugging/manual recovery if remux failed
-                    logger.warning(f"Remux failed - keeping temp TS file for recovery: {temp_ts_path}")
-                    # Clean up any partial MKV
-                    try:
-                        if os.path.exists(final_path):
-                            os.remove(final_path)
-                            logger.debug(f"Cleaned up partial MKV file: {final_path}")
-                    except Exception:
-                        pass
-            break  # Completed (success or deterministic failure)
-
-        except (OSError, subprocess.SubprocessError) as e:
-            # Clean up partial output before potential retry
-            try:
-                if os.path.exists(final_path):
-                    os.remove(final_path)
-            except Exception:
-                pass
-            if _remux_attempt + 1 < _dvr_remux_max_retries:
-                _wait = _dvr_remux_retry_interval * (2 ** _remux_attempt)
-                logger.warning(
-                    f"DVR recording {recording_id}: remux failed "
-                    f"({type(e).__name__}), retrying in {_wait}s "
-                    f"({_remux_attempt + 1}/{_dvr_remux_max_retries})..."
-                )
-                time.sleep(_wait)
+                    _pre_rmtree_rec = Recording.objects.filter(id=recording_id).first()
+                    if _pre_rmtree_rec:
+                        _pre_rmtree_cp = _pre_rmtree_rec.custom_properties or {}
+                        _pre_rmtree_cp["file_url"] = f"/api/channels/recordings/{recording_id}/file/"
+                        _pre_rmtree_cp["output_file_url"] = _pre_rmtree_cp["file_url"]
+                        _pre_rmtree_rec.custom_properties = _pre_rmtree_cp
+                        _pre_rmtree_rec.save(update_fields=["custom_properties"])
+                except Exception as _pre_e:
+                    logger.warning(f"DVR recording {recording_id}: pre-rmtree DB update failed: {_pre_e}")
+                # Wait for active HLS viewers to finish downloading segments
+                # before deleting the directory.  The hls endpoint refreshes
+                # _hls_viewer_key on every .ts request with a 20s TTL, so the
+                # key expiring naturally means no segment was fetched in the
+                # last 20 seconds (client has stopped).  We loop until the key
+                # is gone with a 4-hour hard safety cap in case Redis gets stuck.
+                _viewer_wait_start = time.time()
+                _safety_timeout = 14400  # 4 hours absolute maximum
+                try:
+                    from core.utils import RedisClient as _RC
+                    _rv = _RC.get_client(max_retries=1, retry_interval=0)
+                    if _rv and _rv.exists(_hls_viewer_key):
+                        logger.info(
+                            f"DVR recording {recording_id}: active HLS viewer detected, "
+                            f"deferring HLS directory cleanup until client disconnects"
+                        )
+                        while _rv.exists(_hls_viewer_key):
+                            if time.time() - _viewer_wait_start > _safety_timeout:
+                                logger.warning(
+                                    f"DVR recording {recording_id}: viewer wait safety timeout "
+                                    f"({_safety_timeout}s) reached, proceeding with HLS directory cleanup"
+                                )
+                                break
+                            time.sleep(2)
+                        logger.info(
+                            f"DVR recording {recording_id}: viewer wait complete after "
+                            f"{time.time() - _viewer_wait_start:.1f}s"
+                        )
+                except Exception as _ve:
+                    logger.debug(f"DVR recording {recording_id}: viewer wait check failed: {_ve}")
+                try:
+                    shutil.rmtree(hls_dir)
+                    logger.debug(f"Cleaned up HLS directory: {hls_dir}")
+                except Exception as _e:
+                    logger.warning(f"Failed to remove HLS directory {hls_dir}: {_e}")
+                # Now that the HLS dir is gone, clear _hls_dir from custom_properties
+                # so the hls endpoint will redirect (.m3u8) or 404 (.ts) cleanly.
+                try:
+                    _post_rmtree_rec = Recording.objects.filter(id=recording_id).first()
+                    if _post_rmtree_rec:
+                        _post_rmtree_cp = _post_rmtree_rec.custom_properties or {}
+                        if "_hls_dir" in _post_rmtree_cp:
+                            _post_rmtree_cp.pop("_hls_dir", None)
+                            _post_rmtree_rec.custom_properties = _post_rmtree_cp
+                            _post_rmtree_rec.save(update_fields=["custom_properties"])
+                except Exception as _post_e:
+                    logger.warning(f"DVR recording {recording_id}: post-rmtree DB update failed: {_post_e}")
             else:
-                logger.warning(
-                    f"DVR recording {recording_id}: remux failed "
-                    f"after {_dvr_remux_max_retries} attempts: {e}. "
-                    f"Keeping .ts for manual recovery: {temp_ts_path}"
+                logger.error(
+                    f"DVR recording {recording_id}: all HLS\u2192MKV concat attempts "
+                    f"failed (direct rc={concat_result.returncode}, MP4 fallback also "
+                    f"failed). Keeping HLS segments for recovery. stderr: "
+                    f"{(concat_result.stderr or '')[:500]}"
                 )
+        except Exception as _ce:
+            logger.error(f"DVR recording {recording_id}: concat exception: {_ce}")
+        finally:
+            try:
+                os.remove(concat_list_path)
+            except OSError:
+                pass
+    else:
+        logger.warning(f"DVR recording {recording_id}: no HLS segments found. Nothing to concat")
 
     # Persist final metadata to Recording (status, ended_at, and stream stats if available)
     try:
@@ -2465,6 +2618,11 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str):
         cp = recording_obj.custom_properties or {}
         cp["ended_at"] = str(datetime.now())
 
+        # Restore file_url to the permanent MKV endpoint now that recording has ended.
+        # During recording it pointed to /hls/index.m3u8; clients should use /file/ hereafter.
+        cp["file_url"] = f"/api/channels/recordings/{recording_id}/file/"
+        cp["output_file_url"] = cp["file_url"]
+
         # Final status priority: stopped > completed > interrupted.
         # "stopped" is set by the stop endpoint before stream teardown, so
         # refresh_from_db() above guarantees it is visible here.
@@ -2591,7 +2749,7 @@ def recover_recordings_on_startup():
         from django.utils import timezone
         from .models import Recording
         from core.utils import RedisClient
-        from .signals import schedule_recording_task
+        from .signals import schedule_recording_task, revoke_task
 
         redis = RedisClient.get_client()
         if redis:
@@ -2637,14 +2795,12 @@ def recover_recordings_on_startup():
 
                 # Preserve the pre-restart .ts segment path so run_recording
                 # can concatenate it with the resumed segment later.
-                old_ts = cp.get("_temp_file_path")
-                if old_ts and os.path.exists(old_ts) and os.path.getsize(old_ts) > 0:
-                    prior_segments = cp.get("_pre_restart_ts_paths", [])
-                    prior_segments.append(old_ts)
-                    cp["_pre_restart_ts_paths"] = prior_segments
+                hls_dir_path = cp.get("_hls_dir")
+                if hls_dir_path and os.path.isdir(hls_dir_path):
+                    existing_segs = [f for f in os.listdir(hls_dir_path) if f.endswith(".ts")]
                     logger.info(
                         f"recover_recordings_on_startup: recording {rec.id} — "
-                        f"preserving pre-restart TS segment: {old_ts}"
+                        f"HLS dir has {len(existing_segs)} existing segment(s), will resume"
                     )
 
                 rec.custom_properties = cp
@@ -2687,38 +2843,86 @@ def recover_recordings_on_startup():
         for rec in expired:
             try:
                 cp = rec.custom_properties or {}
-                ts_path = cp.get("_temp_file_path")
+                hls_dir_path = cp.get("_hls_dir")
                 mkv_path = cp.get("file_path")
 
-                if ts_path and os.path.exists(ts_path) and os.path.getsize(ts_path) > 0 and mkv_path:
-                    logger.info(
-                        f"recover_recordings_on_startup: recording {rec.id} expired "
-                        f"during downtime — remuxing partial TS ({os.path.getsize(ts_path):,} bytes)"
-                    )
-                    os.makedirs(os.path.dirname(mkv_path), exist_ok=True)
-                    result = subprocess.run(
-                        [
-                            "ffmpeg", "-y",
-                            "-fflags", "+genpts+igndts+discardcorrupt",
-                            "-err_detect", "ignore_err",
-                            "-i", ts_path, "-map", "0", "-c", "copy", mkv_path,
-                        ],
-                        stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
-                    )
-                    if result.returncode == 0 and os.path.exists(mkv_path) and os.path.getsize(mkv_path) > 0:
-                        cp["status"] = "interrupted"
-                        cp["interrupted_reason"] = "server_restarted_after_end"
-                        cp["remux_success"] = True
+                if hls_dir_path and os.path.isdir(hls_dir_path) and mkv_path:
+                    # Parse m3u8 for ordered segment list; fall back to sorted filenames
+                    _m3u8 = os.path.join(hls_dir_path, "index.m3u8")
+                    _segs = []
+                    if os.path.exists(_m3u8):
                         try:
-                            os.remove(ts_path)
-                        except OSError:
+                            with open(_m3u8) as _f:
+                                for _l in _f:
+                                    _l = _l.strip()
+                                    if _l and not _l.startswith('#'):
+                                        _sp = os.path.join(hls_dir_path, _l) if not os.path.isabs(_l) else _l
+                                        if os.path.exists(_sp):
+                                            _segs.append(_sp)
+                        except Exception:
                             pass
-                        logger.info(f"recover_recordings_on_startup: recording {rec.id} remuxed successfully")
+                    if not _segs:
+                        _segs = sorted(
+                            os.path.join(hls_dir_path, f)
+                            for f in os.listdir(hls_dir_path)
+                            if f.startswith("seg_") and f.endswith(".ts")
+                        )
+
+                    if _segs:
+                        logger.info(
+                            f"recover_recordings_on_startup: recording {rec.id} expired "
+                            f"during downtime, concat {len(_segs)} HLS segment(s) \u2192 MKV"
+                        )
+                        os.makedirs(os.path.dirname(mkv_path), exist_ok=True)
+                        _concat_txt = os.path.join(hls_dir_path, "concat.txt")
+                        try:
+                            with open(_concat_txt, "w") as _cl:
+                                for _s in _segs:
+                                    _escaped = _s.replace("'", "'\\''")
+                                    _cl.write(f"file '{_escaped}'\n")
+                            _res = subprocess.run(
+                                [
+                                    "ffmpeg", "-y",
+                                    "-f", "concat", "-safe", "0",
+                                    "-i", _concat_txt,
+                                    "-c", "copy", mkv_path,
+                                ],
+                                stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
+                            )
+                            if _res.returncode == 0 and os.path.exists(mkv_path) and os.path.getsize(mkv_path) > 0:
+                                cp["status"] = "interrupted"
+                                cp["interrupted_reason"] = "server_restarted_after_end"
+                                cp["remux_success"] = True
+                                try:
+                                    shutil.rmtree(hls_dir_path)
+                                except OSError:
+                                    pass
+                                logger.info(
+                                    f"recover_recordings_on_startup: recording {rec.id} HLS\u2192MKV concat succeeded"
+                                )
+                            else:
+                                cp["status"] = "interrupted"
+                                cp["interrupted_reason"] = "server_restarted_after_end"
+                                cp["remux_success"] = False
+                                logger.warning(
+                                    f"recover_recordings_on_startup: recording {rec.id} concat failed, keeping HLS dir"
+                                )
+                        except Exception as _ce:
+                            cp["status"] = "interrupted"
+                            cp["interrupted_reason"] = "server_restarted_after_end"
+                            cp["remux_success"] = False
+                            logger.warning(
+                                f"recover_recordings_on_startup: recording {rec.id} concat error: {_ce}"
+                            )
+                        finally:
+                            try:
+                                os.remove(_concat_txt)
+                            except OSError:
+                                pass
                     else:
                         cp["status"] = "interrupted"
                         cp["interrupted_reason"] = "server_restarted_after_end"
                         cp["remux_success"] = False
-                        logger.warning(f"recover_recordings_on_startup: recording {rec.id} remux failed, keeping .ts")
                 else:
                     cp["status"] = "interrupted"
                     cp["interrupted_reason"] = "server_restarted_after_end"
diff --git a/apps/channels/tests/test_dvr_port_resolution.py b/apps/channels/tests/test_dvr_port_resolution.py
index c7373959..61f74454 100644
--- a/apps/channels/tests/test_dvr_port_resolution.py
+++ b/apps/channels/tests/test_dvr_port_resolution.py
@@ -2,58 +2,68 @@ import os
 from django.test import SimpleTestCase
 from unittest.mock import patch
 
-from apps.channels.tasks import build_dvr_candidates
+from apps.channels.tasks import get_dvr_stream_base_url
 
 
-class DVRPortResolutionTests(SimpleTestCase):
+class DVRStreamBaseURLTests(SimpleTestCase):
     """
-    Tests that DVR recording candidate URLs respect the DISPATCHARR_PORT
-    environment variable instead of hardcoding port 9191.
+    Tests that get_dvr_stream_base_url() returns the correct single URL
+    for each deployment mode.
     """
 
-    @patch.dict(os.environ, {'REDIS_HOST': 'redis'}, clear=True)
-    def test_default_port_uses_9191(self):
-        """Without DISPATCHARR_PORT set, candidates default to 9191."""
-        candidates = build_dvr_candidates()
-        self.assertIn('http://web:9191', candidates)
-        self.assertIn('http://localhost:9191', candidates)
+    @patch.dict(os.environ, {}, clear=True)
+    def test_aio_default_uses_localhost_5656(self):
+        """AIO mode (default) reaches uwsgi directly on loopback port 5656."""
+        url = get_dvr_stream_base_url()
+        self.assertEqual(url, 'http://127.0.0.1:5656')
 
-    @patch.dict(os.environ, {'DISPATCHARR_PORT': '8080', 'REDIS_HOST': 'redis'}, clear=True)
-    def test_custom_port_reflected_in_candidates(self):
-        """DISPATCHARR_PORT=8080 replaces all hardcoded 9191 references."""
-        candidates = build_dvr_candidates()
-        self.assertIn('http://web:8080', candidates)
-        self.assertIn('http://localhost:8080', candidates)
-        self.assertNotIn('http://web:9191', candidates)
-        self.assertNotIn('http://localhost:9191', candidates)
+    @patch.dict(os.environ, {'DISPATCHARR_ENV': 'aio'}, clear=True)
+    def test_aio_explicit_uses_localhost_5656(self):
+        """Explicit DISPATCHARR_ENV=aio also uses loopback port 5656."""
+        url = get_dvr_stream_base_url()
+        self.assertEqual(url, 'http://127.0.0.1:5656')
+
+    @patch.dict(os.environ, {'DISPATCHARR_ENV': 'dev'}, clear=True)
+    def test_dev_mode_uses_localhost_5656(self):
+        """Dev mode shares the container with uwsgi — uses loopback port 5656."""
+        url = get_dvr_stream_base_url()
+        self.assertEqual(url, 'http://127.0.0.1:5656')
+
+    @patch.dict(os.environ, {'DISPATCHARR_ENV': 'modular', 'DISPATCHARR_PORT': '9191'}, clear=True)
+    def test_modular_uses_web_service_name(self):
+        """Modular mode uses the 'web' Docker service name by default."""
+        url = get_dvr_stream_base_url()
+        self.assertEqual(url, 'http://web:9191')
+
+    @patch.dict(os.environ, {'DISPATCHARR_ENV': 'modular', 'DISPATCHARR_PORT': '8080'}, clear=True)
+    def test_modular_custom_port(self):
+        """Modular mode respects DISPATCHARR_PORT."""
+        url = get_dvr_stream_base_url()
+        self.assertEqual(url, 'http://web:8080')
 
     @patch.dict(os.environ, {
-        'DISPATCHARR_PORT': '7777',
-        'DISPATCHARR_ENV': 'dev',
-        'REDIS_HOST': 'redis',
+        'DISPATCHARR_ENV': 'modular',
+        'DISPATCHARR_PORT': '9191',
+        'DISPATCHARR_WEB_HOST': 'dispatcharr_web',
     }, clear=True)
-    def test_dev_mode_includes_5656_and_custom_port(self):
-        """Dev mode includes both uwsgi internal port (5656) and custom port."""
-        candidates = build_dvr_candidates()
-        self.assertIn('http://127.0.0.1:5656', candidates)
-        self.assertIn('http://127.0.0.1:7777', candidates)
+    def test_modular_custom_web_host(self):
+        """DISPATCHARR_WEB_HOST overrides the default 'web' service name."""
+        url = get_dvr_stream_base_url()
+        self.assertEqual(url, 'http://dispatcharr_web:9191')
 
     @patch.dict(os.environ, {
         'DISPATCHARR_INTERNAL_TS_BASE_URL': 'http://custom:1234',
-        'REDIS_HOST': 'redis',
+        'DISPATCHARR_ENV': 'modular',
     }, clear=True)
-    def test_explicit_override_is_first(self):
-        """DISPATCHARR_INTERNAL_TS_BASE_URL should be the first candidate."""
-        candidates = build_dvr_candidates()
-        self.assertEqual(candidates[0], 'http://custom:1234')
+    def test_explicit_override_always_wins(self):
+        """DISPATCHARR_INTERNAL_TS_BASE_URL takes priority over all other settings."""
+        url = get_dvr_stream_base_url()
+        self.assertEqual(url, 'http://custom:1234')
 
     @patch.dict(os.environ, {
-        'DISPATCHARR_PORT': '3000',
-        'DISPATCHARR_INTERNAL_API_BASE': 'http://myhost:4000',
-        'REDIS_HOST': 'redis',
+        'DISPATCHARR_INTERNAL_TS_BASE_URL': 'http://custom:1234/',
     }, clear=True)
-    def test_internal_api_base_overrides_web_fallback(self):
-        """DISPATCHARR_INTERNAL_API_BASE replaces the http://web:{port} default."""
-        candidates = build_dvr_candidates()
-        self.assertIn('http://myhost:4000', candidates)
-        self.assertNotIn('http://web:3000', candidates)
+    def test_explicit_override_strips_trailing_slash(self):
+        """Trailing slash is stripped from DISPATCHARR_INTERNAL_TS_BASE_URL."""
+        url = get_dvr_stream_base_url()
+        self.assertEqual(url, 'http://custom:1234')
diff --git a/dispatcharr/celery.py b/dispatcharr/celery.py
index 8ccb3c33..0d63f063 100644
--- a/dispatcharr/celery.py
+++ b/dispatcharr/celery.py
@@ -49,6 +49,11 @@ app.conf.update(
     worker_task_log_format='%(asctime)s %(levelname)s %(task_name)s: %(message)s',
 )
 
+# Route long-running DVR recordings to a dedicated `dvr` queue consumed by a thread-pool worker.
+app.conf.task_routes = {
+    'apps.channels.tasks.run_recording': {'queue': 'dvr'},
+}
+
 # Add memory cleanup after task completion
 @task_postrun.connect  # Use the imported signal
 def cleanup_task_memory(**kwargs):
diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml
index 547ac249..58b6e4f5 100644
--- a/docker/docker-compose.yml
+++ b/docker/docker-compose.yml
@@ -118,8 +118,12 @@ services:
       - DISPATCHARR_ENV=modular
 
       # Internal Service Communication
-      # Must match the web service port for DVR recording and internal API calls
+      # Celery uses these to reach the web container for DVR recording.
+      # DISPATCHARR_PORT must match the port exposed by the web service.
+      # DISPATCHARR_WEB_HOST defaults to "web" (the service name above).
+      # Only set DISPATCHARR_WEB_HOST if you rename the web service in this file.
       - DISPATCHARR_PORT=9191
+      #- DISPATCHARR_WEB_HOST=web
 
       # PostgreSQL — must match web service settings
       - POSTGRES_HOST=db
diff --git a/docker/entrypoint.celery.sh b/docker/entrypoint.celery.sh
index b38b6ac3..c795d781 100644
--- a/docker/entrypoint.celery.sh
+++ b/docker/entrypoint.celery.sh
@@ -67,4 +67,8 @@ NICE_LEVEL="${CELERY_NICE_LEVEL:-5}"
 if [ "$NICE_LEVEL" -lt 0 ] 2>/dev/null; then
     echo "Warning: CELERY_NICE_LEVEL=$NICE_LEVEL is negative, requires SYS_NICE capability"
 fi
-nice -n "$NICE_LEVEL" celery -A dispatcharr worker -l info --autoscale=6,1
+
+# DVR worker: thread pool for the long-running, I/O-bound run_recording task.
+nice -n "$NICE_LEVEL" celery -A dispatcharr worker -Q dvr -n dvr@%h --pool=threads --concurrency=20 -l info &
+# Default prefork worker: every queue except `dvr`.
+nice -n "$NICE_LEVEL" celery -A dispatcharr worker -Q celery -n default@%h --autoscale=6,1 -l info
diff --git a/docker/uwsgi.debug.ini b/docker/uwsgi.debug.ini
index d2847335..bc21275d 100644
--- a/docker/uwsgi.debug.ini
+++ b/docker/uwsgi.debug.ini
@@ -7,9 +7,10 @@ exec-before = python /app/scripts/wait_for_redis.py
 
 ; Start Redis first
 attach-daemon = redis-server
-; Then start other services with configurable nice level (default: 5 for low priority)
-; Users can override via CELERY_NICE_LEVEL environment variable in docker-compose
-attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker --autoscale=6,1
+; Default prefork worker: every queue except `dvr`.
+attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker -Q celery -n default@%%h --autoscale=6,1
+; DVR worker: thread pool for the long-running, I/O-bound run_recording task.
+attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker -Q dvr -n dvr@%%h --pool=threads --concurrency=20
 attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr beat
 attach-daemon = daphne -b 0.0.0.0 -p 8001 dispatcharr.asgi:application
 attach-daemon = cd /app/frontend && npm run dev
diff --git a/docker/uwsgi.dev.ini b/docker/uwsgi.dev.ini
index 51aae9a5..dd799500 100644
--- a/docker/uwsgi.dev.ini
+++ b/docker/uwsgi.dev.ini
@@ -9,9 +9,10 @@ exec-pre = python /app/scripts/wait_for_redis.py
 
 ; Start Redis first
 attach-daemon = redis-server --protected-mode no
-; Then start other services with configurable nice level (default: 5 for low priority)
-; Users can override via CELERY_NICE_LEVEL environment variable in docker-compose
-attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker --autoscale=6,1
+; Default prefork worker: every queue except `dvr`.
+attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker -Q celery -n default@%%h --autoscale=6,1
+; DVR worker: thread pool for the long-running, I/O-bound run_recording task.
+attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker -Q dvr -n dvr@%%h --pool=threads --concurrency=20
 attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr beat
 attach-daemon = daphne -b 0.0.0.0 -p 8001 dispatcharr.asgi:application
 attach-daemon = cd /app/frontend && npm run dev
diff --git a/docker/uwsgi.ini b/docker/uwsgi.ini
index 920bac48..3972c3d9 100644
--- a/docker/uwsgi.ini
+++ b/docker/uwsgi.ini
@@ -9,9 +9,10 @@ exec-pre = python /app/scripts/wait_for_redis.py
 
 ; Start Redis first
 attach-daemon = redis-server
-; Then start other services with configurable nice level (default: 5 for low priority)
-; Users can override via CELERY_NICE_LEVEL environment variable in docker-compose
-attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker --autoscale=6,1
+; Default prefork worker: every queue except `dvr`.
+attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker -Q celery -n default@%%h --autoscale=6,1
+; DVR worker: thread pool for the long-running, I/O-bound run_recording task.
+attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker -Q dvr -n dvr@%%h --pool=threads --concurrency=20
 attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr beat
 attach-daemon = daphne -b 0.0.0.0 -p 8001 dispatcharr.asgi:application
 
diff --git a/frontend/src/components/FloatingVideo.jsx b/frontend/src/components/FloatingVideo.jsx
index 380f7153..507e0caf 100644
--- a/frontend/src/components/FloatingVideo.jsx
+++ b/frontend/src/components/FloatingVideo.jsx
@@ -4,6 +4,7 @@ import Draggable from 'react-draggable';
 import useVideoStore from '../store/useVideoStore';
 import useAuthStore from '../store/auth';
 import mpegts from 'mpegts.js';
+import Hls from 'hls.js';
 import { CloseButton, Flex, Loader, Text, Box } from '@mantine/core';
 import {
   applyConstraints,
@@ -118,7 +119,6 @@ export default function FloatingVideo() {
   const contentType = useVideoStore((s) => s.contentType);
   const metadata = useVideoStore((s) => s.metadata);
   const hideVideo = useVideoStore((s) => s.hideVideo);
-  const accessToken = useAuthStore((s) => s.accessToken);
 
   const videoRef = useRef(null);
   const playerRef = useRef(null);
@@ -241,10 +241,46 @@ export default function FloatingVideo() {
     const { volume: savedVolume, muted: savedMuted } = getPlayerPrefs();
     if (typeof savedVolume === 'number') video.volume = savedVolume;
     if (typeof savedMuted === 'boolean') video.muted = savedMuted;
+    // Always start playback from the beginning of the seekable range.
+    let hasSeekedToStart = false;
+    const seekToStart = () => {
+      if (hasSeekedToStart) return;
+      try {
+        let target = 0;
+        if (video.seekable && video.seekable.length > 0) {
+          target = video.seekable.start(0);
+        }
+        // Only apply if we're not already at/near the start.  Avoid
+        // setting currentTime when the video has no duration yet.
+        if (
+          Number.isFinite(target) &&
+          Math.abs((video.currentTime || 0) - target) > 0.25
+        ) {
+          video.currentTime = target;
+        }
+        hasSeekedToStart = true;
+      } catch {
+        // ignore
+      }
+    };
+
     const handleLoadStart = () => setIsLoading(true);
-    const handleLoadedData = () => setIsLoading(false);
+    const handleLoadedMetadata = () => {
+      seekToStart();
+    };
+    const handleLoadedData = () => {
+      setIsLoading(false);
+      // hls.js applies its `startPosition` after MEDIA_ATTACHED, which can
+      // run later than `loadedmetadata`. Re-seek here as a safety net so a
+      // hls.js live playlist doesn't snap to the live edge after our first
+      // seek attempt happened against an empty seekable range.
+      seekToStart();
+    };
     const handleCanPlay = () => {
       setIsLoading(false);
+      // Final fallback for the Safari native-HLS path where seekable.start(0)
+      // is sometimes only valid by the time `canplay` fires.
+      seekToStart();
       // Auto-play for VOD content
       video.play().catch((e) => {
         console.log('Auto-play prevented:', e);
@@ -274,23 +310,114 @@ export default function FloatingVideo() {
 
     // Add event listeners
     video.addEventListener('loadstart', handleLoadStart);
+    video.addEventListener('loadedmetadata', handleLoadedMetadata);
     video.addEventListener('loadeddata', handleLoadedData);
     video.addEventListener('canplay', handleCanPlay);
     video.addEventListener('error', handleError);
     video.addEventListener('progress', handleProgress);
 
-    // Set the source
-    video.src = streamUrl;
-    video.load();
+    // HLS handling: in-progress recordings expose .m3u8 playlists (and ended
+    // recordings whose MKV concat hasn't completed yet still serve via /hls/).
+    // Native <video src="*.m3u8"> only works on Safari/iOS, and even there it
+    // cannot follow a growing playlist with a useful seekable window.  Use
+    // hls.js when supported so the user gets a full DVR/timeshift window
+    // across all browsers.  The HLS playlist uses #EXT-X-PLAYLIST-TYPE=EVENT-
+    // style growth (omit_endlist + hls_list_size 0), so hls.js will treat the
+    // entire recorded duration as the seekable range while the recording is
+    // still in progress and as a complete VOD once it ends.
+    const isHls =
+      typeof streamUrl === 'string' &&
+      (streamUrl.includes('.m3u8') ||
+        streamUrl.includes('/hls/index') ||
+        streamUrl.includes('application/vnd.apple.mpegurl'));
+    let hls = null;
+
+    if (isHls && Hls.isSupported()) {
+      hls = new Hls({
+        // Open at the very beginning of the recording rather than the live
+        // edge.  Without this, an in-progress recording would start at "now"
+        // and hide everything already recorded.  hls.js applies this AFTER
+        // MEDIA_ATTACHED, so the listener-driven `seekToStart()` above is
+        // also kept as a safety net for the Safari native-HLS path and for
+        // edge cases where this initial-position logic loses to the user's
+        // first interaction.
+        startPosition: 0,
+        // Allow seeking back to the start of the recording, regardless of
+        // current playhead position.  Recordings can be hours long and the
+        // user may want to scrub anywhere; we explicitly disable buffer
+        // eviction by setting a very large back-buffer length.
+        backBufferLength: 90 * 60, // 90 minutes
+        maxBufferLength: 60,
+        maxMaxBufferLength: 600,
+        // For an in-progress recording, hls.js refreshes the playlist on
+        // its target-duration cadence; let it follow the live edge but keep
+        // the full DVR window seekable.
+        liveSyncDurationCount: 3,
+        liveMaxLatencyDurationCount: 10,
+        enableWorker: true,
+        lowLatencyMode: false,
+        // Inject the JWT into every playlist + segment XHR.  Read the token
+        // from the auth store at request time rather than capturing the
+        // closure value at hls.js init, so a refreshed access token mid-
+        // playback is picked up on the next segment fetch.
+        xhrSetup: (xhr) => {
+          const token = useAuthStore.getState().accessToken;
+          if (token) {
+            xhr.setRequestHeader('Authorization', `Bearer ${token}`);
+          }
+        },
+      });
+      hls.on(Hls.Events.ERROR, (_evt, data) => {
+        if (data.fatal) {
+          // eslint-disable-next-line no-console
+          console.error('HLS fatal error:', data.type, data.details);
+          if (data.type === Hls.ErrorTypes.NETWORK_ERROR) {
+            try {
+              hls.startLoad();
+            } catch {
+              // ignore
+            }
+          } else if (data.type === Hls.ErrorTypes.MEDIA_ERROR) {
+            try {
+              hls.recoverMediaError();
+            } catch {
+              // ignore
+            }
+          } else {
+            setLoadError(`HLS playback error: ${data.details || data.type}`);
+          }
+        }
+      });
+      hls.attachMedia(video);
+      hls.on(Hls.Events.MEDIA_ATTACHED, () => {
+        hls.loadSource(streamUrl);
+      });
+    } else if (isHls && video.canPlayType('application/vnd.apple.mpegurl')) {
+      // Safari path: native HLS support, including seekable DVR windows.
+      video.src = streamUrl;
+      video.load();
+    } else {
+      // Plain progressive file (MKV/MP4): native HTML5.
+      video.src = streamUrl;
+      video.load();
+    }
 
     // Store cleanup function
     playerRef.current = {
       destroy: () => {
         video.removeEventListener('loadstart', handleLoadStart);
+        video.removeEventListener('loadedmetadata', handleLoadedMetadata);
         video.removeEventListener('loadeddata', handleLoadedData);
         video.removeEventListener('canplay', handleCanPlay);
         video.removeEventListener('error', handleError);
         video.removeEventListener('progress', handleProgress);
+        if (hls) {
+          try {
+            hls.destroy();
+          } catch {
+            // ignore
+          }
+        }
         video.removeAttribute('src');
         video.load();
       },
@@ -321,6 +448,14 @@ export default function FloatingVideo() {
           ? `${window.location.origin}${streamUrl}`
           : streamUrl;
 
+      // Read the JWT from the auth store at player-creation time rather than
+      // relying on the closure-captured `accessToken` value.  mpegts.js has
+      // no per-request setup hook (unlike hls.js's xhrSetup), so this header
+      // is baked into the IO loader for the life of the player; we just want
+      // to be sure we use the freshest token available at the moment of
+      // connection rather than whatever React-render snapshot we closed over.
+      const liveAccessToken = useAuthStore.getState().accessToken;
+
       const player = mpegts.createPlayer(
         {
           type: 'mpegts',
@@ -337,8 +472,8 @@ export default function FloatingVideo() {
           autoCleanupMaxBackwardDuration: 120,
           autoCleanupMinBackwardDuration: 60,
           reuseRedirectedURL: true,
-          headers: accessToken
-            ? { Authorization: `Bearer ${accessToken}` }
+          headers: liveAccessToken
+            ? { Authorization: `Bearer ${liveAccessToken}` }
             : undefined,
         }
       );
diff --git a/frontend/src/components/cards/RecordingCard.jsx b/frontend/src/components/cards/RecordingCard.jsx
index 80b5ba80..97481deb 100644
--- a/frontend/src/components/cards/RecordingCard.jsx
+++ b/frontend/src/components/cards/RecordingCard.jsx
@@ -285,7 +285,9 @@ const RecordingCard = ({
       <Tooltip
         label={
           customProps.file_url || customProps.output_file_url
-            ? 'Watch recording'
+            ? isInProgress
+              ? 'Watch in progress recording'
+              : 'Watch recording'
             : 'Recording playback not available yet'
         }
       >
@@ -296,10 +298,7 @@ const RecordingCard = ({
             e.stopPropagation();
             handleWatchRecording();
           }}
-          disabled={
-            customProps.status === 'recording' ||
-            !(customProps.file_url || customProps.output_file_url)
-          }
+          disabled={!(customProps.file_url || customProps.output_file_url)}
         >
           Watch
         </Button>
diff --git a/frontend/src/components/forms/RecordingDetailsModal.jsx b/frontend/src/components/forms/RecordingDetailsModal.jsx
index 39643d27..bc8b245f 100644
--- a/frontend/src/components/forms/RecordingDetailsModal.jsx
+++ b/frontend/src/components/forms/RecordingDetailsModal.jsx
@@ -122,7 +122,8 @@ const RecordingDetailsModal = ({
   const canWatchRecording =
     (customProps.status === 'completed' ||
       customProps.status === 'stopped' ||
-      customProps.status === 'interrupted') &&
+      customProps.status === 'interrupted' ||
+      customProps.status === 'recording') &&
     Boolean(fileUrl);
 
   const isSeriesGroup = Boolean(