diff --git a/backend/app/api/ws/device_namespace.py b/backend/app/api/ws/device_namespace.py index bf4ffb9b86..26b4c9e228 100644 --- a/backend/app/api/ws/device_namespace.py +++ b/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, diff --git a/backend/app/api/ws/local_task_responses.py b/backend/app/api/ws/local_task_responses.py index adb37cc25e..163fa7529e 100644 --- a/backend/app/api/ws/local_task_responses.py +++ b/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,6 +438,10 @@ 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( @@ -438,6 +449,10 @@ def runtime_terminal_event( 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, diff --git a/backend/app/models/im_session.py b/backend/app/models/im_session.py index cd17ba2216..a028e96ec5 100644 --- a/backend/app/models/im_session.py +++ b/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 diff --git a/backend/app/services/channels/dingtalk/handler.py b/backend/app/services/channels/dingtalk/handler.py index 3ad55e9cd1..f4259a26a0 100644 --- a/backend/app/services/channels/dingtalk/handler.py +++ b/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, ) diff --git a/backend/app/services/channels/handler.py b/backend/app/services/channels/handler.py index 693a2a94b5..9d8d601022 100644 --- a/backend/app/services/channels/handler.py +++ b/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 "", ) diff --git a/backend/app/services/im/notification_dispatcher.py b/backend/app/services/im/notification_dispatcher.py index 7cf202dae1..b0adf343ab 100644 --- a/backend/app/services/im/notification_dispatcher.py +++ b/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() diff --git a/backend/app/services/im/session_service.py b/backend/app/services/im/session_service.py index 0276f6699c..e37fd606e9 100644 --- a/backend/app/services/im/session_service.py +++ b/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,6 +83,7 @@ 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, @@ -89,6 +91,8 @@ async def get_or_create_private_session( ) 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 diff --git a/backend/tests/api/ws/test_wework_runtime_namespace.py b/backend/tests/api/ws/test_wework_runtime_namespace.py index e626fd4802..0377e050c3 100644 --- a/backend/tests/api/ws/test_wework_runtime_namespace.py +++ b/backend/tests/api/ws/test_wework_runtime_namespace.py @@ -17,6 +17,19 @@ from app.core.socketio import SOCKETIO_MAX_HTTP_BUFFER_SIZE +@pytest.fixture(autouse=True) +def runtime_notification_sender(monkeypatch): + """Keep namespace tests isolated from notification DB and provider I/O.""" + + sender = AsyncMock(return_value={"sent": 0, "results": []}) + monkeypatch.setattr( + device_namespace.im_notification_dispatcher, + "send_runtime_task_update_for_user", + sender, + ) + return sender + + def _im_source() -> dict: return { "source": "im", @@ -137,6 +150,7 @@ async def test_runtime_event_forwards_im_progress_events_to_channel_callbacks( @pytest.mark.asyncio async def test_runtime_event_completes_im_channel_callback_on_terminal_event( monkeypatch, + runtime_notification_sender, ): """The native runtime reports its answer as ``data.value``, not a response body.""" @@ -162,10 +176,14 @@ async def test_runtime_event_completes_im_channel_callback_on_terminal_event( assert kwargs["task_id"] == "runtime:local-device:runtime-375023196" assert kwargs["status"] == "COMPLETED" assert kwargs["result"] == {"value": "Hi! What would you like to work on?"} + runtime_notification_sender.assert_not_awaited() @pytest.mark.asyncio -async def test_runtime_event_fails_im_channel_callback_on_failed_event(monkeypatch): +async def test_runtime_event_fails_im_channel_callback_on_failed_event( + monkeypatch, + runtime_notification_sender, +): """``response.failed`` must release the card instead of leaving it pending.""" namespace = DeviceNamespace() @@ -188,6 +206,7 @@ async def test_runtime_event_fails_im_channel_callback_on_failed_event(monkeypat kwargs = registry.handle_task_completed.await_args.kwargs assert kwargs["status"] == "FAILED" assert kwargs["error"] == "model request timed out" + runtime_notification_sender.assert_not_awaited() @pytest.mark.asyncio @@ -216,6 +235,157 @@ async def test_runtime_event_skips_channel_callbacks_without_im_source(monkeypat registry.handle_task_completed.assert_not_awaited() +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("event_type", "event_data", "expected_status", "expected_content"), + [ + ( + "response.completed", + {"value": "Codex finished the task"}, + "COMPLETED", + "Codex finished the task", + ), + ( + "response.completed", + { + "response": { + "output": [ + { + "content": [ + { + "type": "output_text", + "text": "Claude finished the task", + } + ] + } + ] + } + }, + "COMPLETED", + "Claude finished the task", + ), + ( + "response.failed", + {"error": {"message": "model request timed out"}}, + "FAILED", + "model request timed out", + ), + ( + "response.incomplete", + {"error": {"message": "turn cancelled"}}, + "CANCELLED", + "turn cancelled", + ), + ( + "error", + {"message": "Claude transport failed"}, + "FAILED", + "Claude transport failed", + ), + ], +) +async def test_runtime_terminal_event_notifies_im_dispatcher( + monkeypatch, + runtime_notification_sender, + event_type, + event_data, + expected_status, + expected_content, +): + namespace = DeviceNamespace() + + result = await _relay_runtime_event( + namespace, + monkeypatch, + { + "event_type": event_type, + "taskId": "runtime-375023196", + "taskTitle": "分析线上问题", + "data": event_data, + }, + ) + + assert result == {"success": True} + runtime_notification_sender.assert_awaited_once_with( + user_id=7, + address={ + "deviceId": "local-device", + "localTaskId": "runtime-375023196", + }, + title="分析线上问题", + status=expected_status, + content=expected_content, + source=None, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("event_type", "event_data"), + [ + ("response.output_text.delta", {"delta": "still running"}), + ("response.completed", {"value": ""}), + ("response.completed", {"response": {"output": []}}), + ], +) +async def test_runtime_event_skips_non_terminal_or_empty_success_notifications( + monkeypatch, + runtime_notification_sender, + event_type, + event_data, +): + namespace = DeviceNamespace() + + result = await _relay_runtime_event( + namespace, + monkeypatch, + { + "event_type": event_type, + "taskId": "runtime-375023196", + "data": event_data, + }, + ) + + assert result == {"success": True} + runtime_notification_sender.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_runtime_notification_failure_does_not_break_wework_relay( + monkeypatch, + runtime_notification_sender, +): + namespace = DeviceNamespace() + sio = AsyncMock() + runtime_notification_sender.side_effect = RuntimeError("DingTalk unavailable") + monkeypatch.setattr( + namespace, + "get_session", + AsyncMock(return_value={"user_id": 7, "device_id": "local-device"}), + ) + monkeypatch.setattr( + device_namespace, + "run_sync_in_executor", + AsyncMock(return_value=None), + ) + monkeypatch.setattr(device_namespace, "get_sio", lambda: sio) + + result = await namespace.on_runtime_event( + "device-sid", + { + "event": "response.completed", + "payload": { + "taskId": "runtime-375023196", + "data": {"value": "Task finished"}, + }, + }, + ) + + assert result == {"success": True} + sio.emit.assert_awaited_once() + assert sio.emit.await_args.args[0] == "runtime:event" + + @pytest.mark.asyncio async def test_device_runtime_event_persists_project_chat_before_browser_relay( monkeypatch, diff --git a/backend/tests/services/channels/dingtalk/test_handler.py b/backend/tests/services/channels/dingtalk/test_handler.py index 52895f19b5..dbc0aa101d 100644 --- a/backend/tests/services/channels/dingtalk/test_handler.py +++ b/backend/tests/services/channels/dingtalk/test_handler.py @@ -29,3 +29,4 @@ def test_parse_message_preserves_dingtalk_message_id() -> None: assert context.content == "continue runtime" assert context.extra_data["message_id"] == "dingtalk-message-1" assert context.extra_data["callback_data"]["msgId"] == "dingtalk-message-1" + assert context.proactive_recipient_id == "staff-a" diff --git a/backend/tests/services/channels/test_private_im_session_integration.py b/backend/tests/services/channels/test_private_im_session_integration.py index 8af3ea89a8..3046abd318 100644 --- a/backend/tests/services/channels/test_private_im_session_integration.py +++ b/backend/tests/services/channels/test_private_im_session_integration.py @@ -35,6 +35,7 @@ def parse_message(self, raw_data: dict[str, Any]) -> MessageContext: is_mention=raw_data.get("is_mention", False), raw_message=raw_data, extra_data=raw_data.get("extra_data", {}), + proactive_recipient_id=raw_data.get("proactive_recipient_id"), images=raw_data.get("images", []), files=raw_data.get("files", []), ) @@ -97,6 +98,7 @@ def _message( *, conversation_type: str = "private", conversation_id: str = "conv-private", + proactive_recipient_id: str | None = None, extra_data: dict[str, Any] | None = None, images: list[dict[str, str]] | None = None, files: list[dict[str, Any]] | None = None, @@ -107,6 +109,7 @@ def _message( "conversation_id": conversation_id, "sender_id": "staff-a", "sender_name": "Alice", + "proactive_recipient_id": proactive_recipient_id, "extra_data": extra_data or {}, "images": images or [], "files": files or [], @@ -196,13 +199,16 @@ async def test_private_bind_creates_session_and_replies_bound( ) -> None: handler = FakeChannelHandler(test_user) - handled = await handler.handle_message(_message("/bind")) + handled = await handler.handle_message( + _message("/bind", proactive_recipient_id="staff-delivery-a") + ) test_db.expire_all() session = await _private_session(test_db, test_user) assert handled is True assert session is not None assert session.sender_id == "staff-a" + assert session.proactive_recipient_id == "staff-delivery-a" assert session.display_name == "Alice" assert any("已绑定" in reply for reply in handler.replies) diff --git a/backend/tests/services/im/test_notification_dispatcher.py b/backend/tests/services/im/test_notification_dispatcher.py index c9591e04a3..0562a4a79f 100644 --- a/backend/tests/services/im/test_notification_dispatcher.py +++ b/backend/tests/services/im/test_notification_dispatcher.py @@ -4,6 +4,7 @@ from datetime import datetime from typing import Any +from unittest.mock import AsyncMock import pytest from sqlalchemy.orm import Session @@ -12,9 +13,19 @@ from app.models.kind import Kind from app.services.im.notification_dispatcher import im_notification_dispatcher 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 encrypt_sensitive_data +@pytest.fixture(autouse=True) +def isolate_im_session_cache(fake_im_session_cache: Any) -> Any: + """Keep dispatcher tests from mutating the developer's Redis state.""" + + return fake_im_session_cache + + def _create_channel( db: Session, *, @@ -50,6 +61,7 @@ def _create_session( channel_id: int, channel_type: str, sender_id: str, + proactive_recipient_id: str = "", ) -> IMPrivateSession: return IMPrivateSession( session_key=im_session_service.build_session_key( @@ -63,13 +75,14 @@ def _create_session( channel_id=channel_id, conversation_id=f"conv-{channel_id}", sender_id=sender_id, + proactive_recipient_id=proactive_recipient_id, display_name=f"sender-{sender_id}", last_seen_at=datetime.now(), ) @pytest.mark.asyncio -async def test_dingtalk_notification_decrypts_channel_secret( +async def test_dingtalk_notification_uses_private_session_staff_id( test_db: Session, test_user, monkeypatch: pytest.MonkeyPatch, @@ -87,7 +100,8 @@ async def test_dingtalk_notification_decrypts_channel_secret( user_id=test_user.id, channel_id=9401, channel_type="dingtalk", - sender_id="staff-1", + sender_id="sender-union-1", + proactive_recipient_id="staff-1", ) test_db.commit() calls: list[dict[str, Any]] = [] @@ -119,6 +133,83 @@ async def send_text_message(self, user_ids: list[str], content: str): assert calls[1] == {"user_ids": ["staff-1"], "content": "已切换"} +@pytest.mark.asyncio +async def test_dingtalk_legacy_binding_backfill_requires_matching_conversation( + test_db: Session, + test_user, + monkeypatch: pytest.MonkeyPatch, +) -> None: + _create_channel( + test_db, + channel_id=9403, + channel_type="dingtalk", + config={ + "client_id": "ding-client-id", + "client_secret": encrypt_sensitive_data("ding-client-secret"), + }, + ) + session = _create_session( + user_id=test_user.id, + channel_id=9403, + channel_type="dingtalk", + sender_id="sender-union-1", + ) + subscription_notification_service.update_user_im_binding( + test_db, + user_id=test_user.id, + channel_id=9403, + channel_type="dingtalk", + sender_id="sender-union-2", + sender_staff_id="staff-2", + conversation_id="another-conversation", + ) + test_db.commit() + calls: list[list[str]] = [] + + class FakeDingTalkRobotSender: + def __init__(self, client_id: str, client_secret: str): + pass + + async def send_text_message(self, user_ids: list[str], content: str): + calls.append(user_ids) + return {"success": True} + + monkeypatch.setattr( + "app.services.channels.dingtalk.sender.DingTalkRobotSender", + FakeDingTalkRobotSender, + ) + + mismatched = await im_notification_dispatcher.send_text( + test_db, + session, + "不会串发", + ) + + assert mismatched["success"] is False + assert mismatched["error"] == "Missing DingTalk staff ID" + assert calls == [] + + subscription_notification_service.update_user_im_binding( + test_db, + user_id=test_user.id, + channel_id=9403, + channel_type="dingtalk", + sender_id="sender-union-1", + sender_staff_id="staff-1", + conversation_id=session.conversation_id, + ) + + matched = await im_notification_dispatcher.send_text( + test_db, + session, + "安全回填", + ) + + assert matched["success"] is True + assert calls == [["staff-1"]] + assert session.proactive_recipient_id == "staff-1" + + @pytest.mark.asyncio async def test_telegram_notification_decrypts_bot_token( test_db: Session, @@ -229,8 +320,9 @@ async def send_text_message(self, chat_id: int, text: str): assert result["sent"] == 1 assert calls[1]["chat_id"] == 100200300 - assert "Native Codex task" in calls[1]["text"] - assert "Implemented from native Codex" in calls[1]["text"] + assert calls[1]["text"] == ( + "任务「Native Codex task」有新的 AI 回复:\n\n" "Implemented from native Codex" + ) @pytest.mark.asyncio @@ -269,6 +361,104 @@ async def test_runtime_task_update_suppresses_global_target_while_client_is_acti assert result["results"] == [] +@pytest.mark.asyncio +@pytest.mark.parametrize("target_kind", ["active", "subscribed"]) +async def test_runtime_task_update_master_switch_suppresses_session_targets( + test_db: Session, + test_user, + fake_im_session_cache, + monkeypatch: pytest.MonkeyPatch, + target_kind: str, +) -> None: + address = { + "deviceId": "device-1", + "localTaskId": "codex-thread-1", + } + session = _create_session( + user_id=test_user.id, + channel_id=9413, + channel_type="dingtalk", + sender_id="sender-union-1", + proactive_recipient_id="staff-1", + ) + await im_session_service.save_session(session) + if target_kind == "active": + await im_session_service.bind_active_runtime_task( + test_db, + session=session, + runtime_task=address, + ) + else: + await im_session_service.subscribe_runtime_task_notification( + test_db, + session=session, + runtime_task=address, + ) + + send_text = AsyncMock(return_value={"success": True}) + monkeypatch.setattr(im_notification_dispatcher, "send_text", send_text) + + result = await im_notification_dispatcher.send_runtime_task_update( + test_db, + user_id=test_user.id, + address=address, + title="Native Codex task", + status="updated", + content="Suppressed update", + source="codex_watcher", + ) + + assert result == {"sent": 0, "results": []} + send_text.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_runtime_task_update_uses_active_session_when_master_switch_enabled( + test_db: Session, + test_user, + fake_im_session_cache, + monkeypatch: pytest.MonkeyPatch, +) -> None: + address = { + "deviceId": "device-1", + "localTaskId": "codex-thread-1", + } + session = _create_session( + user_id=test_user.id, + channel_id=9413, + channel_type="dingtalk", + sender_id="sender-union-1", + proactive_recipient_id="staff-1", + ) + await im_session_service.save_session(session) + await im_session_service.bind_active_runtime_task( + test_db, + session=session, + runtime_task=address, + ) + await im_session_service.enable_global_notification(test_db, session=session) + await im_session_service.update_im_notification_presence( + user_id=test_user.id, + client_id="wework-client", + away=False, + ) + send_text = AsyncMock(return_value={"success": True}) + monkeypatch.setattr(im_notification_dispatcher, "send_text", send_text) + + result = await im_notification_dispatcher.send_runtime_task_update( + test_db, + user_id=test_user.id, + address=address, + title="Native Codex task", + status="updated", + content="Bound update", + source="codex_watcher", + ) + + assert result["sent"] == 1 + send_text.assert_awaited_once() + + @pytest.mark.asyncio async def test_runtime_task_update_uses_subscribed_native_codex_task( test_db: Session, @@ -293,6 +483,7 @@ async def test_runtime_task_update_uses_subscribed_native_codex_task( "workspacePath": "/repo/Wegent", } await im_session_service.save_session(session) + await im_session_service.enable_global_notification(test_db, session=session) await im_session_service.update_im_notification_presence( user_id=test_user.id, client_id="wework-client", diff --git a/backend/tests/services/im/test_session_service.py b/backend/tests/services/im/test_session_service.py index bfadf07683..745408b6ea 100644 --- a/backend/tests/services/im/test_session_service.py +++ b/backend/tests/services/im/test_session_service.py @@ -23,6 +23,7 @@ async def test_get_or_create_private_session_uses_stable_redis_key( channel_id=33, conversation_id="chat-9", sender_id="1001", + proactive_recipient_id="staff-original", display_name="Alice", ) @@ -38,6 +39,7 @@ async def test_get_or_create_private_session_uses_stable_redis_key( assert refreshed.session_key == first.session_key assert refreshed.display_name == "Alice New" + assert refreshed.proactive_recipient_id == "staff-original" assert refreshed.mode == IMSessionMode.CHAT assert refreshed.state == IMSessionState.IDLE assert refreshed.active_task_id is None diff --git a/docs/en/wegent/developer-guide/runtime-local-work.md b/docs/en/wegent/developer-guide/runtime-local-work.md index 231b8c19c3..5c2607a318 100644 --- a/docs/en/wegent/developer-guide/runtime-local-work.md +++ b/docs/en/wegent/developer-guide/runtime-local-work.md @@ -311,8 +311,8 @@ Runtime tasks can send notifications to IM sessions, but notification state is k - In IM, `/notify on` enables the current user's global runtime task notification target for the current IM session. - `/notify off` disables global notifications, and `/notify status` reports the current state. - A single IM session can subscribe to one runtime task and receive only that task's updates. -- When the executor detects a native Codex task timestamp change, it sends `runtime.tasks.updated` without `workspacePath`, but with `status` and `content`, only after the last assistant message reaches a terminal state and contains reply content. Backend ignores running or streaming updates and delivers the terminal reply according to task subscriptions and global notification settings. -- Wegent runtime sends and the native Codex watcher use the same `deviceId + localTaskId` for deduplication, so Codex and Wework do not notify twice for the same task update. +- The executor sends `runtime:event` over the device WebSocket. Backend delivers only `response.completed`, `response.failed`, `response.incomplete`, and `error` terminal updates according to active task bindings, task subscriptions, and global notification settings. Successful terminal events without reply content are not sent. +- Notification identity uses `deviceId + localTaskId` and excludes `workspacePath`. IM-originated turns do not echo another notification to the same IM, while the legacy `runtime.tasks.updated` receiver remains only for older executors. ## URL diff --git a/docs/zh/wegent/developer-guide/runtime-local-work.md b/docs/zh/wegent/developer-guide/runtime-local-work.md index e875a8be77..394d444153 100644 --- a/docs/zh/wegent/developer-guide/runtime-local-work.md +++ b/docs/zh/wegent/developer-guide/runtime-local-work.md @@ -309,8 +309,8 @@ Wework 不再显示“未映射工作区”。executor 返回的线程必须能 - 在 IM 中使用 `/notify on`、`/通知 开` 开启当前用户的全局运行时任务通知目标。 - 使用 `/notify off` 关闭全局通知,使用 `/notify status` 查看当前状态。 - 单个 IM 会话订阅某个运行时任务后,只接收该任务的更新。 -- executor 发现原生 Codex 任务更新时间变化时,只在最后一条 assistant 消息进入终态且有回复内容后,通过设备 WebSocket 发送不含 `workspacePath`、但包含 `status` 和 `content` 的 `runtime.tasks.updated`。Backend 会忽略运行中/流式更新,并按订阅和全局通知设置把终态回复投递到 IM。 -- Wegent 发起的 runtime send 与原生 Codex watcher 使用同一个 `deviceId + localTaskId` 去重,避免 Codex 和 Wework 对同一次任务更新重复通知。 +- executor 通过设备 WebSocket 发送 `runtime:event`;Backend 只把 `response.completed`、`response.failed`、`response.incomplete` 和 `error` 终态按任务绑定、任务订阅和全局通知设置投递到 IM。成功终态没有回复正文时不发送。 +- 通知身份使用 `deviceId + localTaskId`,不包含 `workspacePath`;IM 发起的回合不会再向同一 IM 回声通知。旧 executor 的 `runtime.tasks.updated` 接收端仅作为兼容路径保留。 ## URL diff --git a/executor/src/runtime_work/events.rs b/executor/src/runtime_work/events.rs index 256a1cf032..7f92a47ea9 100644 --- a/executor/src/runtime_work/events.rs +++ b/executor/src/runtime_work/events.rs @@ -31,7 +31,7 @@ use super::{ }, util::{ extract_text, is_codex_context_compaction_item_type, is_completed_plan_item, item_id, - item_type, now_ms, raw_string_field, string_field, + item_type, now_ms, raw_string_field, runtime_task_title, string_field, }, }; @@ -146,6 +146,14 @@ pub(crate) fn emit_response_event( payload_object.insert("source".to_owned(), source.clone()); } } + if terminal { + if let Some(title) = runtime_task_title(request) { + if let Some(payload_object) = payload.get_mut("payload").and_then(Value::as_object_mut) + { + payload_object.insert("taskTitle".to_owned(), Value::String(title)); + } + } + } if let Some(generated_user_message) = request.extra.get("runtime_generated_user_message") { if let Some(payload_object) = payload.get_mut("payload").and_then(Value::as_object_mut) { payload_object.insert( @@ -2141,6 +2149,42 @@ mod tests { assert!(event["payload"]["eventSeq"].as_u64().unwrap_or_default() > 0); } + #[test] + fn emits_runtime_task_title_only_with_terminal_response_events() { + let (event_tx, mut event_rx) = broadcast::channel(2); + let request = ExecutionRequest { + task_id: "task-1".to_owned(), + subtask_id: "codex-turn-1".to_owned(), + extra: Map::from_iter([( + "runtimeTaskTitle".to_owned(), + json!("Analyze production issue"), + )]), + ..ExecutionRequest::default() + }; + + emit_response_event( + &Some(event_tx.clone()), + "device-1", + "response.output_text.delta", + "local-task-1", + &request, + json!({"delta": "working"}), + ); + emit_response_event( + &Some(event_tx), + "device-1", + "response.completed", + "local-task-1", + &request, + json!({"value": "done"}), + ); + + let progress = event_rx.try_recv().expect("progress event"); + let terminal = event_rx.try_recv().expect("terminal event"); + assert!(progress["payload"].get("taskTitle").is_none()); + assert_eq!(terminal["payload"]["taskTitle"], "Analyze production issue"); + } + #[test] fn emits_one_transient_block_while_codex_reconnects() { let (event_tx, mut event_rx) = broadcast::channel(4); diff --git a/executor/src/runtime_work/handler.rs b/executor/src/runtime_work/handler.rs index 22addf7285..580310c81c 100644 --- a/executor/src/runtime_work/handler.rs +++ b/executor/src/runtime_work/handler.rs @@ -102,8 +102,9 @@ use super::{ apply_runtime_payload_metadata, bool_field, cloud_project_id, execution_request, id_field, infer_workspace_kind, integer_field, is_codex_context_compaction_item_type, item_id, item_type, normalize_device_id, normalize_workspace_path, now_ms, prompt_text, - restore_cloud_project_id, restore_origin, runtime_task_id, string_field, - timestamp_ms_field, workspace_group_path, workspace_path, + restore_cloud_project_id, restore_origin, runtime_task_id, runtime_task_title, + set_runtime_task_title, string_field, timestamp_ms_field, workspace_group_path, + workspace_path, }, worktrees::{WorktreeManager, WorktreeSettingsPatch}, }; diff --git a/executor/src/runtime_work/handler/claude_turns.rs b/executor/src/runtime_work/handler/claude_turns.rs index dbf0c90507..4458d900fd 100644 --- a/executor/src/runtime_work/handler/claude_turns.rs +++ b/executor/src/runtime_work/handler/claude_turns.rs @@ -376,6 +376,14 @@ impl RuntimeWorkRpcHandler { if let Some(source) = request.extra.get("source") { payload["payload"]["source"] = source.clone(); } + if matches!( + event_type, + "response.completed" | "response.failed" | "response.incomplete" | "error" + ) { + if let Some(title) = runtime_task_title(request) { + payload["payload"]["taskTitle"] = Value::String(title); + } + } let _ = event_tx.send(payload); } diff --git a/executor/src/runtime_work/handler/helpers/routing.rs b/executor/src/runtime_work/handler/helpers/routing.rs index 29d13e3fd3..f1e0706a70 100644 --- a/executor/src/runtime_work/handler/helpers/routing.rs +++ b/executor/src/runtime_work/handler/helpers/routing.rs @@ -351,6 +351,7 @@ fn runtime_event_request_from_link(link: &RuntimeTaskLink) -> ExecutionRequest { prompt: Value::String(link.title.clone()), ..ExecutionRequest::default() }; + set_runtime_task_title(&mut request, &link.title); if !link.project_plugin_ids.is_empty() { request.extra.insert( "project_plugin_ids".to_owned(), diff --git a/executor/src/runtime_work/handler/tasks.rs b/executor/src/runtime_work/handler/tasks.rs index 99d4f74ee6..6a23f66b08 100644 --- a/executor/src/runtime_work/handler/tasks.rs +++ b/executor/src/runtime_work/handler/tasks.rs @@ -276,6 +276,7 @@ impl RuntimeWorkRpcHandler { let mut request = execution_request(&payload) .ok_or_else(|| AppIpcError::new("bad_request", "executionRequest is required"))?; apply_runtime_payload_metadata(&mut request, &payload); + set_runtime_task_title(&mut request, &title); if is_codex_runtime(&runtime) { if let (Some(project_key), Some(project_name)) = ( request.runtime_project_key.as_deref(), @@ -716,6 +717,9 @@ impl RuntimeWorkRpcHandler { let mut request = payload_execution_request .ok_or_else(|| AppIpcError::new("bad_request", "executionRequest is required"))?; apply_runtime_payload_metadata(&mut request, &payload); + if let Some(link) = existing_link.as_ref() { + set_runtime_task_title(&mut request, &link.title); + } if let Some(link) = existing_link.as_ref() { mark_runtime_model_switch(&mut request, link, &payload); } @@ -1018,6 +1022,7 @@ impl RuntimeWorkRpcHandler { .or_else(|| workspace_path(&payload)) .unwrap_or_default(); apply_runtime_payload_metadata(&mut request, &payload); + set_runtime_task_title(&mut request, &existing_link.title); mark_runtime_model_switch(&mut request, &existing_link, &payload); restore_cloud_project_id(&mut request, &existing_link.runtime_handle); restore_origin(&mut request, &existing_link.runtime_handle); diff --git a/executor/src/runtime_work/util.rs b/executor/src/runtime_work/util.rs index aa9de94a37..61790ddbb5 100644 --- a/executor/src/runtime_work/util.rs +++ b/executor/src/runtime_work/util.rs @@ -20,6 +20,9 @@ pub(crate) fn execution_request(payload: &Value) -> Option { } pub(crate) fn apply_runtime_payload_metadata(request: &mut ExecutionRequest, payload: &Value) { + if let Some(title) = string_field(payload, "title") { + set_runtime_task_title(request, &title); + } if !request.extra.contains_key("cloudProjectId") && !request.extra.contains_key("cloud_project_id") { @@ -135,6 +138,28 @@ pub(crate) fn apply_runtime_payload_metadata(request: &mut ExecutionRequest, pay } } +pub(crate) fn set_runtime_task_title(request: &mut ExecutionRequest, title: &str) { + let title = title.trim(); + if title.is_empty() { + return; + } + request.extra.insert( + "runtimeTaskTitle".to_owned(), + Value::String(title.to_owned()), + ); +} + +pub(crate) fn runtime_task_title(request: &ExecutionRequest) -> Option { + request + .extra + .get("runtimeTaskTitle") + .or_else(|| request.extra.get("runtime_task_title")) + .and_then(Value::as_str) + .map(str::trim) + .filter(|title| !title.is_empty()) + .map(str::to_owned) +} + fn runtime_generated_user_message(payload: &Value) -> Option { let source = payload.get("source").filter(|value| value.is_object())?; if string_field(source, "source").as_deref() != Some("im") { @@ -775,6 +800,18 @@ mod tests { assert_eq!(cloud_project_id(&request), Some(json!(9001))); } + #[test] + fn copies_runtime_task_title_from_runtime_payload() { + let mut request = ExecutionRequest::default(); + + apply_runtime_payload_metadata(&mut request, &json!({"title": " Analyze issue "})); + + assert_eq!( + runtime_task_title(&request).as_deref(), + Some("Analyze issue") + ); + } + #[test] fn copies_runtime_origin_from_runtime_payload() { let mut request = ExecutionRequest::default(); diff --git a/executor/tests/app_runtime_work_send_contract.rs b/executor/tests/app_runtime_work_send_contract.rs index ee9be6515f..6e92e80964 100644 --- a/executor/tests/app_runtime_work_send_contract.rs +++ b/executor/tests/app_runtime_work_send_contract.rs @@ -346,6 +346,11 @@ async fn claude_runtime_task_uses_conversation_events_and_resumes_follow_up() { assert!(runtime_events.iter().any(|event| { event["event"] == "response.created" && event["payload"]["runtime"] == "claude_code" })); + assert!(runtime_events.iter().any(|event| { + event["event"] == "response.completed" + && event["payload"]["runtime"] == "claude_code" + && event["payload"]["taskTitle"] == "first turn" + })); } #[tokio::test]