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 f16afe8653..8314059960 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 @@ -39,6 +40,8 @@ from .providers.openai_provider import looks_like_unparsed_tool_call from .tools import ToolRegistry +logger = logging.getLogger("coworker.engine") + class ApprovalOutcome(str, Enum): ONCE = "once" @@ -237,6 +240,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: @@ -274,6 +283,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, @@ -541,6 +562,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 @@ -729,6 +762,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/inbox_routing.py b/coworker/inbox_routing.py index bcbf33f8ae..4c5ee85ee5 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 @@ -28,8 +29,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: @@ -89,6 +90,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/app.py b/coworker/server/app.py index 457c08110b..706b7679ee 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 @@ -191,6 +194,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: @@ -1327,6 +1334,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 130ede1f23..c0d99fd519 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 @@ -3742,6 +3749,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()} @@ -4374,6 +4387,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: @@ -4435,9 +4457,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 "" @@ -4446,10 +4469,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, @@ -4458,9 +4479,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: @@ -4543,10 +4590,50 @@ 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 + + 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) return try: @@ -4583,6 +4670,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 "?" @@ -4605,6 +4700,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. @@ -4660,35 +4761,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 @@ -4733,13 +4902,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 2ecf28f89b..8eb22a218c 100644 --- a/surfaces/gui/src/components/AccessSection.tsx +++ b/surfaces/gui/src/components/AccessSection.tsx @@ -271,6 +271,7 @@ export function AccessSection({ /> ) : channelsFor ? ( {tt("access.add_channel")}
- + `} + /> diff --git a/surfaces/gui/src/components/SubscriptionsChip.tsx b/surfaces/gui/src/components/SubscriptionsChip.tsx index 0c15d81716..a666e1bc20 100644 --- a/surfaces/gui/src/components/SubscriptionsChip.tsx +++ b/surfaces/gui/src/components/SubscriptionsChip.tsx @@ -30,6 +30,7 @@ export function ChannelPicker({ recent, onSubmit, onPickName, + placeholder, }: { value: string; onChange: (v: string) => void; @@ -38,6 +39,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); 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 617834053f..5bd24bb10d 100644 --- a/surfaces/gui/src/components/connectors/AddConnectionModal.tsx +++ b/surfaces/gui/src/components/connectors/AddConnectionModal.tsx @@ -60,7 +60,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..6630029c80 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,228 @@ 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 + + +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"