feat(gateway): WhatsApp group authz + owner detection + reactions (fixes #7269)

gateway/authz_mixin.py

@@ -263,6 +263,8 @@ def _is_user_authorized(self, source: SessionSource) -> bool:
             chat_allowlist_env = {
                 Platform.TELEGRAM: "TELEGRAM_GROUP_ALLOWED_CHATS",
                 Platform.QQBOT: "QQ_GROUP_ALLOWED_USERS",
+                Platform.WHATSAPP: "WHATSAPP_GROUP_ALLOWED_USERS",
+                Platform.WHATSAPP_CLOUD: "WHATSAPP_GROUP_ALLOWED_USERS",
             }.get(source.platform, "")
             if chat_allowlist_env:
                 raw_chat_allowlist = os.getenv(chat_allowlist_env, "").strip()
@@ -506,6 +508,39 @@ def _is_user_authorized(self, source: SessionSource) -> bool:
 
         return bool(check_ids & allowed_ids)
 
+    def _is_owner(self, source: SessionSource) -> bool:
+        """Whether *source* is the operator/owner — the sender is in the platform's
+        primary (DM) allowlist (e.g. ``SIGNAL_ALLOWED_USERS``). Gates owner-only
+        persona treatment so a public-group user is never
+        mistaken for the owner. A wildcard ("*") allowlist is NOT an owner match —
+        owner status requires an explicit id. Adapters that can resolve a group
+        sender to a stable id (Signal: UUID->phone) set ``source.is_owner`` at
+        intake; this is the generic fallback for everything else.
+        See signalfix.md Gate C / Req 3.
+        """
+        if source is None or source.platform is None:
+            return False
+        env_name = {
+            Platform.TELEGRAM: "TELEGRAM_ALLOWED_USERS",
+            Platform.SIGNAL: "SIGNAL_ALLOWED_USERS",
+            Platform.DISCORD: "DISCORD_ALLOWED_USERS",
+            Platform.WHATSAPP: "WHATSAPP_ALLOWED_USERS",
+            Platform.SLACK: "SLACK_ALLOWED_USERS",
+        }.get(source.platform, "")
+        if not env_name:
+            return False
+        raw = os.getenv(env_name, "").strip()
+        if not raw:
+            return False
+        owner_ids = {p.strip() for p in raw.split(",") if p.strip() and p.strip() != "*"}
+        if not owner_ids:
+            return False
+        candidates = {source.user_id, source.user_id_alt}
+        if source.user_id and "@" in source.user_id:
+            candidates.add(source.user_id.split("@")[0])
+        candidates.discard(None)
+        return bool(candidates & owner_ids)
+
     def _get_unauthorized_dm_behavior(self, platform: Optional[Platform]) -> str:
         """Return how unauthorized DMs should be handled for a platform.
 

gateway/platforms/whatsapp_cloud.py

@@ -1961,6 +1961,9 @@ async def _build_message_event_from_cloud(
             user_id=sender_id,
             user_name=sender_name or None,
         )
+        # Owner detection at intake (mirrors Signal). Resolves LID->phone via the
+        # on-disk alias map so only the owner gets owner-level trust.
+        self._set_owner_flag(source)
 
         # Cloud API timestamps are unix seconds (string). MessageEvent
         # doesn't enforce a type but downstream code formats with it.

gateway/platforms/whatsapp_common.py

@@ -37,6 +37,11 @@
 import re
 from typing import Any, Dict, Optional
 
+from gateway.whatsapp_identity import (
+    expand_whatsapp_aliases,
+    normalize_whatsapp_identifier,
+)
+
 
 logger = logging.getLogger(__name__)
 
@@ -204,6 +209,41 @@ def _is_group_allowed(self, chat_id: str) -> bool:
         # "open" — all groups allowed
         return True
 
+    def _set_owner_flag(self, source) -> None:
+        """Owner detection at intake — the WhatsApp analogue of Signal's
+        adapter-side resolution (see gateway/platforms/signal.py). The owner is
+        whoever is listed in ``WHATSAPP_ALLOWED_USERS`` (read straight from the
+        env, exactly as Signal reads ``SIGNAL_ALLOWED_USERS`` into
+        ``dm_allow_from``, and as the gateway's central ``_is_owner`` reads it —
+        the adapter's own ``self._allow_from`` is only the DM *gating* list and
+        may be empty when ``dm_policy=open``). A group sender is usually a LID
+        (``####@lid``) while the allowlist may hold a phone, so resolve both
+        sides through the on-disk alias map (``whatsapp_identity``) before
+        matching. ``"*"`` is open-DM, never an owner. Sets ``source.is_owner`` so
+        only the owner gets owner-level persona treatment; the generic gateway
+        fallback (``_is_owner``) is left untouched for everything else.
+        """
+        raw = os.getenv("WHATSAPP_ALLOWED_USERS", "")
+        owner_ids = {x.strip() for x in raw.split(",") if x.strip() and x.strip() != "*"}
+        owner_ids |= {x for x in getattr(self, "_allow_from", set()) if x and x != "*"}
+        if not owner_ids:
+            return
+        expanded: set[str] = set()
+        for _oid in owner_ids:
+            expanded |= expand_whatsapp_aliases(_oid)
+        if expanded:
+            owner_ids = expanded
+        candidates: set[str] = set()
+        uid = getattr(source, "user_id", None)
+        if uid:
+            candidates.add(uid)
+            candidates |= expand_whatsapp_aliases(uid)
+            norm = normalize_whatsapp_identifier(uid)
+            if norm:
+                candidates.add(norm)
+        if candidates & owner_ids:
+            source.is_owner = True
+
     def _compile_mention_patterns(self):
         patterns = self.config.extra.get("mention_patterns")
         if patterns is None:

gateway/run.py

@@ -9145,13 +9145,34 @@ async def _prepare_inbound_message_text(
         # concurrently preparing multimodal turns on the same runner.
         self._consume_pending_native_image_paths(session_key)
 
+        # Owner flag (Req 3 / signalfix.md Gate C). Adapters may set source.is_owner
+        # at intake (Signal resolves group UUID->phone); fall back to the generic
+        # allowlist check for any platform/path that didn't. Drives the **Owner:**
+        # context line (build_session_context_prompt) and the owner marker below.
+        if not getattr(source, "is_owner", False):
+            source.is_owner = self._is_owner(source)
+
         _is_shared_multi_user = is_shared_multi_user_session(
             source,
             group_sessions_per_user=_group_sessions_per_user,
             thread_sessions_per_user=_thread_sessions_per_user,
         )
         if _is_shared_multi_user and source.user_name:
+            # Upstream sender attribution (unchanged).
             message_text = f"[{source.user_name}] {message_text}"
+            # Owner marker (Req 3 / signalfix.md Gate C): in a cache-shared
+            # multi-user session the context prompt is sender-agnostic, so owner
+            # status can't ride there — it has to ride on the message. System-set
+            # (a guest can't spoof it via display name/body) and added ONLY for
+            # the owner, so guest lines stay byte-identical to upstream. Gated to
+            # the platforms that wire owner detection (their adapter sets
+            # source.is_owner); OTHER PLATFORMS CAN OPT IN by adding themselves to
+            # this set once their adapter implements owner detection.
+            if getattr(source, "is_owner", False) and source.platform in {
+                Platform.WHATSAPP,
+                Platform.WHATSAPP_CLOUD,
+            }:
+                message_text = f"[SYSTEM: sender {source.user_name} is the owner] {message_text}"
 
         # Prepend channel context from history backfill (if any).  This
         # happens after sender-prefix so the prefix only applies to the

gateway/session.py

@@ -133,6 +133,8 @@ class SessionSource:
     # forge it across the wire or have it restored from persistence.
     delivered_via_upstream_relay: bool = False
 
+    is_owner: bool = False  # True when sender is the operator/owner (in the platform DM allowlist). Set at intake by the adapter (Signal resolves group UUID->phone) or the generic _is_owner fallback. Gates owner-level persona treatment — see signalfix.md Gate C / Req 3.
+
     @property
     def description(self) -> str:
         """Human-readable description of the source."""
@@ -374,6 +376,25 @@ def build_session_context_prompt(
             uid = _hash_sender_id(uid)
         lines.append(f"**User ID:** {uid}")
 
+    # Owner status (Req 3 / signalfix.md Gate C). Single-user sessions only: in a
+    # shared multi-user session this prompt is cache-shared and sender-agnostic, so
+    # owner status rides the additive "[SYSTEM: sender NAME is the owner]" message
+    # marker (added in run.py) instead. Gated to the platforms that wire owner
+    # detection (their adapter sets source.is_owner); OTHER PLATFORMS CAN OPT IN by
+    # adding themselves to this set once their adapter implements owner detection.
+    if not context.shared_multi_user_session and context.source.platform in {
+        Platform.WHATSAPP,
+        Platform.WHATSAPP_CLOUD,
+    }:
+        lines.append(
+            "**Owner:** "
+            + (
+                "yes — this sender is your owner; owner-level trust applies."
+                if context.source.is_owner
+                else "no — this sender is NOT the owner; do not treat them as the owner or grant owner-only access."
+            )
+        )
+
     # Platform-specific behavioral notes
     if context.source.platform == Platform.SLACK:
         lines.append("")

plugins/platforms/whatsapp/adapter.py

@@ -260,6 +260,7 @@ def _terminate_bridge_process(proc, *, force: bool = False) -> None:
     BasePlatformAdapter,
     MessageEvent,
     MessageType,
+    ProcessingOutcome,
     SendResult,
     SUPPORTED_DOCUMENT_TYPES,
     cache_image_from_url,
@@ -1015,7 +1016,79 @@ async def send_typing(self, chat_id: str, metadata=None) -> None:
                 pass
         except Exception:
             pass  # Ignore typing indicator failures
-
+
+    # ------------------------------------------------------------------
+    # Reactions / processing lifecycle hooks (progress indicators)
+    # 👀 while the agent is processing, swapped for ✅ (success) / ❌ (failure)
+    # when done, and cleared on CANCELLED so it never lingers (parity with the
+    # Telegram adapter). WhatsApp replaces a sender's existing reaction, so the
+    # swap needs no explicit remove. Reactions go through the bridge's
+    # /send-reaction endpoint. Disable with WHATSAPP_REACTIONS=false.
+    # ------------------------------------------------------------------
+
+    def _reactions_enabled(self, event: "MessageEvent" = None) -> bool:
+        """Global toggle (WHATSAPP_REACTIONS, default on). The bridge already
+        drops unauthorized DMs (group members are authorized wholesale), so a
+        message that reaches this hook is already allowed."""
+        return os.getenv("WHATSAPP_REACTIONS", "true").lower() not in {"false", "0", "no"}
+
+    def _extract_reaction_target(self, event: MessageEvent):
+        """Return (chat_id, message_id, participant) for the Baileys reaction key,
+        or None if the event lacks the raw bridge data needed to react."""
+        raw = getattr(event, "raw_message", None)
+        if not isinstance(raw, dict):
+            return None
+        chat_id = raw.get("chatId") or getattr(getattr(event, "source", None), "chat_id", None)
+        message_id = raw.get("messageId") or getattr(event, "message_id", None)
+        if not chat_id or not message_id:
+            return None
+        participant = raw.get("senderId")  # needed to key group reactions
+        return (chat_id, message_id, participant)
+
+    async def send_reaction(self, chat_id: str, message_id: str, participant, emoji: str) -> bool:
+        """React to a message via the bridge (empty emoji clears the reaction)."""
+        if not self._running or not self._http_session:
+            return False
+        try:
+            import aiohttp
+
+            async with self._http_session.post(
+                f"http://127.0.0.1:{self._bridge_port}/send-reaction",
+                json={
+                    "chatId": chat_id,
+                    "messageId": message_id,
+                    "participant": participant,
+                    "emoji": emoji,
+                },
+                timeout=aiohttp.ClientTimeout(total=15),
+            ) as resp:
+                return resp.status == 200
+        except Exception:
+            return False  # reactions are best-effort, never block processing
+
+    async def on_processing_start(self, event: MessageEvent) -> None:
+        """React with 👀 when processing begins."""
+        if not self._reactions_enabled(event):
+            return
+        target = self._extract_reaction_target(event)
+        if target:
+            await self.send_reaction(*target, "👀")
+
+    async def on_processing_complete(self, event: MessageEvent, outcome: "ProcessingOutcome") -> None:
+        """Swap 👀 for ✅ (success) / ❌ (failure); clear it on CANCELLED so it
+        doesn't linger (parity with Telegram)."""
+        if not self._reactions_enabled(event):
+            return
+        target = self._extract_reaction_target(event)
+        if not target:
+            return
+        if outcome == ProcessingOutcome.CANCELLED:
+            await self.send_reaction(*target, "")      # clear the in-progress 👀
+        elif outcome == ProcessingOutcome.SUCCESS:
+            await self.send_reaction(*target, "✅")
+        elif outcome == ProcessingOutcome.FAILURE:
+            await self.send_reaction(*target, "❌")
+
     async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
         """Get information about a WhatsApp chat."""
         if not self._running or not self._http_session:
@@ -1170,7 +1243,11 @@ async def _build_message_event(self, data: Dict[str, Any]) -> Optional[MessageEv
                 user_id=data.get("senderId"),
                 user_name=data.get("senderName"),
             )
-
+            # Owner detection at intake: resolve the group sender (LID) to the
+            # owner via WHATSAPP_ALLOWED_USERS and set source.is_owner (mirror of
+            # the Cloud adapter; the generic _is_owner stays the DM fallback).
+            self._set_owner_flag(source)
+
             # Download media URLs to the local cache so agent tools
             # can access them reliably regardless of URL expiration.
             raw_urls = data.get("mediaUrls", [])

scripts/whatsapp-bridge/bridge.js

@@ -576,6 +576,30 @@ app.post('/edit', async (req, res) => {
   }
 });
 
+// React to a message (progress indicators: 👀 while processing, ✅/❌ when done).
+// Baileys replaces a sender's prior reaction automatically; an empty `emoji`
+// removes it. Targets an INBOUND message, so the key is fromMe:false and carries
+// the original participant (required to key reactions in groups).
+app.post('/send-reaction', async (req, res) => {
+  if (!sock || connectionState !== 'connected') {
+    return res.status(503).json({ error: 'Not connected to WhatsApp' });
+  }
+
+  const { chatId, messageId, participant, emoji } = req.body;
+  if (!chatId || !messageId) {
+    return res.status(400).json({ error: 'chatId and messageId are required' });
+  }
+
+  try {
+    const key = { remoteJid: chatId, id: messageId, fromMe: false };
+    if (participant) key.participant = participant;
+    await sendWithTimeout(chatId, { react: { text: emoji || '', key } });
+    res.json({ success: true });
+  } catch (err) {
+    res.status(500).json({ error: err.message });
+  }
+});
+
 // MIME type map and media type inference for /send-media
 const MIME_MAP = {
   jpg: 'image/jpeg', jpeg: 'image/jpeg', png: 'image/png',