diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2622dfc2..4de12ded 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -43,9 +43,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
-- **Live proxy channels could remain running with no clients after disconnect.** `stop_channel` called `log_system_event('channel_stop')` synchronously before local cleanup and Redis key deletion; Connect integrations or DB event writes on that path could block teardown indefinitely while the cleanup thread kept refreshing metadata TTL for a stale `stream_buffers` entry. Teardown now stops ffmpeg/output managers, releases ownership, removes local buffers and managers, deletes Redis keys, and only then logs the stop event asynchronously. Metadata TTL refresh skips channels mid-shutdown and non-owned channels; a stuck-stop watchdog forces cleanup if `stop_channel` does not return within ~30s. Stream buffers also ignore late `add_chunk()` writes once shutdown begins (`buffer.stopping` set at the start of `stream_manager.stop()`).
+- **Live proxy channels could remain running with no clients after disconnect.** `stop_channel` called `log_system_event('channel_stop')` synchronously before local cleanup and Redis key deletion; Connect integrations or DB event writes on that path could block teardown indefinitely while the cleanup thread kept refreshing metadata TTL for a stale `stream_buffers` entry. Teardown now signals shutdown (`buffer.stopping`), runs model `release_stream()` and Redis key deletion, releases ownership, stops ffmpeg/output managers and local buffers, and only then logs the stop event asynchronously. Metadata TTL refresh skips channels mid-shutdown and non-owned channels; a stuck-stop watchdog forces cleanup if `stop_channel` does not return within ~10s (or 2× `channel_shutdown_delay`, whichever is greater). Stream buffers ignore late `add_chunk()` writes once shutdown begins (`buffer.stopping` set at teardown start).
 - **Client reconnect during channel teardown could wedge the channel across uWSGI workers.** The disconnect path called `proxy_server.stop_channel()` directly, so other workers were not notified and reconnecting clients could attach while ownership was released on the owner worker. The cleanup thread then re-acquired expired ownership and looped on orphaned clients while the upstream connection stayed open. All lifecycle stops now go through coordinated Redis teardown (`channel_stopping` flag, metadata `stopping` state, and `CHANNEL_STOP` pubsub). New stream requests are rejected with `503 Retry-After` during pending shutdown or active teardown; `initialize_channel()` and ownership re-acquisition refuse to proceed in those states; non-owner workers clean local resources only on pubsub; and orphaned-client sweeps force cleanup when teardown is already active. (Fixes #1342)
 - **Orphaned upstream threads kept writing Redis after teardown.** Partial cleanup could remove a channel from `stream_managers` while the `stream-{uuid}` OS thread kept running, so orphan metadata sweeps only deleted Redis keys and chunks/metadata were immediately recreated (bare `total_bytes` hash with TTL -1). Stream managers now stop upstream when ownership is lost, orphan cleanup stops local ffmpeg/stream threads even when registry entries are gone (`_live_stream_managers` tracks managers until the thread exits), and forced recovery stops processes before Redis deletion instead of dropping dict entries without calling `stop()`.
+- **Rapid channel switching could leave bare `buffer:index` keys in Redis.** `buffer.stop()` flushed a final partial chunk via `INCR` during teardown, creating a persistent index key (TTL -1) after chunk keys expired; under load overlapping disconnect and cleanup-thread stops could race on `_stop_local_stream_activity`, blocking on ffmpeg stderr join before `_clean_redis_keys` ever ran. Teardown no longer writes to Redis from `buffer.stop()`, disconnect uses coordinated stop only (no duplicate upstream stop), model `release_stream()` runs before Redis keys are scanned/deleted (so `profile_connections` counters are not left stuck), ownership is released after Redis cleanup but before the blocking local ffmpeg stop, and a `finally` block guarantees Redis cleanup if local teardown raises.
+- **Zombie ffmpeg could survive owner-lock expiry and orphan Redis sweeps.** When the 30s owner lock lapsed under single-worker load, disconnect handling treated the worker as non-owner so coordinated stop never ran, while ffmpeg kept writing and the orphan sweeper only deleted Redis keys (recreating them immediately). Disconnect now re-acquires ownership when local upstream is still active, stops locally when the last client leaves without a lock, orphan cleanup stops local ffmpeg before Redis deletion via `_has_local_upstream_activity`, re-init stops lingering upstream before starting a duplicate thread, and `is_channel_teardown_active` includes channels mid-`stop_channel` on this worker so rapid reconnect gets 503 during teardown.
 - **Stale `channel_stream` Redis keys after a channel stopped could skip connection accounting on retune.** On `dev`, `get_stream()` reused any existing `channel_stream` / `stream_profile` assignment without reserving a new slot. If those keys were left behind after a stop, the next tune-in could reach the provider without incrementing Redis counters. `get_stream()` now releases stale assignments when proxy metadata shows the channel is inactive, then reserves fresh slots.
 - **EPG auto-match reliability fixes.**
   - Memory could spike to multiple GB on large EPG sources when building a full in-memory catalog before fuzzy matching; single-channel matching now streams rows and bounds ML work to a small candidate set.
diff --git a/apps/channels/models.py b/apps/channels/models.py
index f900a817..ef1d0cff 100644
--- a/apps/channels/models.py
+++ b/apps/channels/models.py
@@ -626,11 +626,11 @@ class Channel(models.Model):
 
     def _release_stale_stream_assignment(self, redis_client, stream_id: int) -> None:
         """Release pool counters and remove stale channel/stream assignment keys."""
+        profile_id = None
         profile_id_bytes = redis_client.get(f"stream_profile:{stream_id}")
         if profile_id_bytes:
             try:
                 profile_id = int(profile_id_bytes)
-                release_profile_slot(profile_id, redis_client)
             except (ValueError, TypeError):
                 logger.debug(
                     "Invalid profile ID for stale assignment on stream %s: %s",
@@ -638,6 +638,32 @@ class Channel(models.Model):
                     profile_id_bytes,
                 )
 
+        if profile_id is None:
+            metadata_key = RedisKeys.channel_metadata(str(self.uuid))
+            meta_profile_id = redis_client.hget(
+                metadata_key, ChannelMetadataField.M3U_PROFILE
+            )
+            if meta_profile_id:
+                try:
+                    profile_id = int(meta_profile_id)
+                except (ValueError, TypeError):
+                    logger.debug(
+                        "Invalid profile ID in metadata for stale assignment on "
+                        "channel %s: %s",
+                        self.uuid,
+                        meta_profile_id,
+                    )
+
+        if profile_id is not None:
+            release_profile_slot(profile_id, redis_client)
+        else:
+            logger.warning(
+                "Channel %s: releasing stale stream %s assignment without profile "
+                "info - profile_connections may leak",
+                self.uuid,
+                stream_id,
+            )
+
         redis_client.delete(f"channel_stream:{self.id}")
         redis_client.delete(f"stream_profile:{stream_id}")
 
diff --git a/apps/channels/tests/test_ts_proxy_teardown.py b/apps/channels/tests/test_ts_proxy_teardown.py
index 14b4e898..3bc92f18 100644
--- a/apps/channels/tests/test_ts_proxy_teardown.py
+++ b/apps/channels/tests/test_ts_proxy_teardown.py
@@ -5,6 +5,7 @@ from unittest.mock import MagicMock, patch
 from django.test import TestCase
 
 from apps.proxy.live_proxy.constants import ChannelMetadataField, ChannelState
+from apps.proxy.live_proxy.input.buffer import StreamBuffer
 from apps.proxy.live_proxy.input.manager import StreamManager
 from apps.proxy.live_proxy.redis_keys import RedisKeys
 from apps.proxy.live_proxy.server import ProxyServer
@@ -14,6 +15,29 @@ from apps.proxy.live_proxy.services.channel_service import ChannelService
 CHANNEL_ID = "00000000-0000-0000-0000-000000000099"
 
 
+def _configure_ownership_pipeline(
+    redis,
+    *,
+    stop_exists=0,
+    metadata_exists=1,
+    client_count=0,
+    owner=None,
+    disconnect=None,
+    state=None,
+):
+    pipe = MagicMock()
+    redis.pipeline.return_value = pipe
+    pipe.execute.return_value = (
+        stop_exists,
+        metadata_exists,
+        client_count,
+        owner,
+        disconnect,
+        state,
+    )
+    return pipe
+
+
 def _mock_proxy_server(redis_client=None):
     server = MagicMock()
     server.redis_client = redis_client or MagicMock()
@@ -79,11 +103,6 @@ class LocalStreamActivityTests(TestCase):
         server.stop_all_output_profiles = MagicMock()
         return server
 
-    def test_has_local_stream_activity_from_live_registry(self):
-        server = self._make_server()
-        server._live_stream_managers[CHANNEL_ID] = MagicMock()
-        self.assertTrue(server._has_local_stream_activity(CHANNEL_ID))
-
     @patch.object(ProxyServer, "_join_stream_thread")
     def test_stop_local_stream_activity_stops_live_manager(self, mock_join):
         server = self._make_server()
@@ -114,9 +133,9 @@ class OrphanMetadataCleanupTests(TestCase):
 
     @patch.object(ProxyServer, "_clean_redis_keys")
     @patch.object(ProxyServer, "_stop_local_stream_activity")
-    @patch.object(ProxyServer, "_has_local_stream_activity", return_value=True)
+    @patch.object(ProxyServer, "_has_local_upstream_activity", return_value=True)
     def test_orphan_metadata_stops_local_processes_before_redis(
-        self, mock_has_local, mock_stop_local, mock_clean_redis
+        self, mock_has_upstream, mock_stop_local, mock_clean_redis
     ):
         server = self._make_server()
         metadata_key = RedisKeys.channel_metadata(CHANNEL_ID)
@@ -130,15 +149,16 @@ class OrphanMetadataCleanupTests(TestCase):
 
         server._check_orphaned_metadata()
 
-        mock_has_local.assert_called_with(CHANNEL_ID)
+        mock_has_upstream.assert_called_with(CHANNEL_ID)
         mock_stop_local.assert_called_once_with(CHANNEL_ID)
         mock_clean_redis.assert_called_once_with(CHANNEL_ID)
 
     @patch.object(ProxyServer, "_clean_redis_keys")
+    @patch.object(ProxyServer, "_broadcast_upstream_stop")
     @patch.object(ProxyServer, "_stop_local_stream_activity")
-    @patch.object(ProxyServer, "_has_local_stream_activity", return_value=False)
-    def test_orphan_metadata_remote_channel_only_cleans_redis(
-        self, mock_has_local, mock_stop_local, mock_clean_redis
+    @patch.object(ProxyServer, "_has_local_upstream_activity", return_value=False)
+    def test_orphan_metadata_remote_channel_broadcasts_stop(
+        self, mock_has_upstream, mock_stop_local, mock_broadcast, mock_clean_redis
     ):
         server = self._make_server()
         metadata_key = RedisKeys.channel_metadata(CHANNEL_ID)
@@ -149,6 +169,7 @@ class OrphanMetadataCleanupTests(TestCase):
 
         server._check_orphaned_metadata()
 
+        mock_broadcast.assert_called_once_with(CHANNEL_ID)
         mock_stop_local.assert_not_called()
         mock_clean_redis.assert_called_once_with(CHANNEL_ID)
 
@@ -171,9 +192,9 @@ class OrphanChannelCleanupTests(TestCase):
 
     @patch.object(ProxyServer, "_clean_redis_keys")
     @patch.object(ProxyServer, "_stop_local_stream_activity")
-    @patch.object(ProxyServer, "_has_local_stream_activity", return_value=True)
+    @patch.object(ProxyServer, "_has_local_upstream_activity", return_value=True)
     def test_orphan_channel_stops_local_before_redis(
-        self, mock_has_local, mock_stop_local, mock_clean_redis
+        self, mock_has_upstream, mock_stop_local, mock_clean_redis
     ):
         server = self._make_server()
         metadata_key = RedisKeys.channel_metadata(CHANNEL_ID)
@@ -189,10 +210,10 @@ class OrphanChannelCleanupTests(TestCase):
 class StreamManagerOwnershipTests(TestCase):
     def test_still_owner_false_when_different_worker(self):
         buffer = MagicMock()
-        buffer.redis_client.get.side_effect = lambda key: (
-            None if "stopping" in key else b"worker-b"
+        _configure_ownership_pipeline(
+            buffer.redis_client,
+            owner=b"worker-b",
         )
-        buffer.redis_client.exists.return_value = False
         manager = StreamManager(
             CHANNEL_ID, "http://example/stream", buffer, worker_id="worker-a"
         )
@@ -201,8 +222,12 @@ class StreamManagerOwnershipTests(TestCase):
 
     def test_still_owner_true_when_owner_lock_expired_but_not_stopping(self):
         buffer = MagicMock()
-        buffer.redis_client.get.return_value = None
-        buffer.redis_client.exists.return_value = False
+        _configure_ownership_pipeline(
+            buffer.redis_client,
+            client_count=0,
+            owner=None,
+            state=ChannelState.CONNECTING.encode(),
+        )
         manager = StreamManager(
             CHANNEL_ID, "http://example/stream", buffer, worker_id="worker-a"
         )
@@ -211,7 +236,10 @@ class StreamManagerOwnershipTests(TestCase):
 
     def test_still_owner_false_when_channel_stopping_key_set(self):
         buffer = MagicMock()
-        buffer.redis_client.exists.return_value = True
+        _configure_ownership_pipeline(
+            buffer.redis_client,
+            stop_exists=1,
+        )
         manager = StreamManager(
             CHANNEL_ID, "http://example/stream", buffer, worker_id="worker-a"
         )
@@ -220,8 +248,10 @@ class StreamManagerOwnershipTests(TestCase):
 
     def test_update_bytes_skipped_after_ownership_lost(self):
         buffer = MagicMock()
-        buffer.redis_client.get.return_value = b"other-worker"
-        buffer.redis_client.exists.return_value = False
+        _configure_ownership_pipeline(
+            buffer.redis_client,
+            owner=b"other-worker",
+        )
         manager = StreamManager(
             CHANNEL_ID, "http://example/stream", buffer, worker_id="worker-a"
         )
@@ -230,3 +260,383 @@ class StreamManagerOwnershipTests(TestCase):
         manager._update_bytes_processed(500)
 
         buffer.redis_client.hincrby.assert_not_called()
+
+
+class StreamBufferStopTests(TestCase):
+    def test_stop_discards_local_data_without_redis_writes(self):
+        redis = MagicMock()
+        buffer = StreamBuffer(channel_id=CHANNEL_ID, redis_client=redis)
+        buffer._write_buffer = bytearray(b"x" * 376)
+
+        buffer.stop()
+
+        self.assertTrue(buffer.stopping)
+        self.assertEqual(len(buffer._write_buffer), 0)
+        redis.incr.assert_not_called()
+        redis.setex.assert_not_called()
+        redis.delete.assert_not_called()
+
+
+class StopChannelTeardownTests(TestCase):
+    def _make_server(self):
+        with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
+            server = ProxyServer()
+        server.worker_id = "testhost:1"
+        server.stream_managers = {}
+        server.stream_buffers = {}
+        server.client_managers = {}
+        server.profile_managers = {}
+        server.profile_buffers = {}
+        server._live_stream_managers = {}
+        server._stopping_channels = set()
+        server._stopping_since = {}
+        server.redis_client = MagicMock()
+        server.redis_client.exists.return_value = False
+        server.am_i_owner = MagicMock(return_value=False)
+        server._collect_channel_stop_event_data = MagicMock(return_value=None)
+        server.release_ownership = MagicMock()
+        return server
+
+    @patch.object(ProxyServer, "_spawn_channel_stop_event")
+    @patch.object(ProxyServer, "_clean_redis_keys")
+    @patch.object(ProxyServer, "_stop_local_stream_activity")
+    def test_stop_channel_cleans_redis_before_blocking_local_stop(
+        self, mock_stop_local, mock_clean_redis, mock_spawn_event
+    ):
+        server = self._make_server()
+        call_order = []
+
+        def stop_local(channel_id):
+            call_order.append("local")
+
+        def clean_redis(channel_id):
+            call_order.append("redis")
+
+        mock_stop_local.side_effect = stop_local
+        mock_clean_redis.side_effect = clean_redis
+
+        server.stop_channel(CHANNEL_ID)
+
+        self.assertEqual(call_order, ["redis", "local"])
+        mock_spawn_event.assert_called_once_with(None)
+
+    @patch.object(ProxyServer, "_spawn_channel_stop_event")
+    @patch.object(ProxyServer, "_clean_redis_keys")
+    @patch.object(ProxyServer, "_stop_local_stream_activity", side_effect=RuntimeError("boom"))
+    def test_stop_channel_cleans_redis_in_finally_when_local_stop_fails(
+        self, mock_stop_local, mock_clean_redis, mock_spawn_event
+    ):
+        server = self._make_server()
+
+        result = server.stop_channel(CHANNEL_ID)
+
+        self.assertFalse(result)
+        mock_clean_redis.assert_called_once_with(CHANNEL_ID)
+        mock_spawn_event.assert_not_called()
+        self.assertNotIn(CHANNEL_ID, server._stopping_channels)
+
+    @patch.object(ProxyServer, "_spawn_channel_stop_event")
+    @patch.object(ProxyServer, "_clean_redis_keys")
+    @patch.object(ProxyServer, "_stop_local_stream_activity")
+    def test_stop_channel_owner_releases_after_redis_cleanup(
+        self, mock_stop_local, mock_clean_redis, mock_spawn_event
+    ):
+        server = self._make_server()
+        server.am_i_owner.return_value = True
+        stop_data = {"channel_id": CHANNEL_ID}
+        server._collect_channel_stop_event_data.return_value = stop_data
+        call_order = []
+
+        def stop_local(channel_id):
+            call_order.append("local")
+
+        def release(channel_id):
+            call_order.append("release")
+
+        def clean_redis(channel_id):
+            call_order.append("redis")
+
+        mock_stop_local.side_effect = stop_local
+        server.release_ownership.side_effect = release
+        mock_clean_redis.side_effect = clean_redis
+
+        server.stop_channel(CHANNEL_ID)
+
+        self.assertEqual(call_order, ["redis", "release", "local"])
+        server._collect_channel_stop_event_data.assert_called_once_with(CHANNEL_ID)
+        mock_spawn_event.assert_called_once_with(stop_data)
+
+
+class CleanRedisKeysOrderTests(TestCase):
+    @patch("apps.proxy.live_proxy.server.Stream.objects.get")
+    @patch("apps.proxy.live_proxy.server.Channel.objects.get")
+    def test_clean_redis_keys_releases_profile_slot_before_live_keys_deleted(
+        self, mock_channel_get, mock_stream_get
+    ):
+        from apps.channels.models import Channel, Stream
+
+        with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
+            server = ProxyServer()
+        server.redis_client = MagicMock()
+        call_order = []
+
+        channel = MagicMock()
+        channel.release_stream.return_value = True
+
+        def channel_get(uuid):
+            call_order.append("release")
+            return channel
+
+        mock_channel_get.side_effect = channel_get
+        mock_stream_get.side_effect = Stream.DoesNotExist
+
+        def scan(*args, **kwargs):
+            call_order.append("redis")
+            return (0, [b"live:channel:foo:buffer:index"])
+
+        server.redis_client.scan.side_effect = scan
+
+        server._clean_redis_keys(CHANNEL_ID)
+
+        self.assertEqual(call_order, ["release", "redis"])
+        channel.release_stream.assert_called_once()
+        server.redis_client.delete.assert_called_once()
+
+
+class LocalUpstreamActivityTests(TestCase):
+    def _make_server(self):
+        with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
+            server = ProxyServer()
+        server.worker_id = "testhost:1"
+        server.stream_managers = {}
+        server.stream_buffers = {}
+        server.client_managers = {}
+        server._live_stream_managers = {}
+        return server
+
+    def test_upstream_activity_excludes_reader_only_client_manager(self):
+        server = self._make_server()
+        server.client_managers[CHANNEL_ID] = MagicMock()
+        server.stream_buffers[CHANNEL_ID] = MagicMock()
+        self.assertFalse(server._has_local_upstream_activity(CHANNEL_ID))
+
+    def test_upstream_activity_from_live_registry(self):
+        server = self._make_server()
+        server._live_stream_managers[CHANNEL_ID] = MagicMock()
+        self.assertTrue(server._has_local_upstream_activity(CHANNEL_ID))
+
+
+class ClientDisconnectOwnershipTests(TestCase):
+    def test_last_client_triggers_stop_when_upstream_active_without_owner_lock(self):
+        from apps.proxy.live_proxy.client_manager import ClientManager
+
+        proxy_server = MagicMock()
+        proxy_server.worker_id = "testhost:1"
+        proxy_server.am_i_owner.return_value = False
+        proxy_server._has_local_upstream_activity.return_value = True
+        proxy_server.extend_ownership.return_value = False
+
+        redis = MagicMock()
+        redis.hget.return_value = b"viewer"
+        redis.scard.return_value = 0
+
+        manager = ClientManager(
+            channel_id=CHANNEL_ID,
+            redis_client=redis,
+            worker_id="testhost:1",
+        )
+        manager.proxy_server = proxy_server
+        manager.clients = {"client-1"}
+        manager.client_set_key = RedisKeys.clients(CHANNEL_ID)
+        manager._notify_owner_of_activity = MagicMock()
+        manager._trigger_stats_update = MagicMock()
+        manager.get_total_client_count = MagicMock(return_value=0)
+        proxy_server._spawn_on_hub = MagicMock()
+
+        manager.remove_client("client-1")
+
+        proxy_server._spawn_on_hub.assert_called_once_with(
+            proxy_server.handle_client_disconnect, CHANNEL_ID
+        )
+
+
+class TeardownActiveLocalStopTests(TestCase):
+    @patch("apps.proxy.live_proxy.services.channel_service.ProxyServer.get_instance")
+    def test_teardown_active_when_local_stop_in_progress(self, mock_get_instance):
+        server = MagicMock()
+        server._stopping_channels = {CHANNEL_ID}
+        server.redis_client = MagicMock()
+        server.redis_client.exists.return_value = False
+        mock_get_instance.return_value = server
+
+        self.assertTrue(ChannelService.is_channel_teardown_active(CHANNEL_ID))
+
+
+class HandleClientDisconnectUpstreamFirstTests(TestCase):
+    def _make_server(self):
+        with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
+            server = ProxyServer()
+        server.worker_id = "testhost:1"
+        server.client_managers = {CHANNEL_ID: MagicMock()}
+        server.stream_managers = {CHANNEL_ID: MagicMock()}
+        server._live_stream_managers = {}
+        server.profile_managers = {}
+        server.output_managers = {}
+        server.redis_client = MagicMock()
+        server.redis_client.scard.return_value = 0
+        server._stopping_channels = set()
+        return server
+
+    @patch.object(ProxyServer, "_coordinated_stop_channel")
+    @patch.object(ProxyServer, "_stop_upstream_before_redis_cleanup")
+    @patch.object(ProxyServer, "_has_local_upstream_activity", return_value=True)
+    def test_last_client_uses_coordinated_stop_only(
+        self, _mock_upstream, mock_stop_upstream, mock_coordinated
+    ):
+        server = self._make_server()
+
+        with patch(
+            "apps.proxy.live_proxy.server.ConfigHelper.channel_shutdown_delay",
+            return_value=0,
+        ):
+            server.handle_client_disconnect(CHANNEL_ID)
+
+        mock_stop_upstream.assert_not_called()
+        mock_coordinated.assert_called_once_with(CHANNEL_ID)
+
+
+class InitWaitAbortTests(TestCase):
+    def _make_generator(self):
+        from apps.proxy.live_proxy.output.ts.generator import StreamGenerator
+
+        return StreamGenerator(
+            CHANNEL_ID,
+            "client-1",
+            "127.0.0.1",
+            "test-agent",
+            channel_initializing=True,
+        )
+
+    def test_abort_when_client_removed_locally(self):
+        generator = self._make_generator()
+        server = MagicMock()
+        client_manager = MagicMock()
+        client_manager.clients = set()
+        server.client_managers = {CHANNEL_ID: client_manager}
+        server.redis_client = MagicMock()
+
+        self.assertEqual(generator._init_wait_abort_reason(server, time.time()), "client_gone")
+
+    def test_abort_when_connect_stalled_without_buffer(self):
+        from apps.proxy.config import TSConfig as Config
+
+        generator = self._make_generator()
+        server = MagicMock()
+        client_manager = MagicMock()
+        client_manager.clients = {"client-1"}
+        server.client_managers = {CHANNEL_ID: client_manager}
+        buffer = MagicMock()
+        buffer.index = 0
+        server.stream_buffers = {CHANNEL_ID: buffer}
+        server.redis_client = MagicMock()
+        server.redis_client.hget.return_value = b"connecting"
+        server.redis_client.get.return_value = None
+
+        started = time.time() - (getattr(Config, "CONNECTION_TIMEOUT", 10) + 1)
+        self.assertEqual(generator._init_wait_abort_reason(server, started), "stalled")
+
+
+class UpstreamStopBroadcastTests(TestCase):
+    def _make_server(self):
+        with patch("apps.proxy.live_proxy.server.RedisClient.get_client", return_value=MagicMock()):
+            server = ProxyServer()
+        server.worker_id = "testhost:1"
+        server.stream_managers = {}
+        server._live_stream_managers = {}
+        server.redis_client = MagicMock()
+        return server
+
+    @patch.object(ProxyServer, "_stop_local_stream_activity")
+    @patch.object(ProxyServer, "_broadcast_upstream_stop")
+    @patch.object(ProxyServer, "_has_local_upstream_activity", return_value=True)
+    def test_local_upstream_stops_locally_without_broadcast(
+        self, _mock_has, mock_broadcast, mock_stop_local
+    ):
+        server = self._make_server()
+        server._stop_upstream_before_redis_cleanup(CHANNEL_ID)
+        mock_stop_local.assert_called_once_with(CHANNEL_ID)
+        mock_broadcast.assert_not_called()
+
+    @patch.object(ProxyServer, "_stop_local_stream_activity")
+    @patch.object(ProxyServer, "_broadcast_upstream_stop")
+    @patch.object(ProxyServer, "_has_local_upstream_activity", return_value=False)
+    def test_orphan_cleanup_broadcasts_when_no_local_upstream(
+        self, _mock_has, mock_broadcast, mock_stop_local
+    ):
+        server = self._make_server()
+        server._stop_upstream_before_redis_cleanup(CHANNEL_ID)
+        mock_broadcast.assert_called_once_with(CHANNEL_ID)
+        mock_stop_local.assert_not_called()
+
+
+class StreamManagerStillOwnerTests(TestCase):
+    def _make_manager(self, redis_client):
+        buffer = MagicMock()
+        buffer.redis_client = redis_client
+        buffer.channel_id = CHANNEL_ID
+        manager = StreamManager(
+            CHANNEL_ID,
+            "http://example/stream.ts",
+            buffer,
+            worker_id="testhost:1",
+        )
+        return manager
+
+    def test_stops_when_metadata_removed(self):
+        redis = MagicMock()
+        _configure_ownership_pipeline(redis, metadata_exists=0)
+        manager = self._make_manager(redis)
+        self.assertFalse(manager._still_owner())
+
+    def test_keeps_running_during_connecting_before_client_registered(self):
+        redis = MagicMock()
+        _configure_ownership_pipeline(
+            redis,
+            client_count=0,
+            owner=b"testhost:1",
+            state=ChannelState.CONNECTING.encode(),
+        )
+        manager = self._make_manager(redis)
+        self.assertTrue(manager._still_owner())
+
+    def test_stops_after_disconnect_when_shutdown_delay_is_zero(self):
+        redis = MagicMock()
+        _configure_ownership_pipeline(
+            redis,
+            client_count=0,
+            owner=b"testhost:1",
+            disconnect=b"1700000000.0",
+            state=ChannelState.ACTIVE.encode(),
+        )
+        manager = self._make_manager(redis)
+        with patch(
+            "apps.proxy.live_proxy.input.manager.ConfigHelper.channel_shutdown_delay",
+            return_value=0,
+        ):
+            self.assertFalse(manager._still_owner())
+
+    def test_keeps_running_during_shutdown_delay(self):
+        redis = MagicMock()
+        _configure_ownership_pipeline(
+            redis,
+            client_count=0,
+            owner=b"testhost:1",
+            disconnect=str(time.time()).encode(),
+            state=ChannelState.ACTIVE.encode(),
+        )
+        manager = self._make_manager(redis)
+        with patch(
+            "apps.proxy.live_proxy.input.manager.ConfigHelper.channel_shutdown_delay",
+            return_value=5,
+        ):
+            self.assertTrue(manager._still_owner())
diff --git a/apps/proxy/live_proxy/client_manager.py b/apps/proxy/live_proxy/client_manager.py
index 3771d55c..3767a25a 100644
--- a/apps/proxy/live_proxy/client_manager.py
+++ b/apps/proxy/live_proxy/client_manager.py
@@ -102,6 +102,7 @@ class ClientManager:
                         break
 
                     # Send heartbeat for all local clients
+                    clients_to_remove = set()
                     with self.lock:
                         # Skip this cycle if we have no local clients
                         if not self.clients:
@@ -109,7 +110,6 @@ class ClientManager:
 
                         # IMPROVED GHOST DETECTION: Check for stale clients before sending heartbeats
                         current_time = time.time()
-                        clients_to_remove = set()
 
                         # First identify clients that should be removed
                         for client_id in self.clients:
@@ -132,12 +132,11 @@ class ClientManager:
                                     logger.debug(f"Client {client_id} inactive for {current_time - last_active_time:.1f}s, removing as ghost")
                                     clients_to_remove.add(client_id)
 
-                        # Remove ghost clients in a separate step
-                        for client_id in clients_to_remove:
-                            self.remove_client(client_id)
-
                         if clients_to_remove:
-                            logger.info(f"Removed {len(clients_to_remove)} ghost clients from channel {self.channel_id}")
+                            logger.info(
+                                f"Identified {len(clients_to_remove)} ghost clients "
+                                f"for channel {self.channel_id}"
+                            )
 
                         # Now send heartbeats only for remaining clients
                         pipe = self.redis_client.pipeline()
@@ -172,6 +171,9 @@ class ClientManager:
                         if self.clients and not all(c in clients_to_remove for c in self.clients):
                             self._notify_owner_of_activity()
 
+                    for client_id in clients_to_remove:
+                        self.remove_client(client_id)
+
                 except Exception as e:
                     logger.error(f"Error in client heartbeat thread: {e}")
 
@@ -305,6 +307,9 @@ class ClientManager:
 
     def remove_client(self, client_id):
         """Remove a client from this channel and Redis"""
+        client_username = "unknown"
+        remaining = None
+
         with self.lock:
             if client_id in self.clients:
                 self.clients.remove(client_id)
@@ -315,61 +320,80 @@ class ClientManager:
             self.last_active_time = time.time()
 
             if self.redis_client:
-                # Get client data before removing the data
                 client_key = f"live:channel:{self.channel_id}:clients:{client_id}"
                 client_username = self.redis_client.hget(client_key, "username") or "unknown"
                 if isinstance(client_username, bytes):
                     client_username = client_username.decode("utf-8")
 
-                # Remove from channel's client set
                 self.redis_client.srem(self.client_set_key, client_id)
-
-                client_key = f"live:channel:{self.channel_id}:clients:{client_id}"
                 self.redis_client.delete(client_key)
-
-                # Check if this was the last client
                 remaining = self.redis_client.scard(self.client_set_key) or 0
-                if remaining == 0:
-                    logger.warning(f"Last client removed: {client_id} - channel may shut down soon")
 
-                    # Trigger disconnect time tracking even if we're not the owner
-                    disconnect_key = RedisKeys.last_client_disconnect(self.channel_id)
-                    self.redis_client.setex(disconnect_key, 60, str(time.time()))
+            local_count = len(self.clients)
 
-                self._notify_owner_of_activity()
+        if not self.redis_client:
+            return local_count
 
-                # Check if we're the owner - if so, handle locally; if not, publish event
-                am_i_owner = self.proxy_server and self.proxy_server.am_i_owner(self.channel_id)
+        if remaining == 0:
+            logger.warning(f"Last client removed: {client_id} - channel may shut down soon")
+            disconnect_key = RedisKeys.last_client_disconnect(self.channel_id)
+            self.redis_client.setex(disconnect_key, 60, str(time.time()))
 
-                if am_i_owner:
-                    # We're the owner - handle the disconnect directly
-                    logger.debug(f"Owner handling CLIENT_DISCONNECTED for client {client_id} locally (not publishing)")
-                    if (remaining == 0
-                            or self.proxy_server.output_managers.get(self.channel_id)
-                            or self.proxy_server.profile_managers.get(self.channel_id)):
-                        import gevent
-                        gevent.spawn(self.proxy_server.handle_client_disconnect, self.channel_id)
-                else:
-                    # We're not the owner - publish event so owner can handle it
-                    logger.debug(f"Non-owner publishing CLIENT_DISCONNECTED event for client {client_id} on channel {self.channel_id} from worker {self.worker_id}")
-                    event_data = json.dumps({
-                        "event": EventType.CLIENT_DISCONNECTED,
-                        "channel_id": self.channel_id,
-                        "client_id": client_id,
-                        "worker_id": self.worker_id or "unknown",
-                        "timestamp": time.time(),
-                        "remaining_clients": remaining,
-                        "username": client_username
-                    })
-                    self.redis_client.publish(RedisKeys.events_channel(self.channel_id), event_data)
+        self._notify_owner_of_activity()
 
-                # Trigger channel stats update via WebSocket
-                self._trigger_stats_update()
+        am_i_owner = self.proxy_server and self.proxy_server.am_i_owner(self.channel_id)
 
-            total_clients = self.get_total_client_count()
-            logger.info(f"Client disconnected: {client_id} (local: {len(self.clients)}, total: {total_clients})")
+        # Owner lock TTL can expire while local ffmpeg is still running on this worker.
+        if (not am_i_owner and self.proxy_server
+                and self.proxy_server._has_local_upstream_activity(self.channel_id)):
+            if self.proxy_server.extend_ownership(self.channel_id):
+                am_i_owner = True
 
-        return len(self.clients)
+        schedule_disconnect = False
+        if am_i_owner:
+            logger.debug(
+                f"Owner handling CLIENT_DISCONNECTED for client {client_id} locally (not publishing)"
+            )
+            if (remaining == 0
+                    or self.proxy_server.output_managers.get(self.channel_id)
+                    or self.proxy_server.profile_managers.get(self.channel_id)):
+                schedule_disconnect = True
+        elif (remaining == 0 and self.proxy_server
+              and self.proxy_server._has_local_upstream_activity(self.channel_id)):
+            logger.warning(
+                f"Last client removed while local upstream still active for "
+                f"{self.channel_id} without owner lock - stopping channel"
+            )
+            schedule_disconnect = True
+        else:
+            logger.debug(
+                f"Non-owner publishing CLIENT_DISCONNECTED event for client {client_id} "
+                f"on channel {self.channel_id} from worker {self.worker_id}"
+            )
+            event_data = json.dumps({
+                "event": EventType.CLIENT_DISCONNECTED,
+                "channel_id": self.channel_id,
+                "client_id": client_id,
+                "worker_id": self.worker_id or "unknown",
+                "timestamp": time.time(),
+                "remaining_clients": remaining,
+                "username": client_username
+            })
+            self.redis_client.publish(RedisKeys.events_channel(self.channel_id), event_data)
+
+        if schedule_disconnect and self.proxy_server:
+            self.proxy_server._spawn_on_hub(
+                self.proxy_server.handle_client_disconnect, self.channel_id
+            )
+
+        self._trigger_stats_update()
+
+        total_clients = self.get_total_client_count()
+        logger.info(
+            f"Client disconnected: {client_id} (local: {local_count}, total: {total_clients})"
+        )
+
+        return local_count
 
     def get_client_count(self):
         """Get local client count"""
diff --git a/apps/proxy/live_proxy/input/buffer.py b/apps/proxy/live_proxy/input/buffer.py
index b345ead3..3d3fb091 100644
--- a/apps/proxy/live_proxy/input/buffer.py
+++ b/apps/proxy/live_proxy/input/buffer.py
@@ -309,38 +309,16 @@ class StreamBuffer:
         self.fill_timers.clear()
 
         try:
-            # Flush any remaining data in the write buffer
-            if hasattr(self, '_write_buffer') and len(self._write_buffer) > 0:
-                # Ensure remaining data is aligned to TS packets
-                complete_size = (len(self._write_buffer) // 188) * 188
-
-                if complete_size > 0:
-                    final_chunk = self._write_buffer[:complete_size]
-
-                    # Write final chunk to Redis
-                    with self.lock:
-                        if self.redis_client:
-                            try:
-                                chunk_index = self.redis_client.incr(self.buffer_index_key)
-                                chunk_key = f"{self.buffer_prefix}{chunk_index}"
-                                self.redis_client.setex(chunk_key, self.chunk_ttl, bytes(final_chunk))
-                                self.index = chunk_index
-                                logger.info(f"Flushed final chunk of {len(final_chunk)} bytes to Redis")
-                            except Exception as e:
-                                logger.error(f"Error flushing final chunk: {e}")
-
-                # Clear buffers
-                self._write_buffer = bytearray()
-                if hasattr(self, '_partial_packet'):
-                    self._partial_packet = bytearray()
-
-            # Clean up the chunk timestamps sorted set
-            if self.redis_client and self.chunk_timestamps_key:
-                try:
-                    self.redis_client.delete(self.chunk_timestamps_key)
-                except Exception as e:
-                    logger.error(f"Error deleting chunk timestamps key: {e}")
-
+            with self.lock:
+                if hasattr(self, '_write_buffer') and len(self._write_buffer) > 0:
+                    discarded = len(self._write_buffer)
+                    self._write_buffer = bytearray()
+                    if hasattr(self, '_partial_packet'):
+                        self._partial_packet = bytearray()
+                    logger.debug(
+                        f"Discarded {discarded} bytes from local write buffer "
+                        f"for channel {self.channel_id}"
+                    )
         except Exception as e:
             logger.error(f"Error during buffer stop: {e}")
 
diff --git a/apps/proxy/live_proxy/input/manager.py b/apps/proxy/live_proxy/input/manager.py
index 06b12644..ae744473 100644
--- a/apps/proxy/live_proxy/input/manager.py
+++ b/apps/proxy/live_proxy/input/manager.py
@@ -112,6 +112,11 @@ class StreamManager:
         # Add tracking for data throughput
         self.bytes_processed = 0
         self.last_bytes_update = time.time()
+
+        # Cached result of the pipelined Redis ownership audit (hot read path).
+        self._ownership_cache_valid_until = 0.0
+        self._ownership_cached = True
+        self._OWNERSHIP_CHECK_INTERVAL = 1.0
         self.bytes_update_interval = 5  # Update Redis every 5 seconds
 
         # Add stderr reader thread property
@@ -182,7 +187,90 @@ class StreamManager:
         logger.warning(f"Timeout waiting for existing processes to close for channel {self.channel_id} after {timeout}s")
         return False
 
-    def _still_owner(self):
+    def _invalidate_ownership_cache(self):
+        self._ownership_cache_valid_until = 0.0
+
+    @staticmethod
+    def _decode_redis_value(value):
+        if value is None:
+            return None
+        if isinstance(value, bytes):
+            return value.decode()
+        return value
+
+    def _disconnect_shutdown_ready(self, disconnect_value):
+        """True when last-client disconnect has passed the configured shutdown delay."""
+        if not disconnect_value:
+            return False
+
+        shutdown_delay = ConfigHelper.channel_shutdown_delay()
+        if shutdown_delay <= 0:
+            return True
+
+        disconnect_value = self._decode_redis_value(disconnect_value)
+        try:
+            disconnect_time = float(disconnect_value)
+        except (ValueError, TypeError):
+            return False
+        return (time.time() - disconnect_time) >= shutdown_delay
+
+    def _evaluate_ownership_from_redis(self, redis_client):
+        """Single pipelined Redis round-trip for the full ownership audit."""
+        stop_key = RedisKeys.channel_stopping(self.channel_id)
+        metadata_key = RedisKeys.channel_metadata(self.channel_id)
+        clients_key = RedisKeys.clients(self.channel_id)
+        owner_key = RedisKeys.channel_owner(self.channel_id)
+        disconnect_key = RedisKeys.last_client_disconnect(self.channel_id)
+
+        pipe = redis_client.pipeline(transaction=False)
+        pipe.exists(stop_key)
+        pipe.exists(metadata_key)
+        pipe.scard(clients_key)
+        pipe.get(owner_key)
+        pipe.get(disconnect_key)
+        pipe.hget(metadata_key, ChannelMetadataField.STATE)
+        (
+            stop_exists,
+            metadata_exists,
+            client_count,
+            current_owner,
+            disconnect_value,
+            state_raw,
+        ) = pipe.execute()
+
+        if stop_exists:
+            return False
+
+        if not metadata_exists:
+            logger.warning(
+                f"Channel {self.channel_id} metadata removed from Redis - stopping upstream"
+            )
+            return False
+
+        client_count = client_count or 0
+        state = self._decode_redis_value(state_raw)
+        current_owner = self._decode_redis_value(current_owner)
+
+        if current_owner and current_owner != self.worker_id:
+            return False
+
+        if not current_owner:
+            if client_count == 0 and state not in ChannelState.PRE_ACTIVE:
+                logger.warning(
+                    f"Channel {self.channel_id} has no owner and no clients - stopping upstream"
+                )
+                return False
+            return True
+
+        if client_count == 0 and self._disconnect_shutdown_ready(disconnect_value):
+            logger.info(
+                f"Channel {self.channel_id} disconnect shutdown ready - stopping upstream"
+            )
+            return False
+
+        return True
+
+    def _still_owner(self, *, force=False):
         """Return True while this worker should keep the upstream connection open."""
         if self.stopping or self.stop_requested:
             return False
@@ -194,26 +282,49 @@ class StreamManager:
         if not redis_client:
             return True
 
-        try:
-            stop_key = RedisKeys.channel_stopping(self.channel_id)
-            if redis_client.exists(stop_key):
-                return False
+        now = time.time()
+        if not force and now < self._ownership_cache_valid_until:
+            return self._ownership_cached
 
-            owner_key = RedisKeys.channel_owner(self.channel_id)
-            current_owner = redis_client.get(owner_key)
-            if not current_owner:
-                # Owner lock TTL may lapse briefly between cleanup-thread renewals.
-                # Keep running unless coordinated teardown has started (checked above).
-                return True
-            if isinstance(current_owner, bytes):
-                current_owner = current_owner.decode()
-            return current_owner == self.worker_id
+        try:
+            result = self._evaluate_ownership_from_redis(redis_client)
+            self._ownership_cached = result
+            self._ownership_cache_valid_until = now + self._OWNERSHIP_CHECK_INTERVAL
+            return result
         except Exception as e:
             logger.debug(f"Ownership check failed for channel {self.channel_id}: {e}")
+            return True
+
+    def _upstream_may_continue(self):
+        """
+        Per-chunk gate for the hot read path.
+
+        Local stop flags are checked every chunk. The coordinated-teardown Redis
+        flag is checked every chunk (one EXISTS). The full ownership audit is
+        pipelined and cached for ~1s — enough for loop boundaries while avoiding
+        6+ Redis round-trips per chunk during steady streaming.
+        """
+        if self.stopping or self.stop_requested or not self.running:
+            return False
+        if self.buffer is not None and self.buffer.stopping:
             return False
 
+        redis_client = getattr(self.buffer, 'redis_client', None)
+        if redis_client and self.worker_id:
+            try:
+                if redis_client.exists(RedisKeys.channel_stopping(self.channel_id)):
+                    self._ownership_cached = False
+                    self._ownership_cache_valid_until = 0.0
+                    return False
+            except Exception as e:
+                logger.debug(
+                    f"Channel stopping check failed for {self.channel_id}: {e}"
+                )
+
+        return self._still_owner()
+
     def _ensure_owner_or_stop(self):
-        if self._still_owner():
+        if self._still_owner(force=True):
             return True
 
         logger.warning(
@@ -539,6 +650,7 @@ class StreamManager:
                 logger.debug(f"Closing existing HTTP connections before establishing transcode connection for channel {self.channel_id}")
                 self._close_connection()
 
+            close_old_connections()
             channel = get_stream_object(self.channel_id)
 
             # Use FFmpeg specifically for HLS streams
@@ -1071,7 +1183,7 @@ class StreamManager:
 
     def _update_bytes_processed(self, chunk_size):
         """Update the total bytes processed in Redis metadata"""
-        if not self._still_owner():
+        if not self._upstream_may_continue():
             return
 
         try:
@@ -1146,17 +1258,11 @@ class StreamManager:
         """Stop the stream manager and cancel all timers"""
         logger.info(f"Stopping stream manager for channel {self.channel_id}")
 
-        # Add at the beginning of your stop method
         self.stopping = True
+        self._invalidate_ownership_cache()
         if self.buffer is not None:
             self.buffer.stopping = True
 
-        # Release stream resources if we're the owner
-        if self.current_stream_id and hasattr(self, 'worker_id') and self.worker_id:
-            if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
-                owner_key = RedisKeys.channel_owner(self.channel_id)
-                current_owner = self.buffer.redis_client.get(owner_key)
-
         # Cancel all buffer check timers
         for timer in list(self._buffer_check_timers):
             try:
@@ -1545,7 +1651,8 @@ class StreamManager:
             if hasattr(self, 'stderr_reader_thread') and self.stderr_reader_thread and self.stderr_reader_thread.is_alive():
                 try:
                     logger.debug(f"Waiting for stderr reader thread to terminate for channel {self.channel_id}")
-                    self.stderr_reader_thread.join(timeout=2.0)
+                    stderr_join_timeout = 0.25 if self.stopping else 2.0
+                    self.stderr_reader_thread.join(timeout=stderr_join_timeout)
                     if self.stderr_reader_thread.is_alive():
                         logger.warning(f"Stderr reader thread did not terminate within timeout for channel {self.channel_id}")
                 except Exception as e:
@@ -1638,7 +1745,7 @@ class StreamManager:
                 self.connected = False
                 return False
 
-            if not self._still_owner():
+            if not self._upstream_may_continue():
                 self.stop()
                 return False
 
@@ -1649,8 +1756,7 @@ class StreamManager:
             # Add directly to buffer without TS-specific processing
             success = self.buffer.add_chunk(chunk)
 
-            # Update last data timestamp in Redis if successful
-            if success and self._still_owner() and hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
+            if success and hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
                 last_data_key = RedisKeys.last_data(self.buffer.channel_id)
                 self.buffer.redis_client.set(last_data_key, str(time.time()), ex=60)
 
diff --git a/apps/proxy/live_proxy/output/ts/generator.py b/apps/proxy/live_proxy/output/ts/generator.py
index 64bb4e1d..efde02ad 100644
--- a/apps/proxy/live_proxy/output/ts/generator.py
+++ b/apps/proxy/live_proxy/output/ts/generator.py
@@ -139,12 +139,75 @@ class StreamGenerator:
         finally:
             self._cleanup()
 
+    def _init_wait_abort_reason(self, proxy_server, initialization_start):
+        """Return a reason string when init wait should end early, else None."""
+        from ...services.channel_service import ChannelService
+
+        if ChannelService.is_channel_teardown_active(self.channel_id):
+            return "stopping"
+
+        client_mgr = proxy_server.client_managers.get(self.channel_id)
+        if client_mgr is not None:
+            if self.client_id not in client_mgr.clients:
+                return "client_gone"
+        elif proxy_server.redis_client:
+            client_key = RedisKeys.client_metadata(self.channel_id, self.client_id)
+            if not proxy_server.redis_client.exists(client_key):
+                total = proxy_server.redis_client.scard(RedisKeys.clients(self.channel_id)) or 0
+                if total == 0:
+                    return "client_gone"
+
+        elapsed = time.time() - initialization_start
+        connection_timeout = getattr(Config, 'CONNECTION_TIMEOUT', 10)
+        if elapsed < connection_timeout or not proxy_server.redis_client:
+            return None
+
+        metadata_key = RedisKeys.channel_metadata(self.channel_id)
+        state_raw = proxy_server.redis_client.hget(metadata_key, 'state')
+        if not state_raw:
+            return None
+
+        state = state_raw.decode() if isinstance(state_raw, bytes) else state_raw
+        if state not in ('connecting', 'initializing', 'buffering'):
+            return None
+
+        buffer = proxy_server.stream_buffers.get(self.channel_id)
+        if buffer is not None and buffer.index > 0:
+            return None
+
+        if proxy_server.redis_client.get(RedisKeys.last_data(self.channel_id)):
+            return None
+
+        return "stalled"
+
     def _wait_for_initialization(self):
         initialization_start = time.time()
         max_init_wait = ConfigHelper.client_wait_timeout()
         proxy_server = ProxyServer.get_instance()
 
         while time.time() - initialization_start < max_init_wait:
+            abort_reason = self._init_wait_abort_reason(proxy_server, initialization_start)
+            if abort_reason == "stopping":
+                logger.error(
+                    f"[{self.client_id}] Channel {self.channel_id} teardown active during initialization"
+                )
+                yield create_ts_packet('error', "Error: Channel is stopping")
+                return False
+            if abort_reason == "client_gone":
+                logger.info(
+                    f"[{self.client_id}] Client disconnected during initialization wait, aborting"
+                )
+                yield create_ts_packet('error', "Error: Client disconnected")
+                return False
+            if abort_reason == "stalled":
+                logger.warning(
+                    f"[{self.client_id}] Channel {self.channel_id} stalled in connecting state "
+                    f"with no buffer data after "
+                    f"{getattr(Config, 'CONNECTION_TIMEOUT', 10)}s, aborting init wait"
+                )
+                yield create_ts_packet('error', "Error: Connection stalled")
+                return False
+
             if proxy_server.redis_client:
                 metadata_key = RedisKeys.channel_metadata(self.channel_id)
                 metadata = proxy_server.redis_client.hgetall(metadata_key)
@@ -537,35 +600,31 @@ class StreamGenerator:
         stream_released = False
         if proxy_server.redis_client:
             try:
-                metadata_key = RedisKeys.channel_metadata(self.channel_id)
-                metadata = proxy_server.redis_client.hgetall(metadata_key)
-                if metadata:
-                    stream_id_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.STREAM_ID)
-                    if stream_id_bytes:
-                        # Check if we're the last client
-                        if self.channel_id in proxy_server.client_managers:
-                            client_count = proxy_server.client_managers[self.channel_id].get_total_client_count()
-                            # Only the last client or owner should release the stream
-                            if client_count <= 1 and proxy_server.am_i_owner(self.channel_id):
-                                try:
-                                    # Try Channel first (normal flow), fall back to Stream (preview flow)
-                                    try:
-                                        obj = Channel.objects.get(uuid=self.channel_id)
-                                    except (Channel.DoesNotExist, Exception):
-                                        obj = Stream.objects.get(stream_hash=self.channel_id)
-                                    stream_released = obj.release_stream()
-                                    if stream_released:
-                                        logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}")
-                                    else:
-                                        logger.warning(f"[{self.client_id}] release_stream found no keys for channel {self.channel_id}")
-                                except Exception as e:
-                                    logger.error(f"[{self.client_id}] Error releasing stream for channel {self.channel_id}: {e}")
+                if self.channel_id in proxy_server.client_managers:
+                    client_count = proxy_server.client_managers[self.channel_id].get_total_client_count()
+                    # Pool slots are global; the last client on any worker must release.
+                    if client_count <= 1:
+                        try:
+                            try:
+                                obj = Channel.objects.get(uuid=self.channel_id)
+                            except (Channel.DoesNotExist, Exception):
+                                obj = Stream.objects.get(stream_hash=self.channel_id)
+                            stream_released = obj.release_stream()
+                            if stream_released:
+                                logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}")
+                            else:
+                                logger.warning(f"[{self.client_id}] release_stream found no keys for channel {self.channel_id}")
+                        except Exception as e:
+                            logger.error(f"[{self.client_id}] Error releasing stream for channel {self.channel_id}: {e}")
             except Exception as e:
                 logger.error(f"[{self.client_id}] Error checking stream data for release: {e}")
 
         if self.channel_id in proxy_server.client_managers:
             client_manager = proxy_server.client_managers[self.channel_id]
-            local_clients = client_manager.remove_client(self.client_id)
+            if self.client_id in client_manager.clients:
+                local_clients = client_manager.remove_client(self.client_id)
+            else:
+                local_clients = client_manager.get_client_count()
             total_clients = client_manager.get_total_client_count()
             logger.info(f"[{self.client_id}] Disconnected after {elapsed:.2f}s (local: {local_clients}, total: {total_clients})")
 
diff --git a/apps/proxy/live_proxy/server.py b/apps/proxy/live_proxy/server.py
index a348c876..667daf2f 100644
--- a/apps/proxy/live_proxy/server.py
+++ b/apps/proxy/live_proxy/server.py
@@ -72,6 +72,7 @@ class ProxyServer:
         self._channel_names = {}
         self._stopping_channels = set()  # channels with an active stop_channel call in progress
         self._stopping_since = {}  # channel_id -> time.time() when stop_channel began
+        self._local_stop_locks = {}
         # Managers kept until the stream OS thread exits (may outlive stream_managers dict)
         self._live_stream_managers = {}
 
@@ -141,6 +142,18 @@ class ProxyServer:
             logger.error(f"Redis command error: {e}")
             return None
 
+    def _spawn_on_hub(self, fn, *args, **kwargs):
+        """Schedule fn on the gevent hub from any thread without blocking the caller."""
+        import gevent
+
+        try:
+            hub = gevent.get_hub()
+            hub.loop.run_callback_threadsafe(
+                lambda: gevent.spawn(fn, *args, **kwargs)
+            )
+        except Exception as e:
+            logger.error(f"Failed to schedule {getattr(fn, '__name__', fn)} on hub: {e}")
+
     def _start_event_listener(self):
         """Listen for events from other workers"""
         if not self.redis_client:
@@ -507,9 +520,9 @@ class ProxyServer:
             current = self.redis_client.get(lock_key)
 
             if current is None:
-                # Key expired, re-acquire if we have the stream_manager, but never
+                # Key expired, re-acquire if we still run local upstream, but never
                 # during coordinated teardown (multi-worker reconnect-during-stop race).
-                if channel_id in self.stream_managers:
+                if channel_id in self.stream_managers or channel_id in self._live_stream_managers:
                     if self._channel_unavailable_for_new_clients(channel_id):
                         logger.info(
                             f"Refusing ownership re-acquisition for {channel_id}; "
@@ -546,6 +559,12 @@ class ProxyServer:
                 )
                 return False
 
+            if self._has_local_upstream_activity(channel_id):
+                logger.warning(
+                    f"Stopping lingering local upstream before initializing channel {channel_id}"
+                )
+                self._stop_local_stream_activity(channel_id)
+
             if self.redis_client:
                 metadata_key = RedisKeys.channel_metadata(channel_id)
                 if self.redis_client.exists(metadata_key):
@@ -879,26 +898,7 @@ class ProxyServer:
                 owner = metadata.get('owner', 'unknown')
                 logger.info(f"Zombie channel details - state: {state}, owner: {owner}")
 
-            # Clean up Redis keys
             self._clean_redis_keys(channel_id)
-
-            # Force release resources in the Channel model
-            try:
-                channel = Channel.objects.get(uuid=channel_id)
-                if not channel.release_stream():
-                    logger.warning(f"Failed to release stream for zombie channel {channel_id}")
-                else:
-                    logger.info(f"Released stream allocation for zombie channel {channel_id}")
-            except Exception as e:
-                try:
-                    stream = Stream.objects.get(stream_hash=channel_id)
-                    if not stream.release_stream():
-                        logger.warning(f"Failed to release stream for zombie channel {channel_id}")
-                    else:
-                        logger.info(f"Released stream allocation for zombie channel {channel_id}")
-                except Exception as e:
-                    logger.error(f"Error releasing stream for zombie channel {channel_id}: {e}")
-
             return True
         except Exception as e:
             logger.error(f"Error cleaning zombie channel {channel_id}: {e}", exc_info=True)
@@ -914,8 +914,10 @@ class ProxyServer:
         # to manage for this channel at all.
         if (channel_id not in self.client_managers
                 and channel_id not in self.stream_managers
+                and channel_id not in self._live_stream_managers
                 and channel_id not in self.profile_managers
-                and channel_id not in self.output_managers):
+                and channel_id not in self.output_managers
+                and not self._has_local_upstream_activity(channel_id)):
             return
 
         try:
@@ -987,12 +989,12 @@ class ProxyServer:
 
             if total == 0:
                 logger.debug(f"No clients left after disconnect event - stopping channel {channel_id}")
-                disconnect_key = RedisKeys.last_client_disconnect(channel_id)
-                self.redis_client.setex(disconnect_key, 60, str(time.time()))
 
                 shutdown_delay = ConfigHelper.channel_shutdown_delay()
+                disconnect_key = RedisKeys.last_client_disconnect(channel_id)
 
                 if shutdown_delay > 0:
+                    self.redis_client.setex(disconnect_key, 60, str(time.time()))
                     logger.info(f"Waiting {shutdown_delay}s before stopping channel...")
                     gevent.sleep(shutdown_delay)
 
@@ -1002,6 +1004,13 @@ class ProxyServer:
                         self.redis_client.delete(disconnect_key)
                         return
 
+                if shutdown_delay <= 0:
+                    self.redis_client.setex(disconnect_key, 60, str(time.time()))
+
+                # Coordinated stop runs local teardown + Redis cleanup once.
+                # Do not call _stop_upstream_before_redis_cleanup here — it races
+                # with stop_channel and can leave stop_channel blocked on stderr join
+                # before _clean_redis_keys ever runs (orphaned buffer:index keys).
                 self._coordinated_stop_channel(channel_id)
         except Exception as e:
             logger.error(f"Error handling client disconnect for channel {channel_id}: {e}")
@@ -1371,18 +1380,41 @@ class ProxyServer:
                 return thread
         return None
 
-    def _has_local_stream_activity(self, channel_id):
+    def _has_local_upstream_activity(self, channel_id):
+        """True when this worker runs a local ffmpeg / stream-{uuid} OS thread."""
         if channel_id in self.stream_managers:
             return True
         if channel_id in self._live_stream_managers:
             return True
-        if channel_id in self.stream_buffers:
-            return True
-        if channel_id in self.client_managers:
-            return True
         thread = self._get_stream_thread(channel_id)
         return thread is not None and thread.is_alive()
 
+    def _broadcast_upstream_stop(self, channel_id):
+        """Ask every uWSGI worker to stop local ffmpeg for this channel."""
+        from .services.channel_service import ChannelService
+
+        if not self.redis_client:
+            return
+        try:
+            ChannelService.mark_channel_stopping(channel_id, broadcast=True)
+            logger.info(
+                f"Broadcast upstream stop for channel {channel_id} to all workers"
+            )
+        except Exception as e:
+            logger.error(
+                f"Error broadcasting upstream stop for channel {channel_id}: {e}"
+            )
+
+    def _stop_upstream_before_redis_cleanup(self, channel_id):
+        """Stop local ffmpeg before deleting Redis keys (prevents delete/recreate loops)."""
+        if self._has_local_upstream_activity(channel_id):
+            logger.info(
+                f"Channel {channel_id} has local upstream activity - stopping processes"
+            )
+            self._stop_local_stream_activity(channel_id)
+            return
+        self._broadcast_upstream_stop(channel_id)
+
     def _join_stream_thread(self, channel_id, timeout=2.0):
         thread = self._get_stream_thread(channel_id)
         if thread and thread.is_alive():
@@ -1402,8 +1434,41 @@ class ProxyServer:
             manager = self._live_stream_managers.pop(channel_id, None)
         return manager
 
+    def _signal_upstream_shutdown(self, channel_id):
+        """Halt upstream Redis writes immediately without blocking on ffmpeg join."""
+        if channel_id in self.stream_buffers:
+            self.stream_buffers[channel_id].stopping = True
+        for registry in (self.stream_managers, self._live_stream_managers):
+            manager = registry.get(channel_id)
+            if manager is None:
+                continue
+            manager.stopping = True
+            manager.stop_requested = True
+            if getattr(manager, 'buffer', None) is not None:
+                manager.buffer.stopping = True
+
+    def _get_local_stop_lock(self, channel_id):
+        lock = self._local_stop_locks.get(channel_id)
+        if lock is None:
+            lock = threading.Lock()
+            self._local_stop_locks[channel_id] = lock
+        return lock
+
     def _stop_local_stream_activity(self, channel_id):
         """Stop local ffmpeg/stream threads regardless of registry state."""
+        lock = self._get_local_stop_lock(channel_id)
+        if not lock.acquire(blocking=False):
+            logger.debug(
+                f"Local stop already in progress for channel {channel_id}, skipping duplicate"
+            )
+            return
+        try:
+            self._stop_local_stream_activity_locked(channel_id)
+        finally:
+            lock.release()
+            self._local_stop_locks.pop(channel_id, None)
+
+    def _stop_local_stream_activity_locked(self, channel_id):
         stream_manager = self._resolve_stream_manager(channel_id)
         if stream_manager is not None:
             if hasattr(stream_manager, 'stop'):
@@ -1445,17 +1510,13 @@ class ProxyServer:
         self.profile_buffers.pop(channel_id, None)
         self._live_stream_managers.pop(channel_id, None)
 
-    def _cleanup_stopped_channel_local(self, channel_id):
-        """Remove local managers/buffers for a channel that is shutting down."""
-        self._stop_local_stream_activity(channel_id)
-
     def _recover_stuck_channel_stops(self):
         """Force cleanup when stop_channel never finishes (e.g. blocked on logging)."""
         if not self._stopping_channels:
             return
 
         now = time.time()
-        stuck_threshold = max(30, ConfigHelper.channel_shutdown_delay() * 2)
+        stuck_threshold = max(10, ConfigHelper.channel_shutdown_delay() * 2)
 
         for channel_id in list(self._stopping_channels):
             started = self._stopping_since.get(channel_id, now)
@@ -1483,10 +1544,10 @@ class ProxyServer:
         self._stopping_channels.add(channel_id)
         self._stopping_since[channel_id] = time.time()
         stop_event_data = None
+        redis_cleaned = False
         try:
             logger.info(f"Stopping channel {channel_id}")
 
-            # First set a stopping key that clients will check
             if self.redis_client:
                 stop_key = RedisKeys.channel_stopping(channel_id)
                 if self.redis_client.exists(stop_key):
@@ -1494,54 +1555,26 @@ class ProxyServer:
                 else:
                     self.redis_client.setex(stop_key, 60, "true")
 
-            # Only stop the actual stream manager if we're the owner
-            if self.am_i_owner(channel_id):
-                logger.info(f"This worker ({self.worker_id}) is the owner - closing provider connection")
-                if channel_id in self.stream_managers:
-                    stream_manager = self.stream_managers[channel_id]
-
-                    # Signal thread to stop and close resources
-                    if hasattr(stream_manager, 'stop'):
-                        stream_manager.stop()
-                    else:
-                        stream_manager.running = False
-                        if hasattr(stream_manager, '_close_socket'):
-                            stream_manager._close_socket()
-
-                # Wait for stream thread to finish
-                stream_thread_name = f"stream-{channel_id}"
-                stream_thread = None
-
-                for thread in threading.enumerate():
-                    if thread.name == stream_thread_name:
-                        stream_thread = thread
-                        break
-
-                if stream_thread and stream_thread.is_alive():
-                    logger.info(f"Waiting for stream thread to terminate")
-                    try:
-                        # Very short timeout to prevent hanging the app
-                        stream_thread.join(timeout=2.0)
-                        if stream_thread.is_alive():
-                            logger.warning(f"Stream thread did not terminate within timeout")
-                    except RuntimeError:
-                        logger.debug(f"Could not join stream thread (may be current thread)")
-
-                # Stop all output format managers before releasing ownership
-                self.stop_all_output_formats(channel_id)
-
-                # Stop all output profile transcodes before releasing ownership
-                self.stop_all_output_profiles(channel_id)
-
+            was_owner = self.am_i_owner(channel_id)
+            if was_owner:
+                logger.info(
+                    f"This worker ({self.worker_id}) is the owner - closing provider connection"
+                )
                 stop_event_data = self._collect_channel_stop_event_data(channel_id)
 
-                # Release ownership
+            # Stop new chunk writes before Redis cleanup; do not block on ffmpeg join yet.
+            self._signal_upstream_shutdown(channel_id)
+
+            # Release profile slots and delete Redis keys before any blocking local stop.
+            # A concurrent disconnect + cleanup-thread stop used to wedge here behind a
+            # 2s stderr join, never reaching scan/delete (bare buffer:index with TTL -1).
+            self._clean_redis_keys(channel_id)
+            redis_cleaned = True
+
+            if was_owner:
                 self.release_ownership(channel_id)
 
-            # Always clean up local resources before Redis/logging so teardown
-            # cannot be blocked by connect integrations or DB event writes.
-            self._cleanup_stopped_channel_local(channel_id)
-            self._clean_redis_keys(channel_id)
+            self._stop_local_stream_activity(channel_id)
             self._spawn_channel_stop_event(stop_event_data)
 
             return True
@@ -1549,6 +1582,13 @@ class ProxyServer:
             logger.error(f"Error stopping channel {channel_id}: {e}")
             return False
         finally:
+            if not redis_cleaned:
+                try:
+                    self._clean_redis_keys(channel_id)
+                except Exception as e:
+                    logger.error(
+                        f"Error cleaning Redis keys for channel {channel_id} during finally: {e}"
+                    )
             self._stopping_channels.discard(channel_id)
             self._stopping_since.pop(channel_id, None)
 
@@ -1595,7 +1635,11 @@ class ProxyServer:
                     self._recover_stuck_channel_stops()
 
                     # Create a unified list of all channels we have locally
-                    all_local_channels = set(self.stream_managers.keys()) | set(self.client_managers.keys())
+                    all_local_channels = (
+                        set(self.stream_managers.keys())
+                        | set(self.client_managers.keys())
+                        | set(self._live_stream_managers.keys())
+                    )
 
                     # Single loop through all channels - process each exactly once
                     for channel_id in list(all_local_channels):
@@ -1764,7 +1808,8 @@ class ProxyServer:
                             # === NON-OWNER CHANNEL HANDLING ===
                             # Safety: if we have a stream_manager, we ARE the real owner
                             # but the Redis key may have expired. Try to re-acquire.
-                            if channel_id in self.stream_managers:
+                            if (channel_id in self.stream_managers
+                                    or channel_id in self._live_stream_managers):
                                 # Ownership was explicitly released by an active stop_channel call -
                                 # don't fight the shutdown by trying to re-acquire.
                                 if channel_id in self._stopping_channels:
@@ -1930,11 +1975,7 @@ class ProxyServer:
                             # Orphaned channel with no clients - clean it up
                             logger.info(f"Cleaning up orphaned channel {channel_id}")
 
-                            if self._has_local_stream_activity(channel_id):
-                                logger.info(
-                                    f"Orphaned channel {channel_id} has local stream activity - stopping processes"
-                                )
-                                self._stop_local_stream_activity(channel_id)
+                            self._stop_upstream_before_redis_cleanup(channel_id)
                             self._clean_redis_keys(channel_id)
                 except Exception as e:
                     logger.error(f"Error processing channel key {key}: {e}")
@@ -1964,8 +2005,7 @@ class ProxyServer:
                     if not metadata:
                         # Empty metadata - clean it up
                         logger.warning(f"Found empty metadata for channel {channel_id} - cleaning up")
-                        if self._has_local_stream_activity(channel_id):
-                            self._stop_local_stream_activity(channel_id)
+                        self._stop_upstream_before_redis_cleanup(channel_id)
                         self._clean_redis_keys(channel_id)
                         continue
 
@@ -1987,11 +2027,7 @@ class ProxyServer:
                         state = metadata.get('state', 'unknown')
                         logger.warning(f"Found orphaned metadata for channel {channel_id} (state: {state}, owner: {owner}, clients: {client_count}) - cleaning up")
 
-                        if self._has_local_stream_activity(channel_id):
-                            logger.info(
-                                f"Channel {channel_id} has local stream activity - stopping processes"
-                            )
-                            self._stop_local_stream_activity(channel_id)
+                        self._stop_upstream_before_redis_cleanup(channel_id)
                         self._clean_redis_keys(channel_id)
                     elif not owner_alive and client_count > 0:
                         # SCARD may include ghost entries from a dead worker's
@@ -2008,8 +2044,7 @@ class ProxyServer:
                                 f"owner: {owner}) had {client_count} ghost client(s) "
                                 f"- cleaning up"
                             )
-                            if self._has_local_stream_activity(channel_id):
-                                self._stop_local_stream_activity(channel_id)
+                            self._stop_upstream_before_redis_cleanup(channel_id)
                             self._clean_redis_keys(channel_id)
                         else:
                             if self._channel_teardown_active(channel_id):
@@ -2018,8 +2053,7 @@ class ProxyServer:
                                     f"{real_count} client(s) during teardown; forcing cleanup"
                                 )
                                 self._force_clear_channel_clients(channel_id)
-                                if self._has_local_stream_activity(channel_id):
-                                    self._stop_local_stream_activity(channel_id)
+                                self._stop_upstream_before_redis_cleanup(channel_id)
                                 self._clean_redis_keys(channel_id)
                             else:
                                 logger.warning(
@@ -2036,7 +2070,11 @@ class ProxyServer:
 
     def _clean_redis_keys(self, channel_id):
         """Clean up all Redis keys for a channel more efficiently"""
-        # Release the channel, stream, and profile keys from the channel
+        total_deleted = 0
+
+        # Release the M3U profile slot while channel_stream / metadata still exist.
+        # Scanning live:channel keys first deletes metadata and breaks release_stream()
+        # fallback, leaving profile_connections counters stuck (e.g. profile_connections:70 = 1).
         try:
             channel = Channel.objects.get(uuid=channel_id)
             if not channel.release_stream():
@@ -2049,36 +2087,29 @@ class ProxyServer:
             except (Stream.DoesNotExist, Exception):
                 logger.debug(f"No Channel or Stream found for {channel_id}")
 
-        if not self.redis_client:
-            return 0
+        if self.redis_client:
+            try:
+                patterns = [
+                    f"live:channel:{channel_id}:*",
+                    RedisKeys.events_channel(channel_id),
+                ]
 
-        try:
-            # Define key patterns to scan for
-            patterns = [
-                f"live:channel:{channel_id}:*",  # All channel keys
-                RedisKeys.events_channel(channel_id)  # Event channel
-            ]
+                for pattern in patterns:
+                    cursor = 0
+                    while True:
+                        cursor, keys = self.redis_client.scan(cursor, match=pattern, count=100)
+                        if keys:
+                            self.redis_client.delete(*keys)
+                            total_deleted += len(keys)
 
-            total_deleted = 0
+                        if cursor == 0:
+                            break
 
-            for pattern in patterns:
-                cursor = 0
-                while True:
-                    cursor, keys = self.redis_client.scan(cursor, match=pattern, count=100)
-                    if keys:
-                        self.redis_client.delete(*keys)
-                        total_deleted += len(keys)
+                logger.info(f"Cleaned up {total_deleted} Redis keys for channel {channel_id}")
+            except Exception as e:
+                logger.error(f"Error cleaning Redis keys for channel {channel_id}: {e}")
 
-                    # Exit when cursor returns to 0
-                    if cursor == 0:
-                        break
-
-            logger.info(f"Cleaned up {total_deleted} Redis keys for channel {channel_id}")
-            return total_deleted
-
-        except Exception as e:
-            logger.error(f"Error cleaning Redis keys for channel {channel_id}: {e}")
-            return 0
+        return total_deleted
 
     def refresh_channel_registry(self):
         """Refresh TTL for active channels using standard keys"""
diff --git a/apps/proxy/live_proxy/services/channel_service.py b/apps/proxy/live_proxy/services/channel_service.py
index 6671e444..dcd0b5e1 100644
--- a/apps/proxy/live_proxy/services/channel_service.py
+++ b/apps/proxy/live_proxy/services/channel_service.py
@@ -73,6 +73,9 @@ class ChannelService:
     def is_channel_teardown_active(channel_id):
         """True when a coordinated channel stop is in progress (visible to all workers)."""
         proxy_server = ProxyServer.get_instance()
+        if channel_id in proxy_server._stopping_channels:
+            return True
+
         if not proxy_server.redis_client:
             return False
 
@@ -345,41 +348,19 @@ class ChannelService:
             except Exception as e:
                 logger.error(f"Error fetching channel state: {e}")
 
-        # Mark stopping in Redis and notify all workers before local teardown
+        # Mark stopping in Redis and notify all workers before local teardown.
+        # stop_channel() releases profile slots via _clean_redis_keys() before Redis deletion.
         if proxy_server.redis_client:
             ChannelService.mark_channel_stopping(channel_id, broadcast=True)
             logger.info(f"Marked channel {channel_id} stopping and broadcast stop to all workers")
-            local_result = proxy_server.stop_channel(channel_id)
-        else:
-            local_result = proxy_server.stop_channel(channel_id)
-
-        # Release the channel in the channel model if applicable
-        try:
-            channel = Channel.objects.get(uuid=channel_id)
-            model_released = channel.release_stream()
-            if model_released:
-                logger.info(f"Released channel {channel_id} stream allocation")
-            else:
-                logger.warning(f"Channel {channel_id}: release_stream found no keys to clean")
-        except (Channel.DoesNotExist, Exception):
-            logger.warning(f"Could not find Channel model for UUID {channel_id}, attempting stream hash")
-            try:
-                stream = Stream.objects.get(stream_hash=channel_id)
-                model_released = stream.release_stream()
-                if model_released:
-                    logger.info(f"Released stream {channel_id} stream allocation")
-                else:
-                    logger.warning(f"Stream {channel_id}: release_stream found no keys to clean")
-            except (Stream.DoesNotExist, Exception) as e:
-                logger.error(f"No Channel or Stream found for {channel_id}: {e}")
-                model_released = False
+        local_result = proxy_server.stop_channel(channel_id)
 
         return {
             'status': 'success',
             'message': 'Channel stop request sent',
             'channel_id': channel_id,
             'previous_state': channel_info,
-            'model_released': model_released,
+            'model_released': bool(local_result),
             'local_stop_result': local_result
         }
 
diff --git a/apps/proxy/live_proxy/views.py b/apps/proxy/live_proxy/views.py
index b547cefd..2119b4bd 100644
--- a/apps/proxy/live_proxy/views.py
+++ b/apps/proxy/live_proxy/views.py
@@ -211,6 +211,12 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
 
         # Start initialization if needed
         if needs_initialization or not proxy_server.check_if_channel_exists(channel_id):
+            if ChannelService.is_channel_unavailable_for_new_clients(channel_id):
+                logger.info(
+                    f"[{client_id}] Channel {channel_id} became unavailable before init, rejecting"
+                )
+                return _channel_stopping_response()
+
             logger.info(f"[{client_id}] Starting channel {channel_id} initialization")
             # Force cleanup of any previous instance if in terminal state
             if channel_state in [
@@ -433,6 +439,16 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
                     )  # 502 Bad Gateway
 
             # Initialize channel with the stream's user agent (not the client's)
+            if ChannelService.is_channel_unavailable_for_new_clients(channel_id):
+                if connection_allocated:
+                    if not channel.release_stream():
+                        logger.warning(f"[{client_id}] Failed to release stream before teardown reject")
+                    connection_allocated = False
+                logger.info(
+                    f"[{client_id}] Channel {channel_id} unavailable before init call, rejecting"
+                )
+                return _channel_stopping_response()
+
             success = ChannelService.initialize_channel(
                 channel_id,
                 stream_url,
@@ -541,6 +557,16 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
                 f"[{client_id}] Successfully initialized channel {channel_id} locally"
             )
 
+        if ChannelService.is_channel_unavailable_for_new_clients(channel_id):
+            if _client_pre_registered:
+                mgr = proxy_server.client_managers.get(channel_id)
+                if mgr:
+                    mgr.remove_client(client_id)
+            logger.info(
+                f"[{client_id}] Channel {channel_id} became unavailable during setup, rejecting"
+            )
+            return _channel_stopping_response()
+
         # Register client
         output_profile = _resolve_output_profile(request, user)
         if output_profile: