diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml
index 41790485d..b889a7e4e 100644
--- a/.github/workflows/python-package.yml
+++ b/.github/workflows/python-package.yml
@@ -70,7 +70,10 @@ jobs:
   test-pypy:
     name: 'Python pypy3.11/Cython: false'
     runs-on: ubuntu-latest
-    timeout-minutes: 10
+    # PyPy runs the pure-Python paths and is markedly slower than CPython, and
+    # this leg now runs the formerly-skipped tests too; give it headroom over
+    # the old 10-minute cap, which already clipped the run at ~96%.
+    timeout-minutes: 15
     continue-on-error: true
     env:
       USE_CYTHON: 'false'
diff --git a/faust/assignor/copartitioned_assignor.py b/faust/assignor/copartitioned_assignor.py
index ada471793..34969e4fd 100644
--- a/faust/assignor/copartitioned_assignor.py
+++ b/faust/assignor/copartitioned_assignor.py
@@ -259,8 +259,39 @@ def _assign_round_robin(self, unassigned: Iterable[int], active: bool) -> None:
                         and assigment.can_assign(partition, active)
                     )
                 )  # By above assertion, should never throw error
-                unassigned_partition = assign_to.pop_partition(active)
+                unassigned_partition = self._free_up_partition(assign_to, active)
                 unassigned.append(unassigned_partition)
 
             # Assign partition
             assign_to.assign_partition(partition, active)
+
+    def _free_up_partition(
+        self, assignment: CopartitionedAssignment, active: bool
+    ) -> int:
+        # Evict a partition from an exhausted client so the partition we are
+        # placing can take its slot.  Prefer a victim that some *other*
+        # not-yet-exhausted client can immediately take, so the re-queued
+        # victim makes forward progress instead of bouncing straight back
+        # here.  ``CopartitionedAssignment.pop_partition`` pops an arbitrary
+        # set element, and when the arbitrary choice is the victim's only
+        # possible home two partitions ping-pong between the queue and this
+        # client forever.  Whether that happens depends on set iteration
+        # order, so the livelock strikes on PyPy while CPython (with a
+        # different order) escapes it for the same input.  Choosing a
+        # re-homable victim in deterministic (sorted) order makes the loop
+        # terminate on every interpreter.
+        held = assignment.get_assigned_partitions(active)
+        for candidate in sorted(held):
+            for other in self._client_assignments.values():
+                if other is assignment:
+                    continue
+                if not self._client_exhausted(other, active) and other.can_assign(
+                    candidate, active
+                ):
+                    assignment.unassign_partition(candidate, active)
+                    return candidate
+        # No immediately re-homable victim: fall back to a deterministic
+        # arbitrary choice (still correct, just not progress-guaranteed).
+        candidate = min(held)
+        assignment.unassign_partition(candidate, active)
+        return candidate
diff --git a/faust/streams.py b/faust/streams.py
index 9f4286a3f..784bdf376 100644
--- a/faust/streams.py
+++ b/faust/streams.py
@@ -7,8 +7,10 @@
 import weakref
 from asyncio import CancelledError
 from contextvars import ContextVar
+from functools import wraps
 from typing import (
     Any,
+    AsyncGenerator,
     AsyncIterable,
     AsyncIterator,
     Callable,
@@ -88,6 +90,30 @@ async def maybe_forward(value: Any, channel: ChannelT) -> Any:
     return value
 
 
+def _tracks_buffer_agen(
+    fun: Callable[..., AsyncIterable],
+) -> Callable[..., AsyncIterable]:
+    """Register buffering generators (``take()`` and friends) on the stream.
+
+    The cleanup for these generators -- acking consumed events, restoring
+    ``enable_acks`` and detaching the buffering processor -- lives in
+    ``finally`` blocks that only run once the generator is finalized.  When a
+    caller abandons the generator (``break`` inside ``async for``), CPython's
+    reference counting finalizes it right away, but PyPy defers finalization
+    to the next major GC cycle, which may never come.  Tracking every
+    generator in a WeakSet lets ``Stream.on_stop`` close leftovers
+    explicitly, making cleanup deterministic on any interpreter.
+    """
+
+    @wraps(fun)
+    def _create_agen(self: StreamT, *args: Any, **kwargs: Any) -> AsyncIterable:
+        agen = fun(self, *args, **kwargs)
+        cast("Stream", self)._active_agens.add(agen)
+        return agen
+
+    return _create_agen
+
+
 class _LinkedListDirection(NamedTuple):
     attr: str
     getter: Callable[[StreamT], Optional[StreamT]]
@@ -149,6 +175,13 @@ def __init__(
         self._processors = list(processors) if processors else []
         self._on_start = on_start
 
+        # Live async generators handed out by this stream --
+        # __aiter__ iterators and the take() family -- closed explicitly
+        # in on_stop().  A WeakSet so that on CPython an abandoned
+        # generator is still finalized (and thus cleaned up) promptly by
+        # reference counting.
+        self._active_agens: "weakref.WeakSet[AsyncGenerator]" = weakref.WeakSet()
+
         # attach beacon to channel, or if iterable attach to current task.
         task = current_task(loop=self.loop)
         if task is not None:
@@ -300,6 +333,7 @@ async def events(self) -> AsyncIterable[EventT]:
             if self.current_event is not None:
                 yield self.current_event
 
+    @_tracks_buffer_agen
     async def take(self, max_: int, within: Seconds) -> AsyncIterable[Sequence[T_co]]:
         """Buffer n values at a time and yield a list of buffered values.
 
@@ -392,6 +426,7 @@ async def add_to_buffer(value: T) -> T:
             self.enable_acks = stream_enable_acks
             self._processors.remove(add_to_buffer)
 
+    @_tracks_buffer_agen
     async def take_events(
         self, max_: int, within: Seconds
     ) -> AsyncIterable[Sequence[EventT]]:
@@ -485,6 +520,7 @@ async def add_to_buffer(value: T) -> T:
             self.enable_acks = stream_enable_acks
             self._processors.remove(add_to_buffer)
 
+    @_tracks_buffer_agen
     async def take_with_timestamp(
         self, max_: int, within: Seconds, timestamp_field_name: str
     ) -> AsyncIterable[Sequence[T_co]]:
@@ -592,6 +628,7 @@ def enumerate(self, start: int = 0) -> AsyncIterable[Tuple[int, T_co]]:
         """
         return aenumerate(self, start)
 
+    @_tracks_buffer_agen
     async def noack_take(
         self, max_: int, within: Seconds
     ) -> AsyncIterable[Sequence[T_co]]:
@@ -1033,6 +1070,23 @@ async def stop(self) -> None:
 
     async def on_stop(self) -> None:
         """Signal that the stream is stopping."""
+        # Close any buffering generator (take() and friends) the caller
+        # abandoned, so their ``finally`` blocks -- acking consumed events,
+        # restoring ``enable_acks`` and detaching the buffering processor --
+        # run deterministically now instead of whenever the interpreter
+        # finalizes the generator (which on PyPy waits for a GC cycle that
+        # may never come).  ``aclose()`` on an exhausted generator is a
+        # harmless no-op.
+        for agen in list(self._active_agens):
+            try:
+                await agen.aclose()
+            except (RuntimeError, ValueError):
+                # Generator is still running in another task (e.g. an agent
+                # actor iterating this stream during a rebalance); its owner
+                # will clean it up.  CPython raises RuntimeError for this,
+                # PyPy raises ValueError ("async generator already
+                # executing") -- catch both so on_stop never aborts here.
+                pass
         self._passive = False
         self._passive_started.clear()
         for table_or_stream in self.combined:
@@ -1046,9 +1100,14 @@ def __next__(self) -> T:
 
     def __aiter__(self) -> AsyncIterator[T_co]:  # pragma: no cover
         if _CStreamIterator is not None:
-            return self._c_aiter()
+            it = self._c_aiter()
         else:
-            return self._py_aiter()
+            it = self._py_aiter()
+        # Track the iterator so on_stop() can close it if the caller
+        # abandons it (see _tracks_buffer_agen for why this matters on
+        # interpreters without reference counting, e.g. PyPy).
+        self._active_agens.add(cast(AsyncGenerator, it))
+        return it
 
     async def _c_aiter(self) -> AsyncIterator[T_co]:  # pragma: no cover
         self.log.dev("Using Cython optimized __aiter__")
diff --git a/tests/conftest.py b/tests/conftest.py
index 4ce891806..eb256e50e 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,4 +1,7 @@
+import asyncio
+import gc
 import os
+import platform
 import threading
 import time
 from http import HTTPStatus
@@ -44,6 +47,27 @@ def loop(event_loop):
     return event_loop
 
 
+@pytest.fixture()
+def event_loop(request):
+    """Per-test event loop (overrides pytest-asyncio's default).
+
+    Identical to the pytest-asyncio 0.21 fixture, except that on PyPy we
+    finalize leftover async generators while the loop is still alive.
+    PyPy has no reference counting, so an abandoned async generator is
+    only finalized at a later GC cycle -- by which time this loop is
+    closed and its cleanup lands on a dead loop (leaking tasks into later
+    tests and misattributing warnings).  A gc pass plus
+    ``shutdown_asyncgens()`` inside the owning test keeps the cleanup
+    here, matching CPython's prompt-finalization behavior.
+    """
+    loop = asyncio.get_event_loop_policy().new_event_loop()
+    yield loop
+    if platform.python_implementation() == "PyPy":
+        gc.collect()
+        loop.run_until_complete(loop.shutdown_asyncgens())
+    loop.close()
+
+
 class _patching(object):
     def __init__(self, monkeypatch, request):
         self.monkeypatch = monkeypatch
diff --git a/tests/functional/test_streams.py b/tests/functional/test_streams.py
index 851594155..d9439887e 100644
--- a/tests/functional/test_streams.py
+++ b/tests/functional/test_streams.py
@@ -1,5 +1,5 @@
 import asyncio
-import platform
+from contextlib import suppress
 from copy import copy
 from unittest.mock import Mock, patch
 
@@ -40,9 +40,6 @@ def _prepare_app(app):
     return app
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 @pytest.mark.asyncio
 @pytest.mark.allow_lingering_tasks(count=1)
 async def test_simple(app, loop):
@@ -54,9 +51,6 @@ async def test_simple(app, loop):
         assert await channel_empty(stream.channel)
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 @pytest.mark.asyncio
 async def test_async_iterator(app):
     async with new_stream(app) as stream:
@@ -71,9 +65,6 @@ async def test_async_iterator(app):
         assert await channel_empty(stream.channel)
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 @pytest.mark.asyncio
 async def test_throw(app):
     async with new_stream(app) as stream:
@@ -85,9 +76,6 @@ async def test_throw(app):
             await anext(streamit)
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 @pytest.mark.asyncio
 async def test_enumerate(app):
     async with new_stream(app) as stream:
@@ -102,9 +90,6 @@ async def test_enumerate(app):
         assert await channel_empty(stream.channel)
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 @pytest.mark.asyncio
 async def test_items(app):
     async with new_stream(app) as stream:
@@ -120,9 +105,6 @@ async def test_items(app):
         assert await channel_empty(stream.channel)
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 @pytest.mark.asyncio
 async def test_through(app):
     app._attachments.enabled = False
@@ -255,9 +237,6 @@ async def test_stream_filter_acks_filtered_out_messages(app, event_loop):
     assert len(app.consumer.unacked) == 0
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 @pytest.mark.asyncio
 async def test_acks_filtered_out_messages_when_using_take(app, event_loop):
     """
@@ -282,9 +261,6 @@ async def test_acks_filtered_out_messages_when_using_take(app, event_loop):
     assert len(acked) == len(initial_values)
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 @pytest.mark.asyncio
 async def test_events(app):
     async with new_stream(app) as stream:
@@ -321,9 +297,6 @@ def assert_events_acked(events):
         raise
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 class Test_chained_streams:
     def _chain(self, app):
         root = new_stream(app)
@@ -427,9 +400,6 @@ async def assert_was_stopped(self, leader, followers):
             assert node._stopped.is_set()
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 @pytest.mark.asyncio
 async def test_start_and_stop_Stream(app):
     s = new_topic_stream(app)
@@ -445,9 +415,6 @@ async def _start_stop_stream(stream):
     await stream.stop()
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 @pytest.mark.asyncio
 async def test_ack(app):
     async with new_stream(app) as s:
@@ -456,7 +423,8 @@ async def test_ack(app):
         await s.channel.send(value=2)
         event = None
         i = 1
-        async for value in s:
+        it = aiter(s)
+        async for value in it:
             assert value == i
             i += 1
             last_to_ack = value == 2
@@ -464,18 +432,12 @@ async def test_ack(app):
             if value == 2:
                 break
         assert event
-        # need one sleep on Python 3.6.0-3.6.6 + 3.7.0
-        # need two sleeps on Python 3.6.7 + 3.7.1 :-/
-        await asyncio.sleep(0)  # needed for some reason
-        await asyncio.sleep(0)  # needed for some reason
+        await wait_for_stream_ack(event, it)
         if not event.ack.called:
             assert event.message.acked
             assert not event.message.refcount
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 @pytest.mark.asyncio
 async def test_noack(app):
     async with new_stream(app) as s:
@@ -496,9 +458,6 @@ async def test_noack(app):
         event.ack.assert_not_called()
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 @pytest.mark.asyncio
 async def test_acked_when_raising(app):
     async with new_stream(app) as s:
@@ -506,39 +465,32 @@ async def test_acked_when_raising(app):
         await s.channel.send(value=2)
 
         event1 = None
+        it1 = aiter(s)
         with pytest.raises(RuntimeError):
-            async for value in s:
+            async for value in it1:
                 event1 = mock_stream_event_ack(s)
                 assert value == 1
                 raise RuntimeError
         assert event1
-        # need one sleep on Python 3.6.0-3.6.6 + 3.7.0
-        # need two sleeps on Python 3.6.7 + 3.7.1 :-/
-        await asyncio.sleep(0)  # needed for some reason
-        await asyncio.sleep(0)  # needed for some reason
+        await wait_for_stream_ack(event1, it1)
         if not event1.ack.called:
             assert event1.message.acked
             assert not event1.message.refcount
 
         event2 = None
+        it2 = aiter(s)
         with pytest.raises(RuntimeError):
-            async for value in s:
+            async for value in it2:
                 event2 = mock_stream_event_ack(s)
                 assert value == 2
                 raise RuntimeError
         assert event2
-        # need one sleep on Python 3.6.0-3.6.6 + 3.7.0
-        # need two sleeps on Python 3.6.7 + 3.7.1 :-/
-        await asyncio.sleep(0)  # needed for some reason
-        await asyncio.sleep(0)  # needed for some reason
+        await wait_for_stream_ack(event2, it2)
         if not event2.ack.called:
             assert event2.message.acked
             assert not event2.message.refcount
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 @pytest.mark.asyncio
 @pytest.mark.allow_lingering_tasks(count=1)
 async def test_maybe_forward__when_event(app):
@@ -551,9 +503,6 @@ async def test_maybe_forward__when_event(app):
         s.channel.send.assert_not_called()
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 @pytest.mark.asyncio
 async def test_maybe_forward__when_concrete_value(app):
     s = new_stream(app)
@@ -572,6 +521,23 @@ def mock_event_ack(event, return_value=False):
     return event
 
 
+async def wait_for_stream_ack(event, agen):
+    """Ensure a consumed event has been acked, deterministically.
+
+    The ack runs in the ``take()``/iterator generator's ``finally`` block,
+    which only executes when the generator is finalized.  Abandoning the
+    generator (``break`` or an exception in ``async for``) defers that to
+    interpreter finalization -- immediate on CPython via reference counting,
+    but only at a later (maybe never) GC cycle on PyPy.  Closing the generator
+    explicitly runs its ``finally`` now, on every interpreter; ``aclose()`` on
+    an already-finished generator is a harmless no-op.  A still-running
+    generator raises RuntimeError on CPython / ValueError on PyPy; both are
+    ignored (its owner will finalize it).
+    """
+    with suppress(RuntimeError, ValueError):
+        await agen.aclose()
+
+
 async def get_event_from_value(stream, value, key=None):
     await stream.channel.send(key=key, value=value)
     async for value in stream:
@@ -759,17 +725,15 @@ async def test_take(app):
         assert s.enable_acks is True
         await s.channel.send(value=1)
         event = None
-        async for value in s.take(1, within=1):
+        buffer = s.take(1, within=1)
+        async for value in buffer:
             assert value == [1]
             assert s.enable_acks is False
             event = mock_stream_event_ack(s)
             break
 
         assert event
-        # need one sleep on Python 3.6.0-3.6.6 + 3.7.0
-        # need two sleeps on Python 3.6.7 + 3.7.1 :-/
-        await asyncio.sleep(0)
-        await asyncio.sleep(0)
+        await wait_for_stream_ack(event, buffer)
 
         if not event.ack.called:
             assert event.message.acked
@@ -777,6 +741,67 @@ async def test_take(app):
         assert s.enable_acks is True
 
 
+@pytest.mark.asyncio
+async def test_take_cleanup_on_stream_stop(app):
+    """An abandoned take() generator is closed when the stream stops.
+
+    Breaking out of ``async for`` leaves the generator suspended; its cleanup
+    (acking consumed events, restoring ``enable_acks``, detaching the
+    buffering processor) normally runs only when the interpreter finalizes
+    the generator.  Holding a strong reference here blocks finalization on
+    every interpreter -- mimicking PyPy, where finalization waits for a GC
+    cycle that may never come -- so this asserts the stream itself closes
+    leftover generators deterministically on stop.
+    """
+    async with new_stream(app) as s:
+        assert s.enable_acks is True
+        n_processors = len(s._processors)
+        await s.channel.send(value=1)
+        agen = s.take(1, within=1)  # hold a strong reference
+        event = None
+        async for value in agen:
+            assert value == [1]
+            assert s.enable_acks is False
+            event = mock_stream_event_ack(s)
+            break
+
+        assert event
+        # The reference above keeps the generator alive, so no interpreter
+        # has finalized it yet: cleanup must come from stopping the stream.
+        await s.stop()
+
+        if not event.ack.called:
+            assert event.message.acked
+        assert s.enable_acks is True
+        assert len(s._processors) == n_processors
+
+
+@pytest.mark.asyncio
+async def test_aiter_cleanup_on_stream_stop(app):
+    """An abandoned ``__aiter__`` generator is closed when the stream stops.
+
+    Same contract as ``test_take_cleanup_on_stream_stop`` but for the main
+    stream iterator: its inner ``finally`` acks the last yielded event, and
+    that only runs when the generator is finalized.  Holding a strong
+    reference blocks finalization on every interpreter (mimicking PyPy), so
+    the ack must come from the stream closing its tracked iterators on stop.
+    """
+    async with new_stream(app) as s:
+        await s.channel.send(value=1)
+        agen = aiter(s)  # hold a strong reference
+        event = None
+        async for value in agen:
+            assert value == 1
+            event = mock_stream_event_ack(s)
+            break
+
+        assert event
+        await s.stop()
+
+        if not event.ack.called:
+            assert event.message.acked
+
+
 @pytest.mark.asyncio
 async def test_take__10(app, loop):
     s = new_stream(app)
@@ -869,9 +894,10 @@ async def test_take_wit_timestamp(app):
         assert s.enable_acks is True
         await s.channel.send(value={"id": 1})
         event = None
-        async for value in s.take_with_timestamp(
+        buffer = s.take_with_timestamp(
             1, within=1, timestamp_field_name="test_timestamp"
-        ):
+        )
+        async for value in buffer:
             assert "test_timestamp" in value[0].keys()
             assert isinstance(value[0]["test_timestamp"], float)
             assert s.enable_acks is False
@@ -879,10 +905,7 @@ async def test_take_wit_timestamp(app):
             break
 
         assert event
-        # need one sleep on Python 3.6.0-3.6.6 + 3.7.0
-        # need two sleeps on Python 3.6.7 + 3.7.1 :-/
-        await asyncio.sleep(0)
-        await asyncio.sleep(0)
+        await wait_for_stream_ack(event, buffer)
 
         if not event.ack.called:
             assert event.message.acked
@@ -896,19 +919,17 @@ async def test_take_wit_timestamp_wit_simple_value(app):
         assert s.enable_acks is True
         await s.channel.send(value=1)
         event = None
-        async for value in s.take_with_timestamp(
+        buffer = s.take_with_timestamp(
             1, within=1, timestamp_field_name="test_timestamp"
-        ):
+        )
+        async for value in buffer:
             assert value == [1]
             assert s.enable_acks is False
             event = mock_stream_event_ack(s)
             break
 
         assert event
-        # need one sleep on Python 3.6.0-3.6.6 + 3.7.0
-        # need two sleeps on Python 3.6.7 + 3.7.1 :-/
-        await asyncio.sleep(0)
-        await asyncio.sleep(0)
+        await wait_for_stream_ack(event, buffer)
 
         if not event.ack.called:
             assert event.message.acked
@@ -922,19 +943,15 @@ async def test_take_wit_timestamp_without_timestamp_field(app):
         assert s.enable_acks is True
         await s.channel.send(value=1)
         event = None
-        async for value in s.take_with_timestamp(
-            1, within=1, timestamp_field_name=None
-        ):
+        buffer = s.take_with_timestamp(1, within=1, timestamp_field_name=None)
+        async for value in buffer:
             assert value == [1]
             assert s.enable_acks is False
             event = mock_stream_event_ack(s)
             break
 
         assert event
-        # need one sleep on Python 3.6.0-3.6.6 + 3.7.0
-        # need two sleeps on Python 3.6.7 + 3.7.1 :-/
-        await asyncio.sleep(0)
-        await asyncio.sleep(0)
+        await wait_for_stream_ack(event, buffer)
 
         if not event.ack.called:
             assert event.message.acked
diff --git a/tests/meticulous/assignor/test_copartitioned_assignor.py b/tests/meticulous/assignor/test_copartitioned_assignor.py
index 05c2f0ce8..e0fb1f66b 100644
--- a/tests/meticulous/assignor/test_copartitioned_assignor.py
+++ b/tests/meticulous/assignor/test_copartitioned_assignor.py
@@ -1,9 +1,7 @@
 import copy
-import platform
 from collections import Counter
 from typing import MutableMapping
 
-import pytest
 from hypothesis import assume, given, settings
 from hypothesis.strategies import integers
 
@@ -79,9 +77,6 @@ def client_removal_sticky(
     return True
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 @given(
     partitions=integers(min_value=0, max_value=256),
     replicas=integers(min_value=0, max_value=64),
@@ -100,9 +95,6 @@ def test_fresh_assignment(partitions, replicas, num_clients):
     assert is_valid(new_assignments, partitions, replicas)
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 @given(
     partitions=integers(min_value=0, max_value=256),
     replicas=integers(min_value=0, max_value=64),
@@ -139,9 +131,6 @@ def test_add_new_clients(partitions, replicas, num_clients, num_additional_clien
         assert clients_balanced(new_assignments)
 
 
-@pytest.mark.skipif(
-    platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-)
 @given(
     partitions=integers(min_value=0, max_value=256),
     replicas=integers(min_value=0, max_value=64),
@@ -179,3 +168,31 @@ def test_remove_clients(partitions, replicas, num_clients, num_removal_clients):
         assert client_removal_sticky(old_assignments, new_assignments)
     elif partitions >= num_clients - num_removal_clients:
         assert clients_balanced(new_assignments)
+
+
+@given(
+    partitions=integers(min_value=2, max_value=256),
+    num_clients=integers(min_value=2, max_value=64),
+    replicas=integers(min_value=1, max_value=8),
+)
+@settings(deadline=TEST_DEADLINE)
+def test_standby_assignment_terminates(partitions, num_clients, replicas):
+    # Regression test for a standby-assignment livelock.
+    #
+    # When the round-robin pass has to free up a slot on an exhausted client,
+    # it used to evict an arbitrary partition (``set.pop``).  If that arbitrary
+    # choice was the partition's only possible home, two partitions could
+    # ping-pong between the work queue and the same client forever.  Whether it
+    # happened depended on set iteration order, so the assignment hung on PyPy
+    # while completing on CPython for identical input.  Any ``replicas >= 1``
+    # case exercises the standby path; this asserts termination and a valid,
+    # fully-replicated assignment regardless of interpreter.
+    assume(replicas < num_clients)
+    client_assignments = {
+        str(client): CopartitionedAssignment(topics=_topics)
+        for client in range(num_clients)
+    }
+    new_assignments = CopartitionedAssignor(
+        _topics, client_assignments, partitions, replicas=replicas
+    ).get_assignment()
+    assert is_valid(new_assignments, partitions, replicas)
diff --git a/tests/unit/agents/test_agent.py b/tests/unit/agents/test_agent.py
index 25f033ac6..54f881f68 100644
--- a/tests/unit/agents/test_agent.py
+++ b/tests/unit/agents/test_agent.py
@@ -1,5 +1,4 @@
 import asyncio
-import platform
 from unittest.mock import ANY, call, patch
 
 import pytest
@@ -954,9 +953,6 @@ def test_channel_iterator(self, *, agent):
     def test_label(self, *, agent):
         assert label(agent)
 
-    @pytest.mark.skipif(
-        platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-    )
     async def test_context_calls_sink(self, *, agent):
         class SinkCalledException(Exception):
             pass
diff --git a/tests/unit/test_streams.py b/tests/unit/test_streams.py
index 4a6cab481..c28765915 100644
--- a/tests/unit/test_streams.py
+++ b/tests/unit/test_streams.py
@@ -1,5 +1,5 @@
 import asyncio
-import platform
+import gc
 from collections import defaultdict
 from contextlib import ExitStack
 from unittest.mock import Mock, patch
@@ -123,9 +123,6 @@ async def test_echo(self, *, stream, app):
         await echoing("val")
         channel.send.assert_called_once_with(value="val")
 
-    @pytest.mark.skipif(
-        platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-    )
     @pytest.mark.asyncio
     @pytest.mark.allow_lingering_tasks(count=1)
     async def test_aiter_tracked(self, *, stream, app):
@@ -141,9 +138,6 @@ async def test_aiter_tracked(self, *, stream, app):
         else:
             event.ack.assert_called_once_with()
 
-    @pytest.mark.skipif(
-        platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy"
-    )
     @pytest.mark.asyncio
     @pytest.mark.allow_lingering_tasks(count=1)
     async def test_aiter_tracked__CancelledError(self, *, stream, app):
@@ -196,9 +190,11 @@ async def agent(stream):
             await s.channel.put(new_event(app, topic="bar", value=sentinel))
             s._on_message_in = Mock()
             await got_sentinel.wait()
-            await asyncio.sleep(0)
-            await asyncio.sleep(0)
-            await asyncio.sleep(0)
+            for _ in range(20):
+                if event.ack.called or event.message.acked:
+                    break
+                gc.collect()
+                await asyncio.sleep(0.05)
         s._on_message_in.assert_called_once_with(
             event.message.tp,
             event.message.offset,