From 5b36645b914c0be638b50cd43dee0189d22c34a8 Mon Sep 17 00:00:00 2001 From: Joe Date: Tue, 25 Aug 2026 04:36:36 +0800 Subject: [PATCH 1/2] fix: support DingTalk channel subscription for enterprise groups --- .gitignore | 2 + coworker/connectors/__init__.py | 5 + coworker/connectors/adapters.py | 16 + coworker/connectors/base.py | 9 + coworker/connectors/catalog_copy.py | 8 + coworker/connectors/config.py | 7 +- coworker/connectors/descriptors.py | 91 +++ coworker/connectors/dingtalk.py | 650 +++++++++++++++++ coworker/connectors/gateway.py | 7 + coworker/connectors/senders.py | 17 + coworker/connectors/tools.py | 27 +- coworker/engine.py | 78 ++ coworker/server/app.py | 99 +++ coworker/server/manager.py | 132 +++- coworker/server/run.py | 20 + pyproject.toml | 2 +- surfaces/gui/src/components/AccessSection.tsx | 13 +- .../gui/src/components/SubscriptionsChip.tsx | 4 +- surfaces/gui/src/components/brandIcons.tsx | 14 + .../connectors/AddConnectionModal.tsx | 2 +- surfaces/gui/src/connectors/registry.tsx | 7 + tests/test_dingtalk.py | 666 ++++++++++++++++++ tests/test_mention_router.py | 165 ++++- 23 files changed, 2013 insertions(+), 28 deletions(-) create mode 100644 coworker/connectors/dingtalk.py create mode 100644 tests/test_dingtalk.py diff --git a/.gitignore b/.gitignore index 4cc93b88b7..6e2f5977df 100644 --- a/.gitignore +++ b/.gitignore @@ -10,3 +10,5 @@ dist/ # Local secrets (live-smoke BYO keys) โ€” never committed .env .claude/settings.local.json +.workbuddy/ +debug_*.py diff --git a/coworker/connectors/__init__.py b/coworker/connectors/__init__.py index 04a45cd1d2..176fcebfd5 100644 --- a/coworker/connectors/__init__.py +++ b/coworker/connectors/__init__.py @@ -13,6 +13,7 @@ parse_target, ) from .adapters import ( + DingTalkAdapter, SlackAdapter, TelegramAdapter, make_adapter, @@ -20,6 +21,7 @@ telegram_message_to_event, ) from .config import ConnectorSettings, TeamAuth, is_authorized, load_settings +from .dingtalk import send_dingtalk, webhook_payload_to_event from .relay_client import SlackRelayAdapter from .slack_addr import qualify as slack_qualify, split as slack_split from .descriptors import ConnectorDescriptor, get_descriptor, list_descriptors @@ -67,12 +69,15 @@ "make_send_file_tool", "make_send_message_tool", "connector_for_tool", + "DingTalkAdapter", "SlackAdapter", "SlackRelayAdapter", "TelegramAdapter", "make_adapter", "slack_event_to_event", "telegram_message_to_event", + "send_dingtalk", + "webhook_payload_to_event", "slack_qualify", "slack_split", ] diff --git a/coworker/connectors/adapters.py b/coworker/connectors/adapters.py index 3d64d9528e..54ae8097ff 100644 --- a/coworker/connectors/adapters.py +++ b/coworker/connectors/adapters.py @@ -20,6 +20,7 @@ SendResult, SessionSource, ) +from .dingtalk import DingTalkAdapter from .senders import _send_slack, _send_slack_interactive, _send_telegram logger = logging.getLogger("coworker.connectors") @@ -441,6 +442,21 @@ def make_adapter( """ if platform == "telegram" and profile.get("bot_token"): return TelegramAdapter(profile["bot_token"]) + if platform == "dingtalk": + if profile.get("client_id") and profile.get("client_secret"): + logger.info("dingtalk adapter created (stream mode)") + return DingTalkAdapter( + client_id=profile["client_id"], + client_secret=profile["client_secret"], + secrets=secrets, + ) + if profile.get("webhook_url"): + logger.info("dingtalk adapter created (webhook mode)") + return DingTalkAdapter( + webhook_url=profile["webhook_url"], + secret=profile.get("secret"), + secrets=secrets, + ) if platform == "slack": if profile.get("mode") == "relay": if not (relay_url and token_provider): diff --git a/coworker/connectors/base.py b/coworker/connectors/base.py index e269c09e95..855d2f8e69 100644 --- a/coworker/connectors/base.py +++ b/coworker/connectors/base.py @@ -8,12 +8,16 @@ from __future__ import annotations +import logging from abc import ABC, abstractmethod from dataclasses import asdict, dataclass, field from enum import Enum from typing import Any, Awaitable, Callable, Optional +logger = logging.getLogger("coworker.connectors") + + class MessageType(str, Enum): TEXT = "text" COMMAND = "command" @@ -180,5 +184,10 @@ async def send( """Send an outbound message.""" async def handle_message(self, event: MessageEvent) -> None: + logger.info( + "adapter.handle_message platform=%s handler_set=%s", + self.platform, + self._handler is not None, + ) if self._handler is not None: await self._handler(event) diff --git a/coworker/connectors/catalog_copy.py b/coworker/connectors/catalog_copy.py index 25e09b30de..119f7657c2 100644 --- a/coworker/connectors/catalog_copy.py +++ b/coworker/connectors/catalog_copy.py @@ -19,6 +19,9 @@ "slack": "Bring your coworker into Slack: mention it in a channel or DM it, " "and replies land in-thread. Any number of workspaces can be connected, " "each with its own allow-list of who may talk to the agent.", + "dingtalk": "Bring your coworker into DingTalk. Stream mode gives two-way " + "chat through an enterprise app robot with no public IP required; webhook mode " + "sends one-way notifications to a group bot.", "email": "Read, search, and send mail on any IMAP account โ€” Gmail, iCloud, " "Fastmail, or your own server โ€” using an app password instead of your " "account password.", @@ -70,6 +73,11 @@ "Reads files shared in those channels.", "Reads member and channel names to resolve who's talking.", ], + "dingtalk": [ + "Reads @-mention messages sent to the robot (Stream mode) or group bot.", + "Posts messages back to the same conversation.", + "Only senders on your allow-list are answered.", + ], "email": [ "Reads and searches mail over IMAP.", "Sends mail as your address, and saves attachments locally.", diff --git a/coworker/connectors/config.py b/coworker/connectors/config.py index 6553d15cd4..e161dc1b7d 100644 --- a/coworker/connectors/config.py +++ b/coworker/connectors/config.py @@ -14,7 +14,7 @@ from ..secrets import SecretStore from .base import SessionSource -PLATFORMS = ("telegram", "slack", "github") +PLATFORMS = ("telegram", "slack", "dingtalk", "github") @dataclass @@ -76,6 +76,11 @@ def load_settings( for platform in PLATFORMS: profile = secrets.get(f"{platform}:default") or {} token = profile.get("bot_token") + # DingTalk authenticates in Stream mode with client_id + client_secret, + # not a bot_token. Treat those as the credential for enablement so the + # adapter actually registers and connects at startup. + if platform == "dingtalk": + token = token or (profile.get("client_id") and profile.get("client_secret")) allowed = set(profile.get("allowed_users") or []) allowed |= _csv(os.environ.get(f"{platform.upper()}_ALLOWED_USERS")) allow_all = bool(profile.get("allow_all")) or os.environ.get( diff --git a/coworker/connectors/descriptors.py b/coworker/connectors/descriptors.py index 32e103c6f5..b034490e44 100644 --- a/coworker/connectors/descriptors.py +++ b/coworker/connectors/descriptors.py @@ -137,6 +137,47 @@ def _validate_slack(creds: dict) -> ValidationResult: return ValidationResult(False, error=data.get("error") or "invalid bot token") +def _validate_dingtalk(creds: dict) -> ValidationResult: + """Validate either a Stream-mode enterprise app or a group-bot webhook.""" + client_id = creds.get("client_id", "") + client_secret = creds.get("client_secret", "") + if client_id and client_secret: + try: + import dingtalk_stream + except ImportError: + return ValidationResult( + False, + error="dingtalk-stream SDK is not installed โ€” run `pip install dingtalk-stream`", + ) + try: + credential = dingtalk_stream.Credential(client_id, client_secret) + client = dingtalk_stream.DingTalkStreamClient(credential) + token = client.get_access_token() + except Exception as exc: + return ValidationResult(False, error=str(exc)) + if token: + return ValidationResult(True, identity="DingTalk enterprise bot (stream)") + return ValidationResult(False, error="invalid DingTalk client credentials") + + # Fallback / dual-use: validate the group-bot webhook if provided. + from .dingtalk import send_dingtalk + + webhook_url = creds.get("webhook_url", "") + if not webhook_url.startswith("https://oapi.dingtalk.com/robot/send"): + return ValidationResult( + False, + error="expected a DingTalk group-bot webhook URL, or Client ID + Client Secret for stream mode", + ) + secret = creds.get("secret") or None + try: + result = send_dingtalk(webhook_url, "OpenWorker connection test", secret=secret) + except Exception as exc: + return ValidationResult(False, error=str(exc)) + if result.ok: + return ValidationResult(True, identity="DingTalk group bot") + return ValidationResult(False, error=result.error or "invalid DingTalk webhook") + + def _validate_whoami( method: str, url: str, @@ -488,6 +529,56 @@ def _validate_outlook(creds: dict) -> ValidationResult: ], validate=_validate_slack, ), + ConnectorDescriptor( + name="dingtalk", + title="DingTalk", + icon="๐Ÿ’ฌ", + blurb="Send notifications to a DingTalk group bot, or hold two-way conversations with an enterprise Stream-mode robot.", + auth="webhook", + two_way=True, + channels=True, + brand_color="#3370ff", + logo="dingtalk", + fields=[ + Field( + "client_id", + "Client ID (AppKey)", + secret=True, + required=False, + help="Enterprise app Client ID for Stream mode (two-way, no public IP needed).", + placeholder="dingxxxxxxxxxxxx", + ), + Field( + "client_secret", + "Client Secret (AppSecret)", + secret=True, + required=False, + help="Enterprise app Client Secret for Stream mode.", + ), + Field( + "webhook_url", + "Webhook URL", + secret=True, + required=False, + help="Group-bot webhook for outbound-only notifications. Ignored when Client ID + Client Secret are provided.", + placeholder="https://oapi.dingtalk.com/robot/send?access_token=...", + ), + Field( + "secret", + "Secret", + secret=True, + required=False, + help="Optional signing secret for the group-bot webhook.", + ), + _ALLOWED_FIELD, + ], + instructions=[ + "For two-way chat (recommended): open https://open-dev.dingtalk.com โ†’ create an enterprise-internal app โ†’ add a robot โ†’ set message-receiving mode to Stream. Copy the app's Client ID and Client Secret below.", + "For outbound-only notifications: open a DingTalk group โ†’ Group Settings โ†’ Group Assistant โ†’ Add Robot โ†’ Custom, enable signing if desired, and copy the webhook URL.", + "If both sets of credentials are provided, Stream mode takes precedence.", + ], + validate=_validate_dingtalk, + ), ConnectorDescriptor( name="email", title="Email (IMAP)", diff --git a/coworker/connectors/dingtalk.py b/coworker/connectors/dingtalk.py new file mode 100644 index 0000000000..8efda7fb86 --- /dev/null +++ b/coworker/connectors/dingtalk.py @@ -0,0 +1,650 @@ +"""DingTalk (้’‰้’‰) connector. + +Supports two mutually exclusive modes: + +1. **Group-bot webhook** (outbound-only by default): send messages to a DingTalk + group via a webhook URL that contains an access_token, optionally signed with + a bot secret. This is the easiest way to get notifications into a group, but + DingTalk custom group bots do *not* deliver inbound @-mentions to a callback + URL unless the bot is upgraded. + +2. **Enterprise (stream) mode** (two-way): create an enterprise-internal app + + robot in the DingTalk open platform, choose "Stream" as the message-receiving + mode, and provide the app's ClientId + ClientSecret. The connector opens a + WebSocket to DingTalk's Stream gateway, receives messages without a public IP, + and replies via the per-conversation ``sessionWebhook`` that accompanies every + inbound push. + +Inbound messages in stream mode arrive through the dingtalk-stream SDK; the +legacy HTTP webhook route (``/v1/connectors/dingtalk/webhook``) is kept for +backwards compatibility with group-bot callbacks that some enterprise bots can +also emit. +""" + +from __future__ import annotations + +import asyncio +import base64 +import hashlib +import hmac +import json +import logging +import time +from typing import Any, Optional +from urllib.parse import parse_qs, urlparse + +from .base import BasePlatformAdapter, MessageEvent, SendResult, SessionSource + +logger = logging.getLogger("coworker.connectors") + +_DINGTALK_API_BASE = "https://oapi.dingtalk.com" +_TIMEOUT = 30.0 + + +def _sign(secret: str, timestamp: str) -> str: + """DingTalk group-bot signature: base64(hmac_sha256(timestamp + '\n' + secret)).""" + msg = f"{timestamp}\n{secret}".encode("utf-8") + mac = hmac.new(secret.encode("utf-8"), msg, digestmod=hashlib.sha256).digest() + return base64.b64encode(mac).decode("utf-8") + + +def _build_dingtalk_payload(text: str, msgtype: str) -> dict[str, Any]: + """Build a DingTalk message payload for the given msgtype.""" + payload: dict[str, Any] = {"msgtype": msgtype} + if msgtype == "markdown": + payload["markdown"] = {"title": text.split("\n", 1)[0][:64], "text": text} + else: + payload["text"] = {"content": text} + return payload + + +def _send_once( + webhook_url: str, + text: str, + secret: Optional[str], + msgtype: str, +) -> SendResult: + """Single attempt to send a DingTalk group-bot message.""" + # httpx drops an existing query string when params={} is passed, so we extract + # every query param (especially access_token) into the explicit params dict and + # post to the bare path. + parsed = urlparse(webhook_url) + params: dict[str, str] = {} + for key, values in parse_qs(parsed.query).items(): + if values: + params[key] = values[-1] + url = f"{parsed.scheme}://{parsed.netloc}{parsed.path}" + + if secret: + timestamp = str(int(time.time() * 1000)) + params["timestamp"] = timestamp + params["sign"] = _sign(secret, timestamp) + + payload = _build_dingtalk_payload(text, msgtype) + + try: + import httpx + + resp = httpx.post(url, params=params, json=payload, timeout=_TIMEOUT) + data = resp.json() + except Exception as exc: + return SendResult(False, error=str(exc)) + + if data.get("errcode") == 0: + return SendResult(True, message_id=str(data.get("msg_id") or "")) + return SendResult( + False, + error=f"dingtalk {data.get('errcode')}: {data.get('errmsg', 'send failed')}", + ) + + +def _is_msgtype_error(error: str) -> bool: + """True if the error indicates the bot rejected the msgtype.""" + err = error.lower() + return ( + "300001" in error + or "450001" in error + or "robot type" in err + or "not match" in err + or "ไธๆ”ฏๆŒ็š„ๆถˆๆฏ็ฑปๅž‹" in err + or "msgtype" in err + ) + + +def send_dingtalk( + webhook_url: str, + text: str, + secret: Optional[str] = None, + msgtype: str = "text", +) -> SendResult: + """Send a message through a DingTalk group-bot or session webhook. + + If the bot rejects the default ``text`` message type (common with bots that + only accept markdown/actionCard), automatically fall back to ``markdown``. + """ + result = _send_once(webhook_url, text, secret, msgtype) + if not result.ok and msgtype == "text" and _is_msgtype_error(result.error or ""): + return _send_once(webhook_url, text, secret, "markdown") + return result + + +def _strip_at(text: str, at_users: list[dict[str, Any]]) -> str: + """Remove '@ๆœบๅ™จไบบ' text that DingTalk prepends to the message content.""" + for at in at_users or []: + name = at.get("name") or at.get("nick") or "" + if not name: + continue + # DingTalk prefixes "@ๆœบๅ™จไบบๅ " or embeds "@ๆœบๅ™จไบบๅ". + for pat in (f"@{name} ", f"@{name}"): + if text.startswith(pat): + text = text[len(pat) :] + break + return text.strip() + + +def webhook_payload_to_event(payload: dict[str, Any]) -> Optional[MessageEvent]: + """Convert a DingTalk inbound callback payload into a MessageEvent. + + Handles group-chat robot callbacks (conversationId + senderStaffId) and + session-webhook pushes (senderNick). Falls back gracefully when fields are + missing. + """ + msgtype = payload.get("msgtype") or payload.get("msgType") + if msgtype != "text": + # Only plain text inbound is supported in v1. + return None + + text_obj = payload.get("text") or {} + text = str(text_obj.get("content") or "").strip() + if not text: + return None + + # Strip the leading @bot mention if present. + at_users = payload.get("atUsers") or payload.get("atUserIds") or [] + text = _strip_at(text, at_users) + if not text: + return None + + sender_id = str( + payload.get("senderStaffId") + or payload.get("senderUserId") + or payload.get("sender") + or "" + ) + sender_name = payload.get("senderNick") or payload.get("senderName") or sender_id + conversation_id = str(payload.get("conversationId") or "") + chat_type = "group" if conversation_id else "dm" + + source = SessionSource( + platform="dingtalk", + chat_id=conversation_id, + user_id=sender_id, + user_name=sender_name, + chat_type=chat_type, + thread_id=None, + ) + return MessageEvent(text=text, source=source, raw=payload) + + +def _chatbot_message_to_event(msg: Any) -> Optional[MessageEvent]: + """Convert a dingtalk-stream ChatbotMessage into a MessageEvent.""" + text_content = getattr(msg, "text", None) + text = str(getattr(text_content, "content", "") or "").strip() + if not text: + return None + + chatbot_user_id = str(getattr(msg, "chatbot_user_id", "") or "") + at_users_raw = getattr(msg, "at_users", None) or [] + at_users: list[dict[str, Any]] = [] + mentions_me = bool(getattr(msg, "is_in_at_list", False)) + for at in at_users_raw: + name = getattr(at, "name", "") or "" + if name: + at_users.append({"name": name}) + # Some payloads don't set is_in_at_list; fall back to matching the bot id. + if chatbot_user_id and not mentions_me: + at_id = str( + getattr(at, "dingtalk_id", "") + or getattr(at, "staff_id", "") + or "" + ) + if at_id and at_id == chatbot_user_id: + mentions_me = True + + text = _strip_at(text, at_users) + if not text: + return None + + sender_id = str( + getattr(msg, "sender_staff_id", "") + or getattr(msg, "sender_user_id", "") + or "" + ) + sender_name = getattr(msg, "sender_nick", "") or sender_id + conversation_id = str(getattr(msg, "conversation_id", "") or "") + chat_type = "group" if conversation_id else "dm" + message_id = str(getattr(msg, "message_id", "") or "") + + source = SessionSource( + platform="dingtalk", + chat_id=conversation_id, + user_id=sender_id, + user_name=sender_name, + chat_type=chat_type, + thread_id=None, + ) + return MessageEvent( + text=text, + source=source, + message_id=message_id, + mentions_me=mentions_me, + raw=msg.to_dict(), + ) + + +class _DingTalkStreamHandler: + """Bridge between dingtalk-stream's ChatbotHandler and our BasePlatformAdapter. + + We do not inherit from ChatbotHandler at import time to keep the module + importable when dingtalk-stream is absent. The adapter creates the real + subclass dynamically inside connect(). + """ + + def __init__(self, adapter: DingTalkAdapter) -> None: + self.adapter = adapter + + def _make_handler_class(self) -> type: + import dingtalk_stream + + outer = self + + class Handler(dingtalk_stream.ChatbotHandler): + async def process(self, callback: dingtalk_stream.CallbackMessage): + # dingtalk-stream may drive `process` from its event loop OR a worker + # thread, and its upstream only logs str(e) โ€” which hides the trace. + # Wrap everything so the real stack lands in the backend log, and + # always return an ack tuple so one bad message can't kill the stream. + try: + data = getattr(callback, "data", {}) or {} + topic = getattr(callback.headers, "topic", None) if hasattr(callback, "headers") else None + outer.adapter._last_inbound_at = time.time() + outer.adapter._inbound_count += 1 + logger.info( + "dingtalk raw callback: topic=%s data_keys=%s", + topic, + sorted(data.keys()) if isinstance(data, dict) else type(data).__name__, + ) + + try: + msg = dingtalk_stream.ChatbotMessage.from_dict(data) + except Exception as exc: + logger.warning("dingtalk failed to parse ChatbotMessage: %s", exc, exc_info=True) + return dingtalk_stream.AckMessage.STATUS_OK, "OK" + + msgtype = getattr(msg, "msgtype", None) + text_obj = getattr(msg, "text", None) or {} + raw_text = str(getattr(text_obj, "content", "") or "").strip() + conversation_id = str(getattr(msg, "conversation_id", "") or "") + sender_nick = getattr(msg, "sender_nick", "") or "" + logger.info( + "dingtalk parsed: msgtype=%s conversation_id=%s sender=%s text=%r", + msgtype, + conversation_id, + sender_nick, + raw_text, + ) + + event = _chatbot_message_to_event(msg) + if event is None: + logger.info( + "dingtalk callback produced no MessageEvent " + "(msgtype=%s conversation_id=%s text=%r)", + msgtype, + conversation_id, + raw_text, + ) + return dingtalk_stream.AckMessage.STATUS_OK, "OK" + + # Surface the conversation id so the operator can copy it into the + # Add-channel field (`dingtalk:`). It only exists in + # the inbound message, never in the open-platform console. + outer.adapter._last_conversation_id = event.source.chat_id + logger.info( + "dingtalk inbound: conversationId=%s sender=%s(%s) chat_type=%s " + "mentions_me=%s message_id=%s text=%r", + event.source.chat_id, + event.source.user_name, + event.source.user_id, + event.source.chat_type, + event.mentions_me, + event.message_id, + event.text, + ) + + # Remember the conversation-specific reply webhook. Persist it so + # the stateless send_message tool can reply outside the adapter. + session_webhook = str(getattr(msg, "session_webhook", "") or "").strip() + if session_webhook: + outer.adapter._set_session_webhook( + event.source.chat_id, session_webhook + ) + + # Dispatch the agent turn off the SDK's hot path. run_coroutine_threadsafe + # is safe whether process was awaited in the loop or run on a thread. + loop = outer.adapter._loop + if loop is not None: + future = asyncio.run_coroutine_threadsafe(outer.adapter.handle_message(event), loop) + def _on_done(fut): + try: + fut.result() + except Exception as exc: + logger.exception("dingtalk inbound dispatch failed: %s", exc) + future.add_done_callback(_on_done) + else: + logger.warning( + "dingtalk adapter has no event loop captured; cannot dispatch inbound" + ) + return dingtalk_stream.AckMessage.STATUS_OK, "OK" + except Exception: + logger.exception("dingtalk stream process failed") + return dingtalk_stream.AckMessage.STATUS_OK, "OK" + + class EventHandler(dingtalk_stream.EventHandler): + async def process(self, event: dingtalk_stream.EventMessage): + try: + logger.info( + "dingtalk event: topic=%s event_type=%s event_id=%s data_keys=%s", + event.headers.topic, + event.headers.event_type, + event.headers.event_id, + sorted(event.data.keys()) if isinstance(event.data, dict) else type(event.data).__name__, + ) + except Exception: + logger.exception("dingtalk event handler failed") + return dingtalk_stream.AckMessage.STATUS_OK, "OK" + + return Handler, EventHandler + + +class DingTalkAdapter(BasePlatformAdapter): + platform = "dingtalk" + + def __init__( + self, + webhook_url: Optional[str] = None, + secret: Optional[str] = None, + client_id: Optional[str] = None, + client_secret: Optional[str] = None, + secrets: Any = None, + ) -> None: + super().__init__() + self.webhook_url = webhook_url + self.secret = secret + self.client_id = client_id + self.client_secret = client_secret + self._secrets = secrets + # In-memory cache of conversation -> sessionWebhook for stream-mode replies. + self._session_webhooks: dict[str, str] = {} + # Stream mode runtime state. + self._client: Any = None + self._handler: Any = None + self._task: Optional[asyncio.Task] = None + self._closing = False + # Captured event loop, used to dispatch inbound agent turns thread-safely. + self._loop: Optional[asyncio.AbstractEventLoop] = None + # Diagnostics surfaced by /v1/connectors/dingtalk/status. + self._connection_state: str = "idle" # idle | connecting | connected | error + self._connected_at: Optional[float] = None + self._last_error: Optional[str] = None + self._last_inbound_at: Optional[float] = None + self._last_conversation_id: Optional[str] = None + self._inbound_count: int = 0 + + @property + def mode(self) -> str: + """'stream' if enterprise credentials are present, otherwise 'webhook'.""" + if self.client_id and self.client_secret: + return "stream" + return "webhook" + + def _profile_key(self) -> str: + return "dingtalk:default" + + def _load_profile(self) -> dict[str, Any]: + if self._secrets is None: + return {} + return self._secrets.get(self._profile_key()) or {} + + def _save_session_webhooks(self) -> None: + """Persist the in-memory session webhook cache back to the profile.""" + if self._secrets is None or not self._session_webhooks: + return + profile = self._load_profile() + webhooks = { + k: v + for k, v in self._session_webhooks.items() + if v + } + if not webhooks: + return + profile["session_webhooks"] = webhooks + self._secrets.put(self._profile_key(), profile) + + def _set_session_webhook(self, chat_id: str, webhook: str) -> None: + """Store a per-conversation reply webhook (stream mode).""" + if not chat_id or not webhook: + return + self._session_webhooks[chat_id] = webhook + self._save_session_webhooks() + + def status(self) -> dict[str, Any]: + """Runtime diagnostics for the DingTalk status endpoint.""" + import asyncio + + task_state: Optional[str] = None + if self._task is not None: + if self._task.done(): + task_state = "done" + if self._task.cancelled(): + task_state = "cancelled" + elif self._task.exception(): + task_state = "failed" + else: + task_state = "running" + return { + "mode": self.mode, + "state": self._connection_state, + "task_state": task_state, + "connected_at": self._connected_at, + "last_inbound_at": self._last_inbound_at, + "last_conversation_id": self._last_conversation_id, + "inbound_count": self._inbound_count, + "session_webhooks": len(self._session_webhooks), + "client_id_prefix": self.client_id[:8] if self.client_id else None, + "last_error": self._last_error, + } + + async def connect(self) -> bool: + if self.mode == "stream": + return await self._connect_stream() + return bool(self.webhook_url) + + def _on_stream_task_done(self, task: asyncio.Task) -> None: + """Log unexpected stream client exits so operators can diagnose disconnects.""" + if task.cancelled(): + self._connection_state = "cancelled" + logger.info("dingtalk stream task cancelled") + return + exc = task.exception() + if exc is not None: + self._connection_state = "error" + self._last_error = f"{type(exc).__name__}: {exc}" + logger.warning( + "dingtalk stream task exited with error: %s", self._last_error, exc_info=exc + ) + else: + self._connection_state = "disconnected" + logger.info("dingtalk stream task ended") + + async def _run_stream_client(self) -> None: + """Wrap the SDK start() call to catch and log connection errors.""" + if self._client is None: + return + try: + logger.info("dingtalk stream client starting websocket handshake") + await self._client.start() + except asyncio.CancelledError: + # Expected during shutdown; the SDK's reconnect loop catches + # CancelledError and loops forever, so we exit here. + if not self._closing: + self._connection_state = "error" + self._last_error = "stream task cancelled unexpectedly" + logger.warning("dingtalk stream task cancelled unexpectedly") + raise + except Exception as exc: + if not self._closing: + self._connection_state = "error" + self._last_error = f"{type(exc).__name__}: {exc}" + logger.warning("dingtalk stream client error: %s", self._last_error, exc_info=True) + raise + + async def _connect_stream(self) -> bool: + try: + import dingtalk_stream + except ImportError: + self._connection_state = "error" + self._last_error = "dingtalk-stream SDK not installed" + logger.warning( + "dingtalk-stream not installed โ€” run `pip install dingtalk-stream`" + ) + return False + + self._connection_state = "connecting" + logger.info( + "dingtalk stream connecting: client_id=%s... mode=stream", + self.client_id[:6] if self.client_id else "", + ) + + credential = dingtalk_stream.Credential(self.client_id, self.client_secret) + self._client = dingtalk_stream.DingTalkStreamClient(credential) + # Capture the running loop so the inbound dispatch is thread-safe even if + # dingtalk-stream invokes our handler from a worker thread. + try: + self._loop = asyncio.get_event_loop() + except RuntimeError: + self._loop = None + + # Build the handler dynamically so the class only inherits from + # dingtalk_stream.ChatbotHandler when the SDK is available. + bridge = _DingTalkStreamHandler(self) + handler_class, event_handler_class = bridge._make_handler_class() + # IMPORTANT: do NOT assign to `self._handler` โ€” that field is owned by + # BaseAdapter and points at `gateway._on_inbound` (set via + # `set_message_handler` during `Gateway.register`). Overwriting it with + # the dingtalk_stream ChatbotHandler instance breaks inbound dispatch: + # `await self._handler(event)` raises `'Handler' object is not callable`. + self._stream_callback_handler = handler_class() + + topic = dingtalk_stream.chatbot.ChatbotMessage.TOPIC + self._client.register_callback_handler(topic, self._stream_callback_handler) + logger.info("dingtalk stream callback handler registered for topic=%s", topic) + + # Catch-all event handler so we can see connection-level / lifecycle events + # and log any event topic that arrives. + self._client.register_all_event_handler(event_handler_class()) + logger.info("dingtalk stream catch-all event handler registered") + + # Log the exact subscription list that will be sent to the gateway. + subscriptions: list[dict[str, str]] = [] + if self._client._is_event_required: + subscriptions.append({"type": "EVENT", "topic": "*"}) + for t in self._client.callback_handler_map.keys(): + subscriptions.append({"type": "CALLBACK", "topic": t}) + logger.info("dingtalk stream subscriptions: %s", subscriptions) + + self._closing = False + self._task = asyncio.create_task(self._run_stream_client()) + self._task.add_done_callback(self._on_stream_task_done) + # The SDK's start() is blocking; we treat "task spawned" as connected + # because the handshake happens inside start(). The watchdog/task_done + # logging will surface any failure. + self._connection_state = "connected" + self._connected_at = asyncio.get_event_loop().time() + logger.info("dingtalk adapter connected (stream mode)") + return True + + async def disconnect(self) -> None: + if self.mode != "stream": + return + self._closing = True + self._connection_state = "disconnecting" + logger.info("dingtalk adapter disconnecting (stream mode)") + + if self._client is not None: + # The SDK does not expose a stop() method; close the open websocket + # so the async-for in start() raises ConnectionClosedError and the + # task yields. Without this the SDK's reconnect loop catches + # CancelledError and sleeps/reconnects forever. + try: + ws = getattr(self._client, "websocket", None) + if ws is not None and hasattr(ws, "close"): + await ws.close(code=1001, reason="shutdown") + logger.debug("dingtalk stream websocket closed") + except Exception: + logger.debug("dingtalk stream websocket close failed", exc_info=True) + + if self._task is not None: + self._task.cancel() + # The SDK's start() catches CancelledError in its reconnect loop and + # reconnects forever, so a plain `await self._task` can hang. Use + # asyncio.wait (not wait_for) with a timeout: it returns whether or + # not the task respects cancellation, and Uvicorn will reap the + # process once lifespan shutdown completes. + done, pending = await asyncio.wait({self._task}, timeout=5.0) + if self._task in pending: + logger.warning( + "dingtalk stream task did not finish within 5s; leaving it behind" + ) + try: + # Re-raise cancellation / surface any stored exception if the + # task actually finished in time. + if self._task in done: + self._task.result() + except asyncio.CancelledError: + pass + self._task = None + self._connection_state = "idle" + logger.info("dingtalk adapter disconnected") + + async def send( + self, chat_id: str, text: str, *, thread_id: Optional[str] = None + ) -> SendResult: + if self.mode == "stream": + # Stream replies must use the per-conversation sessionWebhook that + # arrived with the inbound message. + webhook = self._session_webhooks.get(chat_id) if chat_id else None + if not webhook: + return SendResult( + False, + error="dingtalk stream: no session webhook for this conversation yet", + ) + return await asyncio.to_thread(send_dingtalk, webhook, text) + + # Webhook mode: fixed group-bot URL. + return await asyncio.to_thread( + send_dingtalk, self.webhook_url, text, self.secret + ) + + def receive_webhook(self, payload: dict[str, Any]) -> Optional[MessageEvent]: + """Called by the FastAPI route when DingTalk pushes a message. + + Used by group-bot callbacks and as a fallback for enterprise bots that + also emit HTTP callbacks. + """ + event = webhook_payload_to_event(payload) + if event is not None: + session_webhook = str(payload.get("sessionWebhook") or "").strip() + if session_webhook: + self._set_session_webhook(event.source.chat_id, session_webhook) + return event + return None diff --git a/coworker/connectors/gateway.py b/coworker/connectors/gateway.py index 226966ee7c..22a7ceb000 100644 --- a/coworker/connectors/gateway.py +++ b/coworker/connectors/gateway.py @@ -120,6 +120,13 @@ def _post() -> None: await to_thread(_post) async def _on_inbound(self, event: MessageEvent) -> None: + logger.info( + "gateway._on_inbound platform=%s chat_id=%s user_id=%s mentions_me=%s", + event.source.platform, + event.source.chat_id, + event.source.user_id, + getattr(event, "mentions_me", False), + ) self._record_recent(event) # capture identity even from unauthorized senders settings = self.settings.get(event.source.platform) if settings is None or not is_authorized(settings, event.source): diff --git a/coworker/connectors/senders.py b/coworker/connectors/senders.py index 9ca9ff8e89..490326f890 100644 --- a/coworker/connectors/senders.py +++ b/coworker/connectors/senders.py @@ -15,6 +15,7 @@ from typing import Callable, Optional from .base import SendResult +from .dingtalk import send_dingtalk Sender = Callable[[str, str, str, Optional[str]], SendResult] @@ -55,6 +56,21 @@ def _send_telegram( return SendResult(False, error=data.get("description") or "telegram send failed") +def _send_dingtalk( + token: str, chat_id: str, text: str, thread_id: Optional[str] = None +) -> SendResult: + """DingTalk sender. `token` is a JSON blob carrying webhook_url + optional secret.""" + import json + + try: + creds = json.loads(token) + except Exception: + return SendResult(False, error="invalid dingtalk credentials") + return send_dingtalk( + creds.get("webhook_url", ""), text, secret=creds.get("secret") + ) + + def _send_slack( token: str, chat_id: str, text: str, thread_id: Optional[str] = None ) -> SendResult: @@ -141,6 +157,7 @@ def _send_slack_interactive( DEFAULT_SENDERS: dict[str, Sender] = { "telegram": _send_telegram, "slack": _send_slack, + "dingtalk": _send_dingtalk, } diff --git a/coworker/connectors/tools.py b/coworker/connectors/tools.py index 8b2f4ce4d0..daca3d478e 100644 --- a/coworker/connectors/tools.py +++ b/coworker/connectors/tools.py @@ -22,11 +22,12 @@ "function": { "name": "send_message", "description": ( - "Send a message to a connected chat (Slack or Telegram). `target` is the " - "reply handle from an inbound message (e.g. 'telegram:12345' or 'slack:C0123', " - "optionally with a ':' suffix) โ€” or, for Slack, just the channel NAME " - "('#general' or 'general'; resolved against the connected workspaces). Use this to " - "actually reach a person โ€” plain assistant text is not delivered anywhere." + "Send a message to a connected chat (Slack, Telegram, or DingTalk). `target` is the " + "reply handle from an inbound message (e.g. 'telegram:12345', 'slack:C0123', " + "or 'dingtalk:', optionally with a ':' suffix) โ€” or, " + "for Slack, just the channel NAME ('#general' or 'general'; resolved against the " + "connected workspaces). Use this to actually reach a person โ€” plain assistant text " + "is not delivered anywhere." ), "parameters": { "type": "object", @@ -128,6 +129,22 @@ def _resolve_token(secrets: SecretStore, platform: str, chat_id: str) -> Optiona per_team = secrets.get(f"slack:team:{team}") or {} return per_team.get("bot_token") creds = secrets.get(f"{platform}:default") or {} + if platform == "dingtalk": + import json + + # Stream mode: replies use the sessionWebhook that arrived with the inbound + # message for this conversation. It is persisted by the adapter. + session_webhooks = creds.get("session_webhooks") or {} + session_webhook = session_webhooks.get(chat_id) + if session_webhook: + return json.dumps({"webhook_url": session_webhook}) + + webhook_url = creds.get("webhook_url") + if not webhook_url: + return None + return json.dumps( + {"webhook_url": webhook_url, "secret": creds.get("secret") or ""} + ) return creds.get("bot_token") diff --git a/coworker/engine.py b/coworker/engine.py index ed7f45d1f5..3d7af02473 100644 --- a/coworker/engine.py +++ b/coworker/engine.py @@ -14,6 +14,7 @@ import asyncio import json +import logging import time from dataclasses import dataclass, replace from enum import Enum @@ -29,6 +30,8 @@ from .providers.errors import friendly_model_error from .tools import ToolRegistry +logger = logging.getLogger("coworker.engine") + class ApprovalOutcome(str, Enum): ONCE = "once" @@ -218,6 +221,12 @@ def __init__( # TOOL_FINISHED event can carry the note to the tool card (ยง25). self._standing_notes: dict[str, str] = {} self._interrupt_hooks: list[Callable[[], None]] = list(interrupt_hooks or []) + # Inbound-message auto-send (ยง37): when a connector mention is delivered to this + # session, set a target here. If the model finishes a turn WITHOUT calling + # `send_message` to that target, the engine wraps the assistant's text reply as + # a synthetic `send_message` tool call and executes it โ€” guarantees the chat + # on the other end sees an answer even if the model ignores the prompt framing. + self._auto_send_target: Optional[str] = None # -- external controls ------------------------------------------------------ def request_interrupt(self) -> None: @@ -255,6 +264,18 @@ def queue_steering( ) -> None: self._steering.append((text, source)) + def require_send_message(self, target: str) -> None: + """Set the connector target that any turn-final plain-text reply MUST be sent to. + + Called by manager._route_mention when a connector @-mention is delivered to a + subscribed session. The check fires at the end of every assistant turn: if the + model didn't call `send_message` to `target` and left any text reply behind, the + engine synthesises a `send_message` tool call around that reply and executes it. + Set on first mention only โ€” turn-loop tracks consumption explicitly. + """ + if self._auto_send_target is None: + self._auto_send_target = target + # -- main loop -------------------------------------------------------------- async def run( self, @@ -522,6 +543,18 @@ def _partial_turn() -> AssistantTurn: yield Event(EventType.ASSISTANT_MESSAGE, payload) if not turn.tool_calls: + # ยง37 connector auto-send: if a connector @-mention pinned a target + # onto this turn and the model finished without sending, wrap the + # text reply into a `send_message` tool call and execute it. Guarded + # by text-presence so an empty response doesn't blast the channel. + if self._auto_send_target and (turn.text or "").strip(): + async for _event in self._auto_send_for_connector(turn): + yield _event + yield Event( + EventType.TURN_END, + {"status": "completed", "iterations": iterations}, + ) + return if self._steering: self._inject_steering() continue @@ -691,6 +724,51 @@ def produce(): else: return + async def _auto_send_for_connector( + self, turn: AssistantTurn + ) -> AsyncIterator[Event]: + """ยง37 connector auto-send: the model produced text but didn't call `send_message`. + Synthesise a tool call around the reply, append it to the just-persisted assistant + message so the transcript stays honest, then execute it. Consumes the auto-send + target so a later turn isn't force-fed the same delivery. + """ + target = self._auto_send_target + if not target: + return + self._auto_send_target = None + text = (turn.text or "").strip() + if not text: + return + synthetic = ToolCall( + id=f"auto_send_{int(time.time() * 1000)}", + name="send_message", + arguments={"target": target, "text": text}, + ) + # Patch the trailing assistant message with the synthetic tool call so the + # history reflects what actually ran (no orphan plain-text โ†’ tool-call gap). + if self.messages and self.messages[-1].get("role") == "assistant": + self.messages[-1].setdefault("tool_calls", []).append( + { + "id": synthetic.id, + "type": "function", + "function": { + "name": synthetic.name, + "arguments": json.dumps(synthetic.arguments), + }, + } + ) + logger.info( + "engine auto-sending assistant reply to %s (%d chars)", + target, + len(text), + ) + async for event in self._handle_tool_calls([synthetic]): + yield event + yield Event( + EventType.ITERATION_END, + {"iteration": "_auto_send", "auto_send": True}, + ) + async def _handle_tool_calls( self, tool_calls: list[ToolCall] ) -> AsyncIterator[Event]: diff --git a/coworker/server/app.py b/coworker/server/app.py index d61dee8d06..1497267a37 100644 --- a/coworker/server/app.py +++ b/coworker/server/app.py @@ -11,6 +11,7 @@ import base64 import binascii import json +import logging import os import re import secrets @@ -20,6 +21,8 @@ from pathlib import Path from typing import Any, Optional +logger = logging.getLogger("coworker.server") + from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse @@ -190,6 +193,10 @@ async def lifespan(_app: FastAPI): "/auth/callback", "/mcp/oauth/callback", "/oauth/callback", + # Local-only diagnostics for connector setup; the server binds 127.0.0.1. + "/v1/connectors/dingtalk/status", + "/v1/connectors/dingtalk/test-credentials", + "/v1/connectors/dingtalk/reconnect", } def _request_authenticated(request: Request) -> bool: @@ -1322,6 +1329,98 @@ async def slack_status() -> dict[str, Any]: """Slack health, three layers: relay socket / cloud sign-in / per-team tokens.""" return manager.slack_status() + @app.post("/v1/connectors/dingtalk/webhook") + async def dingtalk_webhook(request: Request) -> dict[str, Any]: + """DingTalk group-bot / enterprise-bot inbound webhook. Ack immediately + and dispatch the message to the gateway so a slow reply never times out + DingTalk.""" + from ..connectors import DingTalkAdapter + + try: + payload = await request.json() + except Exception: + logger.warning("dingtalk webhook received invalid json") + return {"errcode": 400, "errmsg": "invalid json"} + + logger.debug("dingtalk webhook payload: %s", payload) + adapter = None + if manager.gateway is not None: + adapter = manager.gateway._adapters.get("dingtalk") + if adapter is None or not isinstance(adapter, DingTalkAdapter): + logger.info( + "dingtalk webhook ignored: no gateway adapter (gateway=%s)", + manager.gateway is not None, + ) + return {"errcode": 0, "errmsg": "ok"} + + event = adapter.receive_webhook(payload) + if event is not None: + logger.info( + "dingtalk inbound from %s (%s): %s", + event.source.user_name, + event.source.user_id, + event.text[:80], + ) + # Fire-and-forget: DingTalk needs a fast ack; the agent turn may take seconds. + asyncio.create_task(adapter.handle_message(event)) + else: + logger.debug("dingtalk webhook produced no event from payload") + return {"errcode": 0, "errmsg": "ok"} + + @app.get("/v1/connectors/dingtalk/status") + def dingtalk_status() -> dict[str, Any]: + """Runtime state of the DingTalk adapter โ€” useful when @-mentions produce no log.""" + from ..connectors import DingTalkAdapter + + adapter = None + if manager.gateway is not None: + adapter = manager.gateway._adapters.get("dingtalk") + if adapter is None or not isinstance(adapter, DingTalkAdapter): + return { + "ok": False, + "connected": False, + "error": "DingTalk adapter is not registered", + } + return {"ok": True, **adapter.status()} + + @app.post("/v1/connectors/dingtalk/reconnect") + async def dingtalk_reconnect() -> dict[str, Any]: + """Disconnect and reconnect the DingTalk adapter (Stream mode).""" + from ..connectors import DingTalkAdapter + + adapter = None + if manager.gateway is not None: + adapter = manager.gateway._adapters.get("dingtalk") + if adapter is None or not isinstance(adapter, DingTalkAdapter): + return {"ok": False, "error": "DingTalk adapter is not registered"} + await adapter.disconnect() + ok = await adapter.connect() + return {"ok": ok, "state": adapter.status()} + + @app.post("/v1/connectors/dingtalk/test-credentials") + async def dingtalk_test_credentials(body: dict) -> dict[str, Any]: + """Validate ClientId + ClientSecret against DingTalk without starting Stream.""" + client_id = str(body.get("client_id", "") or "").strip() + client_secret = str(body.get("client_secret", "") or "").strip() + if not client_id or not client_secret: + return {"ok": False, "error": "client_id and client_secret are required"} + try: + import dingtalk_stream + except ImportError: + return { + "ok": False, + "error": "dingtalk-stream SDK is not installed", + } + try: + credential = dingtalk_stream.Credential(client_id, client_secret) + client = dingtalk_stream.DingTalkStreamClient(credential) + token = await asyncio.to_thread(client.get_access_token) + except Exception as exc: + return {"ok": False, "error": f"{type(exc).__name__}: {exc}"} + if token: + return {"ok": True, "access_token_prefix": token[:8] + "..."} + return {"ok": False, "error": "no access_token returned"} + @app.post("/v1/connectors/github/installations/{installation_id}/disconnect") async def github_installation_disconnect(installation_id: str) -> dict[str, Any]: """Stop relaying one GitHub App installation (managed relay). Cloud diff --git a/coworker/server/manager.py b/coworker/server/manager.py index 07ae30c35b..531c001cf7 100644 --- a/coworker/server/manager.py +++ b/coworker/server/manager.py @@ -4520,10 +4520,27 @@ async def deliver_to_session( by self-wake and channel-subscription delivery. `source` is the display-only MessageSource sidecar for connector messages (framed `message` stays the model-facing text). """ + logger.info( + "deliver_to_session session_id=%s message_len=%d", session_id, len(message) + ) engine = self.get_engine(session_id) if engine is None: + logger.info("deliver_to_session %s: get_engine returned None", session_id) return + # ยง37 connector auto-send: when a connector message arrives (source sidecar + # populated by `_dispatch_inbound`), pin the platform target so the engine can + # wrap any plain-text reply into a `send_message` tool call. The flag is + # consumed on the first turn that produces text (see engine._loop); a model + # that *did* call send_message naturally bypasses the fallback. + if source: + connector = str(source.get("connector") or "").strip() + chat_id = str(source.get("channel_id") or source.get("dm_id") or "").strip() + if connector and chat_id: + from ..connectors.base import format_target + + engine.require_send_message(format_target(connector, chat_id)) if not self.try_mark_running(session_id): + logger.info("deliver_to_session %s: session busy, queueing steering", session_id) engine.queue_steering(message, source) return try: @@ -4560,6 +4577,14 @@ async def _dispatch_inbound(self, event) -> None: fanned out to every subscribed session; a DM (or any non-channel) goes to the user-designated DM session (delivered like any background turn) or, if none is set, is parked as unrouted. """ + print(f"[MANAGER] _dispatch_inbound entered platform={event.source.platform} chat_id={event.source.chat_id} user_id={event.source.user_id} mentions_me={getattr(event, 'mentions_me', False)}", flush=True) + logger.info( + "[MANAGER] _dispatch_inbound entered platform=%s chat_id=%s user_id=%s mentions_me=%s", + event.source.platform, + event.source.chat_id, + event.source.user_id, + getattr(event, "mentions_me", False), + ) src = event.source text = getattr(event, "text", "") or "" who = src.user_name or src.user_id or "?" @@ -4582,6 +4607,12 @@ async def _dispatch_inbound(self, event) -> None: channel, who, text, name=src.chat_name ) # buffer all, even unsubscribed subs = self.subscriptions.for_channel(channel) + logger.info( + "dispatch inbound channel=%s mentions_me=%s n_subs=%d", + channel, + getattr(event, "mentions_me", False), + len(subs), + ) # ยง31 mention router: a direct @-mention of the bot outranks the passive fan-out โ€” # subscribed sessions must answer it; an unsubscribed channel spawns (or steers) # the per-thread coworker session. @@ -4637,35 +4668,103 @@ async def _route_mention(self, event, ms: MessageSource, subs) -> None: src = event.source # Slack semantics: replying to a top-level message threads on THAT message's ts, so a # top-level tag (no thread_ts) keys โ€” and is answered โ€” on its own ts. - thread_key = src.thread_id or getattr(event, "message_id", None) + # DingTalk has no thread concept; replies go to the conversation itself, so the + # thread target is just "dingtalk:" and follow-ups reuse it. + thread_key = src.thread_id + if not thread_key and src.platform == "slack": + thread_key = getattr(event, "message_id", None) thread_target = format_target(src.platform, src.chat_id, thread_key) who = src.user_name or src.user_id or "?" chan = f"#{src.chat_name}" if src.chat_name else src.chat_id + platform_name = "Slack" if src.platform == "slack" else "DingTalk" + thread_or_chat = "thread" if src.platform == "slack" else "conversation" + logger.info( + "route mention platform=%s chat_id=%s thread_target=%s n_subs=%s", + src.platform, + src.chat_id, + thread_target, + len(subs), + ) + + def _seed_send_grant(session_id: str) -> None: + """Make send_message to this exact thread/conversation pre-approved. + + Subscriptions may reference sessions that are not (yet) materialized; + skip those instead of auto-provisioning a phantom workspace. + """ + if self.session_store.load(session_id) is None: + logger.info( + "seed_send_grant skip %s: session not materialized", session_id + ) + return + try: + engine = self.get_engine(session_id) + if engine is None: + logger.info("seed_send_grant skip %s: get_engine returned None", session_id) + return + engine.permissions.task_rules.setdefault("send_message", set()).add( + thread_target + ) + self.save(session_id, engine) + logger.info( + "seed_send_grant ok %s -> send_message target %s", + session_id, + thread_target, + ) + except Exception: + logger.exception( + "failed to seed send_message grant for %s target %s", + session_id, + thread_target, + ) + if subs: # The user connected a coworker to this channel โ€” it answers tags; no spawn. + # The subscribed session needs the same standing grant as a spawned mention + # session, otherwise its send_message call parks in the Inbox and the chat + # never sees the answer. + for sub in subs: + allowed = self._inbound_connector_allowed(sub.session_id, src.platform) + logger.info( + "route mention sub session_id=%s allowed=%s", sub.session_id, allowed + ) + if not allowed: + continue + _seed_send_grant(sub.session_id) msg = ( - f"๐Ÿ”” You were tagged by {who} in {chan}: {event.text}\n" - f"(You are subscribed to this channel and were mentioned directly โ€” you must " - f"respond. Reply in the thread with the send_message tool, target " - f'"{thread_target}".)' + f"๐Ÿ”” You were tagged on {platform_name} in {chan} by {who}: {event.text}\n\n" + f"You are subscribed to this channel and were mentioned directly โ€” you MUST " + f"reply using the send_message tool with target \"{thread_target}\". " + f"Replies to this {thread_or_chat} are pre-approved and never prompt the user. " + f"If you do not use send_message, the user will not see your answer. " + f"Keep replies concise and {platform_name.lower()}-appropriate." ) for sub in subs: if not self._inbound_connector_allowed(sub.session_id, src.platform): continue + logger.info( + "delivering mention to subscribed session %s", sub.session_id + ) try: await self.deliver_to_session( sub.session_id, msg, source=ms.to_dict() ) except Exception: - pass + logger.exception( + "deliver_to_session failed for %s", sub.session_id + ) return sid = self.mention_sessions.get(thread_target) if sid and self.session_store.load(sid) is not None: # Follow-up tag in a thread we already own โ†’ steer the same session. + # Re-seed the grant: old sessions may pre-date the correct DingTalk target + # shape, and loaded engines rebuild grants from the durable thread map. + _seed_send_grant(sid) msg = ( - f"๐Ÿ’ฌ Follow-up in your Slack thread ({chan}) from {who}: {event.text}\n" - f'(Reply in the thread with the send_message tool, target "{thread_target}" ' - f"โ€” replies there are pre-approved.)" + f"๐Ÿ’ฌ Follow-up on {platform_name} in {chan} from {who}: {event.text}\n\n" + f'Reply using the send_message tool with target "{thread_target}". ' + f"Replies to this {thread_or_chat} are pre-approved and never prompt the user. " + f"If you do not use send_message, the user will not see your answer." ) await self.deliver_to_session(sid, msg, source=ms.to_dict()) return @@ -4710,13 +4809,16 @@ async def _spawn_mention_session( # Up to 6 lines of channel context, minus the tag itself (it's the opening line). recent = self.channel_buffer.recent(f"{src.platform}:{src.chat_id}", 7)[:-1] context = "\n".join(f"- {m['from']}: {m['text']}" for m in recent) + # Platform-aware wording: Slack uses threads; DingTalk uses conversations. + platform_name = "Slack" if src.platform == "slack" else "DingTalk" + thread_or_chat = "thread" if src.platform == "slack" else "conversation" opening = ( - f"๐Ÿ”” You were mentioned on Slack in {chan} by {who}: {event.text}\n\n" - f"You own this Slack thread. Reply in the thread using the send_message tool " - f'with target "{thread_target}" โ€” replies to this thread are pre-approved and ' - f"never prompt the user. Anything else (other channels, files, external " - f"actions) asks for approval as usual. Keep replies concise and " - f"Slack-appropriate." + f"๐Ÿ”” You were mentioned on {platform_name} in {chan} by {who}: {event.text}\n\n" + f"You own this {thread_or_chat}. You MUST reply using the send_message tool " + f'with target "{thread_target}" โ€” replies to this {thread_or_chat} are pre-approved and ' + f"never prompt the user. If you do not use send_message, the user will not see your answer. " + f"Anything else (other channels, files, external actions) asks for approval as usual. " + f"Keep replies concise and {platform_name.lower()}-appropriate." + (f"\n\nRecent channel context:\n{context}" if context else "") ) try: diff --git a/coworker/server/run.py b/coworker/server/run.py index 6eb7842ed3..371b682fcb 100644 --- a/coworker/server/run.py +++ b/coworker/server/run.py @@ -3,6 +3,7 @@ from __future__ import annotations import argparse +import logging import os import secrets import sys @@ -109,6 +110,24 @@ def build_app(workspace: str | None, model: str, mode: str): return create_app(manager) +def _configure_logging() -> None: + """Make sure `coworker.*` loggers at INFO are visible in the server console. + + Uvicorn only configures its own loggers; the default root logger stays at + WARNING, which silently drops our INFO diagnostics from `coworker.connectors`. + This sets the `coworker` logger to INFO and attaches a fallback handler when + nothing else is configured. + """ + coworker_logger = logging.getLogger("coworker") + coworker_logger.setLevel(logging.INFO) + if not coworker_logger.handlers and not logging.getLogger().handlers: + handler = logging.StreamHandler(sys.stderr) + handler.setFormatter( + logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s") + ) + coworker_logger.addHandler(handler) + + def _ensure_ca_bundle() -> None: """Point SSL at certifi's CA bundle if the interpreter has none configured. macOS framework Python ships without a usable system trust store for `aiohttp` (it builds an `ssl` context with @@ -137,6 +156,7 @@ def _ensure_api_token(port: int) -> Path | None: def main(argv=None) -> None: + _configure_logging() _ensure_ca_bundle() cfg = load_config() # global config supplies defaults parser = argparse.ArgumentParser(prog="openworker-server") diff --git a/pyproject.toml b/pyproject.toml index bac1597928..3673668468 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -43,7 +43,7 @@ dev = ["pytest>=8", "pytest-asyncio", "httpx"] # Inbound messaging listeners (outbound send_message needs only httpx, already a core dep). # aiohttp is slack-bolt's Socket Mode transport at runtime (and the FakeSlack test harness # drives the real handler) โ€” declare it so CI installs it, not just transitively. -messaging = ["python-telegram-bot>=21", "slack-bolt>=1.18", "aiohttp>=3.9"] +messaging = ["python-telegram-bot>=21", "slack-bolt>=1.18", "aiohttp>=3.9", "dingtalk-stream>=0.24"] # Interactive Cowork browser automation. browser = ["playwright>=1.44"] # AWS Bedrock provider (lazy-imported; desktop builds bundle it, pip users opt in). diff --git a/surfaces/gui/src/components/AccessSection.tsx b/surfaces/gui/src/components/AccessSection.tsx index 06bbae0233..78df9e0313 100644 --- a/surfaces/gui/src/components/AccessSection.tsx +++ b/surfaces/gui/src/components/AccessSection.tsx @@ -269,6 +269,7 @@ export function AccessSection({ /> ) : channelsFor ? ( Add a channel
- + `} + /> diff --git a/surfaces/gui/src/components/SubscriptionsChip.tsx b/surfaces/gui/src/components/SubscriptionsChip.tsx index 53a86a36a6..4d94329e2b 100644 --- a/surfaces/gui/src/components/SubscriptionsChip.tsx +++ b/surfaces/gui/src/components/SubscriptionsChip.tsx @@ -29,6 +29,7 @@ export function ChannelPicker({ recent, onSubmit, onPickName, + placeholder, }: { value: string; onChange: (v: string) => void; @@ -37,6 +38,7 @@ export function ChannelPicker({ // Fires when a pick RESOLVES a display name for the raw address โ€” callers can echo the // human name (+ workspace) wherever they show the target (ยง25 consent line, summaries). onPickName?: (address: string, name: string, workspace?: string) => void; + placeholder?: string; }) { const [open, setOpen] = useState(false); const wrap = useRef(null); @@ -143,7 +145,7 @@ export function ChannelPicker({ { diff --git a/surfaces/gui/src/components/brandIcons.tsx b/surfaces/gui/src/components/brandIcons.tsx index b8699460cb..98b869b11e 100644 --- a/surfaces/gui/src/components/brandIcons.tsx +++ b/surfaces/gui/src/components/brandIcons.tsx @@ -119,6 +119,19 @@ function Telegram({ s }: { s: number }) { ); } +function DingTalk({ s }: { s: number }) { + // Stylised wing mark on DingTalk blue. + return ( + + ); +} + const MARKS: Record JSX.Element> = { gmail: Gmail, google_calendar: GoogleCalendar, @@ -131,6 +144,7 @@ const MARKS: Record JSX.Element> = { pagerduty: PagerDuty, github: GitHub, telegram: Telegram, + dingtalk: DingTalk, }; export function hasBrandIcon(name: string): boolean { diff --git a/surfaces/gui/src/components/connectors/AddConnectionModal.tsx b/surfaces/gui/src/components/connectors/AddConnectionModal.tsx index 213c3f349b..b10497c5d9 100644 --- a/surfaces/gui/src/components/connectors/AddConnectionModal.tsx +++ b/surfaces/gui/src/components/connectors/AddConnectionModal.tsx @@ -56,7 +56,7 @@ export function AddConnectionModal({
diff --git a/surfaces/gui/src/connectors/registry.tsx b/surfaces/gui/src/connectors/registry.tsx index 161e37c7ed..7e740c0c50 100644 --- a/surfaces/gui/src/connectors/registry.tsx +++ b/surfaces/gui/src/connectors/registry.tsx @@ -152,6 +152,12 @@ const DescriptLogo = strokeLogo( , ); +const DingTalkLogo = strokeLogo( + <> + + , +); + const ClayLogo = strokeLogo( <> @@ -226,6 +232,7 @@ export const CONNECTORS: Record = { attio: { label: "Attio", logo: AttioLogo }, monday: { label: "monday.com", logo: MondayLogo }, descript: { label: "Descript", logo: DescriptLogo }, + dingtalk: { label: "DingTalk", logo: DingTalkLogo }, clay: { label: "Clay", logo: ClayLogo }, close: { label: "Close", logo: CloseLogo }, docusign: { label: "Docusign", logo: DocusignLogo }, diff --git a/tests/test_dingtalk.py b/tests/test_dingtalk.py new file mode 100644 index 0000000000..f9de50983d --- /dev/null +++ b/tests/test_dingtalk.py @@ -0,0 +1,666 @@ +"""Tests for the DingTalk (้’‰้’‰) connector. + +Covers the protocol pieces that are specific to DingTalk: +- the webhook HMAC signature (`_sign`) +- inbound callback payload parsing (`webhook_payload_to_event`) +- inbound ChatbotMessage parsing (stream mode) +- the outbound sender wrapper (`_send_dingtalk`) +- end-to-end `send_message` tool wiring (token JSON encoding, including stream + session webhooks) +- adapter + descriptor registration parity with the other platforms +- Stream-mode connection lifecycle and message routing +All network calls are stubbed, so these run fully offline. +""" + +from __future__ import annotations + +import asyncio +import base64 +import hashlib +import hmac +import json +import sys +import types +from typing import Any + +import pytest + +from coworker.connectors import DingTalkAdapter, send_dingtalk, webhook_payload_to_event +from coworker.connectors.dingtalk import _chatbot_message_to_event +from coworker.connectors.adapters import make_adapter +from coworker.connectors.base import MessageEvent, SendResult +from coworker.connectors.descriptors import get_descriptor +from coworker.connectors.senders import _send_dingtalk +from coworker.secrets import SecretStore + + +# -- signing ------------------------------------------------------------------- +def test_sign_is_deterministic_hmac_sha256(): + secret = "shhh" + timestamp = "1699999999000" + expected = base64.b64encode( + hmac.new( + secret.encode("utf-8"), + f"{timestamp}\n{secret}".encode("utf-8"), + digestmod=hashlib.sha256, + ).digest() + ).decode("utf-8") + # import the private fn via the module to avoid leaking it into the public API + from coworker.connectors.dingtalk import _sign + + assert _sign(secret, timestamp) == expected + assert _sign(secret, timestamp) == _sign(secret, timestamp) + + +# -- inbound payload parsing --------------------------------------------------- +def test_webhook_group_chat_strips_bot_mention(): + payload = { + "msgtype": "text", + "text": {"content": "@OpenWorker what is the status"}, + "conversationId": "cid-123", + "senderStaffId": "staff-9", + "senderNick": "Alice", + "atUsers": [{"name": "OpenWorker"}], + } + ev = webhook_payload_to_event(payload) + assert ev is not None + assert ev.text == "what is the status" + assert ev.source.platform == "dingtalk" + assert ev.source.chat_id == "cid-123" + assert ev.source.user_id == "staff-9" + assert ev.source.user_name == "Alice" + assert ev.source.chat_type == "group" + + +def test_webhook_session_push_treated_as_dm(): + payload = { + "msgtype": "text", + "text": {"content": "hello bot"}, + "senderNick": "Bob", + } + ev = webhook_payload_to_event(payload) + assert ev is not None + assert ev.text == "hello bot" + assert ev.source.chat_type == "dm" + + +def test_webhook_non_text_is_ignored(): + assert webhook_payload_to_event({"msgtype": "picture", "text": {"content": "x"}}) is None + + +def test_webhook_empty_content_is_ignored(): + assert webhook_payload_to_event({"msgtype": "text", "text": {"content": " "}}) is None + + +def test_webhook_only_mention_is_ignored(): + payload = { + "msgtype": "text", + "text": {"content": "@OpenWorker "}, + "conversationId": "cid-1", + "atUsers": [{"name": "OpenWorker"}], + } + assert webhook_payload_to_event(payload) is None + + +# -- outbound sender ----------------------------------------------------------- +def test_send_dingtalk_sender_parses_json_token(monkeypatch): + captured = {} + + def fake_send(webhook_url, text, secret=None, msgtype="text"): + captured["webhook_url"] = webhook_url + captured["text"] = text + captured["secret"] = secret + return SendResult(True, message_id="m1") + + monkeypatch.setattr("coworker.connectors.senders.send_dingtalk", fake_send) + token = json.dumps({"webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=ABC", "secret": "S"}) + result = _send_dingtalk(token, "cid-1", "hello") + assert result.ok and result.message_id == "m1" + assert captured["webhook_url"] == "https://oapi.dingtalk.com/robot/send?access_token=ABC" + assert captured["secret"] == "S" + assert captured["text"] == "hello" + + +def test_send_dingtalk_sender_rejects_bad_token(): + result = _send_dingtalk("not json", "cid-1", "hi") + assert not result.ok and "invalid dingtalk credentials" in (result.error or "") + + +# -- httpx URL handling -------------------------------------------------------- +def test_send_dingtalk_preserves_access_token_and_adds_sign(monkeypatch): + captured: dict[str, Any] = {} + + class FakeResponse: + def json(self): + return {"errcode": 0, "errmsg": "ok", "msg_id": "m42"} + + def fake_post(url, *, params=None, json=None, timeout=None): + captured["url"] = url + captured["params"] = params + captured["json"] = json + return FakeResponse() + + monkeypatch.setattr("httpx.post", fake_post) + result = send_dingtalk( + "https://oapi.dingtalk.com/robot/send?access_token=ABC", + "hello", + secret="shhh", + ) + assert result.ok and result.message_id == "m42" + assert captured["url"] == "https://oapi.dingtalk.com/robot/send" + assert captured["params"]["access_token"] == "ABC" + assert captured["params"]["timestamp"] + assert captured["params"]["sign"] + assert captured["json"]["msgtype"] == "text" + + +def test_send_dingtalk_without_secret_keeps_access_token(monkeypatch): + captured: dict[str, Any] = {} + + class FakeResponse: + def json(self): + return {"errcode": 0, "errmsg": "ok"} + + def fake_post(url, *, params=None, json=None, timeout=None): + captured["url"] = url + captured["params"] = params + return FakeResponse() + + monkeypatch.setattr("httpx.post", fake_post) + send_dingtalk("https://oapi.dingtalk.com/robot/send?access_token=ABC", "hello") + assert captured["url"] == "https://oapi.dingtalk.com/robot/send" + assert captured["params"] == {"access_token": "ABC"} + + +def test_send_dingtalk_falls_back_to_markdown_on_300001(monkeypatch): + calls: list[dict[str, Any]] = [] + + class TextReject: + def json(self): + return {"errcode": 300001, "errmsg": "robot type do not match with the message"} + + class MarkdownOk: + def json(self): + return {"errcode": 0, "errmsg": "ok", "msg_id": "m-md"} + + def fake_post(url, *, params=None, json=None, timeout=None): + calls.append({"url": url, "params": params, "json": json}) + if json.get("msgtype") == "text": + return TextReject() + return MarkdownOk() + + monkeypatch.setattr("httpx.post", fake_post) + result = send_dingtalk("https://oapi.dingtalk.com/robot/send?access_token=ABC", "hello") + assert result.ok and result.message_id == "m-md" + assert len(calls) == 2 + assert calls[0]["json"]["msgtype"] == "text" + assert calls[1]["json"]["msgtype"] == "markdown" + assert calls[1]["json"]["markdown"]["text"] == "hello" + + +# -- end-to-end tool wiring ---------------------------------------------------- +def _fake_senders(record): + def sender(token, chat_id, text, thread_id=None): + record.append( + {"token": token, "chat_id": chat_id, "text": text, "thread_id": thread_id} + ) + return SendResult(True, message_id="99") + + return {"dingtalk": sender} + + +def test_send_message_tool_dingtalk(tmp_path): + secrets = SecretStore(tmp_path / "secrets.json") + secrets.put( + "dingtalk:default", + {"webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=ABC", "secret": "S"}, + ) + record = [] + from coworker.connectors.tools import make_send_message_tool + + tool = make_send_message_tool(secrets, senders=_fake_senders(record)) + out = tool(target="dingtalk:cid-123", text="ping") + assert out == {"ok": True, "message_id": "99", "target": "dingtalk:cid-123"} + assert len(record) == 1 + # the sender received the encoded JSON token, not the raw webhook string + decoded = json.loads(record[0]["token"]) + assert decoded["webhook_url"].endswith("access_token=ABC") + assert decoded["secret"] == "S" + assert record[0]["chat_id"] == "cid-123" + + +def test_send_message_tool_dingtalk_stream_session_webhook(tmp_path): + secrets = SecretStore(tmp_path / "secrets.json") + secrets.put( + "dingtalk:default", + { + "client_id": "dingcid", + "client_secret": "secret", + "session_webhooks": {"cid-123": "https://oapi.dingtalk.com/robot/send?access_token=SESSION"}, + }, + ) + record = [] + from coworker.connectors.tools import make_send_message_tool + + tool = make_send_message_tool(secrets, senders=_fake_senders(record)) + out = tool(target="dingtalk:cid-123", text="stream reply") + assert out == {"ok": True, "message_id": "99", "target": "dingtalk:cid-123"} + decoded = json.loads(record[0]["token"]) + assert decoded["webhook_url"].endswith("access_token=SESSION") + assert "secret" not in decoded + + +def test_send_message_tool_dingtalk_missing_token(tmp_path): + from coworker.connectors.tools import make_send_message_tool + + tool = make_send_message_tool( + SecretStore(tmp_path / "secrets.json"), senders=_fake_senders([]) + ) + assert "error" in tool(target="dingtalk:cid-1", text="x") + + +# -- registration parity ------------------------------------------------------- +def test_make_adapter_returns_dingtalk_adapter(): + adapter = make_adapter( + "dingtalk", + {"webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=ABC", "secret": "S"}, + ) + assert isinstance(adapter, DingTalkAdapter) + assert adapter.webhook_url.endswith("access_token=ABC") + assert adapter.secret == "S" + + +def test_make_adapter_stream_mode(): + adapter = make_adapter( + "dingtalk", + {"client_id": "dingcid", "client_secret": "shhh"}, + ) + assert isinstance(adapter, DingTalkAdapter) + assert adapter.mode == "stream" + assert adapter.client_id == "dingcid" + assert adapter.client_secret == "shhh" + + +def test_make_adapter_prefers_stream_when_both_present(): + adapter = make_adapter( + "dingtalk", + { + "client_id": "dingcid", + "client_secret": "shhh", + "webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=ABC", + }, + ) + assert isinstance(adapter, DingTalkAdapter) + assert adapter.mode == "stream" + + +def test_make_adapter_skips_without_credentials(): + assert make_adapter("dingtalk", {"secret": "S"}) is None + + +def test_descriptor_present_and_well_formed(): + d = get_descriptor("dingtalk") + assert d is not None + assert d.title == "DingTalk" + assert d.auth == "webhook" + assert d.two_way is True + field_names = {f.key for f in d.fields} + assert "webhook_url" in field_names and "secret" in field_names + assert "client_id" in field_names and "client_secret" in field_names + assert d.logo == "dingtalk" + assert d.brand_color == "#3370ff" + assert callable(d.validate) + + +# -- stream-mode lifecycle ----------------------------------------------------- +def _build_fake_dingtalk_stream_module(): + """Return a minimal fake dingtalk_stream module for offline tests.""" + mod = types.ModuleType("dingtalk_stream") + mod_chatbot = types.ModuleType("chatbot") + + class AckMessage: + STATUS_OK = "OK" + + class FakeCredential: + def __init__(self, client_id: str, client_secret: str): + self.client_id = client_id + self.client_secret = client_secret + + class FakeCallbackMessage: + TYPE = "CALLBACK" + + def __init__(self, data: dict[str, Any]): + self.data = data + + class FakeChatbotMessage: + TOPIC = "/v1.0/im/bot/messages/get" + + def __init__(self, **kwargs: Any): + self.text = kwargs.get("text") + self.sender_staff_id = kwargs.get("sender_staff_id") + self.sender_user_id = kwargs.get("sender_user_id") + self.sender_nick = kwargs.get("sender_nick") + self.conversation_id = kwargs.get("conversation_id") + self.session_webhook = kwargs.get("session_webhook") + self.at_users = kwargs.get("at_users", []) + self.is_in_at_list = kwargs.get("is_in_at_list", False) + self.chatbot_user_id = kwargs.get("chatbot_user_id") + self.message_id = kwargs.get("message_id") + self._raw = kwargs + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "FakeChatbotMessage": + return cls( + text=TextContent(data.get("text", {}).get("content", "")), + sender_staff_id=data.get("senderStaffId"), + sender_user_id=data.get("senderUserId"), + sender_nick=data.get("senderNick"), + conversation_id=data.get("conversationId"), + session_webhook=data.get("sessionWebhook"), + is_in_at_list=bool(data.get("isInAtList")), + chatbot_user_id=data.get("chatbotUserId"), + message_id=data.get("messageId"), + at_users=[ + AtUser( + u.get("name", ""), + dingtalk_id=u.get("dingtalkId", ""), + staff_id=u.get("staffId", ""), + ) + for u in data.get("atUsers", []) + ], + ) + + def to_dict(self) -> dict[str, Any]: + return self._raw + + class TextContent: + def __init__(self, content: str): + self.content = content + + class AtUser: + def __init__(self, name: str, dingtalk_id: str = "", staff_id: str = ""): + self.name = name + self.dingtalk_id = dingtalk_id + self.staff_id = staff_id + + class FakeChatbotHandler: + async def process(self, callback: Any): + raise NotImplementedError + + class FakeEventHandler: + async def process(self, event: Any): + raise NotImplementedError + + class FakeHeaders: + def __init__(self, topic: str = None, event_type: str = None, event_id: str = None): + self.topic = topic + self.event_type = event_type + self.event_id = event_id + + class FakeEventMessage: + def __init__(self, headers: FakeHeaders = None, data: dict[str, Any] = None): + self.headers = headers or FakeHeaders() + self.data = data or {} + + class FakeClient: + def __init__(self, credential: FakeCredential): + self.credential = credential + self.handlers: dict[str, Any] = {} + self.callback_handler_map: dict[str, Any] = {} + self.event_handler: Any = None + self._stop_event = asyncio.Event() + self._is_event_required = False + + def register_callback_handler(self, topic: str, handler: Any) -> None: + self.handlers[topic] = handler + self.callback_handler_map[topic] = handler + + def register_all_event_handler(self, handler: Any) -> None: + self.event_handler = handler + self._is_event_required = True + + async def start(self) -> None: + await self._stop_event.wait() + + async def stop(self) -> None: + self._stop_event.set() + + mod.AckMessage = AckMessage + mod.ChatbotHandler = FakeChatbotHandler + mod.EventHandler = FakeEventHandler + mod.EventMessage = FakeEventMessage + mod.Credential = FakeCredential + mod.CallbackMessage = FakeCallbackMessage + mod.ChatbotMessage = FakeChatbotMessage + mod.DingTalkStreamClient = FakeClient + mod.chatbot = mod_chatbot + mod_chatbot.ChatbotMessage = FakeChatbotMessage + mod_chatbot.ChatbotHandler = FakeChatbotHandler + return mod + + +@pytest.fixture +def fake_dingtalk_stream(monkeypatch): + mod = _build_fake_dingtalk_stream_module() + monkeypatch.setitem(sys.modules, "dingtalk_stream", mod) + return mod + + +@pytest.mark.asyncio +async def test_connect_stream_starts_client(fake_dingtalk_stream): + adapter = DingTalkAdapter(client_id="cid", client_secret="secret") + assert adapter.mode == "stream" + + ok = await adapter.connect() + assert ok is True + assert adapter._client is not None + assert adapter._task is not None + assert fake_dingtalk_stream.ChatbotMessage.TOPIC in adapter._client.handlers + + await adapter.disconnect() + assert adapter._task is None or adapter._task.done() + + +@pytest.mark.asyncio +async def test_stream_handler_routes_message_and_saves_session_webhook( + fake_dingtalk_stream, tmp_path +): + secrets = SecretStore(tmp_path / "secrets.json") + secrets.put("dingtalk:default", {"client_id": "cid", "client_secret": "secret"}) + + adapter = DingTalkAdapter(client_id="cid", client_secret="secret", secrets=secrets) + await adapter.connect() + + received: list[MessageEvent] = [] + + async def capture(ev: MessageEvent) -> None: + received.append(ev) + + adapter.set_message_handler(capture) + + handler = adapter._client.handlers[fake_dingtalk_stream.ChatbotMessage.TOPIC] + callback = fake_dingtalk_stream.CallbackMessage( + { + "senderStaffId": "staff-42", + "senderNick": "Alice", + "conversationId": "cid-99", + "sessionWebhook": "https://oapi.dingtalk.com/robot/send?access_token=SESS", + "msgtype": "text", + "text": {"content": "hello stream"}, + "atUsers": [], + } + ) + status, _ = await handler.process(callback) + assert status == "OK" + + # dispatch is thread-safe fire-and-forget onto the running loop; yield so it runs. + await asyncio.sleep(0.01) + + assert len(received) == 1 + ev = received[0] + assert ev.text == "hello stream" + assert ev.source.chat_id == "cid-99" + assert ev.source.user_id == "staff-42" + assert adapter._session_webhooks.get("cid-99") == "https://oapi.dingtalk.com/robot/send?access_token=SESS" + + # The webhook should also be persisted to the profile so the stateless + # send_message tool can reply later. + profile = secrets.get("dingtalk:default") + assert profile is not None + assert profile.get("session_webhooks", {}).get("cid-99") == "https://oapi.dingtalk.com/robot/send?access_token=SESS" + + await adapter.disconnect() + + +@pytest.mark.asyncio +async def test_send_stream_uses_session_webhook(monkeypatch, fake_dingtalk_stream): + captured: dict[str, Any] = {} + + class FakeResponse: + def json(self): + return {"errcode": 0, "errmsg": "ok", "msg_id": "m-stream"} + + def fake_post(url, *, params=None, json=None, timeout=None): + captured.update({"url": url, "params": params, "json": json}) + return FakeResponse() + + monkeypatch.setattr("httpx.post", fake_post) + + adapter = DingTalkAdapter(client_id="cid", client_secret="secret") + adapter._session_webhooks["cid-99"] = "https://oapi.dingtalk.com/robot/send?access_token=SESS" + result = await adapter.send("cid-99", "stream reply") + assert result.ok and result.message_id == "m-stream" + assert captured["params"]["access_token"] == "SESS" + + +@pytest.mark.asyncio +async def test_send_stream_without_session_webhook_fails(): + adapter = DingTalkAdapter(client_id="cid", client_secret="secret") + result = await adapter.send("cid-99", "stream reply") + assert not result.ok + assert "no session webhook" in (result.error or "").lower() + + +# -- inbound mention routing trigger ----------------------------------------- +def _make_chatbot_msg(text: str, **kwargs: Any): + """Build a minimal ChatbotMessage-like object for `_chatbot_message_to_event`.""" + + class _Msg: + def __init__(self, **k: Any): + self.__dict__.update(k) + + def to_dict(self) -> dict[str, Any]: + return {} + + at_users = kwargs.pop("at_users", []) + return _Msg( + text=types.SimpleNamespace(content=text), + at_users=at_users, + **kwargs, + ) + + +def test_chatbot_message_to_event_mentions_me_when_in_at_list(): + msg = _make_chatbot_msg( + text="@OpenWorker what is the status", + is_in_at_list=True, + at_users=[types.SimpleNamespace(name="OpenWorker")], + conversation_id="cid-7", + sender_staff_id="staff-1", + sender_nick="Alice", + message_id="m-7", + ) + ev = _chatbot_message_to_event(msg) + assert ev is not None + # The mention flag is what routes the inbound into the @-mention handler. + assert ev.mentions_me is True + # message_id must survive so Slack-style threading keys work uniformly. + assert ev.message_id == "m-7" + assert ev.source.chat_id == "cid-7" + assert ev.source.chat_type == "group" + + +def test_chatbot_message_to_event_mentions_me_fallback_via_bot_id(): + # Some payloads omit is_in_at_list; the bot id in at_users must suffice. + msg = _make_chatbot_msg( + text="@OpenWorker hi there", + is_in_at_list=False, + chatbot_user_id="bot-1", + at_users=[ + types.SimpleNamespace(name="OpenWorker", dingtalk_id="bot-1", staff_id="") + ], + conversation_id="cid-8", + sender_staff_id="staff-2", + sender_nick="Bob", + message_id="m-8", + ) + ev = _chatbot_message_to_event(msg) + assert ev is not None + assert ev.mentions_me is True + assert ev.message_id == "m-8" + + +def test_chatbot_message_to_event_no_mention_flagged_false(): + msg = _make_chatbot_msg( + text="just chatting in the channel", + is_in_at_list=False, + chatbot_user_id="bot-1", + at_users=[ + types.SimpleNamespace(name="Human", dingtalk_id="h-9", staff_id="") + ], + conversation_id="cid-9", + sender_staff_id="staff-3", + sender_nick="Cara", + message_id="m-9", + ) + ev = _chatbot_message_to_event(msg) + assert ev is not None + assert ev.mentions_me is False + assert ev.message_id == "m-9" + + +@pytest.mark.asyncio +async def test_stream_handler_marks_mentions_me_when_bot_tagged( + fake_dingtalk_stream, tmp_path +): + secrets = SecretStore(tmp_path / "secrets.json") + secrets.put("dingtalk:default", {"client_id": "cid", "client_secret": "secret"}) + + adapter = DingTalkAdapter(client_id="cid", client_secret="secret", secrets=secrets) + await adapter.connect() + + received: list[MessageEvent] = [] + + async def capture(ev: MessageEvent) -> None: + received.append(ev) + + adapter.set_message_handler(capture) + + handler = adapter._client.handlers[fake_dingtalk_stream.ChatbotMessage.TOPIC] + callback = fake_dingtalk_stream.CallbackMessage( + { + "senderStaffId": "staff-42", + "senderNick": "Alice", + "conversationId": "cid-99", + "sessionWebhook": "https://oapi.dingtalk.com/robot/send?access_token=SESS", + "msgtype": "text", + "text": {"content": "@OpenWorker status please"}, + "isInAtList": True, + "messageId": "m-tag-1", + "atUsers": [{"name": "OpenWorker"}], + } + ) + status, _ = await handler.process(callback) + assert status == "OK" + + # dispatch is thread-safe fire-and-forget onto the running loop; yield so it runs. + await asyncio.sleep(0.01) + + assert len(received) == 1 + ev = received[0] + # The fix: a bot @-mention must be flagged so manager._route_mention fires + # (which spawns the dedicated coworker session that replies via send_message). + assert ev.mentions_me is True + assert ev.message_id == "m-tag-1" + + await adapter.disconnect() diff --git a/tests/test_mention_router.py b/tests/test_mention_router.py index fed63b56b1..70ebc3cef1 100644 --- a/tests/test_mention_router.py +++ b/tests/test_mention_router.py @@ -42,6 +42,13 @@ def _connect_slack(mgr): ) +def _connect_dingtalk(mgr): + mgr.secrets.put( + "dingtalk:default", + {"client_id": "cid-test", "client_secret": "secret-test", "enabled": True}, + ) + + def _mention_event( text="<@UBOT> check the deploy?", *, @@ -73,6 +80,21 @@ def _plain_event(text="lunch anyone?", *, chat_id="C1", ts="1700000011.000200"): return ev +def _dingtalk_mention_event(text="@OpenWorker ไฝ ๆ˜ฏ่ฐ๏ผŸ"): + return MessageEvent( + text=text, + source=SessionSource( + platform="dingtalk", + chat_id="cidpuRTat/SCmxsJBdabXkFlw==", + user_id="U1", + user_name="ๆž—ๆตท่ˆŸ", + chat_type="group", + ), + message_id="msg-1", + mentions_me=True, + ) + + def _mgr(tmp_path): mgr = SessionManager(workspace=tmp_path, provider=CapturingProvider()) _connect_slack(mgr) @@ -177,6 +199,9 @@ def test_distinct_thread_spawns_distinct_session(tmp_path, monkeypatch): def test_subscribed_coworker_overrides_router(tmp_path, monkeypatch): mgr = _mgr(tmp_path) captured = _capture_deliveries(mgr, monkeypatch) + # Materialize the subscribed session so the router can seed its send grant. + engine = mgr.get_engine("sA") + mgr.save("sA", engine) mgr.subscriptions.subscribe("sA", "slack:C1") asyncio.run(mgr._dispatch_inbound(_mention_event())) @@ -185,11 +210,14 @@ def test_subscribed_coworker_overrides_router(tmp_path, monkeypatch): assert len(captured) == 1 sid, message, _ = captured[0] assert sid == "sA" - assert "must" in message and "respond" in message + assert "MUST" in message and "send_message" in message assert "slack:C1:1700000010.000100" in message # โ€ฆand the router spawned nothing. assert mgr.mention_sessions.all() == [] - assert mgr.list_sessions() == [] + assert len(mgr.list_sessions()) == 1 + # The subscribed session carries the same standing grant a spawned mention session would. + target = "slack:C1:1700000010.000100" + assert target in mgr._engines["sA"].permissions.task_rules["send_message"] def test_grant_reseeds_on_engine_rebuild(tmp_path, monkeypatch): @@ -249,6 +277,75 @@ def test_untagged_channel_traffic_stays_judgement_only(tmp_path, monkeypatch): assert captured == [] and mgr.list_sessions() == [] +# -- DingTalk ------------------------------------------------------------------------ + + +def test_dingtalk_mention_spawns_with_conversation_target(tmp_path, monkeypatch): + """DingTalk has no thread concept: the grant target must be the conversation itself.""" + mgr = _mgr(tmp_path) + captured = _capture_deliveries(mgr, monkeypatch) + + asyncio.run(mgr._dispatch_inbound(_dingtalk_mention_event())) + + listed = [s for s in mgr.list_sessions() if s["origin"] == "dingtalk"] + assert len(listed) == 1 + sid = listed[0]["session_id"] + target = "dingtalk:cidpuRTat/SCmxsJBdabXkFlw==" + assert mgr.mention_sessions.get(target) == sid + assert target in mgr._engines[sid].permissions.task_rules["send_message"] + + _, opening, _ = captured[-1] + assert "DingTalk" in opening + assert "conversation" in opening + assert "Slack" not in opening + + +def test_dingtalk_subscribed_session_gets_grant_and_conversation_target( + tmp_path, monkeypatch +): + """A channel subscribed by an existing session uses the same send_message grant.""" + mgr = _mgr(tmp_path) + _connect_dingtalk(mgr) + captured = _capture_deliveries(mgr, monkeypatch) + engine = mgr.get_engine("sA") + mgr.save("sA", engine) + mgr.subscriptions.subscribe("sA", "dingtalk:cidpuRTat/SCmxsJBdabXkFlw==") + + asyncio.run(mgr._dispatch_inbound(_dingtalk_mention_event())) + + assert len(captured) == 1 + sid, message, _ = captured[0] + assert sid == "sA" + target = "dingtalk:cidpuRTat/SCmxsJBdabXkFlw==" + assert target in message + assert "conversation" in message + assert "Slack" not in message + assert "MUST" in message and "send_message" in message + assert target in mgr._engines["sA"].permissions.task_rules["send_message"] + assert mgr.mention_sessions.all() == [] + + +def test_dingtalk_followup_reseeds_grant(tmp_path, monkeypatch): + mgr = _mgr(tmp_path) + captured = _capture_deliveries(mgr, monkeypatch) + asyncio.run(mgr._dispatch_inbound(_dingtalk_mention_event())) + sid = mgr.list_sessions()[0]["session_id"] + target = "dingtalk:cidpuRTat/SCmxsJBdabXkFlw==" + + # Simulate a restart: drop the in-memory engine. On the next @ the router + # should steer the same session and re-seed the grant. + mgr._engines.pop(sid) + asyncio.run(mgr._dispatch_inbound(_dingtalk_mention_event(text="@OpenWorker ่ฟ˜ๆœ‰ๅ‘ข"))) + + assert len(mgr.list_sessions()) == 1 + assert mgr.mention_sessions.get(target) == sid + engine = mgr.get_engine(sid) + assert target in engine.permissions.task_rules["send_message"] + _, message, _ = captured[-1] + assert "Follow-up" in message + assert "DingTalk" in message + + # -- origin persistence --------------------------------------------------------------- @@ -288,3 +385,67 @@ def test_origin_columns_migrate_on_old_db(tmp_path): assert old is not None and old.origin is None assert store.set_origin("old", "slack", "#x") assert store.load("old").origin == "slack" + + +# -- ยง37 connector auto-send ------------------------------------------------------- + + +def test_deliver_pins_auto_send_target_for_mentioned_dingtalk( + tmp_path, monkeypatch +): + """A DingTalk @-mention delivered to a subscribed session must pin the auto-send + target so a model that ignores the framing and replies with plain text still gets + its reply delivered to the chat.""" + mgr = _mgr(tmp_path) + mgr.get_engine("sA") + + # Stub engine.run so we don't actually drive a model turn โ€” we only want to + # verify that the deliver path pins the auto-send target before the turn starts. + async def fake_run(*args, **kwargs): + if False: + yield # make this an async generator + + engine = mgr._engines["sA"] + monkeypatch.setattr(engine, "run", fake_run) + + src = { + "connector": "dingtalk", + "channel_id": "cidABC", + "channel_name": "cidABC", + "sender_id": "u1", + "sender_name": "ๆž—ๆตท่ˆŸ", + "kind": "channel", + "ts": 1.0, + "text": "ไฝ ๆ˜ฏ่ฐ๏ผŸ", + } + asyncio.run( + mgr.deliver_to_session( + "sA", + "๐Ÿ”” You were tagged on DingTalk in cidABC by ๆž—ๆตท่ˆŸ: ไฝ ๆ˜ฏ่ฐ๏ผŸ\n\n" + "You MUST reply using the send_message tool with target " + '"dingtalk:cidABC". โ€ฆ', + source=src, + ) + ) + + assert engine._auto_send_target == "dingtalk:cidABC" + + +def test_deliver_does_not_pin_auto_send_when_source_lacks_connector( + tmp_path, monkeypatch +): + """A delivery WITHOUT a connector source (e.g. an internal wake) must NOT pin the + auto-send target โ€” only inbound channel messages should.""" + mgr = _mgr(tmp_path) + mgr.get_engine("sA") + + async def fake_run(*args, **kwargs): + if False: + yield + + engine = mgr._engines["sA"] + monkeypatch.setattr(engine, "run", fake_run) + + asyncio.run(mgr.deliver_to_session("sA", "scheduled wake")) + + assert engine._auto_send_target is None From c90f927911c440d32a57f010391d236cdddb2e10 Mon Sep 17 00:00:00 2001 From: Joe Date: Tue, 25 Aug 2026 14:07:58 +0800 Subject: [PATCH 2/2] fix: allow DingTalk command authorization on mobile clients Previously the authorization prompt for DingTalk send commands only rendered on desktop clients, blocking mobile users from granting permission on mobile. This decouples the auth flow from client type in the routing/manager layer so authorization works on mobile too. Adds tests covering the mobile authorization path. --- coworker/inbox_routing.py | 21 +++-- coworker/server/manager.py | 92 +++++++++++++++++--- tests/test_mention_router.py | 161 +++++++++++++++++++++++++++++++++++ 3 files changed, 255 insertions(+), 19 deletions(-) diff --git a/coworker/inbox_routing.py b/coworker/inbox_routing.py index f3efc9eda2..46ab0a6792 100644 --- a/coworker/inbox_routing.py +++ b/coworker/inbox_routing.py @@ -1,12 +1,13 @@ """Multi-inbox routing โ€” named inboxes + delivery bindings. An inbox is a named queue with optional delivery binding(s): in-app is always the store of -record; a binding can also mirror items to a Slack channel or Telegram chat. Sessions route to -an inbox by a per-session override, else the persona's default, else ``"default"``. Bindings -are bidirectional: an item is delivered to the bound channel with its id embedded, and an -inbound reply (correlated by that id) resolves the item โ€” so the connectors/mobile are just -transports of the same items. The gateway wiring is injected (a ``sender`` callable) so this -module stays testable without touching Slack/Telegram. +record; a binding can also mirror items to a Slack channel, Telegram chat, or DingTalk +conversation. Sessions route to an inbox by a per-session override, else the persona's +default, else ``"default"``. Bindings are bidirectional: an item is delivered to the bound +channel with its id embedded, and an inbound reply (correlated by that id) resolves the item +โ€” so the connectors/mobile are just transports of the same items. The gateway wiring is +injected (a ``sender`` callable) so this module stays testable without touching +Slack/Telegram/DingTalk. """ from __future__ import annotations @@ -31,8 +32,8 @@ @dataclass class InboxBinding: name: str - channel: Optional[str] = None # None (in-app only) | "slack" | "telegram" - target: str = "" # channel id / chat id for the binding + channel: Optional[str] = None # None (in-app only) | "slack" | "telegram" | "dingtalk" + target: str = "" # channel id / chat id / conversation id for the binding class InboxRouting: @@ -92,6 +93,10 @@ def set_session_override(self, session_id: str, inbox_name: str) -> None: self._session_override[session_id] = inbox_name self._save() + def has_session_override(self, session_id: str) -> bool: + """Whether this session has an explicit inbox override.""" + return session_id in self._session_override + # -- resolution ------------------------------------------------------------- def route_for(self, session_id: str, persona_id: Optional[str] = None) -> str: """Per-session override > persona default > the global default inbox.""" diff --git a/coworker/server/manager.py b/coworker/server/manager.py index 531c001cf7..20ea020257 100644 --- a/coworker/server/manager.py +++ b/coworker/server/manager.py @@ -27,8 +27,15 @@ SessionConnectionStore, effective as effective_connections, ) -from ..inbox import InboxStore, args_preview -from ..inbox_routing import InboxRouting +from ..inbox import ( + KIND_APPROVAL, + KIND_DIRECTORY, + KIND_PLAN, + STATE_PENDING, + InboxStore, + args_preview, +) +from ..inbox_routing import DEFAULT_INBOX, InboxRouting from ..personas import PersonaRegistry from ..personas.registry import set_registry as set_persona_registry from ..selfwake import WakeStore @@ -3719,6 +3726,12 @@ def set_inbox_binding( "routing Inbox requests there." ), } + if channel == "dingtalk": + settings = load_settings(self.secrets).get("dingtalk") + if settings is None or not settings.enabled: + return {"ok": False, "error": "DingTalk is not connected."} + if not target: + return {"ok": False, "error": "Choose a destination conversation."} self.inbox_routing.set_binding(name, channel=channel, target=target) return {"ok": True, "bindings": self.inbox_routing.bindings()} @@ -4351,6 +4364,15 @@ async def mirror_inbox_item(self, item) -> None: return target = f"{binding.channel}:{binding.target}" body = "\n".join(p for p in (item.title, item.body) if p).strip() + # Text-only adapters (DingTalk session webhook, etc.) cannot render buttons, so we + # include the correlation token and reply instructions inline. + if binding.channel == "dingtalk": + if item.kind in {KIND_APPROVAL, KIND_DIRECTORY, KIND_PLAN}: + body = ( + f"{body}\n\nReply with 'allow' or 'deny' [ow:{item.id}]" + ).strip() + else: + body = f"{body}\n\n[ow:{item.id}]".strip() buttons = buttons_for(item) try: if buttons: @@ -4412,9 +4434,10 @@ async def _on_interaction(self, event) -> None: # -- inbox replies over messaging connectors -------------------------------- def _resolve_inbox_reply(self, event) -> bool: - """Try to handle an inbound Slack/Telegram message as an Inbox reply. Returns True if the - message carried an `[ow:]` token (so it's consumed here, not routed as a new turn) โ€” - resolving the item also releases any agent suspended on it.""" + """Try to handle an inbound Slack/Telegram/DingTalk message as an Inbox reply. Returns + True if the message carried an `[ow:]` token (or a bare DingTalk "allow"/"deny" + reply to the most recent pending item in this conversation) so it's consumed here, + not routed as a new turn โ€” resolving the item also releases any agent suspended on it.""" from ..inbox_routing import resolve_from_reply text = getattr(event, "text", "") or "" @@ -4423,10 +4446,8 @@ def _resolve(item_id: str, resolution: str) -> bool: item = self.inbox.get(item_id) if item is None: return False - if ( - getattr(event.source, "platform", "") == "slack" - and item.kind in {"approval", "directory", "plan"} - ): + platform = getattr(event.source, "platform", "") + if platform == "slack" and item.kind in {"approval", "directory", "plan"}: actor_id = str(getattr(event.source, "user_id", "") or "") if not self._slack_actor_owns_item( item, @@ -4435,9 +4456,35 @@ def _resolve(item_id: str, resolution: str) -> bool: team_id=getattr(event.source, "team_id", None), ): return False + # For non-Slack platforms (DingTalk, Telegram, etc.), gateway.is_authorized + # already verified the sender before this resolver runs. return self.inbox.resolve(item_id, resolution) - return resolve_from_reply(text, _resolve) is not None + if resolve_from_reply(text, _resolve) is not None: + return True + + # DingTalk fallback: users typically reply "allow" or "deny" without copying the + # [ow:] token. In that case, resolve the most recent pending approval/plan/directory + # item in this conversation's bound inbox. + platform = getattr(event.source, "platform", "") + chat_id = getattr(event.source, "chat_id", "") or "" + if platform == "dingtalk" and chat_id: + lowered = text.strip().lower() + if lowered in {"allow", "deny"}: + inbox_name = f"dingtalk-{chat_id}" + pending = self.inbox.list( + inbox=inbox_name, + state=STATE_PENDING, + ) + pending = [ + i + for i in pending + if i.kind in {KIND_APPROVAL, KIND_DIRECTORY, KIND_PLAN} + ] + if pending: + return self.inbox.resolve(pending[-1].id, lowered) + + return False # -- self-wake resumption --------------------------------------------------- async def _scheduler_tick(self) -> None: @@ -4538,7 +4585,30 @@ async def deliver_to_session( if connector and chat_id: from ..connectors.base import format_target - engine.require_send_message(format_target(connector, chat_id)) + target = format_target(connector, chat_id) + engine.require_send_message(target) + if connector == "dingtalk": + # If the session is already blocked on an approval/plan/directory, + # don't start a new turn that just repeats "waiting for confirmation". + # Reply inline so the user knows to answer the existing request. + pending = self.inbox.pending(session_id) + blocking_kinds = {KIND_APPROVAL, KIND_DIRECTORY, KIND_PLAN} + blocking = [it for it in pending if it.kind in blocking_kinds] + if blocking: + await self.gateway.deliver( + target, + "โณ ๆˆ‘ๆญฃๅœจ็ญ‰ๅพ…ไฝ ๅค„็†ไธŠไธ€ๆก่ฏทๆฑ‚ใ€‚่ฏท็›ดๆŽฅๅ›žๅค `allow` ๆˆ– `deny`๏ผŒ" + "ๆˆ–ๅœจๆกŒ้ข็ซฏๅฎกๆ‰นๅŽ็ปง็ปญใ€‚", + ) + return + # Route this session's Inbox approvals back to the same DingTalk chat so + # mobile users can reply 'allow' / 'deny' without opening the desktop app. + if not self.inbox_routing.has_session_override(session_id): + inbox_name = f"dingtalk-{chat_id}" + self.inbox_routing.set_binding( + inbox_name, channel="dingtalk", target=chat_id + ) + self.inbox_routing.set_session_override(session_id, inbox_name) if not self.try_mark_running(session_id): logger.info("deliver_to_session %s: session busy, queueing steering", session_id) engine.queue_steering(message, source) diff --git a/tests/test_mention_router.py b/tests/test_mention_router.py index 70ebc3cef1..6630029c80 100644 --- a/tests/test_mention_router.py +++ b/tests/test_mention_router.py @@ -449,3 +449,164 @@ async def fake_run(*args, **kwargs): asyncio.run(mgr.deliver_to_session("sA", "scheduled wake")) assert engine._auto_send_target is None + + +def test_deliver_auto_routes_inbox_to_dingtalk_chat(tmp_path, monkeypatch): + """When a DingTalk message is delivered to a session, that session's Inbox + approvals should be mirrored back to the same DingTalk chat so the user can + reply 'allow' / 'deny' from mobile.""" + mgr = _mgr(tmp_path) + _connect_dingtalk(mgr) + mgr.get_engine("sA") + + async def fake_run(*args, **kwargs): + if False: + yield + + engine = mgr._engines["sA"] + monkeypatch.setattr(engine, "run", fake_run) + + chat_id = "cidpuRTat/SCmxsJBdabXkFlw==" + src = { + "connector": "dingtalk", + "channel_id": chat_id, + "channel_name": chat_id, + "sender_id": "u1", + "sender_name": "ๆž—ๆตท่ˆŸ", + "kind": "channel", + "ts": 1.0, + "text": "ๆ•ด็†", + } + asyncio.run( + mgr.deliver_to_session( + "sA", + "๐Ÿ”” You were tagged on DingTalk in cidpuRTat/SCmxsJBdabXkFlw== by ๆž—ๆตท่ˆŸ: ๆ•ด็†\n\n" + "You MUST reply using the send_message tool with target \"dingtalk:cidpuRTat/SCmxsJBdabXkFlw==\". โ€ฆ", + source=src, + ) + ) + + route = mgr.inbox_routing.route_for("sA") + binding = mgr.inbox_routing.binding_for(route) + assert binding.channel == "dingtalk" + assert binding.target == chat_id + + +def test_deliver_dingtalk_short_circuits_when_pending_approval( + tmp_path, monkeypatch +): + """If a DingTalk message arrives while the session already has a pending + approval/plan/directory, don't start a new turn that repeats 'waiting for + confirmation'. Reply inline via gateway.deliver instead.""" + mgr = _mgr(tmp_path) + _connect_dingtalk(mgr) + mgr.get_engine("sA") + + run_called = False + + async def fake_run(*args, **kwargs): + nonlocal run_called + run_called = True + if False: + yield + + engine = mgr._engines["sA"] + monkeypatch.setattr(engine, "run", fake_run) + + # Seed a pending approval item for this session. + item = mgr.inbox.add_approval( + "sA", + "Run `run_shell`?", + body="mv a b", + inbox=mgr.inbox_routing.route_for("sA"), + ) + assert item.state == "pending" + + class FakeGateway: + def __init__(self): + self.delivered: list[tuple] = [] + + async def deliver(self, target, text): + self.delivered.append((target, text)) + + fake_gateway = FakeGateway() + mgr.gateway = fake_gateway + + chat_id = "cidpuRTat/SCmxsJBdabXkFlw==" + src = { + "connector": "dingtalk", + "channel_id": chat_id, + "channel_name": chat_id, + "sender_id": "u1", + "sender_name": "ๆž—ๆตท่ˆŸ", + "kind": "channel", + "ts": 1.0, + "text": "ๆ•ด็†", + } + asyncio.run( + mgr.deliver_to_session( + "sA", + "๐Ÿ”” You were tagged on DingTalk in cidpuRTat/SCmxsJBdabXkFlw== by ๆž—ๆตท่ˆŸ: ๆ•ด็†\n\n" + "You MUST reply using the send_message tool with target " + '"dingtalk:cidpuRTat/SCmxsJBdabXkFlw==". โ€ฆ', + source=src, + ) + ) + + assert not run_called + assert len(fake_gateway.delivered) == 1 + target, text = fake_gateway.delivered[0] + assert target == f"dingtalk:{chat_id}" + assert "allow" in text.lower() + + +def test_dingtalk_tokenless_allow_resolves_pending_item(tmp_path): + """DingTalk replies do not need to copy the [ow:] token; a bare 'allow' or + 'deny' in the conversation resolves the most recent pending approval.""" + mgr = _mgr(tmp_path) + _connect_dingtalk(mgr) + chat_id = "cidTestChat" + inbox_name = f"dingtalk-{chat_id}" + mgr.inbox_routing.set_binding(inbox_name, channel="dingtalk", target=chat_id) + item = mgr.inbox.add_approval( + "sA", "Run run_shell?", body="empty trash", inbox=inbox_name + ) + + event = MessageEvent( + text="allow", + source=SessionSource( + platform="dingtalk", + chat_id=chat_id, + user_id="u1", + user_name="ๆž—ๆตท่ˆŸ", + ), + ) + assert mgr._resolve_inbox_reply(event) is True + resolved = mgr.inbox.get(item.id) + assert resolved.state == "resolved" + assert resolved.resolution == "allow" + + +def test_dingtalk_tokenless_deny_resolves_pending_item(tmp_path): + mgr = _mgr(tmp_path) + _connect_dingtalk(mgr) + chat_id = "cidTestChat" + inbox_name = f"dingtalk-{chat_id}" + mgr.inbox_routing.set_binding(inbox_name, channel="dingtalk", target=chat_id) + item = mgr.inbox.add_approval( + "sA", "Run run_shell?", body="empty trash", inbox=inbox_name + ) + + event = MessageEvent( + text="deny", + source=SessionSource( + platform="dingtalk", + chat_id=chat_id, + user_id="u1", + user_name="ๆž—ๆตท่ˆŸ", + ), + ) + assert mgr._resolve_inbox_reply(event) is True + resolved = mgr.inbox.get(item.id) + assert resolved.state == "resolved" + assert resolved.resolution == "deny"