diff --git a/docs/developerguide/cython.rst b/docs/developerguide/cython.rst
index 4548fa4b0..7facfde9f 100644
--- a/docs/developerguide/cython.rst
+++ b/docs/developerguide/cython.rst
@@ -24,6 +24,87 @@ import falls back:
 That fallback is what makes the accelerators optional, and it is also the
 single biggest hazard in maintaining them.  This page is about the hazard.
 
+.. _cython-optin:
+
+The ``cython_optimizations`` opt-in
+===================================
+
+Two of the Cython fast paths never ran -- each was guarded by a condition that
+could not become true (see :ref:`cython-drift`).  Repairing them activates code
+that has, by definition, never executed in production, so the repairs are
+behind a setting that defaults to **off**:
+
+.. sourcecode:: python
+
+    app = faust.App('myapp', cython_optimizations=True)
+
+or ``CYTHON_OPTIMIZATIONS=1`` in the environment (``FAUST_CYTHON_OPTIMIZATIONS``
+when :setting:`env_prefix` is set).  With it off, the extensions behave exactly
+as the released versions do.
+
+What it gates:
+
+* ``StreamIterator._try_get_quick_value`` -- taking values already in the
+  channel queue instead of always awaiting.
+* ``ConductorHandler`` event reuse -- decoding a message once and reusing the
+  event across channels with matching key/value types, instead of
+  deserializing once per subscribed channel.
+
+What it does **not** gate: the ``on_topic_buffer_full`` argument fix.  That one
+was wrong in *both* implementations, is not a Cython-specific change, and
+produces a metric that was simply incorrect before -- so it applies
+unconditionally.
+
+One consequence to be aware of.  While the setting is off, the Cython path and
+the pure-Python path genuinely differ.  That is not new -- it is what has
+shipped for years -- and the flag does not introduce the divergence, it makes
+it selectable.  The sharpest case is in the conductor: a reused event is never
+decoded a second time, so a channel whose payload would fail to deserialize
+raises no error when the event is reused, and raises one when it is not.  That
+changes which channels receive a message, and how many acks it takes.
+
+Consequently the parity suites run with the setting **on** -- that is the
+configuration in which the two implementations are supposed to agree.  A
+separate test in each suite pins the default-off behaviour, so the historical
+path stays covered too.
+
+.. _cython-optin-lifecycle:
+
+Retiring the setting
+--------------------
+
+The setting is **transitional**.  It exists to make adopting the repaired
+paths a decision rather than something that arrives in an upgrade, and it is
+meant to be removed, not kept.  The intended sequence:
+
+1. **Now** -- ships defaulting to ``False``.  Upgrading changes nothing.
+2. **Default flipped to** ``True`` once there is real-world evidence the
+   repaired paths behave: the parity suites passing is necessary but not
+   sufficient, since they only prove the two implementations agree under test.
+   Record the flip as ``version_changed={'<ver>': 'Enabled by default.'}``.
+3. **Deprecated** -- set ``version_deprecated`` and ``deprecation_reason`` on
+   the setting.  Users who set it explicitly get a warning; nobody else
+   notices.
+4. **Removed** -- delete the setting, both ``bint`` attributes, the branches
+   guarding the fast paths, :mod:`faust.utils.optin`, and the two
+   default-off tests.  At that point the fast paths are simply the behaviour,
+   and the parity suites no longer need a ``conf`` marker.
+
+One thing to know before step 3.
+:meth:`~faust.types.settings.params.Param.__get__` emits a
+:exc:`UserWarning` on *every read* of a deprecated setting, and faust reads
+this one itself -- once per :class:`~faust.Stream`, once per assigned
+partition.  Deprecating it naively would make faust warn at itself, at a rate
+that scales with the deployment, about a setting the user most likely never
+set.
+
+That is why the two extensions read it through
+:func:`faust.utils.optin.cython_optimizations_enabled` rather than
+``app.conf.cython_optimizations``: the helper takes the stored value and so
+stays silent, while user-facing reads still warn, which is the point of
+deprecating it.  ``tests/unit/utils/test_optin.py`` pins both halves, so step 3
+is genuinely a two-line change.
+
 .. _cython-testing:
 
 Testing the compiled code
diff --git a/docs/includes/settingref.txt b/docs/includes/settingref.txt
index 669894cff..df39d53c8 100644
--- a/docs/includes/settingref.txt
+++ b/docs/includes/settingref.txt
@@ -282,6 +282,61 @@ the second version is 2, and so on.
    use: ``app.topic(..., internal=True)``.
 
 
+.. setting:: cython_optimizations
+
+``cython_optimizations``
+------------------------
+
+:type: :class:`bool`
+:default: :const:`False`
+:environment: :envvar:`CYTHON_OPTIMIZATIONS`
+:version-introduced: 0.12.2
+
+Enable the repaired fast paths in the Cython extensions.
+
+Disabled by default, and has no effect at all unless the optional
+Cython extension modules were built.
+
+Faust ships a few hot paths twice: a pure-Python implementation, and a
+Cython one used instead when the extensions are available.  Two of the
+Cython fast paths never actually ran -- each was guarded by a condition
+that could not become true -- so for years the extensions quietly did
+more work than the Python they were meant to accelerate:
+
+* ``StreamIterator`` always awaited the channel rather than taking
+  values already sitting in the queue, and
+* ``ConductorHandler`` re-deserialized the payload once per subscribed
+  channel instead of decoding once and reusing the event.
+
+Both are repaired, but the repaired code has by definition never run in
+production, so it is opt-in.  Leaving this ``False`` keeps the
+extensions behaving exactly as the released versions do.
+
+Note this makes the Cython path differ from the pure-Python path while
+disabled -- which has always been true; the flag does not introduce the
+divergence, it just makes it selectable.  The most visible difference is
+in the conductor: a reused event is never decoded again, so a channel
+whose payload would fail to deserialize raises no error when the event
+is reused, and does when it is not.  That changes which channels receive
+a message and how many acks it takes.
+
+Enable it to get the fast paths::
+
+    app = faust.App('myapp', cython_optimizations=True)
+
+This setting is transitional: it is expected to default to ``True`` in
+a later release, then be deprecated and removed, at which point the
+fast paths are simply the behaviour.  See the developer guide for the
+sequence -- and for why faust reads this setting internally through
+:func:`faust.utils.optin.cython_optimizations_enabled` rather than
+directly, which is what keeps the eventual deprecation from warning
+once per stream and once per assigned partition.
+
+.. seealso::
+
+    The developer guide's :ref:`developers-cython` page, for what the
+    two faults were and how the implementations are held level.
+
 .. setting:: blocking_timeout
 
 ``blocking_timeout``
diff --git a/faust/_cython/streams.pyx b/faust/_cython/streams.pyx
index c2ccb3f23..ef0644ccc 100644
--- a/faust/_cython/streams.pyx
+++ b/faust/_cython/streams.pyx
@@ -7,6 +7,7 @@ from mode.utils.futures import maybe_async, notify
 
 from faust.exceptions import Skip
 from faust.types import ChannelT, EventT
+from faust.utils.optin import cython_optimizations_enabled
 
 
 cdef class StreamIterator:
@@ -35,6 +36,7 @@ cdef class StreamIterator:
         object topics
         object acks_enabled_for
         object _skipped_value
+        bint cython_optimizations
 
     def __init__(self, object stream):
         self.stream = stream
@@ -53,6 +55,11 @@ cdef class StreamIterator:
         self.unacked = self.consumer._unacked_messages
         self.add_unacked = self.unacked.add
         self._skipped_value = self.stream._skipped_value
+        # Opt-in: see the `cython_optimizations` setting.  Read once here
+        # rather than per message, so the hot path costs a `bint` test.
+        # Via the helper, not `app.conf.<name>`, so that deprecating the
+        # setting does not emit a warning per stream -- see faust/utils/optin.py.
+        self.cython_optimizations = cython_optimizations_enabled(self.app.conf)
 
         if isinstance(self.channel, ChannelT):
             self.chan_is_channel = True
@@ -206,9 +213,16 @@ cdef class StreamIterator:
         # chan_queue_empty():`` ... ``channel_value = chan_quick_get()``), so
         # this restores the fast path the extension was meant to provide and
         # brings the two implementations back into agreement.
+        #
+        # Behind the `cython_optimizations` setting, off by default: the
+        # repaired path has never run in production, so taking it is opt-in.
+        # Disabled, this reproduces the released behaviour exactly -- always
+        # reporting "use the slow path", never reaching `get_nowait()`.
         if self.chan_is_channel:
             if self.chan_errors:
                 raise self.chan_errors.popleft()
+            if not self.cython_optimizations:
+                return (True, None)
             if self.chan_queue_empty():
                 return (True, None)
             else:
diff --git a/faust/transport/_cython/conductor.pyx b/faust/transport/_cython/conductor.pyx
index 302100c48..cad2c3871 100644
--- a/faust/transport/_cython/conductor.pyx
+++ b/faust/transport/_cython/conductor.pyx
@@ -3,6 +3,7 @@
 from asyncio import ALL_COMPLETED, ensure_future, wait
 
 from faust.exceptions import KeyDecodeError, ValueDecodeError
+from faust.utils.optin import cython_optimizations_enabled
 
 
 cdef class ConductorHandler:
@@ -20,6 +21,7 @@ cdef class ConductorHandler:
         object wait_until_producer_ebb
         object consumer_on_buffer_full
         object consumer_on_buffer_drop
+        bint cython_optimizations
 
 
     def __init__(self, object conductor, object tp, object channels):
@@ -33,6 +35,11 @@ cdef class ConductorHandler:
         self.acquire_flow_control = self.app.flow_control.acquire
         self.wait_until_producer_ebb = self.app.producer.buffer.wait_until_ebb
         self.consumer = self.app.consumer
+        # Opt-in: see the `cython_optimizations` setting.  Read once per
+        # handler (one per assigned TP), not per message.  Via the helper, not
+        # `app.conf.<name>`, so that deprecating the setting does not emit a
+        # warning per partition -- see faust/utils/optin.py.
+        self.cython_optimizations = cython_optimizations_enabled(self.app.conf)
         # We divide `stream_buffer_maxsize` with Queue.pressure_ratio
         # find a limit to the number of messages we will buffer
         # before considering the buffer to be under high pressure.
@@ -89,9 +96,13 @@ cdef class ConductorHandler:
                         event = await chan.decode(message, propagate=True)
                         event_keyid = keyid
                         dest_event = event
-                    elif keyid == event_keyid:
+                    elif self.cython_optimizations and keyid == event_keyid:
                         dest_event = event
                     else:
+                        # Reuse is behind the `cython_optimizations` setting,
+                        # off by default: the repaired path has never run in
+                        # production.  Disabled, every channel deserializes its
+                        # own event, reproducing the released behaviour.
                         dest_event = await chan.decode(message, propagate=True)
                     if not self._put(dest_event, chan, full):
                         continue
diff --git a/faust/types/settings/settings.py b/faust/types/settings/settings.py
index 8f68b2cf4..42f6746dd 100644
--- a/faust/types/settings/settings.py
+++ b/faust/types/settings/settings.py
@@ -604,6 +604,59 @@ def agent_supervisor(self) -> Type[SupervisorStrategyT]:
             restarted).
         """
 
+    @sections.Common.setting(
+        params.Bool,
+        version_introduced="0.12.2",
+        env_name="CYTHON_OPTIMIZATIONS",
+        default=False,
+    )
+    def cython_optimizations(self) -> bool:
+        """Enable the repaired fast paths in the Cython extensions.
+
+        Disabled by default, and has no effect at all unless the optional
+        Cython extension modules were built.
+
+        Faust ships a few hot paths twice: a pure-Python implementation, and a
+        Cython one used instead when the extensions are available.  Two of the
+        Cython fast paths never actually ran -- each was guarded by a condition
+        that could not become true -- so for years the extensions quietly did
+        more work than the Python they were meant to accelerate:
+
+        * ``StreamIterator`` always awaited the channel rather than taking
+          values already sitting in the queue, and
+        * ``ConductorHandler`` re-deserialized the payload once per subscribed
+          channel instead of decoding once and reusing the event.
+
+        Both are repaired, but the repaired code has by definition never run in
+        production, so it is opt-in.  Leaving this ``False`` keeps the
+        extensions behaving exactly as the released versions do.
+
+        Note this makes the Cython path differ from the pure-Python path while
+        disabled -- which has always been true; the flag does not introduce the
+        divergence, it just makes it selectable.  The most visible difference is
+        in the conductor: a reused event is never decoded again, so a channel
+        whose payload would fail to deserialize raises no error when the event
+        is reused, and does when it is not.  That changes which channels receive
+        a message and how many acks it takes.
+
+        Enable it to get the fast paths::
+
+            app = faust.App('myapp', cython_optimizations=True)
+
+        This setting is transitional: it is expected to default to ``True`` in
+        a later release, then be deprecated and removed, at which point the
+        fast paths are simply the behaviour.  See the developer guide for the
+        sequence -- and for why faust reads this setting internally through
+        :func:`faust.utils.optin.cython_optimizations_enabled` rather than
+        directly, which is what keeps the eventual deprecation from warning
+        once per stream and once per assigned partition.
+
+        .. seealso::
+
+            The developer guide's :ref:`developers-cython` page, for what the
+            two faults were and how the implementations are held level.
+        """
+
     @sections.Common.setting(
         params.Seconds,
         env_name="BLOCKING_TIMEOUT",
diff --git a/faust/utils/optin.py b/faust/utils/optin.py
new file mode 100644
index 000000000..4c13cc5a2
--- /dev/null
+++ b/faust/utils/optin.py
@@ -0,0 +1,48 @@
+"""Reading opt-in settings from faust's own internals.
+
+``cython_optimizations`` is transitional: it exists so the repaired Cython
+fast paths can be adopted deliberately rather than arriving in an upgrade, and
+it is expected to be deprecated and removed once they are the default (see
+``docs/developerguide/cython.rst``).
+
+That plan has a trap in it, which is what this module exists to avoid.
+:meth:`faust.types.settings.params.Param.__get__` emits a :exc:`UserWarning`
+on **every read** of a setting once ``version_deprecated`` is set on it -- and
+faust reads this one itself, once per :class:`~faust.Stream` and once per
+assigned partition.  Deprecating the setting would therefore make faust warn
+at itself, repeatedly, about a setting the user very likely never set and
+cannot act on.
+
+So internal reads go through :func:`cython_optimizations_enabled`, which takes
+the stored value rather than the descriptor.  User-facing reads of
+``app.conf.cython_optimizations`` are untouched and *should* warn once the
+setting is deprecated -- that is the whole point of deprecating it.
+
+Note this deliberately does not use :func:`warnings.catch_warnings` to
+suppress the warning instead: that manipulates global state and is not
+thread-safe, which matters on the free-threaded builds faust now supports.
+
+When the setting is finally removed, delete this module and the two calls to
+it.
+"""
+
+from typing import Any
+
+__all__ = ["cython_optimizations_enabled"]
+
+
+def cython_optimizations_enabled(conf: Any) -> bool:
+    """Return whether the repaired Cython fast paths are enabled.
+
+    Arguments:
+        conf: The app's :class:`~faust.types.settings.Settings`.
+
+    Reads the value the descriptor stores rather than going through the
+    descriptor, so that deprecating the setting does not make every stream and
+    every partition assignment emit a warning from inside faust.  The storage
+    attribute is looked up through the settings registry rather than
+    hard-coded, so renaming the setting cannot silently turn this into a
+    read of a non-existent attribute.
+    """
+    param = type(conf).SETTINGS["cython_optimizations"]
+    return bool(getattr(conf, param.storage_name))
diff --git a/tests/unit/test_cython_parity.py b/tests/unit/test_cython_parity.py
index c505885c2..86b5d6b39 100644
--- a/tests/unit/test_cython_parity.py
+++ b/tests/unit/test_cython_parity.py
@@ -155,6 +155,7 @@ def counting_anext():
 
 @requires_cython
 @pytest.mark.asyncio
+@pytest.mark.conf(cython_optimizations=True)
 async def test_cython_stream_uses_queue_fast_path(*, app) -> None:
     """The compiled iterator must take the non-blocking queue path.
 
@@ -187,6 +188,7 @@ async def test_cython_stream_uses_queue_fast_path(*, app) -> None:
 
 @requires_cython
 @pytest.mark.asyncio
+@pytest.mark.conf(cython_optimizations=True)
 async def test_cython_stream_falls_back_to_slow_path_when_empty(*, app) -> None:
     """An empty queue must still take the awaiting path.
 
@@ -215,3 +217,32 @@ async def test_cython_stream_falls_back_to_slow_path_when_empty(*, app) -> None:
         pending.cancel()
         with pytest.raises(asyncio.CancelledError):
             await pending
+
+
+@requires_cython
+@pytest.mark.asyncio
+async def test_cython_stream_fast_path_is_off_by_default(*, app) -> None:
+    """Without the opt-in, the iterator behaves as the released versions do.
+
+    `cython_optimizations` defaults to False, so the repaired fast path stays
+    dormant: every value goes through `await Channel.__anext__` exactly as it
+    did before the fix.  No `conf` marker here on purpose -- this is the
+    default an unmodified app gets.
+    """
+    assert app.conf.cython_optimizations is False
+
+    it, queue, anext_calls = _new_iterator(app)
+    for i in range(5):
+        queue.put_nowait(i)
+
+    seen = []
+    for _ in range(5):
+        value, _sensor_state = await asyncio.wait_for(it.next(), timeout=5)
+        seen.append(value)
+
+    # Same values either way; only the route differs.
+    assert seen == [0, 1, 2, 3, 4]
+    assert len(anext_calls) == 5, (
+        f"expected the slow path for all 5 values with the optimizations off, "
+        f"got {len(anext_calls)} awaits: the fast path is no longer opt-in"
+    )
diff --git a/tests/unit/transport/test_conductor_parity.py b/tests/unit/transport/test_conductor_parity.py
index 4f3a8b928..54c3ef060 100644
--- a/tests/unit/transport/test_conductor_parity.py
+++ b/tests/unit/transport/test_conductor_parity.py
@@ -240,6 +240,7 @@ async def run_both(harness: Harness, scenario) -> Dict[str, Any]:
 # ------------------------------------------------------------------ delivery
 @requires_cython_conductor
 @pytest.mark.asyncio
+@pytest.mark.conf(cython_optimizations=True)
 @pytest.mark.parametrize("harness", [1, 2, 3], indirect=True)
 async def test_parity__fan_out(harness) -> None:
     """Every subscribed channel gets the event, and refcount matches."""
@@ -259,6 +260,7 @@ async def scenario(handler, h):
 
 @requires_cython_conductor
 @pytest.mark.asyncio
+@pytest.mark.conf(cython_optimizations=True)
 async def test_parity__no_channels(harness) -> None:
     """A TP with no subscribers must not touch the message."""
     harness.channel_set = set()
@@ -276,6 +278,7 @@ async def scenario(handler, h):
 
 @requires_cython_conductor
 @pytest.mark.asyncio
+@pytest.mark.conf(cython_optimizations=True)
 @pytest.mark.parametrize("harness", [3], indirect=True)
 async def test_parity__multiple_messages(harness) -> None:
     """A batch, to catch state carried between calls."""
@@ -292,6 +295,7 @@ async def scenario(handler, h):
 
 @requires_cython_conductor
 @pytest.mark.asyncio
+@pytest.mark.conf(cython_optimizations=True)
 @pytest.mark.parametrize("harness", [2, 4], indirect=True)
 async def test_parity__event_reuse_for_matching_keyid(harness) -> None:
     """Channels with the same (key_type, value_type) share one decode.
@@ -319,6 +323,7 @@ async def scenario(handler, h):
 
 @requires_cython_conductor
 @pytest.mark.asyncio
+@pytest.mark.conf(cython_optimizations=True)
 @pytest.mark.parametrize("harness", [(2, True), (4, True)], indirect=True)
 async def test_parity__no_reuse_for_differing_keyid(harness) -> None:
     """Channels with different (key_type, value_type) each decode their own.
@@ -348,6 +353,7 @@ async def scenario(handler, h):
 # -------------------------------------------------------------- decode errors
 @requires_cython_conductor
 @pytest.mark.asyncio
+@pytest.mark.conf(cython_optimizations=True)
 @pytest.mark.parametrize("harness", [1, 3], indirect=True)
 @pytest.mark.parametrize(
     "exc_cls,bucket",
@@ -378,6 +384,7 @@ async def scenario(handler, h):
 
 @requires_cython_conductor
 @pytest.mark.asyncio
+@pytest.mark.conf(cython_optimizations=True)
 @pytest.mark.parametrize("harness", [3], indirect=True)
 async def test_parity__decode_error_on_one_channel(harness) -> None:
     """One channel's decode fails; the rest of the fan-out must match.
@@ -400,7 +407,7 @@ async def scenario(handler, h):
 # ------------------------------------------------------------ buffer pressure
 @requires_cython_conductor
 @pytest.mark.asyncio
-@pytest.mark.conf(stream_buffer_maxsize=2)
+@pytest.mark.conf(cython_optimizations=True, stream_buffer_maxsize=2)
 async def test_parity__queue_full_path(harness) -> None:
     """When a channel queue is full the handler must await ``chan.put``.
 
@@ -494,7 +501,7 @@ async def test_monitor_counts_buffer_full_by_tp(app, impl) -> None:
 
 @requires_cython_conductor
 @pytest.mark.asyncio
-@pytest.mark.conf(stream_buffer_maxsize=8)
+@pytest.mark.conf(cython_optimizations=True, stream_buffer_maxsize=8)
 async def test_parity__pressure_callbacks(harness) -> None:
     """High-pressure and pressure-drop callbacks must fire identically.
 
@@ -518,3 +525,35 @@ async def scenario(handler, h):
 
     results = await run_both(harness, scenario)
     assert_parity(results)
+
+
+@requires_cython_conductor
+@pytest.mark.asyncio
+@pytest.mark.parametrize("harness", [3], indirect=True)
+async def test_event_reuse_is_off_by_default(harness) -> None:
+    """Without the opt-in, the conductor behaves as the released versions do.
+
+    `cython_optimizations` defaults to False, so the repaired reuse stays
+    dormant and every channel deserializes its own event -- exactly as before
+    the fix.  No `conf` marker here on purpose: this is what an unmodified app
+    gets.
+
+    This is also where the Cython and pure-Python conductors legitimately
+    differ, so it is not a parity test.  That divergence is not new; the flag
+    only makes it selectable.
+    """
+    assert harness.app.conf.cython_optimizations is False
+
+    handler = harness.build("cython")
+    message = harness.message()
+    await handler(message)
+    obs = harness.observations(message)
+
+    n = len(harness.channels)
+    assert obs["n_decodes"] == n, (
+        f"expected one decode per channel with the optimizations off, got "
+        f"{obs['n_decodes']} for {n} channels: reuse is no longer opt-in"
+    )
+    # Delivery itself is unchanged -- only how many times the payload is read.
+    assert obs["n_delivered_total"] == n
+    assert obs["refcount"] == n
diff --git a/tests/unit/utils/test_optin.py b/tests/unit/utils/test_optin.py
new file mode 100644
index 000000000..e021ae33f
--- /dev/null
+++ b/tests/unit/utils/test_optin.py
@@ -0,0 +1,115 @@
+"""The internal read of ``cython_optimizations`` must survive deprecation.
+
+The setting is transitional and expected to be deprecated and then removed.
+That plan has a trap in it: ``Param.__get__`` emits a ``UserWarning`` on every
+read once ``version_deprecated`` is set, and faust reads this setting itself --
+once per stream, and once per assigned partition.  Deprecating it naively would
+make faust warn at itself, repeatedly, about a setting the user probably never
+set.
+
+These tests pin the two halves of the contract:
+
+* internal reads (via ``cython_optimizations_enabled``) stay silent, and
+* user-facing reads (``app.conf.cython_optimizations``) still warn, because
+  warning is the entire point of deprecating a setting.
+"""
+
+import warnings
+
+import pytest
+
+from faust.utils.optin import cython_optimizations_enabled
+
+
+@pytest.fixture()
+def param(app):
+    """The ``cython_optimizations`` Param descriptor."""
+    return type(app.conf).SETTINGS["cython_optimizations"]
+
+
+@pytest.fixture()
+def deprecated(param):
+    """Mark the setting deprecated for the duration of a test."""
+    saved = (param.version_deprecated, param.deprecation_reason)
+    param.version_deprecated = "0.99.0"
+    param.deprecation_reason = "the fast paths are now unconditional"
+    try:
+        yield param
+    finally:
+        param.version_deprecated, param.deprecation_reason = saved
+
+
+def test_matches_the_public_read(app) -> None:
+    assert cython_optimizations_enabled(app.conf) is app.conf.cython_optimizations
+
+
+def test_default_is_false(app) -> None:
+    assert cython_optimizations_enabled(app.conf) is False
+
+
+@pytest.mark.conf(cython_optimizations=True)
+def test_reads_true_when_enabled(app) -> None:
+    assert cython_optimizations_enabled(app.conf) is True
+
+
+def test_internal_read_is_silent_when_deprecated(app, deprecated) -> None:
+    """The whole reason this helper exists."""
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        value = cython_optimizations_enabled(app.conf)
+
+    assert value is False
+    ours = [w for w in caught if "cython_optimizations" in str(w.message)]
+    assert not ours, (
+        f"reading the setting from inside faust warned: "
+        f"{[str(w.message) for w in ours]}.  faust reads this per stream and "
+        f"per partition, so a deprecation would flood logs with a warning "
+        f"the user cannot act on."
+    )
+
+
+def test_public_read_still_warns_when_deprecated(app, deprecated) -> None:
+    """The helper must not disarm the deprecation for users.
+
+    Without this, a future deprecation could be silently ineffective -- which
+    is worse than noisy, because nobody would ever be told to stop using it.
+    """
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        app.conf.cython_optimizations
+
+    ours = [w for w in caught if "cython_optimizations" in str(w.message)]
+    assert ours, "deprecating the setting no longer warns users who read it"
+    assert "deprecated" in str(ours[0].message)
+
+
+def test_extensions_are_silent_when_deprecated(app, deprecated) -> None:
+    """The end-to-end version: neither extension warns per construction.
+
+    The helper is only useful if the extensions actually go through it.  This
+    builds the objects that read the setting -- one per stream, one per
+    assigned partition -- and asserts the deprecation stays quiet.
+    """
+    from faust.streams import _CStreamIterator
+    from faust.transport.conductor import Conductor, ConductorHandler
+    from faust.types import TP
+
+    if _CStreamIterator is None or ConductorHandler is None:
+        pytest.skip("extensions not built in place")
+
+    conductor = Conductor(app)
+    topic = app.topic("foo")
+
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        for _ in range(3):
+            _CStreamIterator(app.stream(app.channel()))
+        for i in range(3):
+            ConductorHandler(conductor, TP("foo", i), {topic})
+
+    ours = [w for w in caught if "cython_optimizations" in str(w.message)]
+    assert not ours, (
+        f"{len(ours)} deprecation warnings from 3 stream iterators and 3 "
+        f"conductor handlers.  A real worker builds one of each per stream and "
+        f"per assigned partition, so this scales with the deployment."
+    )