fix(whatsapp-cloud): review follow-ups for #43921

- nous_subscription: gate the STT managed-default flip on openai-audio
  entitlement and skip when a local backend (faster-whisper or custom
  command) works; new _local_stt_backend_available() helper + tests
- whatsapp_cloud: WHATSAPP_CLOUD_{DM_POLICY,ALLOW_FROM,GROUP_POLICY,
  GROUP_ALLOW_FROM} env overrides so both adapters can run in parallel;
  normalize allowlist entries (JID/punctuation) to bare wa_id
- whatsapp_cloud: wrap per-message event build in try/except (dedup-marked
  wamids would be silently dropped on Meta's batch retry otherwise)
- whatsapp_cloud: validate media_id before URL/filename interpolation,
  delete transient .ogg after voice upload, FIFO-cap interactive-button
  state dicts and per-chat wamid cache
- whatsapp_common: '# **Title**' headers no longer double-wrap asterisks
- setup wizard: read access token / app secret via getpass on TTYs
- docs: new WHATSAPP_CLOUD_* gating env vars

tests/gateway/test_whatsapp_cloud.py

@@ -2248,3 +2248,74 @@ class TestSendTyping:
         await adapter.send_typing("15551234567")
         headers = adapter._http_client.post.call_args.kwargs["headers"]
         assert headers["Authorization"] == "Bearer my-test-token"
+
+
+# ---------------------------------------------------------------------------
+# Allowlist normalization + env decoupling (salvage follow-up)
+# ---------------------------------------------------------------------------
+
+class TestAllowlistNormalization:
+    def test_normalize_allow_ids_strips_jid_suffix_and_punctuation(self):
+        from gateway.platforms.whatsapp_cloud import WhatsAppCloudAdapter
+
+        ids = {"15551234567@s.whatsapp.net", "+1 (555) 765-4321", "15550000000"}
+        normalized = WhatsAppCloudAdapter._normalize_allow_ids(ids)
+        assert normalized == {"15551234567", "15557654321", "15550000000"}
+
+    def test_dm_allowlist_matches_bare_wa_id_against_jid_entry(self):
+        """A Baileys-style JID in the allowlist must match the Cloud API's
+        bare wa_id sender — users share allowlists between both adapters."""
+        from gateway.platforms.whatsapp_cloud import WhatsAppCloudAdapter
+
+        adapter = _make_adapter()
+        adapter._dm_policy = "allowlist"
+        adapter._allow_from = WhatsAppCloudAdapter._normalize_allow_ids(
+            {"15551234567@s.whatsapp.net"}
+        )
+        assert adapter._is_dm_allowed("15551234567") is True
+        assert adapter._is_dm_allowed("19998887777") is False
+
+    def test_cloud_env_overrides_take_precedence(self, monkeypatch):
+        """WHATSAPP_CLOUD_DM_POLICY wins over the shared WHATSAPP_DM_POLICY
+        so both adapters can run in parallel with independent policies."""
+        from gateway.platforms.whatsapp_cloud import WhatsAppCloudAdapter
+
+        monkeypatch.setenv("WHATSAPP_DM_POLICY", "allowlist")
+        monkeypatch.setenv("WHATSAPP_CLOUD_DM_POLICY", "open")
+        monkeypatch.setenv("WHATSAPP_CLOUD_ALLOW_FROM", "+1 555 123 4567")
+
+        config = MagicMock()
+        config.extra = {
+            "phone_number_id": "123",
+            "access_token": "tok",
+        }
+        adapter = WhatsAppCloudAdapter(config)
+        assert adapter._dm_policy == "open"
+        assert adapter._allow_from == {"15551234567"}
+
+
+class TestBoundedInteractiveState:
+    def test_bounded_put_evicts_oldest(self):
+        from collections import OrderedDict
+
+        from gateway.platforms.whatsapp_cloud import (
+            INTERACTIVE_STATE_CACHE_SIZE,
+            WhatsAppCloudAdapter,
+        )
+
+        cache: OrderedDict = OrderedDict()
+        for i in range(INTERACTIVE_STATE_CACHE_SIZE + 10):
+            WhatsAppCloudAdapter._bounded_put(cache, f"id-{i}", "sess")
+        assert len(cache) == INTERACTIVE_STATE_CACHE_SIZE
+        assert "id-0" not in cache
+        assert f"id-{INTERACTIVE_STATE_CACHE_SIZE + 9}" in cache
+
+
+class TestMediaIdValidation:
+    @pytest.mark.asyncio
+    async def test_traversal_media_id_refused(self):
+        adapter = _make_adapter()
+        adapter._http_client = MagicMock()  # would be used if not refused
+        path, mime = await adapter._download_media_to_cache("../../etc/passwd")
+        assert path is None and mime is None
+        adapter._http_client.get.assert_not_called()

tests/gateway/test_whatsapp_formatting.py

@@ -91,6 +91,13 @@ class TestFormatMessage:
         assert adapter.format_message("## Subtitle") == "*Subtitle*"
         assert adapter.format_message("### Deep") == "*Deep*"
 
+    def test_bold_header_does_not_double_wrap(self):
+        """"# **Title**" must become *Title*, not **Title** (WhatsApp would
+        render the doubled asterisks literally)."""
+        adapter = _make_adapter()
+        assert adapter.format_message("# **Title**") == "*Title*"
+        assert adapter.format_message("## __Strong__") == "*Strong*"
+
     def test_links_converted(self):
         adapter = _make_adapter()
         result = adapter.format_message("[click here](https://example.com)")

tests/hermes_cli/test_nous_subscription.py

@@ -723,6 +723,10 @@ def test_apply_nous_managed_defaults_flips_stt_provider_to_openai_for_nous_users
     gateway transcribes their voice notes without needing faster-whisper
     installed."""
     monkeypatch.setattr(ns, "get_env_value", lambda name: "")
+    monkeypatch.setattr(ns, "resolve_openai_audio_api_key", lambda: "")
+    # CI installs [all] extras, so faster-whisper is importable there —
+    # force the "no local backend" case this test is about.
+    monkeypatch.setattr(ns, "_local_stt_backend_available", lambda: False)
     # Avoid the heavy real probing in get_nous_subscription_features.
     monkeypatch.setattr(
         ns,
@@ -751,6 +755,66 @@ def test_apply_nous_managed_defaults_flips_stt_provider_to_openai_for_nous_users
     assert config["stt"]["provider"] == "openai"
 
 
+def _stt_features_stub(*, account_info):
+    return ns.NousSubscriptionFeatures(
+        subscribed=True,
+        nous_auth_present=True,
+        provider_is_nous=True,
+        account_info=account_info,
+        features={
+            key: ns.NousFeatureState(
+                key=key, label=key, included_by_default=True,
+                available=False, active=False, managed_by_nous=False,
+                direct_override=False, toolset_enabled=False,
+                explicit_configured=False,
+            )
+            for key in ("web", "image_gen", "video_gen", "tts", "stt", "browser", "modal")
+        },
+    )
+
+
+def test_apply_nous_managed_defaults_keeps_local_stt_when_backend_works(monkeypatch):
+    """A working local backend (faster-whisper installed or custom command)
+    is a strong intent signal — never flip it to the managed gateway."""
+    monkeypatch.setattr(ns, "get_env_value", lambda name: "")
+    monkeypatch.setattr(ns, "resolve_openai_audio_api_key", lambda: "")
+    monkeypatch.setattr(ns, "_local_stt_backend_available", lambda: True)
+    monkeypatch.setattr(
+        ns,
+        "get_nous_subscription_features",
+        lambda config, **kw: _stt_features_stub(
+            account_info=_account(logged_in=True, paid=True)
+        ),
+    )
+
+    config = {"stt": {"provider": "local"}}
+    changed = ns.apply_nous_managed_defaults(config, enabled_toolsets=[])
+
+    assert "stt" not in changed
+    assert config["stt"]["provider"] == "local"
+
+
+def test_apply_nous_managed_defaults_skips_stt_when_not_entitled(monkeypatch):
+    """A subscriber whose tool pool doesn't cover openai-audio must not be
+    pointed at a managed gateway that will refuse them."""
+    monkeypatch.setattr(ns, "get_env_value", lambda name: "")
+    monkeypatch.setattr(ns, "resolve_openai_audio_api_key", lambda: "")
+    monkeypatch.setattr(ns, "_local_stt_backend_available", lambda: False)
+    monkeypatch.setattr(
+        ns,
+        "get_nous_subscription_features",
+        lambda config, **kw: _stt_features_stub(
+            account_info=_account(logged_in=True, paid=False)
+        ),
+    )
+
+    config = {"stt": {"provider": "local"}}
+    changed = ns.apply_nous_managed_defaults(config, enabled_toolsets=[])
+
+    assert "stt" not in changed
+    assert config["stt"]["provider"] == "local"
+
+
 def test_apply_nous_managed_defaults_skips_stt_when_groq_key_present(monkeypatch):
     """Don't override a user who explicitly set up Groq for STT."""
     env = {"GROQ_API_KEY": "groq-key"}