Fix MessageDeduplicator max_size enforcement

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):

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)