fix(timeshift): enhance session handling and playback logic

Updated the session handling mechanism to improve user experience during reconnects. The first request without a `session_id` now receives a `301` redirect with a minted `session_id`, while reconnects that omit `session_id` but match an existing pool entry are served immediately without a redirect. Additionally, refined playback logic for plain GET requests to restart from byte 0, aligning with provider behavior. Updated tests to reflect these changes and ensure proper session reuse and playback anchoring.

CHANGELOG.md

@@ -18,7 +18,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - **Strictly-UTC API surface, automatic per-provider timezone.** `server_info.timezone` is always `UTC` and the XC EPG `start`/`end` strings are emitted in UTC; the proxy converts the requested instant to the serving provider's own reported timezone (`server_info.timezone` captured on account refresh) at request time. No timezone to configure.
   - **Multi-provider failover.** Catch-up walks the channel's catch-up streams in order (like live playback): if one provider cannot serve the archive the next is tried, each attempt with its own account context (credentials, reported timezone, user-agent). Accounts that fail with an auth/ban-class status are not retried via their other streams.
   - **Provider pool accounting.** Catch-up reserves a provider profile slot (`connection_pool`) before connecting upstream — same contract as live and VOD — and releases it when the session ends. When the default profile is at capacity it walks the account's alternate profiles with their own resolved credentials, and answers `503` when every profile is full.
-  - **Per-client session pool.** The first request without `session_id` receives a `301` redirect to the same URL with a stable `?session_id=` query parameter; that ID becomes the client identity for stats, stop keys, and pool coordination. Each viewer gets their own Redis-backed pool entry even when watching the same programme; idle sessions can be reclaimed via fingerprint matching (same user and programme, IP + user-agent score). Real scrubs within a session stop the in-flight stream through the standard stop-key mechanism; parallel startup probes (full-file, open-ended, and EOF tail `Range` requests) are deferred with `503` instead of opening extra upstream connections. Parallel HTTP probes from the same `session_id` for the same programme do not count as extra streams toward the user limit; distinct sessions still each consume a slot.
+  - **Per-client session pool.** The first request without `session_id` and no fingerprint match receives a `301` redirect to the same URL with a stable `?session_id=` query parameter; that ID becomes the client identity for stats, stop keys, and pool coordination. Reconnects that omit `session_id` but match an existing pool entry (same user, IP, and user-agent) are served immediately without a redirect, matching IPTV client fast-forward behaviour. Each viewer gets their own Redis-backed pool entry even when watching the same programme; idle sessions can be reclaimed via fingerprint matching. Plain GET restarts stream from byte 0 with upstream `Content-Length` when known (provider-faithful). Real scrubs within a session stop the in-flight stream through the standard stop-key mechanism; parallel startup probes (full-file, open-ended, and EOF tail `Range` requests) are deferred with `503` instead of opening extra upstream connections. Parallel HTTP probes from the same `session_id` for the same programme do not count as extra streams toward the user limit; distinct sessions still each consume a slot.
   - **`xmltv_prev_days_override` under `Settings → EPG`** (default `0` = auto-detect). Verbose timeshift logging follows the standard logger DEBUG level.
   - **EPG XMLTV `prev_days` auto-detection** from the provider's largest `tv_archive_duration` (capped at 30 days), overridable via setting or `?prev_days=` URL parameter.
   - **Byte path streams directly via `iter_content` + generator yield** (same pattern as the VOD proxy), with an upfront MPEG-TS sync peek that strips any PHP-warning preamble before bytes reach strict demuxers. Throughput stays comfortably above the typical 5 Mbps FHD bitrate so clients don't buffer.

apps/timeshift/stats.py

@@ -178,14 +178,9 @@ def resolve_stats_playback_fields(
     if range_start == 0:
         return None, now
 
-    if existing_programme_start == timestamp_utc and existing_position_anchor:
-        existing_base = None
-        if existing_playback_base is not None:
-            try:
-                existing_base = float(existing_playback_base)
-            except (TypeError, ValueError):
-                existing_base = None
-        return existing_base, existing_position_anchor
+    # Plain GET (no Range): providers restart from byte 0 on reconnect.
+    if range_start is None:
+        return None, now
 
     return None, now
 

apps/timeshift/tests/test_stats.py

@@ -130,7 +130,7 @@ class ByteRangePlaybackTests(TestCase):
         self.assertAlmostEqual(base, 2371.0, delta=2.0)
         self.assertEqual(anchor, "2000.0")
 
-    def test_resolve_same_programme_preserves_without_range(self):
+    def test_resolve_plain_get_reanchors_on_same_programme(self):
         base, anchor = resolve_stats_playback_fields(
             timestamp_utc="2026-06-08:17-00",
             existing_programme_start="2026-06-08:17-00",
@@ -141,8 +141,8 @@ class ByteRangePlaybackTests(TestCase):
             programme_duration_secs=3900,
             now="2000.0",
         )
-        self.assertAlmostEqual(base, 900.0)
-        self.assertEqual(anchor, "1000.0")
+        self.assertIsNone(base)
+        self.assertEqual(anchor, "2000.0")
 
 
 class TimeshiftStreamStatsTests(TestCase):

apps/timeshift/tests/test_views.py

@@ -2154,7 +2154,7 @@ class TimeshiftSessionReuseTests(TestCase):
 
 
 class TimeshiftSessionRedirectTests(TestCase):
-    """First request must establish a session via 301 redirect (VOD-style)."""
+    """Session mint (301) and inline pool adopt when session_id is omitted."""
 
     def setUp(self):
         self.factory = RequestFactory()
@@ -2178,7 +2178,7 @@ class TimeshiftSessionRedirectTests(TestCase):
         self.assertEqual(response.status_code, 301)
         self.assertIn("session_id=", response["Location"])
 
-    def test_missing_session_id_redirects_to_existing_busy_pool(self):
+    def test_missing_session_id_serves_existing_busy_pool_without_redirect(self):
         existing = "existingbusy1"
         redis = _FakeRedis()
         _seed_pool_session(
@@ -2194,6 +2194,9 @@ class TimeshiftSessionRedirectTests(TestCase):
             HTTP_USER_AGENT="vlc-test",
             REMOTE_ADDR="127.0.0.1",
         )
+        ok = MagicMock(status_code=200)
+        profile = MagicMock(id=31)
+        descriptor = {"account_id": "1", "stream_id": "111", "profile_id": "31"}
         with patch.object(views, "_authenticate_user", return_value=MagicMock(id=1)), \
              patch.object(views, "network_access_allowed", return_value=True), \
              patch.object(views, "Channel") as channel_cls, \
@@ -2202,16 +2205,23 @@ class TimeshiftSessionRedirectTests(TestCase):
                           return_value=[_make_catchup_stream()]), \
              patch.object(views, "get_programme_duration", return_value=40), \
              patch.object(views, "parse_catchup_timestamp", return_value=True), \
-             patch.object(views, "RedisClient") as redis_cls:
+             patch.object(views, "check_user_stream_limits", return_value=True), \
+             patch.object(views, "RedisClient") as redis_cls, \
+             patch.object(
+                 views, "_try_reacquire_idle_pool",
+                 return_value=(descriptor, profile),
+             ) as reacquire_mock, \
+             patch.object(views, "_stream_reused_session", return_value=ok) as reuse_mock:
             redis_cls.get_client.return_value = redis
             channel_cls.objects.get.return_value = MagicMock(id=8)
             response = views.timeshift_proxy(
                 request, "u", "p", "8", "2026-06-08:17-00", "8.ts",
             )
-        self.assertEqual(response.status_code, 301)
-        self.assertIn(f"session_id={existing}", response["Location"])
+        self.assertIs(response, ok)
+        reacquire_mock.assert_called_once()
+        reuse_mock.assert_called_once()
 
-    def test_missing_session_id_redirects_to_busy_pool_on_channel_hop(self):
+    def test_missing_session_id_serves_busy_pool_on_channel_hop_without_redirect(self):
         existing = "channelhop1"
         redis = _FakeRedis()
         _seed_pool_session(
@@ -2228,6 +2238,9 @@ class TimeshiftSessionRedirectTests(TestCase):
             HTTP_USER_AGENT="vlc-test",
             REMOTE_ADDR="127.0.0.1",
         )
+        ok = MagicMock(status_code=200)
+        profile = MagicMock(id=31)
+        descriptor = {"account_id": "1", "stream_id": "111", "profile_id": "31"}
         with patch.object(views, "_authenticate_user", return_value=MagicMock(id=1)), \
              patch.object(views, "network_access_allowed", return_value=True), \
              patch.object(views, "Channel") as channel_cls, \
@@ -2236,14 +2249,21 @@ class TimeshiftSessionRedirectTests(TestCase):
                           return_value=[_make_catchup_stream()]), \
              patch.object(views, "get_programme_duration", return_value=40), \
              patch.object(views, "parse_catchup_timestamp", return_value=True), \
-             patch.object(views, "RedisClient") as redis_cls:
+             patch.object(views, "check_user_stream_limits", return_value=True), \
+             patch.object(views, "RedisClient") as redis_cls, \
+             patch.object(
+                 views, "_try_reacquire_idle_pool",
+                 return_value=(descriptor, profile),
+             ) as reacquire_mock, \
+             patch.object(views, "_stream_reused_session", return_value=ok) as reuse_mock:
             redis_cls.get_client.return_value = redis
             channel_cls.objects.get.return_value = MagicMock(id=8)
             response = views.timeshift_proxy(
                 request, "u", "p", "8", "2026-06-08:17-30", "8.ts",
             )
-        self.assertEqual(response.status_code, 301)
-        self.assertIn(f"session_id={existing}", response["Location"])
+        self.assertIs(response, ok)
+        reacquire_mock.assert_called_once()
+        reuse_mock.assert_called_once()
 
     def test_redirect_preserves_existing_query_params(self):
         request = self.factory.get(
@@ -2692,12 +2712,13 @@ class TimeshiftStatsClientTests(TestCase):
 
         self.assertNotIn(old_gen, self.redis.store)
 
-    def test_register_stats_preserves_position_anchor_on_same_programme(self):
+    def test_register_stats_reanchors_position_on_plain_get_reconnect(self):
         from apps.timeshift.redis_keys import TimeshiftRedisKeys as RedisKeys, TimeshiftRedisKeys
 
         client_key = RedisKeys.client_metadata(self.stats_channel_id, self.client_id)
         self.redis.hset(client_key, "programme_start", "2026-06-08:17-00")
         self.redis.hset(client_key, "position_anchor_at", "1000.0")
+        self.redis.hset(client_key, "playback_base_secs", "900.0")
 
         views._register_stats_client(
             self.redis,
@@ -2713,7 +2734,8 @@ class TimeshiftStatsClientTests(TestCase):
             channel_id=8,
             channel_uuid="00000000-0000-0000-0000-000000000008",
         )
-        self.assertEqual(self.redis.hget(client_key, "position_anchor_at"), "1000.0")
+        self.assertNotEqual(self.redis.hget(client_key, "position_anchor_at"), "1000.0")
+        self.assertIsNone(self.redis.hget(client_key, "playback_base_secs"))
 
     def test_register_stats_byte_range_seek_sets_playback_base(self):
         from apps.timeshift.redis_keys import TimeshiftRedisKeys as RedisKeys, TimeshiftRedisKeys
@@ -3922,6 +3944,45 @@ class TimeshiftScrubPreemptTests(TestCase):
             wait_seconds=views._POOL_SCRUB_WAIT_SECONDS,
         )
 
+    def test_plain_reconnect_preempts_busy_pool_without_range(self):
+        """TiviMate FF reconnects with plain GET; match provider byte-0 restart."""
+        _seed_pool_session(self.redis, session_id=TEST_SESSION_ID)
+        request = self.factory.get(_proxy_url(TEST_SESSION_ID))
+        streams = [_make_catchup_stream(account_id=1, stream_id="111", profile_id=31)]
+        ok = MagicMock(status_code=200)
+        profile = MagicMock(id=31)
+        descriptor = {"account_id": "1", "stream_id": "111", "profile_id": "31"}
+        with patch.object(views, "_authenticate_user", return_value=MagicMock(id=5)), \
+             patch.object(views, "network_access_allowed", return_value=True), \
+             patch.object(views, "Channel") as channel_cls, \
+             patch.object(views, "_user_can_access_channel", return_value=True), \
+             patch.object(views, "get_channel_catchup_streams", return_value=streams), \
+             patch.object(views, "get_programme_duration", return_value=40), \
+             patch.object(views, "check_user_stream_limits", return_value=True), \
+             patch.object(views, "RedisClient") as redis_cls, \
+             patch.object(views, "reserve_profile_slot", return_value=(True, 1, None)), \
+             patch.object(views, "release_profile_slot"), \
+             patch.object(views, "get_transformed_credentials", side_effect=_fake_creds), \
+             patch.object(views, "get_user_active_connections", return_value=[]), \
+             patch.object(
+                 views, "_try_reacquire_idle_pool",
+                 return_value=(descriptor, profile),
+             ) as reacquire_mock, \
+             patch.object(views, "_stream_reused_session", return_value=ok) as reuse_mock, \
+             patch.object(views, "_attempt_timeshift_stream") as attempt_mock:
+            redis_cls.get_client.return_value = self.redis
+            channel_cls.objects.get.return_value = MagicMock(
+                id=8, name="Test", logo_id=None,
+            )
+            response = views.timeshift_proxy(
+                request, "u", "p", "8", "2026-06-08:17-00", "8.ts",
+            )
+        self.assertIs(response, ok)
+        reacquire_mock.assert_called_once()
+        reuse_mock.assert_called_once()
+        self.assertIsNone(reuse_mock.call_args.kwargs["range_header"])
+        attempt_mock.assert_not_called()
+
     def test_busy_pool_reuses_slot_after_scrub_preempt(self):
         _seed_pool_session(self.redis, session_id=TEST_SESSION_ID)
         request = self.factory.get(
@@ -4326,6 +4387,87 @@ class CatchupProxyTests(TestCase):
         serve.assert_called_once()
 
 
+class TimeshiftProviderFaithfulPlainGetTests(_ProxyLoopTestMixin, TestCase):
+    """Contract tests for TiviMate-style plain GET reconnect (no Range header)."""
+
+    def setUp(self):
+        self.redis = _FakeRedis()
+        self.virtual_channel_id = f"{TEST_MEDIA_ID}_111"
+        self.stats_channel_id = views.stats_channel_id(8, TEST_SESSION_ID)
+        self.client_id = TEST_SESSION_ID
+        self.user = MagicMock(id=5, username="viewer")
+
+    @patch.object(views, "_trigger_timeshift_stats_update")
+    @patch.object(views, "_open_upstream")
+    def test_plain_get_does_not_send_range_to_upstream(self, mocked_open, _trigger_mock):
+        ts = _make_ts_payload(188 * 7)
+        mocked_open.return_value = _fake_upstream(200, body=ts)
+
+        views._stream_from_provider(
+            candidate_urls=["http://example.test/timeshift.ts"],
+            user_agent="provider-agent",
+            range_header=None,
+            virtual_channel_id=self.virtual_channel_id,
+            stats_channel_id=self.stats_channel_id,
+            client_id=self.client_id,
+            client_ip="1.2.3.4",
+            client_user_agent="tivimate",
+            user=self.user,
+            channel_display_name="A&E",
+            timestamp_utc="2026-06-08:17-00",
+            channel_logo_id=None,
+            m3u_profile_id=31,
+            debug=False,
+            account_id=None,
+            redis_client=self.redis,
+            pool_session_id=self.client_id,
+            channel_id=8,
+            channel_uuid="00000000-0000-0000-0000-000000000008",
+            duration_minutes=40,
+        )
+
+        mocked_open.assert_called_once()
+        self.assertIsNone(mocked_open.call_args.args[2])
+
+    @patch.object(views, "_trigger_timeshift_stats_update")
+    @patch.object(views, "_open_upstream")
+    def test_plain_get_passes_through_provider_200_without_content_range(
+        self, mocked_open, _trigger_mock,
+    ):
+        ts = _make_ts_payload(188 * 7)
+        upstream = _fake_upstream(200, body=ts)
+        upstream.headers["Content-Length"] = str(len(ts))
+        mocked_open.return_value = upstream
+
+        response = views._stream_from_provider(
+            candidate_urls=["http://example.test/timeshift.ts"],
+            user_agent="provider-agent",
+            range_header=None,
+            virtual_channel_id=self.virtual_channel_id,
+            stats_channel_id=self.stats_channel_id,
+            client_id=self.client_id,
+            client_ip="1.2.3.4",
+            client_user_agent="tivimate",
+            user=self.user,
+            channel_display_name="A&E",
+            timestamp_utc="2026-06-08:17-00",
+            channel_logo_id=None,
+            m3u_profile_id=31,
+            debug=False,
+            account_id=None,
+            redis_client=self.redis,
+            pool_session_id=self.client_id,
+            channel_id=8,
+            channel_uuid="00000000-0000-0000-0000-000000000008",
+            duration_minutes=40,
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertNotIn("Content-Range", response)
+        self.assertEqual(response["Accept-Ranges"], "bytes")
+        self.assertEqual(response["Content-Length"], str(len(ts)))
+
+
 class TimeshiftDownstreamLengthHeaderTests(TestCase):
     def test_open_ended_range_passthrough_upstream_headers(self):
         headers = views._build_downstream_length_headers(
@@ -4372,7 +4514,7 @@ class TimeshiftDownstreamLengthHeaderTests(TestCase):
         self.assertEqual(headers["Content-Length"], "10000")
         self.assertNotIn("Content-Range", headers)
 
-    def test_streaming_full_file_omits_content_length(self):
+    def test_streaming_plain_get_includes_content_length(self):
         headers = views._build_downstream_length_headers(
             range_header=None,
             status_code=200,
@@ -4381,8 +4523,9 @@ class TimeshiftDownstreamLengthHeaderTests(TestCase):
             upstream_content_length="10000",
             streaming=True,
         )
-        self.assertNotIn("Content-Length", headers)
+        self.assertEqual(headers["Content-Length"], "10000")
         self.assertEqual(headers["Accept-Ranges"], "bytes")
+        self.assertNotIn("Content-Range", headers)
 
     def test_passthrough_includes_accept_ranges(self):
         response = views._passthrough_response(416, "bytes */10000")

apps/timeshift/views.py

@@ -103,6 +103,12 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration)
         ``duration``: Dispatcharr ``Channel.id`` (XC API exposes channel.id as stream_id).
         ``timestamp``: UTC programme start (``YYYY-MM-DD:HH-MM`` or XC colon form
             ``YYYY-MM-DD:HH:MM:SS``).
+
+    Session handling (``?session_id=``):
+        First request with no ``session_id`` and no matching pool entry → ``301`` with a
+        minted ``session_id``. Reconnects that omit ``session_id`` but fingerprint-match
+        an in-flight or idle pool entry for the same viewer are served immediately
+        (no redirect). Reuse ``session_id`` for all range/seek requests in a programme.
     """
     raw_id = duration[:-3] if duration.endswith(".ts") else duration
 
@@ -137,8 +143,15 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration)
         "time) plus ``Authorization: Bearer``, ``X-API-Key``, or "
         "``?token=<jwt>``. Optionally include a client ``session_id`` for "
         "provider pooling.\n\n"
-        "If ``session_id`` is omitted on a direct-auth request, the server "
-        "responds with **301** adding a server-minted ``session_id``."
+        "**``session_id`` without a matching pool:** the server responds with "
+        "**301** and a minted ``session_id`` (direct-auth / first play only).\n\n"
+        "**``session_id`` omitted but pool match:** when the same viewer "
+        "reconnects (e.g. IPTV fast-forward) and fingerprint-matches an "
+        "existing pool entry, the stream is served on that request with no "
+        "redirect round trip.\n\n"
+        "**Plain GET (no ``Range``):** upstream archive is streamed from byte "
+        "0 with ``Content-Length`` when the provider reports file size "
+        "(provider-faithful behaviour for clients such as TiviMate)."
     ),
     parameters=[
         OpenApiParameter(
@@ -147,10 +160,11 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration)
             location=OpenApiParameter.QUERY,
             required=False,
             description=(
-                "Playback session from ``POST /api/catchup/sessions/``. "
-                "When present, JWT is optional. The session authenticates "
-                "the player. Reuse for all range/seek requests during the "
-                "same programme."
+                "Playback session from ``POST /api/catchup/sessions/`` or a "
+                "prior **301** redirect. When present, JWT is optional. Reuse "
+                "for all range/seek requests during the same programme. "
+                "May be omitted on reconnect when the server adopts a "
+                "fingerprint-matched pool entry."
             ),
         ),
         OpenApiParameter(
@@ -176,11 +190,18 @@ def timeshift_proxy(request, username, password, stream_id, timestamp, duration)
         ),
     ],
     responses={
-        200: {"description": "MPEG-TS stream (may be partial content for Range requests)."},
+        200: {
+            "description": (
+                "MPEG-TS stream. Partial content (``206``) when the client "
+                "sent ``Range``. Full-file ``200`` on plain GET includes "
+                "``Content-Length`` when the upstream archive size is known."
+            ),
+        },
         301: {
             "description": (
                 "Redirect to the same URL with a server-minted ``session_id`` "
-                "when the client did not supply one (direct-auth flow only)."
+                "when the client did not supply one and no fingerprint-matched "
+                "pool entry exists (first play / direct-auth only)."
             ),
         },
         401: {"description": "Missing or expired authentication / session."},
@@ -278,15 +299,15 @@ def _serve_catchup(request, user, channel, timestamp):
             include_busy=True,
         )
         if matched:
-            logger.debug(
-                "Timeshift session redirect: reusing %s for %s",
-                matched, request.path,
-            )
-            return _finalize_timeshift_response(
-                _redirect_with_session(request, matched),
-            )
-        logger.debug("Timeshift session redirect: %s (new session)", request.path)
-        return _finalize_timeshift_response(_redirect_with_new_session(request))
+            if debug:
+                logger.debug(
+                    "Timeshift session adopt: reusing %s for %s",
+                    matched, request.path,
+                )
+            session_id = matched
+        else:
+            logger.debug("Timeshift session redirect: %s (new session)", request.path)
+            return _finalize_timeshift_response(_redirect_with_new_session(request))
 
     session_entry = _get_pool_entry(redis_client, session_id)
     if session_entry and not _pool_entry_owned_by_user(session_entry, user.id):
@@ -391,6 +412,17 @@ def _serve_catchup(request, user, channel, timestamp):
                 redis_client, effective_session_id,
                 user_id=user.id, user=user,
             )
+        elif _should_preempt_plain_reconnect(
+            range_header,
+            pool_exists=pool_exists,
+            pool_busy=pool_busy,
+            pool_media_id=pool_media_id,
+            media_id=media_id,
+        ):
+            acquired = _try_reacquire_idle_pool(
+                redis_client, effective_session_id,
+                user_id=user.id, user=user,
+            )
         elif scrub_displacement:
             acquired = _try_reacquire_idle_pool(
                 redis_client, effective_session_id,
@@ -634,7 +666,8 @@ def _user_can_access_channel(user, channel):
     )
 
 
-# Per-client pool (session_id from 301 redirect). Fingerprint-match when ?session_id= is dropped.
+# Per-client pool (session_id from 301 redirect or inline adopt when ?session_id=
+# is dropped on reconnect). Fingerprint-match when ?session_id= is absent.
 _POOL_ENTRY_TTL = 10 * 60  # Refreshed on playback GET and active-stream heartbeats.
 _STREAM_READ_INACTIVITY_SECONDS = CLIENT_TTL_SECONDS  # Shorter than pool TTL; data should flow.
 _POOL_IDLE_TTL = CLIENT_TTL_SECONDS  # Must match CLIENT_TTL_SECONDS (stats + fingerprint agree).
@@ -1102,7 +1135,16 @@ def _build_downstream_length_headers(
             headers["Content-Length"] = str(up_end - up_start + 1)
         return headers
 
-    # Omit Content-Length on streaming full-file responses (seeks may preempt).
+    # Plain GET streaming 200: match provider/CDN Content-Length (TiviMate uses
+    # it for archive duration; omitting it breaks FF reconnect playback).
+    if streaming and status_code == 200 and not range_header:
+        if representation_length is not None:
+            headers["Content-Length"] = str(representation_length)
+        elif upstream_content_length:
+            headers["Content-Length"] = str(upstream_content_length)
+        return headers
+
+    # Omit Content-Length on other streaming full-file responses (seeks may preempt).
     if not streaming:
         if representation_length is not None:
             headers["Content-Length"] = str(representation_length)
@@ -1177,6 +1219,24 @@ def _try_reacquire_idle_pool(
     )
 
 
+def _should_preempt_plain_reconnect(
+    range_header, *, pool_exists, pool_busy, pool_media_id, media_id,
+):
+    """True when a plain GET should restart playback like the provider does.
+
+    TiviMate fast-forward closes the HTTP connection and reopens the same
+    programme URL without a Range header. Providers answer that with a full
+    200 from byte 0, not an internal byte-range resume.
+    """
+    if range_header:
+        return False
+    if not pool_exists or not pool_busy:
+        return False
+    if pool_media_id is not None and str(pool_media_id) != str(media_id):
+        return False
+    return True
+
+
 def _should_displace_busy_playback(
     range_header, content_length=None, busy_serving_range=None,
 ):