diff --git a/kalshi/perps/ws/channels.py b/kalshi/perps/ws/channels.py
index 64e6e01..62e792e 100644
--- a/kalshi/perps/ws/channels.py
+++ b/kalshi/perps/ws/channels.py
@@ -181,6 +181,13 @@ async def _wait_for_response(
             if self._stashing:
                 self._maybe_stash(raw, data)
             else:
+                # By design (#405): non-matching data frames are stashed only
+                # during resubscribe_all (rebuilding a book from scratch, where a
+                # lost frame would corrupt the fresh book with no baseline to
+                # gap-detect against). During a normal command the book is
+                # already established, so a dropped sequenced frame surfaces as a
+                # sequence gap on the next frame and is recovered — not silently
+                # lost. Discarding here avoids accumulating stale state.
                 logger.debug(
                     "Discarding non-matching frame during command: type=%s",
                     data.get("type"),
diff --git a/kalshi/perps/ws/client.py b/kalshi/perps/ws/client.py
index fe4c6cf..3c261c8 100644
--- a/kalshi/perps/ws/client.py
+++ b/kalshi/perps/ws/client.py
@@ -69,9 +69,11 @@
 # Cooperative-pause polling interval for the recv loop.
 _RECV_POLL_S: float = 0.05
 
-# Message types that ride the perps margin order book and carry ``seq``. This
-# set lets the recv loop gate orderbook-apply + stale-sid checks by wire type.
-_ORDERBOOK_TYPES = ("orderbook_snapshot", "orderbook_delta")
+# Sequenced data frames — they carry ``seq`` and advance the per-sid watermark,
+# so the recv loop must stale-sid-gate them for unmapped sids (orderbook frames
+# AND order_group_updates). Non-sequenced channels (ticker/trade/fill/
+# user_orders) carry no seq and are handled at the dispatch layer instead.
+_SEQUENCED_TYPES = ("orderbook_snapshot", "orderbook_delta", "order_group_updates")
 
 
 class PerpsWebSocket:
@@ -348,7 +350,7 @@ async def _process_frame(self, raw: str) -> None:
         The orderbook-apply seam validates ``orderbook_snapshot`` /
         ``orderbook_delta`` into their concrete payload models and applies them
         to the local book before dispatch; the stale-sid gate and seq tracking
-        below run for orderbook frames.
+        below run for sequenced frames (orderbook_* + order_group_updates).
         """
         assert self._dispatcher is not None
         data = self._json_loads(raw)
@@ -356,15 +358,18 @@ async def _process_frame(self, raw: str) -> None:
         seq = data.get("seq")
         msg_type = data.get("type", "")
 
-        # Stale orderbook frames arriving after teardown must not be processed.
+        # Stale sequenced frames (orderbook_* + order_group_updates) arriving for
+        # an unmapped sid after teardown must not be processed: tracking their
+        # seq would re-arm the watermark for a dead/recycled sid and manufacture
+        # a spurious gap on a later subscription to that sid.
         if (
-            msg_type in _ORDERBOOK_TYPES
+            msg_type in _SEQUENCED_TYPES
             and isinstance(sid, int)
             and self._sub_mgr is not None
             and self._sub_mgr.get_subscription_by_sid(sid) is None
         ):
             logger.debug(
-                "Dropping stale %s for unmapped sid=%d (post-teardown race)",
+                "Dropping stale sequenced %s for unmapped sid=%d (post-teardown race)",
                 msg_type, sid,
             )
             return
diff --git a/kalshi/ws/client.py b/kalshi/ws/client.py
index 4db7626..9793412 100644
--- a/kalshi/ws/client.py
+++ b/kalshi/ws/client.py
@@ -215,6 +215,12 @@ async def _stop(self) -> None:
                     self._recv_task.cancel()
                     with contextlib.suppress(asyncio.CancelledError, Exception):
                         await self._recv_task
+            elif self._recv_task is not None:
+                # #413: task already finished (e.g. the recv loop raised on a
+                # permanent close code). Retrieve the stored exception so asyncio
+                # doesn't log "Task exception was never retrieved" when it is GC'd.
+                with contextlib.suppress(asyncio.CancelledError, Exception):
+                    self._recv_task.exception()
 
             # Sentinels last (#357): iterator consumers have already seen
             # the post-close drained frames; the sentinel now terminates
diff --git a/tests/perps/ws/test_perps_ws_channels.py b/tests/perps/ws/test_perps_ws_channels.py
index 6e8ea6f..6e958d3 100644
--- a/tests/perps/ws/test_perps_ws_channels.py
+++ b/tests/perps/ws/test_perps_ws_channels.py
@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import asyncio
+import json
 from decimal import Decimal
 
 import pytest
@@ -141,6 +142,30 @@ async def test_subscribe_order_group_yields_typed_message(
         assert frame.msg.contracts_limit == Decimal("150.00")
 
 
+async def test_stale_order_group_updates_for_unmapped_sid_dropped(
+    fake_perps_ws: FakePerpsWS, perps_ws_config: PerpsConfig, perps_auth
+) -> None:
+    # #411: order_group_updates is sequenced (carries `seq`), so a stale frame
+    # for a sid with no current subscription must be dropped BEFORE seq tracking.
+    # Otherwise track_sync() re-arms the watermark for the dead/recycled sid and
+    # manufactures a spurious gap on a later subscription to that sid.
+    ws = PerpsWebSocket(auth=perps_auth, config=perps_ws_config)
+    async with ws.connect() as session:
+        assert session._sub_mgr is not None and session._seq_tracker is not None
+        unmapped_sid = 9999
+        assert session._sub_mgr.get_subscription_by_sid(unmapped_sid) is None
+        raw = json.dumps({
+            "type": "order_group_updates", "sid": unmapped_sid, "seq": 7,
+            "msg": {
+                "event_type": "limit_updated", "order_group_id": "og",
+                "contracts_limit_fp": "10.00", "ts_ms": 1700000000000,
+            },
+        })
+        await session._process_frame(raw)
+        # The stale frame must NOT have armed the watermark for the unmapped sid.
+        assert session._seq_tracker.peek(unmapped_sid) is None
+
+
 async def test_subscribe_orderbook_delta_snapshot_then_delta(
     fake_perps_ws: FakePerpsWS, perps_ws_config: PerpsConfig, perps_auth
 ) -> None:
diff --git a/tests/ws/test_client.py b/tests/ws/test_client.py
index 3ed771c..9da4abf 100644
--- a/tests/ws/test_client.py
+++ b/tests/ws/test_client.py
@@ -3,12 +3,14 @@
 from __future__ import annotations
 
 import asyncio
+import gc
+import logging
 from typing import Any
 
 import pytest
 
 from kalshi.config import KalshiConfig
-from kalshi.errors import KalshiSubscriptionError
+from kalshi.errors import KalshiConnectionError, KalshiSubscriptionError
 from kalshi.ws.client import KalshiWebSocket, _WebSocketSession
 from kalshi.ws.connection import ConnectionState
 from tests._model_fixtures import (
@@ -1084,3 +1086,31 @@ async def tracking_sentinel(
             "Sentinels must broadcast even when _connection.close() raises; "
             "otherwise iterator consumers hang waiting on the closed queue."
         )
+
+
+class TestIssue413StopRetrievesException:
+    """#413: ``_stop()`` retrieves an already-finished recv task's exception."""
+
+    async def test_stop_retrieves_dead_recv_task_exception(
+        self,
+        test_auth,  # type: ignore[no-untyped-def]
+        caplog: pytest.LogCaptureFixture,
+    ) -> None:
+        # When the recv loop already finished with an exception (e.g. a permanent
+        # close code), _stop must retrieve it — otherwise asyncio logs "Task
+        # exception was never retrieved" when the task is garbage-collected.
+        ws = KalshiWebSocket(auth=test_auth, config=KalshiConfig(timeout=5.0))
+
+        async def _boom() -> None:
+            raise KalshiConnectionError("permanent close")
+
+        task: asyncio.Task[None] = asyncio.ensure_future(_boom())
+        await asyncio.sleep(0)  # let it finish; exception now stored, unretrieved
+        assert task.done()
+        ws._recv_task = task
+
+        with caplog.at_level(logging.ERROR, logger="asyncio"):
+            await ws._stop()  # nils ws._recv_task, releasing its only other ref
+            del task
+            gc.collect()
+        assert "never retrieved" not in caplog.text.lower()