Yuanbao adapter singleton loss after WS reconnect

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

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

  merged to main

  files: gateway/platforms/whatsapp_cloud.py, gateway/platforms/yuanbao.py, gateway/run.py, scripts/release.py, tests/gateway/test_proxy_mode.py, tests/gateway/test_whatsapp_cloud_allowed_users.py, tests/test_yuanbao_reconnect_set_active.py

  Show the diff

  • 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 = {
         "[redacted:email]": "JAlmanzarMint",  # PR #52688 salvage (vision: rasterize SVG / re-encode unsupported raster formats to PNG before embedding), folded into #57890
    -    "[redacted:email]": "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).
    +    "[redacted:email]": "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).
         "[email protected]": "Jigoooo",  # PR #58474 salvage (auxiliary: fall back to token resolver when anthropic pool has no usable entry)
         "[redacted:email]": "ostravajih",  # PR #58374 salvage (poolside: coerce integer finish_reason and tool_call id to strings)
    +    "[redacted:email]": "sahil-shubham",  # PR #58448 salvage (whatsapp_cloud: honor documented WHATSAPP_CLOUD_ALLOWED_USERS / ALLOW_ALL_USERS in the DM intake gate)
         "[redacted:email]": "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).
         "[redacted:email]": "danilofalcao",  # PR #56674 salvage (update: skip unsupported platform.matrix lazy refresh on native Windows — python-olm has no Windows wheel)
         "[email protected]": "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": "[redacted:secret-assignment]",
    +        **(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 = "[redacted:secret-assignment]"
    +    adapter._app_secret = "[redacted:secret-assignment]"
    +    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
    +
    +
    [redacted:email]
    +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": "[redacted:secret-assignment]"},
    +        ),
    +        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
    +
    +
    [redacted:email]
    +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