fix(gateway): SSE buffer cap, yuanbao reconnect singleton, WhatsApp Cloud allowlist env vars (3-PR salvage)

gateway/platforms/whatsapp_cloud.py

@@ -225,18 +225,35 @@ def __init__(self, config: PlatformConfig):
         import os
 
         self._reply_prefix: Optional[str] = extra.get("reply_prefix")
-        self._dm_policy: str = str(
-            extra.get("dm_policy")
-            or os.getenv("WHATSAPP_CLOUD_DM_POLICY")
-            or os.getenv("WHATSAPP_DM_POLICY", "open")
-        ).strip().lower()
+        # Allowlist: honor the *documented* WHATSAPP_CLOUD_ALLOWED_USERS (the
+        # var the setup wizard writes) in addition to WHATSAPP_CLOUD_ALLOW_FROM.
+        # The adapter historically read only ALLOW_FROM, so an allowlist
+        # configured via the documented var silently dropped every inbound.
         self._allow_from: set[str] = self._normalize_allow_ids(
             self._coerce_allow_list(
                 extra.get("allow_from")
                 or extra.get("allowFrom")
                 or os.getenv("WHATSAPP_CLOUD_ALLOW_FROM")
+                or os.getenv("WHATSAPP_CLOUD_ALLOWED_USERS")
             )
         )
+        # DM policy: explicit config wins; otherwise choose a safe, working
+        # default -- "open" if the operator opted into allow-all, else
+        # "allowlist" when an allowlist is configured (so it is actually
+        # enforced instead of silently dropping), else "open".
+        _allow_all_optin = str(
+            os.getenv("WHATSAPP_CLOUD_ALLOW_ALL_USERS", "")
+        ).strip().lower() in {"true", "1", "yes"}
+        _default_dm_policy = (
+            "open" if _allow_all_optin
+            else ("allowlist" if self._allow_from else "open")
+        )
+        self._dm_policy: str = str(
+            extra.get("dm_policy")
+            or os.getenv("WHATSAPP_CLOUD_DM_POLICY")
+            or os.getenv("WHATSAPP_DM_POLICY")
+            or _default_dm_policy
+        ).strip().lower()
         self._group_policy: str = str(
             extra.get("group_policy")
             or os.getenv("WHATSAPP_CLOUD_GROUP_POLICY")
@@ -347,6 +364,17 @@ def _is_dm_allowed(self, sender_id: str) -> bool:
             return (bare or sender_id) in self._allow_from
         return super()._is_dm_allowed(sender_id)
 
+    def _open_dm_opted_in(self) -> bool:
+        """Also honor the documented WHATSAPP_CLOUD_ALLOW_ALL_USERS opt-in.
+
+        The shared mixin only checks GATEWAY_ALLOW_ALL_USERS /
+        WHATSAPP_ALLOW_ALL_USERS; the Cloud adapter's documented open-access
+        opt-in is WHATSAPP_CLOUD_ALLOW_ALL_USERS, so honor it here too.
+        """
+        if str(os.getenv("WHATSAPP_CLOUD_ALLOW_ALL_USERS", "")).strip().lower() in {"true", "1", "yes"}:
+            return True
+        return super()._open_dm_opted_in()
+
     # ------------------------------------------------------------------ lifecycle
     async def connect(self, *, is_reconnect: bool = False) -> bool:
         if not check_whatsapp_cloud_requirements():

gateway/platforms/yuanbao.py

@@ -3868,6 +3868,7 @@ async def _do_reconnect(self) -> bool:
                     "[%s] Reconnected on attempt %d. connectId=%s",
                     adapter.name, attempt + 1, self._connect_id,
                 )
+                YuanbaoAdapter.set_active(adapter)
                 return True
 
             except asyncio.TimeoutError:

gateway/run.py

@@ -67,6 +67,7 @@
 _AGENT_CACHE_IDLE_TTL_SECS = 3600.0  # evict agents idle for >1h
 _PLATFORM_CONNECT_TIMEOUT_SECS_DEFAULT = 30.0
 _ADAPTER_DISCONNECT_TIMEOUT_SECS_DEFAULT = 5.0
+_GATEWAY_PROXY_SSE_BUFFER_MAX_CHARS = 16 * 1024 * 1024
 _TELEGRAM_COMMAND_MENTION_RE = re.compile(r"(?<![\w:/])/([A-Za-z0-9][A-Za-z0-9_-]*)")
 
 _TELEGRAM_NOISY_STATUS_RE = re.compile(
@@ -16219,6 +16220,10 @@ def _pause_typing_before_finalize(
                                                 _stream_consumer.on_delta(content)
                                 except json.JSONDecodeError:
                                     pass
+                        if len(buffer) > _GATEWAY_PROXY_SSE_BUFFER_MAX_CHARS:
+                            raise ValueError(
+                                "Proxy SSE stream exceeded max buffer size without a line boundary"
+                            )
 
         except asyncio.CancelledError:
             raise

scripts/release.py

@@ -46,9 +46,10 @@
 # Auto-extracted from noreply emails + manual overrides
 AUTHOR_MAP = {
     "jonathan@mintrx.com": "JAlmanzarMint",  # PR #52688 salvage (vision: rasterize SVG / re-encode unsupported raster formats to PNG before embedding), folded into #57890
-    "al3060388206@gmail.com": "ooiuuii",  # PR #58466/#58377 salvage (redact: fireworks fw-/fpk_ prefixes; telegram: redact bot tokens out of transport error strings). Also PR #58433 salvage (codex: accept recorded final_text when app-server omits turn/completed).
+    "al3060388206@gmail.com": "ooiuuii",  # PR #58466/#58377 salvage (redact: fireworks fw-/fpk_ prefixes; telegram: redact bot tokens out of transport error strings). Also PR #58433 salvage (codex: accept recorded final_text when app-server omits turn/completed) and PR #58472 salvage (gateway: cap proxy SSE line buffer at 16MiB).
     "Jigoooo@users.noreply.github.com": "Jigoooo",  # PR #58474 salvage (auxiliary: fall back to token resolver when anthropic pool has no usable entry)
     "root@vmi3351581.contaboserver.net": "ostravajih",  # PR #58374 salvage (poolside: coerce integer finish_reason and tool_call id to strings)
+    "hello@sahil-shubham.in": "sahil-shubham",  # PR #58448 salvage (whatsapp_cloud: honor documented WHATSAPP_CLOUD_ALLOWED_USERS / ALLOW_ALL_USERS in the DM intake gate)
     "yingwaizhiying@gmail.com": "msh01",  # PR #58250 salvage (telegram: wall-clock init timeout via daemon-thread deadline + abandon the shielded initialize task on timeout so the retry ladder advances instead of hanging on attempt 1/8 under s6 supervision; #58236). Also covers PR #58276 salvage (compression: preserve a real user turn after compaction; #55677).
     "danilo@falcao.org": "danilofalcao",  # PR #56674 salvage (update: skip unsupported platform.matrix lazy refresh on native Windows — python-olm has no Windows wheel)
     "huanshan5195@users.noreply.github.com": "huanshan5195",  # PR #57601 salvage (custom-provider: emit reasoning_effort at the live CustomProfile path so GLM-5.2/ARK/vLLM/Ollama endpoints receive it; + "max" reasoning level)

tests/gateway/test_proxy_mode.py

@@ -334,6 +334,32 @@ async def __aexit__(self, *args):
 
         assert "Proxy connection error" in result["final_response"]
 
+    @pytest.mark.asyncio
+    async def test_rejects_proxy_sse_without_line_boundary_after_buffer_cap(self, monkeypatch):
+        monkeypatch.setenv("GATEWAY_PROXY_URL", "http://host:8642")
+        monkeypatch.delenv("GATEWAY_PROXY_KEY", raising=False)
+        monkeypatch.setattr("gateway.run._GATEWAY_PROXY_SSE_BUFFER_MAX_CHARS", 16)
+        runner = _make_runner()
+        source = _make_source()
+
+        resp = _FakeSSEResponse(status=200, sse_chunks=[b"data: ", b"x" * 20])
+        session = _FakeSession(resp)
+
+        with patch("gateway.run._load_gateway_config", return_value={}):
+            with _patch_aiohttp(session):
+                with patch("aiohttp.ClientTimeout"):
+                    result = await runner._run_agent_via_proxy(
+                        message="hi",
+                        context_prompt="",
+                        history=[],
+                        source=source,
+                        session_id="test",
+                    )
+
+        assert "Proxy connection error" in result["final_response"]
+        assert "exceeded max buffer size" in result["final_response"]
+        assert result["api_calls"] == 0
+
     @pytest.mark.asyncio
     async def test_skips_tool_messages_in_history(self, monkeypatch):
         monkeypatch.setenv("GATEWAY_PROXY_URL", "http://host:8642")

tests/gateway/test_whatsapp_cloud_allowed_users.py

@@ -0,0 +1,109 @@
+"""Regression tests for PR #58448 salvage: the documented
+WHATSAPP_CLOUD_ALLOWED_USERS / WHATSAPP_CLOUD_ALLOW_ALL_USERS env vars
+must actually drive the DM intake gate.
+
+Before the fix, the adapter only read WHATSAPP_CLOUD_ALLOW_FROM and the
+dm_policy default was "open" (which fails closed without an allow-all
+opt-in), so a wizard-configured install using the documented vars
+silently dropped every inbound message.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from gateway.config import Platform
+
+
+def _build_adapter(monkeypatch, env: dict[str, str], extra: dict | None = None):
+    """Construct a real WhatsAppCloudAdapter through __init__ with env vars."""
+    from gateway.platforms.whatsapp_cloud import WhatsAppCloudAdapter
+
+    for var in (
+        "WHATSAPP_CLOUD_ALLOW_FROM",
+        "WHATSAPP_CLOUD_ALLOWED_USERS",
+        "WHATSAPP_CLOUD_ALLOW_ALL_USERS",
+        "WHATSAPP_CLOUD_DM_POLICY",
+        "WHATSAPP_DM_POLICY",
+        "GATEWAY_ALLOW_ALL_USERS",
+        "WHATSAPP_ALLOW_ALL_USERS",
+    ):
+        monkeypatch.delenv(var, raising=False)
+    for key, value in env.items():
+        monkeypatch.setenv(key, value)
+
+    config = MagicMock()
+    config.extra = {
+        "phone_number_id": "1234567890",
+        "access_token": "test-token",
+        **(extra or {}),
+    }
+    return WhatsAppCloudAdapter(config)
+
+
+def _dm_message(sender: str) -> dict:
+    return {"from": sender, "id": "wamid.test", "type": "text"}
+
+
+def test_allowed_users_env_populates_allowlist_and_enforces_it(monkeypatch):
+    adapter = _build_adapter(
+        monkeypatch, {"WHATSAPP_CLOUD_ALLOWED_USERS": "15551234567"}
+    )
+
+    # The documented var must populate the allowlist...
+    assert "15551234567" in adapter._allow_from
+    # ...and flip the default dm_policy to allowlist so it is enforced.
+    assert adapter._dm_policy == "allowlist"
+    # Allowlisted sender passes the intake gate; others are dropped.
+    assert adapter._is_dm_allowed("15551234567") is True
+    assert adapter._is_dm_allowed("19998887777") is False
+
+
+def test_allow_all_users_env_opts_into_open_dms(monkeypatch):
+    adapter = _build_adapter(
+        monkeypatch, {"WHATSAPP_CLOUD_ALLOW_ALL_USERS": "true"}
+    )
+
+    assert adapter._dm_policy == "open"
+    assert adapter._open_dm_opted_in() is True
+    assert adapter._is_dm_allowed("19998887777") is True
+
+
+def test_explicit_dm_policy_still_wins_over_derived_default(monkeypatch):
+    adapter = _build_adapter(
+        monkeypatch,
+        {
+            "WHATSAPP_CLOUD_ALLOWED_USERS": "15551234567",
+            "WHATSAPP_CLOUD_DM_POLICY": "disabled",
+        },
+    )
+
+    # Operator's explicit policy beats the allowlist-derived default.
+    assert adapter._dm_policy == "disabled"
+
+
+def test_unconfigured_default_unchanged(monkeypatch):
+    adapter = _build_adapter(monkeypatch, {})
+
+    # No allowlist, no opt-in: default stays "open" (which fails closed
+    # in the shared mixin without an allow-all opt-in) — pre-fix behavior
+    # for unconfigured installs is preserved.
+    assert adapter._dm_policy == "open"
+    assert adapter._allow_from == set()
+    assert adapter._open_dm_opted_in() is False
+
+
+def test_allow_from_still_takes_precedence(monkeypatch):
+    adapter = _build_adapter(
+        monkeypatch,
+        {
+            "WHATSAPP_CLOUD_ALLOW_FROM": "15550000001",
+            "WHATSAPP_CLOUD_ALLOWED_USERS": "15559999999",
+        },
+    )
+
+    # Legacy ALLOW_FROM wins when both are set (documented precedence).
+    assert "15550000001" in adapter._allow_from
+    assert "15559999999" not in adapter._allow_from

tests/test_yuanbao_reconnect_set_active.py

@@ -0,0 +1,106 @@
+"""test_yuanbao_reconnect_set_active.py - Verify _do_reconnect restores the active singleton.
+
+Regression test for #58363: after a WS disconnect/reconnect cycle,
+``get_active_adapter()`` must return the live adapter (not ``None``).
+The original ``_do_reconnect()`` succeeded but never called
+``YuanbaoAdapter.set_active()``, leaving the singleton permanently
+``None`` until a full gateway restart.
+"""
+
+import sys
+import os
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if _REPO_ROOT not in sys.path:
+    sys.path.insert(0, _REPO_ROOT)
+
+import pytest
+from gateway.platforms.yuanbao import (
+    YuanbaoAdapter,
+    ConnectionManager,
+    get_active_adapter,
+)
+
+
+def _make_adapter(**kwargs):
+    """Create a minimal YuanbaoAdapter mock."""
+    adapter = MagicMock(spec=YuanbaoAdapter)
+    adapter.name = "yuanbao"
+    adapter._app_key = "test_key"
+    adapter._app_secret = "test_secret"
+    adapter._api_domain = "https://test.example.com"
+    adapter._route_env = None
+    adapter._bot_id = "test_bot"
+    adapter._ws_url = "wss://test.example.com/ws"
+    adapter._mark_connected = MagicMock()
+    adapter._mark_disconnected = MagicMock()
+    adapter._release_platform_lock = MagicMock()
+    return adapter
+
+
+@pytest.mark.asyncio
+async def test_do_reconnect_calls_set_active_on_success():
+    """After a successful reconnect, set_active(adapter) must be called."""
+    adapter = _make_adapter()
+    cm = ConnectionManager(adapter)
+
+    # Mock the reconnect internals to succeed on first attempt
+    mock_ws = AsyncMock()
+    mock_ws.close = AsyncMock()
+
+    with (
+        patch.object(cm, "_cleanup_ws", new_callable=AsyncMock) as mock_cleanup,
+        patch(
+            "gateway.platforms.yuanbao.SignManager.force_refresh",
+            new_callable=AsyncMock,
+            return_value={"bot_id": "test_bot", "token": "test_token"},
+        ),
+        patch("gateway.platforms.yuanbao.websockets.connect", new_callable=AsyncMock, return_value=mock_ws),
+        patch.object(cm, "_authenticate", new_callable=AsyncMock, return_value=True),
+        patch.object(cm, "_heartbeat_loop", new_callable=AsyncMock),
+        patch.object(cm, "_receive_loop", new_callable=AsyncMock),
+        patch("gateway.platforms.yuanbao.MAX_RECONNECT_ATTEMPTS", 1),
+    ):
+        # Clear any existing active instance
+        YuanbaoAdapter.set_active(None)
+        assert get_active_adapter() is None
+
+        # Run reconnect
+        result = await cm._do_reconnect()
+
+        # Reconnect should succeed
+        assert result is True
+
+        # After successful reconnect, get_active() must return the adapter
+        assert get_active_adapter() is adapter
+
+
+@pytest.mark.asyncio
+async def test_do_reconnect_does_not_set_active_on_failure():
+    """When all reconnect attempts fail, set_active should NOT be called."""
+    adapter = _make_adapter()
+    cm = ConnectionManager(adapter)
+
+    with (
+        patch.object(cm, "_cleanup_ws", new_callable=AsyncMock),
+        patch(
+            "gateway.platforms.yuanbao.SignManager.force_refresh",
+            new_callable=AsyncMock,
+            side_effect=Exception("auth failed"),
+        ),
+        patch("gateway.platforms.yuanbao.MAX_RECONNECT_ATTEMPTS", 1),
+    ):
+        # Clear any existing active instance
+        YuanbaoAdapter.set_active(None)
+        assert get_active_adapter() is None
+
+        # Run reconnect - should fail
+        result = await cm._do_reconnect()
+
+        # Reconnect should fail
+        assert result is False
+
+        # get_active() should still be None
+        assert get_active_adapter() is None