Skip to content

fix(feishu): route topic sends via reply API; remove invalid thread_id receive_id #59444

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
DarkMagicCK wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(feishu): route topic sends via reply API; remove invalid thread_id receive_id #59444

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 14 additions & 1 deletion gateway/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down
159 changes: 97 additions & 62 deletions plugins/platforms/feishu/adapter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down
46 changes: 44 additions & 2 deletions tests/gateway/test_background_process_notifications.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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


Expand Down Expand Up @@ -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",
)
)

Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down
Loading