feat(gateway): bring Signal groups to Telegram parity (authz, mention, owner detection)

gateway/authz_mixin.py

@@ -415,6 +415,7 @@ def _is_user_authorized(self, source: SessionSource) -> bool:
             chat_allowlist_env = {
                 Platform.TELEGRAM: "TELEGRAM_GROUP_ALLOWED_CHATS",
                 Platform.QQBOT: "QQ_GROUP_ALLOWED_USERS",
+                Platform.SIGNAL: "SIGNAL_GROUP_ALLOWED_USERS",
             }.get(source.platform, "")
             if chat_allowlist_env:
                 raw_chat_allowlist = os.getenv(chat_allowlist_env, "").strip()
@@ -424,7 +425,15 @@ def _is_user_authorized(self, source: SessionSource) -> bool:
                         for cid in raw_chat_allowlist.split(",")
                         if cid.strip()
                     }
-                    if "*" in allowed_group_ids or source.chat_id in allowed_group_ids:
+                    # Match the platform chat_id OR its raw form. Signal emits
+                    # chat_id="group:<id>" while its allowlist holds the raw <id>
+                    # (also exposed as chat_id_alt), so compare both the raw
+                    # alt id and the "group:"-stripped form, not just chat_id.
+                    group_id_candidates = {source.chat_id, source.chat_id_alt}
+                    if source.chat_id and source.chat_id.startswith("group:"):
+                        group_id_candidates.add(source.chat_id.split("group:", 1)[1])
+                    group_id_candidates.discard(None)
+                    if "*" in allowed_group_ids or (group_id_candidates & allowed_group_ids):
                         return True
 
             # Fallback: also check adapter-level config (config.yaml)
@@ -722,6 +731,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],

gateway/platforms/signal.py

@@ -621,6 +621,18 @@ async def _handle_envelope(self, envelope: dict) -> None:
         if text and mentions:
             text = _render_mentions(text, mentions)
 
+        # Extract quote (reply-to) context from Signal dataMessage. Signal's
+        # quote.id is the timestamp of the quoted message; quote.author points
+        # at the quoted sender when available. Preserve both so the gateway can
+        # tell the agent when the user replied to a specific assistant message.
+        # Resolved here (before the mention filter) so reply-to-bot can gate on it.
+        quote_data = data_message.get("quote") or {}
+        reply_to_id = str(quote_data.get("id")) if quote_data.get("id") else None
+        reply_to_text = quote_data.get("text")
+        reply_to_author = self._extract_quote_author(quote_data)
+        reply_to_author_name = quote_data.get("authorName") or quote_data.get("authorProfileName")
+        reply_to_is_own = self._quote_references_own_message(reply_to_id, reply_to_author)
+
         # Mention filter: in groups, only process messages that @mention the bot account
         if is_group and self.require_mention:
             account_norm = self._account_normalized
@@ -632,7 +644,22 @@ async def _handle_envelope(self, envelope: dict) -> None:
                 m.get("number") == account_norm or m.get("uuid") == account_norm
                 for m in (data_message.get("mentions") or [])
             )
-            if not mentioned_in_text and not mentioned_in_metadata:
+            # Reply-to-bot counts as a mention (parity with Telegram/WhatsApp):
+            # treat a quote of a message the bot itself sent as "addressed to me".
+            # Reuse the robust resolver (outbound-timestamp cache + number<->uuid
+            # mapping) instead of a raw phone compare, so UUID-only quote authors
+            # and timestamp-only quotes are matched too.
+            replied_to_bot = reply_to_is_own
+            # Slash-bypass: let "/cmd" run without an @mention (parity with
+            # Telegram/WhatsApp). Slash-access / group_user_allowed_commands still
+            # gates which commands a non-owner may actually run.
+            is_slash = (text or "").strip().startswith("/")
+            if not (
+                mentioned_in_text
+                or mentioned_in_metadata
+                or replied_to_bot
+                or is_slash
+            ):
                 logger.debug(
                     "Signal: ignoring group message (require_mention=true, bot not mentioned)"
                 )
@@ -659,17 +686,6 @@ async def _handle_envelope(self, envelope: dict) -> None:
                 # intentional newlines in a multi-line message are preserved.
                 text = text.replace("  ", " ").strip()
 
-        # Extract quote (reply-to) context from Signal dataMessage. Signal's
-        # quote.id is the timestamp of the quoted message; quote.author points
-        # at the quoted sender when available. Preserve both so the gateway can
-        # tell the agent when the user replied to a specific assistant message.
-        quote_data = data_message.get("quote") or {}
-        reply_to_id = str(quote_data.get("id")) if quote_data.get("id") else None
-        reply_to_text = quote_data.get("text")
-        reply_to_author = self._extract_quote_author(quote_data)
-        reply_to_author_name = quote_data.get("authorName") or quote_data.get("authorProfileName")
-        reply_to_is_own = self._quote_references_own_message(reply_to_id, reply_to_author)
-
         # Process attachments
         attachments_data = data_message.get("attachments", [])
         media_urls = []
@@ -717,6 +733,22 @@ async def _handle_envelope(self, envelope: dict) -> None:
             chat_id_alt=group_id if is_group else None,
         )
 
+        # Owner detection (Req 3 / signalfix.md Gate C). The owner is whoever is
+        # listed in SIGNAL_ALLOWED_USERS (self.dm_allow_from). In groups the sender
+        # is frequently a UUID, so resolve UUID->phone via the number<->uuid cache
+        # before matching. "*" is open-DM, never an owner. Surfaced to the model so
+        # only the owner receives owner-level persona treatment.
+        _owner_ids = {x for x in self.dm_allow_from if x and x != "*"}
+        if _owner_ids:
+            _sender_phone = sender if _looks_like_e164_number(sender) else (
+                self._recipient_number_by_uuid.get(sender_uuid or "", "")
+            )
+            source.is_owner = bool(
+                sender in _owner_ids
+                or (sender_uuid and sender_uuid in _owner_ids)
+                or (_sender_phone and _sender_phone in _owner_ids)
+            )
+
         # Determine message type from media
         msg_type = MessageType.TEXT
         if media_types:

gateway/run.py

@@ -12389,6 +12389,13 @@ 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,
@@ -12414,6 +12421,18 @@ async def _prepare_inbound_message_text(
                     f"{_safe_user_name} | Slack user <@{source.user_id}>"
                 )
             message_text = f"[{_safe_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.SIGNAL,
+            }:
+                message_text = f"[SYSTEM: sender {_safe_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

@@ -204,6 +204,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.
+
     def __post_init__(self) -> None:
         # D-Q2.5 dual-field reconciliation: `scope_id` is canonical, `guild_id`
         # is the deprecated alias. Mirror whichever was provided onto the other
@@ -561,6 +563,24 @@ def build_session_context_prompt(
             uid = _hash_sender_id(uid)
         lines.append(f"**User ID:** {_format_untrusted_prompt_value(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.SIGNAL,
+    }:
+        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:
         # Inject the Slack capability note only when the agent actually has

tests/gateway/test_shared_group_sender_prefix.py

@@ -70,6 +70,71 @@ async def test_preprocess_keeps_plain_text_for_default_group_sessions():
     assert result == "hello"
 
 
+@pytest.mark.asyncio
+async def test_owner_gets_additive_marker_in_shared_session():
+    """Owner messages get a system owner marker IN ADDITION to the [name] prefix.
+
+    Uses Signal because the owner marker is gated to the platforms that wire owner
+    detection (Signal here); other platforms stay byte-identical.
+    """
+    runner = _make_runner(
+        GatewayConfig(
+            platforms={
+                Platform.SIGNAL: PlatformConfig(enabled=True),
+            },
+            group_sessions_per_user=False,
+        )
+    )
+    source = SessionSource(
+        platform=Platform.SIGNAL,
+        chat_id="group:abc123==",
+        chat_name="Test Group",
+        chat_type="group",
+        user_name="Bob",
+        is_owner=True,
+    )
+    event = MessageEvent(text="hello", source=source)
+
+    result = await runner._prepare_inbound_message_text(
+        event=event,
+        source=source,
+        history=[],
+    )
+
+    # [name] prefix preserved (upstream); owner marker prepended only for owner.
+    assert result == "[SYSTEM: sender Bob is the owner] [Bob] hello"
+
+
+@pytest.mark.asyncio
+async def test_guest_has_no_owner_marker_in_shared_session():
+    """Non-owner (guest) messages keep the plain upstream [name] prefix only."""
+    runner = _make_runner(
+        GatewayConfig(
+            platforms={
+                Platform.TELEGRAM: PlatformConfig(enabled=True, token="fake"),
+            },
+            group_sessions_per_user=False,
+        )
+    )
+    source = SessionSource(
+        platform=Platform.TELEGRAM,
+        chat_id="-1002285219667",
+        chat_name="Test Group",
+        chat_type="group",
+        user_name="Alice",
+    )
+    event = MessageEvent(text="hello", source=source)
+
+    result = await runner._prepare_inbound_message_text(
+        event=event,
+        source=source,
+        history=[],
+    )
+
+    assert result == "[Alice] hello"
+    assert "is the owner" not in result
+
+
 @pytest.mark.asyncio
 async def test_preprocess_includes_slack_author_mention_for_shared_thread():
     """Shared Slack threads expose the current author's verifiable user ID
@@ -129,3 +194,4 @@ async def test_preprocess_slack_shared_thread_without_user_id_keeps_name_only():
     )
 
     assert result == "[Alice] hello"
+

tests/gateway/test_signal.py

@@ -2587,3 +2587,80 @@ def test_ttl_evicts_stale_entries(self, monkeypatch):
         adapter._track_sent_timestamp({"timestamp": 3})
         # Both 1 and 2 should be evicted on TTL, only 3 remains
         assert list(adapter._recent_sent_timestamps.keys()) == [3]
+
+
+# ---------------------------------------------------------------------------
+# Reply-to-bot counts as a mention (Bug 2 / PR #53348 review)
+# ---------------------------------------------------------------------------
+
+class TestSignalReplyToBotMention:
+    """In a require_mention group, quoting a message the bot itself sent must
+    bypass the @mention requirement — resolved via the robust timestamp +
+    number<->uuid path (`_quote_references_own_message`), not a raw phone
+    compare, so UUID-only quote authors and timestamp-only quotes are matched.
+    """
+
+    @staticmethod
+    def _group_envelope(quote):
+        return {
+            "envelope": {
+                "sourceNumber": "+15550002222",
+                "sourceUuid": "guest-uuid",
+                "sourceName": "Guest",
+                "timestamp": 1700000000,
+                "dataMessage": {
+                    "message": "thanks",
+                    "groupInfo": {"groupId": "grp=="},
+                    "quote": quote,
+                },
+            }
+        }
+
+    @pytest.mark.asyncio
+    async def test_reply_to_bot_by_timestamp_bypasses_mention(self, monkeypatch):
+        adapter = _make_signal_adapter(monkeypatch, require_mention=True, group_allowed="*")
+        adapter._remember_sent_message_timestamp(555000)
+        captured = {}
+
+        async def fake_handle(event):
+            captured["event"] = event
+
+        adapter.handle_message = fake_handle
+        # No @mention; quote.id matches a timestamp the bot sent (author is the guest).
+        await adapter._handle_envelope(
+            self._group_envelope({"id": 555000, "text": "assistant answer", "author": "guest-uuid"})
+        )
+        assert "event" in captured, "reply-to-own-timestamp should count as a mention"
+        assert captured["event"].reply_to_is_own_message is True
+
+    @pytest.mark.asyncio
+    async def test_reply_to_bot_by_uuid_author_bypasses_mention(self, monkeypatch):
+        adapter = _make_signal_adapter(monkeypatch, require_mention=True, group_allowed="*")
+        # Bot's own number<->uuid mapping observed from an envelope.
+        adapter._recipient_uuid_by_number[adapter._account_normalized] = "bot-uuid"
+        captured = {}
+
+        async def fake_handle(event):
+            captured["event"] = event
+
+        adapter.handle_message = fake_handle
+        # No @mention; quote author is the bot's UUID (never its phone).
+        await adapter._handle_envelope(
+            self._group_envelope({"id": 42, "text": "assistant answer", "author": "bot-uuid"})
+        )
+        assert "event" in captured, "reply whose quote author is the bot UUID should count as a mention"
+
+    @pytest.mark.asyncio
+    async def test_reply_to_non_bot_still_requires_mention(self, monkeypatch):
+        adapter = _make_signal_adapter(monkeypatch, require_mention=True, group_allowed="*")
+        captured = {}
+
+        async def fake_handle(event):
+            captured["event"] = event
+
+        adapter.handle_message = fake_handle
+        # No @mention; quote points at another member and an unknown timestamp.
+        await adapter._handle_envelope(
+            self._group_envelope({"id": 111, "text": "someone else", "author": "other-uuid"})
+        )
+        assert "event" not in captured, "reply to a non-bot message must NOT bypass require_mention"