fix(runtime): deliver away notifications to IM

backend/app/api/ws/device_namespace.py

@@ -45,6 +45,11 @@
 from app.api.ws.local_task_responses import (
     LocalTaskResponsesHandler,
     emit_response_api_event,
+    is_runtime_terminal_event_type,
+    is_terminal_event,
+    local_task_terminal_status,
+    runtime_subtask_id,
+    runtime_terminal_event,
 )
 from app.api.ws.wework_runtime_namespace import (
     PROJECT_CHAT_AGENT_CHUNK_EVENT,
@@ -2209,8 +2214,120 @@ async def _forward_runtime_event(
             device_id=device_id,
             payload=payload["payload"],
         )
+        await self._notify_runtime_event(
+            user_id=user_id,
+            device_id=device_id,
+            payload=payload["payload"],
+        )
         return {"success": True}
 
+    async def _notify_runtime_event(
+        self,
+        *,
+        user_id: int,
+        device_id: str,
+        payload: dict[str, Any],
+    ) -> None:
+        """Best-effort IM delivery for one terminal native Runtime event."""
+
+        source = payload.get("source")
+        source_name = source.get("source") if isinstance(source, dict) else None
+        if source_name == "im":
+            return
+
+        local_task_id = str(payload.get("taskId") or "").strip()
+        event_type = str(payload.get("event_type") or "").strip()
+        if (
+            not local_task_id
+            or not event_type
+            or not is_runtime_terminal_event_type(event_type)
+        ):
+            return
+
+        event_data = payload.get("data")
+        event_data = event_data if isinstance(event_data, dict) else {}
+        subtask_id = runtime_subtask_id(payload, device_id, local_task_id)
+        event = runtime_terminal_event(
+            event_type=event_type,
+            event_data=event_data,
+            subtask_id=subtask_id,
+        ) or self._local_task_responses.execution_event(
+            event_type=event_type,
+            event_data=event_data,
+            subtask_id=subtask_id,
+            message_id=None,
+        )
+        if event is None or not is_terminal_event(event):
+            return
+
+        status = local_task_terminal_status(event)
+        result = event.result if isinstance(event.result, dict) else {}
+        content = str(result.get("value") or event.error or "")
+        if status == "COMPLETED" and not content.strip():
+            logger.info(
+                "[RuntimeTaskNotification] Skipped empty Runtime reply: "
+                "user_id=%s device_id=%s local_task_id=%s event_type=%s",
+                user_id,
+                device_id,
+                local_task_id,
+                event_type,
+            )
+            return
+
+        title = str(
+            payload.get("taskTitle")
+            or payload.get("task_title")
+            or event_data.get("title")
+            or local_task_id
+        ).strip()
+        try:
+            notification = (
+                await im_notification_dispatcher.send_runtime_task_update_for_user(
+                    user_id=user_id,
+                    address={
+                        "deviceId": device_id,
+                        "localTaskId": local_task_id,
+                    },
+                    title=title or local_task_id,
+                    status=status,
+                    content=content,
+                    source=str(source_name) if source_name else None,
+                )
+            )
+        except Exception:
+            logger.exception(
+                "[RuntimeTaskNotification] Runtime event IM delivery failed: "
+                "user_id=%s device_id=%s local_task_id=%s event_type=%s",
+                user_id,
+                device_id,
+                local_task_id,
+                event_type,
+            )
+            return
+
+        notified = int(notification.get("sent") or 0)
+        results = _summarize_runtime_notification_results(notification)
+        if results and notified <= 0:
+            logger.warning(
+                "[RuntimeTaskNotification] Runtime event was not delivered: "
+                "user_id=%s device_id=%s local_task_id=%s event_type=%s results=%s",
+                user_id,
+                device_id,
+                local_task_id,
+                event_type,
+                results,
+            )
+        else:
+            logger.info(
+                "[RuntimeTaskNotification] Runtime event dispatched: "
+                "user_id=%s device_id=%s local_task_id=%s event_type=%s sent=%s",
+                user_id,
+                device_id,
+                local_task_id,
+                event_type,
+                notified,
+            )
+
     async def _publish_task_completed_event(
         self,
         task_id: int,

backend/app/api/ws/local_task_responses.py

@@ -30,9 +30,16 @@
     ResponsesAPIStreamEvents.RESPONSE_COMPLETED.value: EventType.DONE,
     ResponsesAPIStreamEvents.RESPONSE_FAILED.value: EventType.ERROR,
     ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE.value: EventType.CANCELLED,
+    ResponsesAPIStreamEvents.ERROR.value: EventType.ERROR,
 }
 
 
+def is_runtime_terminal_event_type(event_type: str) -> bool:
+    """Return whether a native Runtime event can settle a task turn."""
+
+    return event_type in RUNTIME_TERMINAL_EVENT_TYPES
+
+
 class LocalTaskResponsesHandler:
     """Translate local-task Responses API events into chat/channel events."""
 
@@ -431,13 +438,21 @@ def runtime_terminal_event(
     internal_type = RUNTIME_TERMINAL_EVENT_TYPES.get(event_type)
     if internal_type is None:
         return None
+    if internal_type is EventType.DONE and "value" not in event_data:
+        return None
+    if event_type == ResponsesAPIStreamEvents.ERROR.value:
+        return None
     error = event_data.get("error")
     if internal_type is EventType.DONE:
         return ExecutionEvent(
             type=internal_type.value,
             subtask_id=subtask_id,
             result={"value": event_data.get("value") or ""},
         )
+    if not isinstance(error, dict):
+        response = event_data.get("response")
+        if isinstance(response, dict):
+            error = response.get("error")
     return ExecutionEvent(
         type=internal_type.value,
         subtask_id=subtask_id,

backend/app/models/im_session.py

@@ -35,6 +35,7 @@ class IMPrivateSession:
     channel_id: int
     conversation_id: str
     sender_id: str
+    proactive_recipient_id: str = ""
     display_name: str = ""
     mode: str = IMSessionMode.CHAT
     state: str = IMSessionState.IDLE

backend/app/services/channels/dingtalk/handler.py

@@ -174,6 +174,7 @@ def parse_message(self, raw_data: Any) -> MessageContext:
             is_mention=is_in_at_list,
             raw_message=message,
             extra_data=extra_data,
+            proactive_recipient_id=str(sender_staff_id or "").strip() or None,
             images=images,
             files=files,
         )

backend/app/services/channels/handler.py

@@ -97,6 +97,7 @@ class MessageContext:
     is_mention: bool  # Whether the bot was mentioned (for group chats)
     raw_message: Any  # Original message object from the channel SDK
     extra_data: Dict[str, Any]  # Channel-specific extra data
+    proactive_recipient_id: Optional[str] = None
     images: List[Dict[str, str]] = field(default_factory=list)
     # Each image dict: {"mime_type": "image/png", "base64_data": "iVBOR..."}
     files: List[Dict[str, Any]] = field(default_factory=list)
@@ -733,6 +734,7 @@ async def handle_message(self, raw_data: Any) -> bool:
                     channel_id=self._channel_id,
                     conversation_id=message_context.conversation_id,
                     sender_id=message_context.sender_id,
+                    proactive_recipient_id=message_context.proactive_recipient_id,
                     display_name=message_context.sender_name or "",
                 )
 

backend/app/services/im/notification_dispatcher.py

@@ -14,6 +14,9 @@
 from app.models.im_session import IMPrivateSession
 from app.models.kind import Kind
 from app.services.im.session_service import im_session_service
+from app.services.subscription.notification_service import (
+    subscription_notification_service,
+)
 from shared.utils.crypto import decrypt_sensitive_data
 
 logger = logging.getLogger(__name__)
@@ -134,7 +137,7 @@ async def send_text(
 
             config = _get_channel_config(channel)
             if session.channel_type == "dingtalk":
-                return await self._send_dingtalk(session, config, text)
+                return await self._send_dingtalk(db, session, config, text)
             if session.channel_type == "telegram":
                 return await self._send_telegram(session, config, text)
             if session.channel_type == "discord":
@@ -167,6 +170,10 @@ async def _runtime_notification_sessions(
         user_id: int,
         address: dict[str, Any],
     ) -> list[IMPrivateSession]:
+        settings = await im_session_service.get_global_notification_settings(user_id)
+        if not settings.enabled:
+            return []
+
         active_sessions = await im_session_service.list_active_runtime_task_sessions(
             db,
             user_id=user_id,
@@ -185,8 +192,7 @@ async def _runtime_notification_sessions(
         if subscribed_sessions:
             return _dedupe_sessions(subscribed_sessions)
 
-        settings = await im_session_service.get_global_notification_settings(user_id)
-        if not settings.enabled or not settings.session_key:
+        if not settings.session_key:
             return []
         if not await im_session_service.is_user_away_for_im_notifications(user_id):
             return []
@@ -233,6 +239,7 @@ def _get_channel(self, db: Session, channel_id: int) -> Kind | None:
 
     async def _send_dingtalk(
         self,
+        db: Session,
         session: IMPrivateSession,
         config: dict[str, Any],
         text: str,
@@ -249,9 +256,18 @@ async def _send_dingtalk(
                 "error": "Missing DingTalk credentials",
             }
 
+        recipient_id = await self._dingtalk_recipient_id(db, session)
+        if not recipient_id:
+            return {
+                "success": False,
+                "channel_id": session.channel_id,
+                "channel_type": session.channel_type,
+                "error": "Missing DingTalk staff ID",
+            }
+
         sender = DingTalkRobotSender(client_id, client_secret)
         result = await sender.send_text_message(
-            user_ids=[session.sender_id],
+            user_ids=[recipient_id],
             content=text,
         )
         return {
@@ -260,6 +276,36 @@ async def _send_dingtalk(
             **result,
         }
 
+    async def _dingtalk_recipient_id(
+        self,
+        db: Session,
+        session: IMPrivateSession,
+    ) -> str:
+        recipient_id = str(session.proactive_recipient_id or "").strip()
+        if recipient_id:
+            return recipient_id
+
+        bindings = subscription_notification_service.get_user_im_bindings(
+            db,
+            user_id=session.user_id,
+        )
+        binding = bindings.get(str(session.channel_id))
+        if (
+            binding is None
+            or binding.channel_type != session.channel_type
+            or str(binding.last_conversation_id or "").strip()
+            != session.conversation_id
+        ):
+            return ""
+
+        recipient_id = str(binding.sender_staff_id or "").strip()
+        if not recipient_id:
+            return ""
+
+        session.proactive_recipient_id = recipient_id
+        await im_session_service.save_session(session)
+        return recipient_id
+
     async def _send_telegram(
         self,
         session: IMPrivateSession,
@@ -396,15 +442,7 @@ def _runtime_task_update_message(
         return f"任务「{task_title}」已取消。"
 
     body = content or "任务有新的更新，请打开 Wework 查看完整对话。"
-    return "\n".join(
-        [
-            f"任务「{task_title}」有新的 AI 回复：",
-            "",
-            body,
-            "",
-            "回复这条通知可继续该任务。",
-        ]
-    )
+    return f"任务「{task_title}」有新的 AI 回复：\n\n{body}"
 
 
 im_notification_dispatcher = IMNotificationDispatcher()

backend/app/services/im/session_service.py

@@ -64,6 +64,7 @@ async def get_or_create_private_session(
         channel_id: int,
         conversation_id: str,
         sender_id: str,
+        proactive_recipient_id: str | None = None,
         display_name: str = "",
     ) -> IMPrivateSession:
         now = datetime.now()
@@ -82,13 +83,16 @@ async def get_or_create_private_session(
                 channel_id=channel_id,
                 conversation_id=conversation_id,
                 sender_id=sender_id,
+                proactive_recipient_id=proactive_recipient_id or "",
                 display_name=display_name,
                 last_seen_at=now,
                 created_at=now,
                 updated_at=now,
             )
         else:
             session.sender_id = sender_id
+            if proactive_recipient_id:
+                session.proactive_recipient_id = proactive_recipient_id
             session.display_name = display_name
             session.last_seen_at = now
             session.updated_at = now