diff --git a/gateway/platforms/helpers.py b/gateway/platforms/helpers.py
index 18d97fcb7a..17bc490174 100644
--- a/gateway/platforms/helpers.py
+++ b/gateway/platforms/helpers.py
@@ -57,6 +57,15 @@ class MessageDeduplicator:
         if len(self._seen) > self._max_size:
             cutoff = now - self._ttl
             self._seen = {k: v for k, v in self._seen.items() if v > cutoff}
+            if len(self._seen) > self._max_size:
+                # TTL pruning alone does not cap the cache when every entry is
+                # still fresh. Keep the newest entries so the helper's
+                # max_size bound is enforced under sustained traffic.
+                newest = sorted(
+                    self._seen.items(),
+                    key=lambda item: item[1],
+                )[-self._max_size:]
+                self._seen = dict(newest)
         return False
 
     def clear(self):
diff --git a/tests/gateway/test_message_deduplicator.py b/tests/gateway/test_message_deduplicator.py
index 59fe7e3949..4a140f2761 100644
--- a/tests/gateway/test_message_deduplicator.py
+++ b/tests/gateway/test_message_deduplicator.py
@@ -77,6 +77,19 @@ class TestMessageDeduplicatorTTL:
         assert "old-0" not in dedup._seen
         assert "new-0" in dedup._seen
 
+    def test_max_size_eviction_caps_fresh_entries(self):
+        """Fresh entries must still be capped to max_size on overflow."""
+        dedup = MessageDeduplicator(max_size=2, ttl_seconds=60)
+
+        dedup.is_duplicate("msg-1")
+        dedup.is_duplicate("msg-2")
+        dedup.is_duplicate("msg-3")
+
+        assert len(dedup._seen) == 2
+        assert "msg-1" not in dedup._seen
+        assert "msg-2" in dedup._seen
+        assert "msg-3" in dedup._seen
+
     def test_ttl_zero_means_no_dedup(self):
         """With TTL=0, all entries expire immediately."""
         dedup = MessageDeduplicator(ttl_seconds=0)