From b1efd6571b5a41ded662122c2efbd56b9b7a3695 Mon Sep 17 00:00:00 2001 From: CK Date: Mon, 6 Jul 2026 14:11:59 +0800 Subject: [PATCH] fix(feishu): route topic sends via reply API; remove invalid thread_id receive_id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Async-delegation completions and terminal background notifications re-enter the originating session as synthetic events with no om_ reply anchor. The Feishu adapter's fallback branch then sent these via receive_id_type=thread_id on the create-message API, which the Feishu server rejects with [99992402] field validation failed — thread_id is not a valid receive_id_type (only open_id/union_id/user_id/email/chat_id are). A message can only land in a topic through the reply API (reply_in_thread=true) against a real om_ id. The broken branch was added in #13077 (ff14666cd) and its regression test was a pure mock that only asserted the request was shaped as thread_id, never exercising the real API — a green-mock-hides-integration-bug case. Root fix — anchor everything on a stable om_ id threaded end-to-end: - Feishu inbound: populate source.message_id with the topic root (om_), so it flows into _SESSION_MESSAGE_ID and is captured by background watchers. - delegate_task(background=true): capture message_id before detaching onto the daemon worker thread (mirrors session_key capture) and carry it onto the completion event. - terminal text notifications: pass reply_to=message_id for thread routing. - _inject_watch_notification: fall back to the persisted session origin's message_id when the event lacks one. - Feishu adapter: remove the illegal receive_id_type=thread_id create branch; fall back to a top-level chat create with a warning when no anchor is available (strictly better than a hard send failure, and unreachable in normal operation once the routing layer populates anchors). Tests: - Replace the test that asserted receive_id_type=thread_id (it was freezing the bug) with reply-API-contract assertions + a no-anchor top-level fallback case. - Add Feishu inbound source.message_id tests (topic root + seed message). - Add async-delegation message_id propagation tests (single, batch, default). --- gateway/run.py | 15 +- plugins/platforms/feishu/adapter.py | 159 ++++++++++------- .../test_background_process_notifications.py | 46 ++++- tests/gateway/test_feishu.py | 139 ++++++++++++++- .../gateway/test_relay_delivery_followups.py | 16 +- .../test_stream_consumer_thread_routing.py | 165 +++++++++++++++--- tests/tools/test_async_delegation.py | 76 +++++++- tools/async_delegation.py | 33 ++-- 8 files changed, 539 insertions(+), 110 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index 657f706fd1aa..ceda63a87e75 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -24791,12 +24791,23 @@ async def _inject_watch_notification( parent_session_id = str(evt.get("parent_session_id") or "").strip() if parent_session_id: metadata["gateway_session_id"] = parent_session_id + # Resolve a reply anchor for the synthetic event. Prefer the event's + # explicit message_id (terminal watchers and async-delegation + # completions carry the triggering ``om_`` anchor from the + # session context). When that's missing (older background + # processes dispatched before the anchor was captured, or a + # session origin whose message_id wasn't populated), fall back to + # the persisted session-store origin's message_id — e.g. a Feishu + # thread root. This keeps topic/thread-capable platforms routing + # the re-entry message via the reply API instead of an invalid + # create-by-thread-id path. + _synth_msg_id = str(evt.get("message_id") or "").strip() or getattr(source, "message_id", None) or None synth_event = MessageEvent( text=synth_text, message_type=MessageType.TEXT, source=source, internal=True, - message_id=str(evt.get("message_id") or "").strip() or None, + message_id=_synth_msg_id, metadata=metadata, ) logger.info( @@ -25661,6 +25672,7 @@ async def _run_process_watcher(self, watcher: dict) -> None: await adapter.send( chat_id, message_text, + reply_to=message_id, metadata=_non_conversational_metadata(send_meta, platform=platform_name), ) except Exception as e: @@ -25692,6 +25704,7 @@ async def _run_process_watcher(self, watcher: dict) -> None: await adapter.send( chat_id, message_text, + reply_to=message_id, metadata=_non_conversational_metadata(send_meta, platform=platform_name), ) except Exception as e: diff --git a/plugins/platforms/feishu/adapter.py b/plugins/platforms/feishu/adapter.py index 5b2e4ec4bd17..e43d5d28cebd 100644 --- a/plugins/platforms/feishu/adapter.py +++ b/plugins/platforms/feishu/adapter.py @@ -3346,14 +3346,29 @@ async def _process_inbound_message( if hint: text = f"{hint}\n\n{text}" if text else hint - thread_id = getattr(message, "thread_id", None) or getattr(message, "root_id", None) or None + root_id = getattr(message, "root_id", None) + thread_id = getattr(message, "thread_id", None) or root_id or None reply_to_message_id = ( getattr(message, "parent_id", None) or getattr(message, "upper_message_id", None) - or getattr(message, "root_id", None) + or root_id or None ) reply_to_text = await self._fetch_message_text(reply_to_message_id) if reply_to_message_id else None + # Feishu has no "send by thread_id" API — a message lands in a topic + # only via the reply API with reply_in_thread=true against an ``om_`` + # message id. Thread replies on real inbound messages route off + # ``event.reply_to_message_id``; synthetic / resumed sends (async + # delegation completions, terminal background notifications, cron + # deliveries) only see ``event.message_id`` plus ``source.message_id`` + # (the latter is persisted on the session origin and rehydrated by the + # gateway). Populate ``source.message_id`` with a stable thread anchor + # — the topic root when present, otherwise the message itself (which + # IS the root for a seed message) — so every downstream path that + # resolves a reply anchor for a Feishu thread can find a valid ``om_`` + # id instead of falling into an invalid ``receive_id_type=thread_id`` + # create branch. + thread_reply_anchor = root_id or message_id sender_primary = ( getattr(sender_id, "open_id", None) @@ -3385,6 +3400,7 @@ async def _process_inbound_message( thread_id=thread_id, user_id_alt=sender_profile["user_id_alt"], is_bot=is_bot, + message_id=thread_reply_anchor, ) normalized = MessageEvent( text=text, @@ -4747,43 +4763,51 @@ async def _send_uploaded_file_message( metadata=metadata, ) else: + payload = json.dumps({"file_key": file_key}, ensure_ascii=False) + send_reply_to = reply_to + resolved_thread_anchor = False + if ( + resolved_message_type == "audio" + and (metadata or {}).get("thread_id") + and not send_reply_to + ): + # Audio previously relied on the invalid thread_id create + # request failing with 99992402 before resolving a real + # om_ reply anchor. Resolve first now that anchorless + # threaded sends correctly avoid that invalid API call. + resolved_thread_anchor = True + send_reply_to = (metadata or {}).get("reply_to_message_id") + if not send_reply_to: + send_reply_to = await self._fetch_last_message_in_thread( + (metadata or {}).get("thread_id") + ) + if send_reply_to: + logger.info("[Feishu] Audio: sending via reply API in thread") + message_response = await self._feishu_send_with_retry( chat_id=chat_id, msg_type=resolved_message_type, - payload=json.dumps({"file_key": file_key}, ensure_ascii=False), - reply_to=reply_to, - metadata=metadata, + payload=payload, + reply_to=send_reply_to, + # No valid thread anchor means there is no legal threaded + # request. Send top-level directly instead of first + # emitting receive_id_type=thread_id and waiting for the + # server to reject it. + metadata=metadata if send_reply_to or not resolved_thread_anchor else None, ) - # Audio messages may fail with 99992402 when using thread_id routing. - # Try replying to the last message in the thread, then fall back to chat_id. - if (not self._response_succeeded(message_response) - and getattr(message_response, "code", None) == 99992402 - and resolved_message_type == "audio" - and (metadata or {}).get("thread_id")): - # Try reply API with thread_id as reply anchor - thread_msg_id = (metadata or {}).get("reply_to_message_id") - if not thread_msg_id: - thread_msg_id = await self._fetch_last_message_in_thread( - (metadata or {}).get("thread_id") - ) - if thread_msg_id: - logger.info("[Feishu] Audio: retrying via reply API in thread") - message_response = await self._feishu_send_with_retry( - chat_id=chat_id, - msg_type=resolved_message_type, - payload=json.dumps({"file_key": file_key}, ensure_ascii=False), - reply_to=thread_msg_id, - metadata=metadata, - ) - if not self._response_succeeded(message_response): - logger.warning("[Feishu] Audio send failed in thread, retrying with chat_id") - message_response = await self._feishu_send_with_retry( - chat_id=chat_id, - msg_type=resolved_message_type, - payload=json.dumps({"file_key": file_key}, ensure_ascii=False), - reply_to=None, - metadata=None, - ) + if ( + resolved_thread_anchor + and send_reply_to + and not self._response_succeeded(message_response) + ): + logger.warning("[Feishu] Audio send failed in thread, retrying with chat_id") + message_response = await self._feishu_send_with_retry( + chat_id=chat_id, + msg_type=resolved_message_type, + payload=payload, + reply_to=None, + metadata=None, + ) return self._finalize_send_result(message_response, "file send failed") except Exception as exc: logger.error("[Feishu] Failed to send file %s: %s", file_path, exc, exc_info=True) @@ -4834,34 +4858,45 @@ async def _send_raw_message( request = self._build_reply_message_request(effective_reply_to, body) return await self._run_blocking(self._client.im.v1.message.reply, request) - # For topic/thread messages that fell back from reply→create, use - # thread_id as receive_id so the message lands in the topic instead of - # the main chat. - _thread_id = (metadata or {}).get("thread_id") - if _thread_id: - body = self._build_create_message_body( - receive_id=_thread_id, - msg_type=msg_type, - content=payload, - uuid_value=str(uuid.uuid4()), - ) - request = self._build_create_message_request("thread_id", body) - else: - receive_id = chat_id - receive_id_type = "chat_id" - if chat_id.startswith("feishu_user_id:"): - receive_id = chat_id.split(":", 1)[1] - receive_id_type = "user_id" - elif chat_id.startswith("ou_"): - receive_id_type = "open_id" - - body = self._build_create_message_body( - receive_id=receive_id, - msg_type=msg_type, - content=payload, - uuid_value=str(uuid.uuid4()), + # No reply anchor available. Feishu's create-message API only + # accepts receive_id_type in {open_id, union_id, user_id, email, + # chat_id} — there is NO ``thread_id`` receive_id_type, so a topic + # message cannot be created by threading off the ``omt_`` thread id. + # Landing in a topic requires the reply API above against a real + # ``om_`` message id. When a threaded send reaches this point anyway + # (a synthetic / resumed event whose source.message_id and metadata + # both lacked an anchor), fall back to a top-level chat create and + # warn loudly rather than emitting an invalid ``receive_id_type= + # thread_id`` request that the server rejects with + # ``[99992402] field validation failed``. Thread context is lost on + # this fallback, which is strictly better than a hard send failure — + # the routing layer is expected to keep source.message_id populated + # so this branch stays unreached in normal operation. + if (metadata or {}).get("thread_id") and not effective_reply_to: + logger.warning( + "[Feishu] Thread send with no reply anchor for chat %s thread %s; " + "falling back to top-level chat send (thread context will be lost). " + "Ensure the inbound source.message_id and async-delegation / " + "terminal notification message_id are populated so threaded " + "sends route via the reply API.", + chat_id, + (metadata or {}).get("thread_id"), ) - request = self._build_create_message_request(receive_id_type, body) + receive_id = chat_id + receive_id_type = "chat_id" + if chat_id.startswith("feishu_user_id:"): + receive_id = chat_id.split(":", 1)[1] + receive_id_type = "user_id" + elif chat_id.startswith("ou_"): + receive_id_type = "open_id" + + body = self._build_create_message_body( + receive_id=receive_id, + msg_type=msg_type, + content=payload, + uuid_value=str(uuid.uuid4()), + ) + request = self._build_create_message_request(receive_id_type, body) return await self._run_blocking(self._client.im.v1.message.create, request) @staticmethod diff --git a/tests/gateway/test_background_process_notifications.py b/tests/gateway/test_background_process_notifications.py index 76941bb71064..fb2dff4e44c8 100644 --- a/tests/gateway/test_background_process_notifications.py +++ b/tests/gateway/test_background_process_notifications.py @@ -56,7 +56,7 @@ def _build_runner(monkeypatch, tmp_path, mode: str) -> GatewayRunner: return runner -def _watcher_dict(session_id="proc_test", thread_id=""): +def _watcher_dict(session_id="proc_test", thread_id="", message_id=""): d = { "session_id": session_id, "check_interval": 0, @@ -65,6 +65,8 @@ def _watcher_dict(session_id="proc_test", thread_id=""): } if thread_id: d["thread_id"] = thread_id + if message_id: + d["message_id"] = message_id return d @@ -193,6 +195,7 @@ async def test_inject_watch_notification_routes_from_session_store_origin(monkey thread_id="42", user_id="123", user_name="Emiliyan", + message_id="om_thread_root", ) ) @@ -212,6 +215,7 @@ async def test_inject_watch_notification_routes_from_session_store_origin(monkey assert synth_event.source.thread_id == "42" assert synth_event.source.user_id == "123" assert synth_event.source.user_name == "Emiliyan" + assert synth_event.message_id == "om_thread_root" @pytest.mark.asyncio @@ -431,13 +435,51 @@ async def _instant_sleep(*_a, **_kw): runner = _build_runner(monkeypatch, tmp_path, "concise") adapter = runner.adapters[Platform.TELEGRAM] - await runner._run_process_watcher(_watcher_dict()) + await runner._run_process_watcher( + _watcher_dict(thread_id="omt_topic", message_id="om_thread_root") + ) adapter.send.assert_awaited_once() sent_text = adapter.send.await_args.args[1] assert sent_text.startswith("✅ Background task finished") assert "Here's the final output" not in sent_text assert "5000" not in sent_text + assert adapter.send.await_args.kwargs["reply_to"] == "om_thread_root" + + +@pytest.mark.asyncio +async def test_all_mode_threads_interim_and_final_notifications(monkeypatch, tmp_path): + """Both direct watcher send paths preserve the captured reply anchor.""" + import tools.process_registry as pr_module + + running = SimpleNamespace( + output_buffer="building\n", exited=False, exit_code=None, + command="make", started_at=None, + ) + done = SimpleNamespace( + output_buffer="building\ndone\n", exited=True, exit_code=0, + command="make", started_at=None, + ) + monkeypatch.setattr( + pr_module, "process_registry", _FakeRegistry([running, done], consumed=False) + ) + + async def _instant_sleep(*_a, **_kw): + pass + monkeypatch.setattr(asyncio, "sleep", _instant_sleep) + + runner = _build_runner(monkeypatch, tmp_path, "all") + adapter = runner.adapters[Platform.TELEGRAM] + + await runner._run_process_watcher( + _watcher_dict(thread_id="omt_topic", message_id="om_thread_root") + ) + + assert adapter.send.await_count == 2 + assert all( + call.kwargs["reply_to"] == "om_thread_root" + for call in adapter.send.await_args_list + ) @pytest.mark.asyncio diff --git a/tests/gateway/test_feishu.py b/tests/gateway/test_feishu.py index a923ea5f5d46..c006e2065e56 100644 --- a/tests/gateway/test_feishu.py +++ b/tests/gateway/test_feishu.py @@ -799,6 +799,144 @@ def test_extract_text_message_starting_with_slash_becomes_command(self): self.assertEqual(event.message_type.value, "command") self.assertEqual(event.text, "/help test") + def test_inbound_thread_message_populates_source_message_id_anchor(self): + """A topic/thread inbound message must populate source.message_id + with a stable om_ thread anchor (the topic root) so synthetic / + resumed sends (async-delegation completions, terminal background + notifications) route into the topic via the reply API instead of an + invalid create-by-thread-id path.""" + from gateway.config import PlatformConfig + from plugins.platforms.feishu.adapter import FeishuAdapter + + adapter = FeishuAdapter(PlatformConfig()) + adapter._dispatch_inbound_event = AsyncMock() + adapter.get_chat_info = AsyncMock( + return_value={"chat_id": "oc_chat", "name": "Group", "type": "group"} + ) + adapter._resolve_sender_profile = AsyncMock( + return_value={"user_id": "ou_user", "user_name": "张三", "user_id_alt": None} + ) + adapter._fetch_message_text = AsyncMock(return_value=None) + message = SimpleNamespace( + chat_id="oc_chat", + thread_id="omt_topic_abc", + root_id="om_root_msg", + parent_id=None, + upper_message_id=None, + message_type="text", + content='{"text":"hi in topic"}', + message_id="om_user_msg", + ) + + asyncio.run( + adapter._process_inbound_message( + data=SimpleNamespace(event=SimpleNamespace(message=message)), + message=message, + sender_id=SimpleNamespace(open_id="ou_user", user_id=None, union_id=None), + is_bot=False, + chat_type="group", + message_id="om_user_msg", + ) + ) + + event = adapter._dispatch_inbound_event.await_args.args[0] + # source.message_id carries the topic root, not the omt_ thread id. + self.assertEqual(event.source.thread_id, "omt_topic_abc") + self.assertEqual(event.source.message_id, "om_root_msg") + # event.reply_to_message_id is unchanged — still the root for context. + self.assertEqual(event.reply_to_message_id, "om_root_msg") + + def test_inbound_thread_seed_message_populates_source_message_id_self(self): + """A seed message (the first message of a new topic, no root_id yet) + populates source.message_id with the message itself — it IS the root.""" + from gateway.config import PlatformConfig + from plugins.platforms.feishu.adapter import FeishuAdapter + + adapter = FeishuAdapter(PlatformConfig()) + adapter._dispatch_inbound_event = AsyncMock() + adapter.get_chat_info = AsyncMock( + return_value={"chat_id": "oc_chat", "name": "Group", "type": "group"} + ) + adapter._resolve_sender_profile = AsyncMock( + return_value={"user_id": "ou_user", "user_name": "张三", "user_id_alt": None} + ) + adapter._fetch_message_text = AsyncMock(return_value=None) + message = SimpleNamespace( + chat_id="oc_chat", + thread_id="omt_topic_new", + root_id=None, + parent_id=None, + upper_message_id=None, + message_type="text", + content='{"text":"new topic"}', + message_id="om_seed_msg", + ) + + asyncio.run( + adapter._process_inbound_message( + data=SimpleNamespace(event=SimpleNamespace(message=message)), + message=message, + sender_id=SimpleNamespace(open_id="ou_user", user_id=None, union_id=None), + is_bot=False, + chat_type="group", + message_id="om_seed_msg", + ) + ) + + event = adapter._dispatch_inbound_event.await_args.args[0] + self.assertEqual(event.source.message_id, "om_seed_msg") + + def test_audio_thread_without_anchor_resolves_message_before_send(self): + """Removing create-by-thread-id must not bypass the audio fallback. + + Audio sends without a captured anchor resolve a real om_ message first + and enter the reply API path instead of succeeding as a top-level chat + create before the old 99992402 retry can run. + """ + from plugins.platforms.feishu.adapter import FeishuAdapter + + adapter = Mock(spec=FeishuAdapter) + adapter._client = Mock() + adapter._resolve_outbound_file_routing.return_value = ("opus", "audio") + adapter._get_audio_duration_ms.return_value = 250 + adapter._build_file_upload_body.return_value = object() + adapter._build_file_upload_request.return_value = object() + adapter._run_blocking = AsyncMock( + return_value=SimpleNamespace(success=lambda: True) + ) + adapter._extract_response_field.return_value = "file_key" + adapter._fetch_last_message_in_thread = AsyncMock( + return_value="om_last_in_thread" + ) + adapter._feishu_send_with_retry = AsyncMock( + return_value=SimpleNamespace(success=lambda: True) + ) + adapter._response_succeeded.return_value = True + adapter._finalize_send_result.return_value = SimpleNamespace(success=True) + + with tempfile.TemporaryDirectory() as tmp_dir: + audio_path = Path(tmp_dir) / "voice.ogg" + audio_path.write_bytes(b"opus") + result = asyncio.run( + FeishuAdapter._send_uploaded_file_message( + adapter, + chat_id="oc_main_chat", + file_path=str(audio_path), + reply_to=None, + metadata={"thread_id": "omt_topic_abc"}, + outbound_message_type="audio", + ) + ) + + self.assertTrue(result.success) + adapter._fetch_last_message_in_thread.assert_awaited_once_with( + "omt_topic_abc" + ) + adapter._feishu_send_with_retry.assert_awaited_once() + send_kwargs = adapter._feishu_send_with_retry.await_args.kwargs + self.assertEqual(send_kwargs["reply_to"], "om_last_in_thread") + self.assertEqual(send_kwargs["metadata"], {"thread_id": "omt_topic_abc"}) + @patch.dict(os.environ, {}, clear=True) def test_extract_text_file_injects_content(self): from gateway.config import PlatformConfig @@ -2466,4 +2604,3 @@ def test_chat_locks_is_ordered_dict(self): adapter = self._make_adapter() self.assertIsInstance(adapter._chat_locks, _collections.OrderedDict) - diff --git a/tests/gateway/test_relay_delivery_followups.py b/tests/gateway/test_relay_delivery_followups.py index 6982361a46ab..edb14ea6a54e 100644 --- a/tests/gateway/test_relay_delivery_followups.py +++ b/tests/gateway/test_relay_delivery_followups.py @@ -375,13 +375,14 @@ async def test_disconnect_idempotent_second_pass(): # --------------------------------------------------------------------------- -# 6. Durable routing origin: scope_id survives dispatch -> restart -> replay +# 6. Durable routing origin survives dispatch -> restart -> replay # --------------------------------------------------------------------------- -def test_durable_dispatch_persists_and_recovers_scope_id(tmp_path, monkeypatch): +def test_durable_dispatch_persists_and_recovers_routing_origin(tmp_path, monkeypatch): """End-to-end restart shape: dispatch with a scoped session context bound, simulate owner death, recover — the recovered completion event must carry - scope_id/user_id, and the reconstructed SessionSource must prime them.""" + scope_id/user_id/message_id, and the reconstructed event must preserve + both relay identity and the platform reply anchor.""" import tools.async_delegation as ad from gateway.session_context import clear_session_vars, set_session_vars @@ -394,6 +395,7 @@ def test_durable_dispatch_persists_and_recovers_scope_id(tmp_path, monkeypatch): chat_type="group", user_id="U9", scope_id="G777", + message_id="om_thread_root", session_key="agent:main:discord:group:C123:U9", ) try: @@ -410,6 +412,7 @@ def test_durable_dispatch_persists_and_recovers_scope_id(tmp_path, monkeypatch): assert record.get("scope_id") == "G777", ( "dispatch-time capture must snapshot HERMES_SESSION_SCOPE_ID" ) + assert record.get("message_id") == "om_thread_root" ad._persist_dispatch(record) finally: clear_session_vars(tokens) @@ -432,6 +435,9 @@ def test_durable_dispatch_persists_and_recovers_scope_id(tmp_path, monkeypatch): "relay egress would be declined by the connector's tenant guard" ) assert evt.get("user_id") == "U9" + assert evt.get("message_id") == "om_thread_root", ( + "recovered completion event lost the platform reply anchor" + ) # The gateway-side fallback reconstruction must carry it into the source. runner = _fallback_runner() @@ -441,7 +447,7 @@ def test_durable_dispatch_persists_and_recovers_scope_id(tmp_path, monkeypatch): assert source.user_id == "U9" -def test_live_completion_event_carries_scope_id(tmp_path, monkeypatch): +def test_live_completion_event_carries_routing_origin(tmp_path, monkeypatch): """The live (non-restart) completion event must carry the dispatch-time routing origin too, so priming works even when the in-memory source cache was evicted.""" @@ -452,6 +458,7 @@ def test_live_completion_event_carries_scope_id(tmp_path, monkeypatch): "session_key": "agent:main:discord:group:C123:U9", "scope_id": "G777", "user_id": "U9", + "message_id": "om_thread_root", "goal": "g", "dispatched_at": 100.0, "completed_at": 101.0, @@ -473,3 +480,4 @@ class _PR: ad._push_completion_event(record, {"summary": "ok"}, "completed") assert captured.get("scope_id") == "G777" assert captured.get("user_id") == "U9" + assert captured.get("message_id") == "om_thread_root" diff --git a/tests/gateway/test_stream_consumer_thread_routing.py b/tests/gateway/test_stream_consumer_thread_routing.py index 62e44f459eb9..5c933fbb2006 100644 --- a/tests/gateway/test_stream_consumer_thread_routing.py +++ b/tests/gateway/test_stream_consumer_thread_routing.py @@ -107,39 +107,119 @@ async def test_overflow_first_send_uses_initial_reply_to_id(self): class TestFeishuFallbackThreadRouting: - """Verify FeishuAdapter._send_raw_message routes to topic on fallback.""" + """Verify FeishuAdapter._send_raw_message routes thread sends via reply API. + + Feishu's create-message API only accepts receive_id_type in + {open_id, union_id, user_id, email, chat_id} — there is NO ``thread_id`` + receive_id_type, so a topic message can only land in a topic through the + reply API (``reply_in_thread=true``) against a real ``om_`` message id. + These tests assert that contract: a thread send with an anchor uses the + reply API, and a thread send with no anchor falls back to a top-level + chat create (never an invalid ``receive_id_type=thread_id``). + """ @pytest.mark.asyncio - async def test_create_uses_thread_id_when_available(self): - """When reply_to=None and metadata has thread_id, message.create - should use receive_id_type='thread_id'.""" + async def test_thread_send_with_anchor_uses_reply_api(self): + """When reply_to is set and metadata has thread_id, the reply API is + used with reply_in_thread=True so the message lands in the topic.""" from plugins.platforms.feishu.adapter import FeishuAdapter - # We test the _send_raw_message method directly by mocking the client + mock_client = MagicMock() + mock_reply_response = SimpleNamespace( + success=lambda: True, + data=SimpleNamespace(message_id="new_msg_1"), + ) + mock_client.im.v1.message.reply = MagicMock(return_value=mock_reply_response) + mock_client.im.v1.message.create = MagicMock() + adapter = MagicMock(spec=FeishuAdapter) + adapter._client = mock_client + adapter._build_reply_message_body = FeishuAdapter._build_reply_message_body + adapter._build_reply_message_request = FeishuAdapter._build_reply_message_request + async def _run_blocking_passthrough(func, *args): + return func(*args) + adapter._run_blocking = _run_blocking_passthrough + + import json + await FeishuAdapter._send_raw_message( + adapter, + chat_id="oc_main_chat", + msg_type="text", + payload=json.dumps({"text": "hello"}), + reply_to="om_thread_root", + metadata={"thread_id": "omt_topic_abc"}, + ) + + # Reply API is the path that lands a message in a topic. + mock_client.im.v1.message.reply.assert_called_once() + mock_client.im.v1.message.create.assert_not_called() + # request must target the supplied om_ message id. + reply_request = mock_client.im.v1.message.reply.call_args[0][0] + assert getattr(reply_request, "message_id", None) == "om_thread_root" + + @pytest.mark.asyncio + async def test_thread_send_with_metadata_reply_to_uses_reply_api(self): + """When reply_to is None but metadata carries reply_to_message_id + (the Feishu status-metadata fallback), the reply API is still used.""" + from plugins.platforms.feishu.adapter import FeishuAdapter + + mock_client = MagicMock() + mock_reply_response = SimpleNamespace( + success=lambda: True, + data=SimpleNamespace(message_id="new_msg_1"), + ) + mock_client.im.v1.message.reply = MagicMock(return_value=mock_reply_response) + mock_client.im.v1.message.create = MagicMock() + + adapter = MagicMock(spec=FeishuAdapter) + adapter._client = mock_client + adapter._build_reply_message_body = FeishuAdapter._build_reply_message_body + adapter._build_reply_message_request = FeishuAdapter._build_reply_message_request + async def _run_blocking_passthrough(func, *args): + return func(*args) + adapter._run_blocking = _run_blocking_passthrough + + import json + await FeishuAdapter._send_raw_message( + adapter, + chat_id="oc_main_chat", + msg_type="text", + payload=json.dumps({"text": "hello"}), + reply_to=None, + metadata={ + "thread_id": "omt_topic_abc", + "reply_to_message_id": "om_thread_root", + }, + ) + + mock_client.im.v1.message.reply.assert_called_once() + mock_client.im.v1.message.create.assert_not_called() + + @pytest.mark.asyncio + async def test_thread_send_without_anchor_falls_back_to_chat_create(self): + """When reply_to is None and metadata has thread_id but no anchor, + fall back to a top-level chat create — NOT an invalid + receive_id_type=thread_id. The Feishu API rejects thread_id.""" + from plugins.platforms.feishu.adapter import FeishuAdapter - # Set up the real _send_raw_message logic manually mock_client = MagicMock() mock_create_response = SimpleNamespace( success=lambda: True, data=SimpleNamespace(message_id="new_msg_1"), ) mock_client.im.v1.message.create = MagicMock(return_value=mock_create_response) + mock_client.im.v1.message.reply = MagicMock() - # Use the real implementation path + adapter = MagicMock(spec=FeishuAdapter) adapter._client = mock_client adapter._build_create_message_body = FeishuAdapter._build_create_message_body adapter._build_create_message_request = FeishuAdapter._build_create_message_request - # _send_raw_message routes blocking SDK calls through _run_blocking - # (adapter-owned executor). On a MagicMock(spec=...) that method is - # auto-mocked and would swallow the real call, so wire a passthrough. async def _run_blocking_passthrough(func, *args): return func(*args) adapter._run_blocking = _run_blocking_passthrough - # Call _send_raw_message with reply_to=None and thread_id in metadata import json - result = await FeishuAdapter._send_raw_message( + await FeishuAdapter._send_raw_message( adapter, chat_id="oc_main_chat", msg_type="text", @@ -148,27 +228,60 @@ async def _run_blocking_passthrough(func, *args): metadata={"thread_id": "omt_topic_abc"}, ) - # Verify message.create was called (not message.reply) mock_client.im.v1.message.create.assert_called_once() - - # The request should have receive_id_type="thread_id" + mock_client.im.v1.message.reply.assert_not_called() call_args = mock_client.im.v1.message.create.call_args[0][0] - # Lark SDK builder exposes .body; the in-tree fallback exposes .request_body. - # The contributor's branch had the lark SDK installed, the test environment - # may not — handle both shapes. + receive_id_type = getattr(call_args, "receive_id_type", None) + assert receive_id_type != "thread_id", ( + f"receive_id_type must NOT be 'thread_id' (Feishu rejects it); " + f"got '{receive_id_type}'" + ) + assert receive_id_type == "chat_id", ( + f"Expected top-level fallback receive_id_type='chat_id', " + f"got '{receive_id_type}'" + ) + # receive_id must be the chat_id, never the omt_ thread id. body = getattr(call_args, "body", None) or getattr(call_args, "request_body", None) - assert body is not None, "request has neither .body nor .request_body" - # receive_id should be the thread_id, not the chat_id receive_id = getattr(body, "receive_id", None) if receive_id is None and isinstance(body, str): import json as _json receive_id = _json.loads(body).get("receive_id") - assert receive_id == "omt_topic_abc", ( - f"Expected receive_id='omt_topic_abc', got '{receive_id}'" + assert receive_id == "oc_main_chat", ( + f"Expected receive_id='oc_main_chat', got '{receive_id}'" ) - # And receive_id_type must be 'thread_id', not 'chat_id' - receive_id_type = getattr(call_args, "receive_id_type", None) - assert receive_id_type == "thread_id", ( - f"Expected receive_id_type='thread_id', got '{receive_id_type}'" + + @pytest.mark.asyncio + async def test_create_uses_chat_id_when_no_thread(self): + """When reply_to=None and metadata has no thread_id, message.create + should use receive_id_type='chat_id' (original behavior).""" + from plugins.platforms.feishu.adapter import FeishuAdapter + + mock_client = MagicMock() + mock_create_response = SimpleNamespace( + success=lambda: True, + data=SimpleNamespace(message_id="new_msg_1"), ) + mock_client.im.v1.message.create = MagicMock(return_value=mock_create_response) + adapter = MagicMock(spec=FeishuAdapter) + adapter._client = mock_client + adapter._build_create_message_body = FeishuAdapter._build_create_message_body + adapter._build_create_message_request = FeishuAdapter._build_create_message_request + async def _run_blocking_passthrough(func, *args): + return func(*args) + adapter._run_blocking = _run_blocking_passthrough + + import json + await FeishuAdapter._send_raw_message( + adapter, + chat_id="oc_main_chat", + msg_type="text", + payload=json.dumps({"text": "hello"}), + reply_to=None, + metadata=None, + ) + + mock_client.im.v1.message.create.assert_called_once() + call_args = mock_client.im.v1.message.create.call_args[0][0] + receive_id_type = getattr(call_args, "receive_id_type", None) + assert receive_id_type == "chat_id" diff --git a/tests/tools/test_async_delegation.py b/tests/tools/test_async_delegation.py index d91ff8799034..1b3e290ad2ec 100644 --- a/tests/tools/test_async_delegation.py +++ b/tests/tools/test_async_delegation.py @@ -175,6 +175,81 @@ def runner(): assert evt["delegation_id"] == res["delegation_id"] +def test_completion_event_carries_message_id_single(): + """Single-task dispatch threads message_id onto the completion event so + the synthetic re-entry routes into the original topic/thread via the + platform reply API.""" + def runner(): + return {"status": "completed", "summary": "ok", "api_calls": 1, + "duration_seconds": 1.0, "model": "m"} + + from gateway.session_context import clear_session_vars, set_session_vars + + tokens = set_session_vars(message_id="om_thread_root") + try: + res = ad.dispatch_async_delegation( + goal="g", context=None, toolsets=None, role="leaf", model="m", + session_key="agent:main:feishu:group:oc_chat:omt_topic", + runner=runner, max_async_children=3, + ) + finally: + clear_session_vars(tokens) + assert res["status"] == "dispatched" + + evt = _drain_one() + assert evt is not None + assert evt.get("message_id") == "om_thread_root" + + +def test_completion_event_carries_message_id_batch(): + """Batch dispatch threads message_id onto the combined completion event.""" + def runner(): + return {"results": [{"status": "completed", "summary": "ok"}], + "total_duration_seconds": 1.0} + + from gateway.session_context import clear_session_vars, set_session_vars + + tokens = set_session_vars(message_id="om_thread_root") + try: + res = ad.dispatch_async_delegation_batch( + goals=["g1", "g2"], context=None, toolsets=None, role="orchestrator", + model="m", session_key="agent:main:feishu:group:oc_chat:omt_topic", + runner=runner, max_async_children=3, + ) + finally: + clear_session_vars(tokens) + assert res["status"] == "dispatched" + + evt = _drain_one() + assert evt is not None + assert evt.get("is_batch") is True + assert evt.get("message_id") == "om_thread_root" + + +def test_completion_event_message_id_defaults_empty(): + """When no message_id is supplied (CLI / cron / pre-anchor sessions), + the event carries an empty string, not a missing key — the gateway reads + it unconditionally and an absent key would raise.""" + def runner(): + return {"status": "completed", "summary": "ok", "api_calls": 0, + "duration_seconds": 0.0, "model": "m"} + + from gateway.session_context import clear_session_vars, set_session_vars + + tokens = set_session_vars(message_id="") + try: + ad.dispatch_async_delegation( + goal="g", context=None, toolsets=None, role="leaf", model="m", + session_key="", runner=runner, max_async_children=3, + ) + finally: + clear_session_vars(tokens) + evt = _drain_one() + assert evt is not None + assert "message_id" in evt + assert evt["message_id"] == "" + + def test_rich_reinjection_block_is_self_contained(): def runner(): return {"status": "completed", "summary": "The answer is 42.", @@ -824,4 +899,3 @@ def test_batch_truncation_banner_marks_only_truncated_task(): banner_pos = text.index("TRUNCATED") # The header banner for task 2 appears after task 1's summary. assert banner_pos > clean_pos - diff --git a/tools/async_delegation.py b/tools/async_delegation.py index 4965363a07ee..853ce12c4b18 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -209,11 +209,9 @@ def _capture_routing_origin() -> Dict[str, Any]: carry the contextvars) and persisted with the durable record, so a completion replayed after a restart can reconstruct a full SessionSource even when the session-store origin and in-memory source cache are gone. - scope_id matters most: on a relay-fronted deployment the connector's - fail-closed egress guard needs the tenant discriminator (or a user - binding) to route a scoped reply; without it, post-restart scoped - completions bounce with "target not routed to an onboarded tenant" - (staging 2026-08-09 defect #4). Best-effort — empty values are simply + scope_id matters most for relay egress, while message_id preserves the + platform reply anchor used by topic/thread-capable adapters after a live + completion or process restart. Best-effort — empty values are simply omitted so CLI/contextvar-unaware paths persist nothing new. """ origin: Dict[str, Any] = {} @@ -224,6 +222,7 @@ def _capture_routing_origin() -> Dict[str, Any]: ("scope_id", "HERMES_SESSION_SCOPE_ID"), ("user_id", "HERMES_SESSION_USER_ID"), ("user_name", "HERMES_SESSION_USER_NAME"), + ("message_id", "HERMES_SESSION_MESSAGE_ID"), ): value = get_session_env(env_name, "") if value: @@ -244,10 +243,10 @@ def _persist_dispatch(record: Dict[str, Any]) -> None: key: record.get(key) for key in ( "goal", "goals", "context", "toolsets", "role", "model", "is_batch", - # Routing origin (scope_id/user_id/user_name): persisted so a + # Routing origin (scope/user/message): persisted so a # restart-recovered completion can reconstruct a full # SessionSource — see _capture_routing_origin. - "scope_id", "user_id", "user_name", + "scope_id", "user_id", "user_name", "message_id", ) if key in record } @@ -373,9 +372,9 @@ def recover_abandoned_delegations() -> int: "dispatched_at": dispatched_at, "completed_at": now, } # Routing origin persisted at dispatch (see _capture_routing_origin): - # restores scope_id/user_id for the reconstructed SessionSource so - # relay egress priming works after a restart. - for _k in ("scope_id", "user_id", "user_name"): + # restores relay identity and the platform reply anchor after a + # restart. + for _k in ("scope_id", "user_id", "user_name", "message_id"): if task.get(_k): event[_k] = task[_k] result = {"status": "unknown", "summary": None, "error": event["error"]} @@ -798,9 +797,9 @@ def dispatch_async_delegation( stale-detection block at the top of this module). When omitted, the delegation is not monitored. max_async_children - Concurrency cap. When at capacity the dispatch is REJECTED (the caller - should fall back to sync or tell the user) rather than queued, so a - runaway model can't pile up unbounded background work. + Concurrency cap. When at capacity the dispatch is REJECTED (the + caller should fall back to sync or tell the user) rather than queued, + so a runaway model can't pile up unbounded background work. Returns ------- @@ -969,6 +968,10 @@ def _push_completion_event( "origin_ui_session_id": record.get("origin_ui_session_id", ""), "origin_session_id": record.get("origin_session_id", ""), "parent_session_id": record.get("parent_session_id"), + # message_id carries the triggering message id back onto the + # synthetic re-entry event so topic/thread-capable platforms route + # the result via the reply API instead of an invalid create path. + "message_id": record.get("message_id", ""), "goal": record.get("goal", ""), "context": record.get("context"), "toolsets": record.get("toolsets"), @@ -1181,6 +1184,10 @@ def _push_batch_completion_event( "origin_ui_session_id": event_record.get("origin_ui_session_id", ""), "origin_session_id": event_record.get("origin_session_id", ""), "parent_session_id": event_record.get("parent_session_id"), + # message_id routes the synthetic re-entry message into the original + # topic/thread via the platform reply API; empty when the dispatching + # session had no anchor (CLI / cron / stateless HTTP). + "message_id": event_record.get("message_id", ""), "goal": event_record.get("goal", ""), "goals": event_record.get("goals"), "context": event_record.get("context"),