Make Feishu group listening, contact lookup, and replies reliable

backend/app/api/feishu.py

@@ -29,6 +29,19 @@
 
 router = APIRouter(tags=["feishu"])
 
+_FEISHU_GROUP_PASSIVE_INSTRUCTION = (
+    "You are passively listening in a Feishu group. A message directly addresses you if it "
+    "@mentions you, names you or your Agent name, asks you a question or gives you an "
+    "instruction, or explicitly asks you to reply. You must visibly answer every directly "
+    "addressed message even when it is outside your usual responsibilities. For messages "
+    "that do not directly address you, reply normally only when your responsibilities require "
+    "a visible response; otherwise your entire final response must be exactly NO_REPLY, with "
+    "no other text. Your final response is automatically delivered to the input Feishu group. "
+    "Never call send_channel_message to reply to the current conversation. Use that Tool only "
+    "when the user explicitly asks you to send a separate message to another person or group, "
+    "and then set cross_session_confirmed=true."
+)
+
 _USER_RESOLUTION_ERROR_TIP = (
     "抱歉，我暂时无法稳定识别你的飞书账号，已停止本次处理以避免重复创建账号。"
     "请稍后重试，或联系管理员检查飞书 Contact API 权限。"
@@ -402,10 +415,17 @@ async def _accept_feishu_runtime_message(
             created_by_user_id=user.id,
         )
         _, model, _ = await _load_agent_and_model(db, agent_id)
-        sender_name = (user.display_name or "").strip()
-        executable_content = (
-            f"[发送者: {sender_name}] {content}" if sender_name else content
+        sender_name = (user.display_name or "").strip() or "未知用户"
+        sender_identity = " | ".join(
+            part
+            for part in (
+                f"飞书发送者: {sender_name}",
+                f"user_id: {sender_user_id.strip()}" if sender_user_id.strip() else "",
+                f"open_id: {sender_open_id.strip()}" if sender_open_id.strip() else "",
+            )
+            if part
         )
+        executable_content = f"[{sender_identity}] {content}"
         intake = await enqueue_channel_chat_runtime(
             db,
             agent=agent,
@@ -414,10 +434,18 @@ async def _accept_feishu_runtime_message(
             model=model,
             content=executable_content,
             display_content=display_content,
+            runtime_instruction=(
+                _FEISHU_GROUP_PASSIVE_INSTRUCTION if is_group else ""
+            ),
             source_channel="feishu",
             channel_delivery_target={
                 "receive_id": chat_id if is_group else sender_open_id,
                 "receive_id_type": "chat_id" if is_group else "open_id",
+                **(
+                    {"source_message_id": external_event_id.strip()}
+                    if is_group and external_event_id and external_event_id.strip()
+                    else {}
+                ),
             },
             message_id=channel_message_id(
                 agent_id,
@@ -492,12 +520,20 @@ async def process_feishu_event(agent_id: uuid.UUID, body: dict):
     if event_type == "im.message.receive_v1":
         message = event.get("message", {})
         sender = event.get("sender", {}).get("sender_id", {})
+        sender_type = event.get("sender", {}).get("sender_type", "")
         sender_open_id = sender.get("open_id", "")
         sender_user_id_from_event = sender.get("user_id", "")  # tenant-stable ID, available directly in event body
         msg_type = message.get("message_type", "text")
         chat_type = message.get("chat_type", "p2p")  # p2p or group
         chat_id = message.get("chat_id", "")
 
+        if chat_type == "group" and sender_type and sender_type != "user":
+            logger.info(
+                "[Feishu] Ignoring non-user group message sender_type={}",
+                sender_type,
+            )
+            return {"code": 0, "msg": "non-user group message ignored"}
+
         logger.info(f"[Feishu] Received {msg_type} message, chat_type={chat_type}, open_id={sender_open_id!r}, user_id_from_event={sender_user_id_from_event!r}")
 
         # ── Normalize post (rich text) → extract text + schedule image downloads ──
@@ -574,7 +610,7 @@ async def process_feishu_event(agent_id: uuid.UUID, body: dict):
                 sender_user_id=sender_user_id_from_event,
                 chat_type=chat_type,
                 chat_id=chat_id,
-                external_event_id=event_id or message.get("message_id"),
+                external_event_id=message.get("message_id") or event_id,
             )
             if attachment is not None:
                 if event_id:
@@ -612,7 +648,7 @@ async def process_feishu_event(agent_id: uuid.UUID, body: dict):
                 chat_id=chat_id,
                 content=user_text,
                 display_content=display_content,
-                external_event_id=event_id or message.get("message_id"),
+                external_event_id=message.get("message_id") or event_id,
             )
         except Exception as exc:
             from app.services.channel_user_service import ChannelUserResolutionError

backend/app/services/agent_runtime/channel_chat.py

@@ -114,6 +114,7 @@ async def enqueue_channel_chat_runtime(
     channel_delivery_target: dict,
     display_content: str = "",
     file_name: str = "",
+    runtime_instruction: str = "",
 ) -> ChatRuntimeIntake:
     """Atomically attach a channel message to a new or waiting Chat Run."""
     if agent.tenant_id is None or model is None:
@@ -139,6 +140,7 @@ async def enqueue_channel_chat_runtime(
             content=content,
             display_content=display_content,
             file_name=file_name,
+            runtime_instruction=runtime_instruction,
             message_id=message_id,
             resume_run_id=resume[0] if resume is not None else None,
             resume_correlation_id=resume[1] if resume is not None else None,

backend/app/services/agent_runtime/channel_delivery.py

@@ -148,12 +148,15 @@ def stage_channel_delivery(
     message_id: uuid.UUID,
     idempotency_key: str,
     clock: Callable[[], datetime],
+    target_overrides: dict | None = None,
 ) -> ChannelDelivery | None:
     """Add one provider outbox row to the caller's ChatMessage transaction."""
     route = _route(run, session)
     if route is None:
         return None
     channel, target = route
+    if target_overrides:
+        target.update(target_overrides)
     delivery = ChannelDelivery(
         id=_delivery_id(run.id, idempotency_key),
         tenant_id=run.tenant_id,

backend/app/services/agent_runtime/channel_provider_delivery.py

@@ -7,6 +7,7 @@
 import os
 
 import httpx
+from loguru import logger
 from sqlalchemy import select
 
 from app.models.channel_config import ChannelConfig
@@ -115,6 +116,8 @@ async def _feishu(
                 "channel_target_invalid",
                 "Unsupported Feishu receive_id_type",
             )
+        if receive_id_type == "chat_id":
+            await self._add_feishu_group_reply_reaction(envelope, config)
         response = await feishu_service.send_message(
             config.app_id,
             config.app_secret,
@@ -129,6 +132,36 @@ async def _feishu(
             provider_message_id=str(message_id) if message_id else None,
         )
 
+    @staticmethod
+    async def _add_feishu_group_reply_reaction(
+        envelope: ChannelDeliveryEnvelope,
+        config: _ProviderConfig,
+    ) -> None:
+        source_message_id = envelope.target.get("source_message_id")
+        emoji_type = envelope.target.get("reaction_emoji_type")
+        if (
+            not isinstance(source_message_id, str)
+            or not source_message_id.strip()
+            or emoji_type != "GLANCE"
+        ):
+            return
+        try:
+            from app.services.feishu_service import feishu_service
+
+            await feishu_service.add_message_reaction(
+                config.app_id,
+                config.app_secret,
+                source_message_id.strip(),
+                emoji_type,
+                stage="runtime_group_reply_reaction",
+            )
+        except Exception as exc:
+            # A cosmetic acknowledgement must never block the durable reply.
+            logger.warning(
+                "[Feishu] Failed to add group reply reaction "
+                f"(message_id={source_message_id[:32]}): {exc}"
+            )
+
     async def _dingtalk(
         self,
         envelope: ChannelDeliveryEnvelope,

backend/app/services/agent_runtime/chat_intake.py

@@ -232,6 +232,10 @@ def _direct_lane_key(tenant_id: uuid.UUID, session_id: uuid.UUID) -> str:
     return f"direct_chat_thread:{tenant_id}:{session_id}"
 
 
+def _external_group_lane_key(tenant_id: uuid.UUID, session_id: uuid.UUID) -> str:
+    return f"external_group_thread:{tenant_id}:{session_id}"
+
+
 async def _direct_lane_holder(
     db: AsyncSession,
     *,
@@ -673,6 +677,12 @@ async def enqueue_chat_runtime(
     if channel_delivery_route is not None:
         delivery_target["channel_delivery"] = channel_delivery_route
     is_direct_thread = session.session_type == "direct"
+    is_external_group_thread = (
+        session.session_type == "group"
+        and session.group_id is None
+        and normalized_channel != "web"
+    )
+    uses_session_thread = is_direct_thread or is_external_group_thread
     scheduling_position_created_at = (
         persisted_message.created_at
         if persisted_message is not None
@@ -692,25 +702,46 @@ async def enqueue_chat_runtime(
             goal=_chat_goal(content, display_content, file_name),
             run_kind="foreground",
             model_id=model.id,
-            runtime_thread_id=(str(session.id) if is_direct_thread else None),
+            runtime_thread_id=(str(session.id) if uses_session_thread else None),
             scheduling_lane_key=(
                 _direct_lane_key(tenant_id, session.id)
                 if is_direct_thread
-                else None
+                else (
+                    _external_group_lane_key(tenant_id, session.id)
+                    if is_external_group_thread
+                    else None
+                )
             ),
             scheduling_position_created_at=(
-                scheduling_position_created_at if is_direct_thread else None
+                scheduling_position_created_at
+                if session.session_type in {"direct", "group"}
+                else None
+            ),
+            scheduling_position_id=(
+                resolved_message_id
+                if session.session_type in {"direct", "group"}
+                else None
             ),
-            scheduling_position_id=(resolved_message_id if is_direct_thread else None),
             delivery_status="pending",
             delivery_target=delivery_target,
             idempotency_key=f"start:{source_execution_id}",
             payload={
                 "message_id": str(resolved_message_id),
                 "input_content": runtime_content,
                 "source_channel": normalized_channel,
+                "chat_session_type": session.session_type,
                 "user_id": str(user.id),
                 "application_tools_enabled": application_tools_enabled,
+                **(
+                    {
+                        "context_cutoff": {
+                            "message_id": str(resolved_message_id),
+                            "created_at": scheduling_position_created_at.isoformat(),
+                        }
+                    }
+                    if session.session_type == "group"
+                    else {}
+                ),
                 **(
                     {"runtime_instruction": normalized_runtime_instruction}
                     if normalized_runtime_instruction

backend/app/services/agent_runtime/delivery.py

@@ -919,14 +919,42 @@ async def deliver_runtime_message(
         )
         chat_message_dao.add_scoped(db, message, tenant_id=run.tenant_id)
         session.last_message_at = now()
-    channel_delivery = stage_channel_delivery(
-        db,
-        run=run,
-        session=session,
-        message_id=message.id,
-        idempotency_key=request.idempotency_key,
-        clock=now,
+    route = (run.delivery_target or {}).get("channel_delivery")
+    route_target = route.get("target") if isinstance(route, dict) else None
+    suppress_feishu_group_reply = (
+        request.kind == "terminal"
+        and request.lifecycle_status == "completed"
+        and session.session_type == "group"
+        and session.group_id is None
+        and session.source_channel == "feishu"
+        and isinstance(route, dict)
+        and route.get("channel") == "feishu"
+        and isinstance(route_target, dict)
+        and route_target.get("receive_id_type") == "chat_id"
+        and message.content.strip().casefold() == "no_reply"
     )
+    channel_delivery = None
+    if not suppress_feishu_group_reply:
+        reaction_target_overrides = (
+            {"reaction_emoji_type": "GLANCE"}
+            if (
+                request.kind == "terminal"
+                and request.lifecycle_status == "completed"
+                and session.session_type == "group"
+                and session.group_id is None
+                and session.source_channel == "feishu"
+            )
+            else None
+        )
+        channel_delivery = stage_channel_delivery(
+            db,
+            run=run,
+            session=session,
+            message_id=message.id,
+            idempotency_key=request.idempotency_key,
+            clock=now,
+            target_overrides=reaction_target_overrides,
+        )
     receipt = DeliveryReceipt(
         tenant_id=run.tenant_id,
         run_id=run.id,

backend/app/services/agent_runtime/model_step_service.py

@@ -391,6 +391,20 @@ def _is_group_agent_run(state: RuntimeGraphState) -> bool:
     )
 
 
+def _is_public_group_chat_run(state: RuntimeGraphState) -> bool:
+    initial_input = state["snapshots"].initial_input
+    if _is_group_agent_run(state):
+        return True
+    if initial_input.get("chat_session_type") == "group":
+        return True
+    # Backward compatibility for external-group checkpoints created before
+    # chat_session_type became an explicit immutable Run input.
+    return (
+        initial_input.get("source_channel") not in {None, "web"}
+        and isinstance(initial_input.get("context_cutoff"), Mapping)
+    )
+
+
 def _with_runtime_tools(
     tools: list[dict],
     *,
@@ -1435,7 +1449,8 @@ async def compact_inputs(
     ) -> RunCompactInputs:
         """Profile the exact business request shape used by the Compact node."""
         model, agent, ledger = await self._load(context, state)
-        allow_user_wait = not _is_group_agent_run(state)
+        is_native_group = _is_group_agent_run(state)
+        allow_user_wait = not _is_public_group_chat_run(state)
         application_tools = (
             with_group_runtime_tools(
                 await self._tool_provider(agent.id),
@@ -1451,7 +1466,7 @@ async def compact_inputs(
         tools = _with_runtime_tools(
             application_tools,
             allow_user_wait=allow_user_wait,
-            allow_group_handoff=not allow_user_wait,
+            allow_group_handoff=is_native_group,
         )
         allowed_names = frozenset(
             name for name in (_tool_name(tool) for tool in tools) if name
@@ -1785,7 +1800,8 @@ async def complete_once(
     ) -> ModelStepResult:
         try:
             model, agent, ledger = await self._load(context, state)
-            allow_user_wait = not _is_group_agent_run(state)
+            is_native_group = _is_group_agent_run(state)
+            allow_user_wait = not _is_public_group_chat_run(state)
             application_tools = (
                 with_group_runtime_tools(
                     await self._tool_provider(agent.id),
@@ -1802,7 +1818,7 @@ async def complete_once(
             tools = _with_runtime_tools(
                 application_tools,
                 allow_user_wait=allow_user_wait,
-                allow_group_handoff=not allow_user_wait,
+                allow_group_handoff=is_native_group,
             )
             allowed_names = frozenset(
                 name for name in (_tool_name(tool) for tool in tools) if name
@@ -1887,7 +1903,7 @@ async def complete_once(
                 fallback_tools = _with_runtime_tools(
                     fallback_application_tools,
                     allow_user_wait=allow_user_wait,
-                    allow_group_handoff=not allow_user_wait,
+                    allow_group_handoff=is_native_group,
                 )
                 fallback_allowed_names = frozenset(
                     name
@@ -1970,7 +1986,7 @@ async def complete_once(
                 step,
                 allowed_tool_names=active_allowed_names,
                 allow_user_wait=allow_user_wait,
-                allow_group_handoff=not allow_user_wait,
+                allow_group_handoff=is_native_group,
             )
             reset_reason = _tool_repair_reset_reason(state)
             if reset_reason is not None:
@@ -1985,7 +2001,7 @@ async def complete_once(
                         active_tools,
                     ),
                 )
-            if result.intent == "finish" and not allow_user_wait:
+            if result.intent == "finish" and is_native_group:
                 try:
                     staged_participant_ids = _pending_group_at_participant_ids(state)
                     legacy_participant_ids = result.finish_mention_participant_ids