refactor(gateway): single StreamingConfig.enabled_for gate for both streaming sites (#53697)

Both the TurnRunner and proxy paths now call StreamingConfig.enabled_for(), so the
master-switch + per-platform override logic lives in one place. The proxy path also
returns before re-reading config.yaml when the global switch is off.

gateway/config.py

@@ -508,6 +508,19 @@ class StreamingConfig:
     # fresh-message replacement path; set >0 to opt in.
     fresh_final_after_seconds: float = 0.0
 
+    @property
+    def globally_enabled(self) -> bool:
+        """The ``streaming.enabled`` master switch (``transport: off`` also disables)."""
+        return bool(self.enabled) and self.transport != "off"
+
+    def enabled_for(self, platform_override: Any) -> bool:
+        """Effective streaming for one platform.
+
+        ``platform_override`` is ``display.platforms.<plat>.streaming`` (``None`` = follow global).
+        A per-platform value can only narrow the global switch, never enable streaming on its own.
+        """
+        return self.globally_enabled and (platform_override is None or bool(platform_override))
+
     def to_dict(self) -> Dict[str, Any]:
         return asdict(self)
 

gateway/run_turn.py

@@ -2722,13 +2722,12 @@ class GatewayTurnMixin:
         if _scfg is None:
             from gateway.config import StreamingConfig
             _scfg = StreamingConfig()
+        # Global master switch first: skips the config.yaml re-read on the default (off) path.
+        if not _scfg.globally_enabled:
+            return None
         from gateway.display_config import resolve_display_setting
         _plat_streaming = resolve_display_setting(_load_gateway_config(), _platform_config_key(source.platform), "streaming")
-        _global_streaming_enabled = bool(_scfg.enabled) and _scfg.transport != "off"
-        _streaming_enabled = _global_streaming_enabled and (
-            True if _plat_streaming is None else bool(_plat_streaming)
-        )
-        if not _streaming_enabled:
+        if not _scfg.enabled_for(_plat_streaming):
             return None
         try:
             from gateway.stream_consumer import GatewayStreamConsumer

gateway/run_turn_runner.py

@@ -921,10 +921,7 @@ class TurnRunner:
             scfg = StreamingConfig()
         # display.platforms.<plat>.streaming may disable streaming per platform; None = follow global.
         plat_streaming = ctx.resolve_display_setting(ctx.user_config, platform_key, "streaming")
-        global_streaming_enabled = bool(scfg.enabled) and scfg.transport != "off"
-        want_stream_deltas = not ctx.scheduled_heartbeat and global_streaming_enabled and (
-            True if plat_streaming is None else bool(plat_streaming)
-        )
+        want_stream_deltas = not ctx.scheduled_heartbeat and scfg.enabled_for(plat_streaming)
         want_interim_messages = bool(ctx.interim_assistant_messages_enabled) and not ctx.scheduled_heartbeat
         if want_stream_deltas or want_interim_messages:
             try: