fix(im): dingtalk @ mention + sender identity + codex \n escape

- IM bridge: propagate sender identity (name/id) through all adapters
- Dingtalk: resolve @ mention targets using staffId instead of generic senderId
- Codex runtime: normalize double-escaped \n in message text
- Tests: dingtalk at_mention, sender identity, mcp bool coercion

Author: waterbang

src/cccc/ports/im/adapters/base.py

@@ -85,11 +85,19 @@ def poll(self) -> List[Dict[str, Any]]:
         pass
 
     @abstractmethod
-    def send_message(self, chat_id: str, text: str, thread_id: Optional[int] = None) -> bool:
+    def send_message(
+        self,
+        chat_id: str,
+        text: str,
+        thread_id: Optional[int] = None,
+        *,
+        mention_user_ids: Optional[List[str]] = None,
+    ) -> bool:
         """
         Send a message to a chat.
         Returns True if successful.
         """
+        _ = mention_user_ids
         pass
 
     @abstractmethod

src/cccc/ports/im/adapters/dingtalk.py

@@ -131,7 +131,8 @@ def __init__(
         # so throttle state survives across begin/update/end invocations.
         self._card_client: Optional[Any] = None
 
-        # Cache: last inbound sender per conversation (for outbound @mention)
+        # Cache: last inbound mentionable sender per conversation (for backward-compatible
+        # outbound @mention fallback). Only senderStaffId is safe to reuse here.
         # conversation_id -> (staff_id, nick)   — bounded to _LAST_SENDER_MAX entries
         self._last_sender: Dict[str, tuple[str, str]] = {}
 
@@ -567,8 +568,11 @@ def _enqueue_message(self, event: Dict[str, Any]) -> bool:
             # DingTalk event structure varies by type
             # Robot callback format
             msg_type = event.get("msgtype", "")
-            conversation_id = event.get("conversationId", "")
-            sender_id = event.get("senderStaffId", "") or event.get("senderId", "")
+            conversation_id = str(event.get("conversationId", "") or "").strip()
+            sender_staff_id = str(event.get("senderStaffId", "") or "").strip()
+            sender_id = str(event.get("senderId", "") or "").strip()
+            sender_display_id = sender_staff_id or sender_id
+            mention_user_ids = [sender_staff_id] if sender_staff_id else []
             sender_nick = event.get("senderNick", "user")
             msg_id = event.get("msgId", "")
 
@@ -679,16 +683,19 @@ def _enqueue_message(self, event: Dict[str, Any]) -> bool:
                 self._session_webhook_cache[conversation_id] = (session_webhook, expires_at)
                 self._log(f"[webhook] Cached: id={conversation_id}, expires_raw={session_expires}, expires_at={expires_at:.0f}")
 
-            # Cache sender for outbound @mention (group chats only)
-            if sender_id and conversation_id:
+            # Cache sender for outbound @mention fallback (group chats only).
+            # senderId is not necessarily a valid atUserIds target, so only keep staffId.
+            if sender_staff_id and conversation_id:
                 # Evict oldest entries when cache is full
                 if len(self._last_sender) >= self._LAST_SENDER_MAX:
                     try:
                         oldest_key = next(iter(self._last_sender))
                         del self._last_sender[oldest_key]
                     except StopIteration:
                         pass
-                self._last_sender[conversation_id] = (sender_id, sender_nick)
+                self._last_sender[conversation_id] = (sender_staff_id, sender_nick)
+            elif conversation_id:
+                self._last_sender.pop(conversation_id, None)
 
             # Normalize message
             normalized = {
@@ -699,8 +706,9 @@ def _enqueue_message(self, event: Dict[str, Any]) -> bool:
                 "thread_id": 0,  # DingTalk doesn't have threading like this
                 "text": text,
                 "attachments": attachments,
-                "from_user": sender_nick or sender_id,
-                "from_user_id": sender_id,
+                "from_user": sender_nick or sender_display_id,
+                "from_user_id": sender_display_id,
+                "mention_user_ids": mention_user_ids,
                 "message_id": msg_id,
                 "timestamp": self._parse_event_time(event.get("createAt")),
                 # Keep sessionWebhook for potential reply use
@@ -770,19 +778,27 @@ def _get_conversation_title_cached(self, conversation_id: str) -> str:
         self._conversation_cache[conversation_id] = title
         return title
 
+    def _build_markdown_payload(self, text: str, at_user_ids: Optional[List[str]] = None) -> Dict[str, Any]:
+        """Build a DingTalk markdown payload with optional real-mention metadata."""
+        payload: Dict[str, Any] = {
+            "title": text[:20] if len(text) > 20 else text,
+            "text": text,
+        }
+        cleaned_at_user_ids = [str(x).strip() for x in (at_user_ids or []) if str(x).strip()]
+        if cleaned_at_user_ids:
+            payload["at"] = {"atUserIds": cleaned_at_user_ids}
+        return payload
+
     def _send_via_webhook(self, webhook_url: str, text: str,
                           at_user_ids: Optional[List[str]] = None) -> bool:
         """Send message via sessionWebhook (most reliable for groups)."""
-        title = text[:20] if len(text) > 20 else text
         body: Dict[str, Any] = {
             "msgtype": "markdown",
-            "markdown": {
-                "title": title,
-                "text": text,
-            },
+            "markdown": self._build_markdown_payload(text),
         }
-        if at_user_ids:
-            body["at"] = {"atUserIds": at_user_ids}
+        cleaned_at_user_ids = [str(x).strip() for x in (at_user_ids or []) if str(x).strip()]
+        if cleaned_at_user_ids:
+            body["at"] = {"atUserIds": cleaned_at_user_ids}
         data = json.dumps(body, ensure_ascii=False).encode('utf-8')
 
         req = urllib.request.Request(webhook_url, data=data, method="POST")
@@ -800,7 +816,14 @@ def _send_via_webhook(self, webhook_url: str, text: str,
             self._log(f"[webhook] Error: {e}")
             return False
 
-    def send_message(self, chat_id: str, text: str, thread_id: Optional[int] = None) -> bool:
+    def send_message(
+        self,
+        chat_id: str,
+        text: str,
+        thread_id: Optional[int] = None,
+        *,
+        mention_user_ids: Optional[List[str]] = None,
+    ) -> bool:
         """
         Send a text message to a conversation.
 
@@ -821,11 +844,16 @@ def send_message(self, chat_id: str, text: str, thread_id: Optional[int] = None)
         safe_text = self._compose_safe(text)
 
         # Resolve @mention targets for group conversations
-        at_user_ids: Optional[List[str]] = None
-        if chat_id.startswith("cid") and chat_id in self._last_sender:
-            staff_id, _nick = self._last_sender[chat_id]
-            if staff_id:
-                at_user_ids = [staff_id]
+        at_user_ids: Optional[List[str]]
+        if mention_user_ids is not None:
+            cleaned_explicit_ids = [str(x).strip() for x in mention_user_ids if str(x).strip()]
+            at_user_ids = cleaned_explicit_ids or None
+        else:
+            at_user_ids = None
+            if chat_id.startswith("cid") and chat_id in self._last_sender:
+                staff_id, _nick = self._last_sender[chat_id]
+                if staff_id:
+                    at_user_ids = [staff_id]
 
         # Rate limit
         self._rate_limiter.wait_and_acquire(chat_id)
@@ -847,18 +875,16 @@ def send_message(self, chat_id: str, text: str, thread_id: Optional[int] = None)
         if not self.robot_code:
             if chat_id.startswith("cid"):
                 self._log("[send] Missing robot_code; cannot use new API fallback. Trying legacy API.")
-                return self._send_message_legacy(chat_id, safe_text)
+                return self._send_message_legacy(chat_id, safe_text, at_user_ids=at_user_ids)
             self._log("[send] Missing robot_code; cannot send via API fallback. Configure DINGTALK_ROBOT_CODE.")
             return False
 
         # Use robot message API
+        markdown_payload = self._build_markdown_payload(safe_text, at_user_ids=at_user_ids)
         body: Dict[str, Any] = {
             "robotCode": self.robot_code,
             "msgKey": "sampleMarkdown",
-            "msgParam": json.dumps({
-                "title": safe_text[:20] if len(safe_text) > 20 else safe_text,
-                "text": safe_text,
-            }, ensure_ascii=False),
+            "msgParam": json.dumps(markdown_payload, ensure_ascii=False),
         }
 
         # Determine if group or 1:1
@@ -878,24 +904,28 @@ def send_message(self, chat_id: str, text: str, thread_id: Optional[int] = None)
 
         # Try alternative API for older bots
         if "code" in resp or "errcode" in resp:
-            return self._send_message_legacy(chat_id, safe_text)
+            return self._send_message_legacy(chat_id, safe_text, at_user_ids=at_user_ids)
 
         self._log(f"[send] Failed to chat {chat_id}: {resp}")
         return False
 
-    def _send_message_legacy(self, chat_id: str, text: str) -> bool:
+    def _send_message_legacy(
+        self,
+        chat_id: str,
+        text: str,
+        at_user_ids: Optional[List[str]] = None,
+    ) -> bool:
         """Send message using legacy API (for older bot types)."""
-        title = text[:20] if len(text) > 20 else text
         body = {
             "chatid": chat_id,
             "msg": {
                 "msgtype": "markdown",
-                "markdown": {
-                    "title": title,
-                    "text": text,
-                },
+                "markdown": self._build_markdown_payload(text),
             },
         }
+        cleaned_at_user_ids = [str(x).strip() for x in (at_user_ids or []) if str(x).strip()]
+        if cleaned_at_user_ids:
+            body["msg"]["at"] = {"atUserIds": cleaned_at_user_ids}
 
         resp = self._api_old("POST", "/chat/send", body)
 

src/cccc/ports/im/adapters/discord.py

@@ -276,10 +276,18 @@ async def _resolve_channel(self, chat_id: str) -> Any:
             self._log(f"[warn] Channel {chat_id} not found (cache miss + fetch failed: {e})")
             return None
 
-    def send_message(self, chat_id: str, text: str, thread_id: Optional[int] = None) -> bool:
+    def send_message(
+        self,
+        chat_id: str,
+        text: str,
+        thread_id: Optional[int] = None,
+        *,
+        mention_user_ids: Optional[List[str]] = None,
+    ) -> bool:
         """
         Send a message to a Discord channel.
         """
+        _ = mention_user_ids
         _ = thread_id  # Discord threads are not wired yet (future work).
         if not self._connected or not self._client or not self._loop:
             return False

src/cccc/ports/im/adapters/feishu.py

@@ -610,7 +610,14 @@ def _get_chat_title_cached(self, chat_id: str) -> str:
         self._chat_title_cache[chat_id] = title
         return title
 
-    def send_message(self, chat_id: str, text: str, thread_id: Optional[int] = None) -> bool:
+    def send_message(
+        self,
+        chat_id: str,
+        text: str,
+        thread_id: Optional[int] = None,
+        *,
+        mention_user_ids: Optional[List[str]] = None,
+    ) -> bool:
         """
         Send a text message to a chat.
 
@@ -619,6 +626,7 @@ def send_message(self, chat_id: str, text: str, thread_id: Optional[int] = None)
             text: Message text
             thread_id: Optional root_id for threading
         """
+        _ = mention_user_ids
         if not text:
             return True
 

src/cccc/ports/im/adapters/slack.py

@@ -245,12 +245,20 @@ def poll(self) -> List[Dict[str, Any]]:
 
         return messages
 
-    def send_message(self, chat_id: str, text: str, thread_id: Optional[int] = None) -> bool:
+    def send_message(
+        self,
+        chat_id: str,
+        text: str,
+        thread_id: Optional[int] = None,
+        *,
+        mention_user_ids: Optional[List[str]] = None,
+    ) -> bool:
         """
         Send a message to a Slack channel.
 
         chat_id is actually a channel ID string in Slack.
         """
+        _ = mention_user_ids
         _ = thread_id  # Slack threads not wired yet (future work).
         if not self._connected or not self._web_client:
             return False

src/cccc/ports/im/adapters/telegram.py

@@ -365,7 +365,14 @@ def send_file(
             self._log(f"[send_file] failed: {e}")
             return False
 
-    def send_message(self, chat_id: str, text: str, thread_id: Optional[int] = None) -> bool:
+    def send_message(
+        self,
+        chat_id: str,
+        text: str,
+        thread_id: Optional[int] = None,
+        *,
+        mention_user_ids: Optional[List[str]] = None,
+    ) -> bool:
         """
         Send a message to a chat.
 
@@ -374,6 +381,7 @@ def send_message(self, chat_id: str, text: str, thread_id: Optional[int] = None)
         - Message length limits
         - Retry on failure
         """
+        _ = mention_user_ids
         if not text:
             return True
 

src/cccc/ports/im/bridge.py

@@ -293,6 +293,10 @@ def __init__(
         # Inbound health monitoring: periodic log every 5 minutes
         self._last_health_log: float = 0.0
         self._inbound_count: int = 0  # messages processed since last health log
+        # Per chat/thread mention targets for outbound replies.
+        # The bridge computes this from inbound metadata and passes it explicitly
+        # to adapters instead of letting adapters guess from their own caches.
+        self._mention_targets: Dict[str, List[str]] = {}
 
     def _should_process_inbound(self, *, chat_id: str, thread_id: int, message_id: str) -> bool:
         """
@@ -432,6 +436,7 @@ def _process_inbound(self) -> None:
             if not self._should_process_inbound(chat_id=chat_id, thread_id=thread_id, message_id=message_id):
                 continue
             self._inbound_count += 1
+            self._remember_mention_targets(chat_id, thread_id, msg)
 
             # Authorization check: unauthorized chats may only /subscribe.
             self._log(f"[inbound] Checking auth for chat_id={chat_id} thread={thread_id}")
@@ -581,6 +586,53 @@ def _display_actor_token(self, token: str, actor_labels: Dict[str, str]) -> str:
     def _stream_target_key(chat_id: str, thread_id: int) -> str:
         return f"{chat_id}:{int(thread_id or 0)}"
 
+    def _remember_mention_targets(self, chat_id: str, thread_id: int, msg: Dict[str, Any]) -> None:
+        """Cache the latest explicit mention target for this chat/thread."""
+        target_key = self._stream_target_key(chat_id, thread_id)
+        raw_ids = msg.get("mention_user_ids")
+        mention_user_ids = (
+            [str(x).strip() for x in raw_ids if str(x).strip()]
+            if isinstance(raw_ids, list)
+            else []
+        )
+        if mention_user_ids:
+            self._mention_targets[target_key] = mention_user_ids
+        else:
+            self._mention_targets.pop(target_key, None)
+
+        # Keep the cache bounded without introducing another dependency.
+        while len(self._mention_targets) > 256:
+            oldest_key = next(iter(self._mention_targets))
+            self._mention_targets.pop(oldest_key, None)
+
+    def _resolve_outbound_mention_targets(
+        self,
+        *,
+        event: Dict[str, Any],
+        sub: Any,
+        is_user_facing: bool,
+    ) -> Optional[List[str]]:
+        """Resolve explicit mention targets for an outbound IM message."""
+        if not is_user_facing:
+            return None
+
+        platform = str(getattr(self.adapter, "platform", "") or "").strip().lower()
+        if platform != "dingtalk":
+            return None
+
+        data = event.get("data", {})
+        if isinstance(data, dict):
+            raw_ids = data.get("mention_user_ids")
+            if isinstance(raw_ids, list):
+                cleaned = [str(x).strip() for x in raw_ids if str(x).strip()]
+                return cleaned
+
+        target_key = self._stream_target_key(sub.chat_id, sub.thread_id)
+        cached = self._mention_targets.get(target_key)
+        if cached is None:
+            return None
+        return list(cached)
+
     def _forward_stream_event(self, event: Dict[str, Any]) -> None:
         """Forward a chat.stream event to subscribed chats via adapter streaming methods.
 
@@ -780,7 +832,22 @@ def _forward_event(self, event: Dict[str, Any], *, actor_labels: Optional[Dict[s
 
             # If we didn't send any files, or if there's text with no files, send message.
             if formatted and not sent_any_file and not skip_text_due_to_stream:
-                sent_msg = bool(self.adapter.send_message(sub.chat_id, formatted, thread_id=sub.thread_id))
+                mention_user_ids = self._resolve_outbound_mention_targets(
+                    event=event,
+                    sub=sub,
+                    is_user_facing=is_user_facing,
+                )
+                if mention_user_ids is None:
+                    sent_msg = bool(self.adapter.send_message(sub.chat_id, formatted, thread_id=sub.thread_id))
+                else:
+                    sent_msg = bool(
+                        self.adapter.send_message(
+                            sub.chat_id,
+                            formatted,
+                            thread_id=sub.thread_id,
+                            mention_user_ids=mention_user_ids,
+                        )
+                    )
                 if sent_msg and is_user_facing:
                     delivered_user_facing = True