Repository navigation
fix(runtime): deliver away notifications to IM #2859
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 1 commit
8716e6e
a66a890
c461907
747551f
fe73682
29fb8a8
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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: | ||
|
Comment on lines
+2224
to
+2230
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Move the new IM-notification orchestration into a focused service instead of adding another roughly 100-line method to AGENTS.md reference: AGENTS.md:L45-L47 Useful? React with 👍 / 👎. |
||
| """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, | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -15,6 +15,19 @@ | |
| from app.api.ws.wework_runtime_namespace import WeworkRuntimeNamespace | ||
|
|
||
|
|
||
| @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 | ||
|
Comment on lines
+20
to
+30
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win Add type annotations to the new fixture. Annotate 🤖 Prompt for AI AgentsSource: Coding guidelines |
||
|
|
||
|
|
||
| def _im_source() -> dict: | ||
| return { | ||
| "source": "im", | ||
|
|
@@ -135,6 +148,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.""" | ||
|
|
||
|
|
@@ -160,10 +174,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() | ||
|
|
@@ -186,6 +204,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 | ||
|
|
@@ -214,6 +233,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, | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Do not wait for IM delivery while the runtime-event lock is held.
Line 2217 awaits
_notify_runtime_eventinside the per-socket ordered relay path. Lines 2283-2296 then await IM delivery. A slow dispatcher delays the acknowledgement and blocks later runtime events from the same device socket until delivery returns.Queue this best-effort work through the existing tracked background-task path. Apply a bounded timeout at the delivery boundary.
Also applies to: 2283-2296
🤖 Prompt for AI Agents