From bc075f24f2367f7d8ae341d8a6a9946c7b35fa22 Mon Sep 17 00:00:00 2001 From: Y1fe1Zh0u Date: Tue, 18 Aug 2026 21:19:44 +0800 Subject: [PATCH 1/5] Let Feishu group Agents listen without noisy replies Feishu group bots can now accept ordinary user messages, freeze each input on a stable external-group lane, retain provider sender identity, and suppress exact NO_REPLY completions before creating an external outbox. Current-conversation replies stay bound to the originating Session, while explicit cross-Session sends require an execution-time confirmation flag. Constraint: Existing Feishu apps require manual im:message.group_msg approval and publication. Rejected: Let models choose a reply group through send_channel_message | directory guesses can send to the wrong Session. Confidence: high Scope-risk: moderate Directive: Keep ordinary replies on the Run delivery target; cross-Session channel sends must remain explicit and validated. Tested: Backend focused pytest suites (113, 90, 66, and 61 passed); scoped Ruff on modified focused files; frontend npm run build; scripts/arch-guard.sh; live 3010 Feishu ordinary-message, NO_REPLY, origin-Session delivery, and calendar E2E. Not-tested: Multi-worker load and high-volume long-running compaction under production traffic. --- backend/app/api/feishu.py | 41 ++++- .../services/agent_runtime/channel_chat.py | 2 + .../app/services/agent_runtime/chat_intake.py | 38 ++++- .../app/services/agent_runtime/delivery.py | 30 +++- .../session_context_background.py | 82 +++++++-- .../session_context_compactor.py | 7 +- .../agent_runtime/session_context_service.py | 7 + backend/app/services/agent_tools.py | 30 ++++ .../app/services/builtin_tool_definitions.py | 6 +- .../tests/test_agent_runtime_chat_intake.py | 14 +- backend/tests/test_agent_runtime_delivery.py | 67 ++++++++ backend/tests/test_feishu_channel_runtime.py | 50 +++++- frontend/src/components/ChannelConfig.tsx | 6 +- .../checklists/requirements.md | 36 ++++ .../contracts/feishu-passive-listening.md | 85 ++++++++++ .../data-model.md | 83 ++++++++++ specs/004-feishu-passive-listening/design.md | 30 ++++ specs/004-feishu-passive-listening/plan.md | 147 +++++++++++++++++ .../quickstart.md | 58 +++++++ .../004-feishu-passive-listening/research.md | 73 ++++++++ specs/004-feishu-passive-listening/spec.md | 156 ++++++++++++++++++ 21 files changed, 1006 insertions(+), 42 deletions(-) create mode 100644 specs/004-feishu-passive-listening/checklists/requirements.md create mode 100644 specs/004-feishu-passive-listening/contracts/feishu-passive-listening.md create mode 100644 specs/004-feishu-passive-listening/data-model.md create mode 100644 specs/004-feishu-passive-listening/design.md create mode 100644 specs/004-feishu-passive-listening/plan.md create mode 100644 specs/004-feishu-passive-listening/quickstart.md create mode 100644 specs/004-feishu-passive-listening/research.md create mode 100644 specs/004-feishu-passive-listening/spec.md diff --git a/backend/app/api/feishu.py b/backend/app/api/feishu.py index a24bd9e86..7b3f5998f 100644 --- a/backend/app/api/feishu.py +++ b/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,6 +434,9 @@ 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, @@ -492,12 +515,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 +605,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 +643,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 diff --git a/backend/app/services/agent_runtime/channel_chat.py b/backend/app/services/agent_runtime/channel_chat.py index 9a5357ce9..b96a3028e 100644 --- a/backend/app/services/agent_runtime/channel_chat.py +++ b/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, diff --git a/backend/app/services/agent_runtime/chat_intake.py b/backend/app/services/agent_runtime/chat_intake.py index 5e9e7c9f1..b4af54acc 100644 --- a/backend/app/services/agent_runtime/chat_intake.py +++ b/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,16 +702,26 @@ 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}", @@ -711,6 +731,16 @@ async def enqueue_chat_runtime( "source_channel": normalized_channel, "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 diff --git a/backend/app/services/agent_runtime/delivery.py b/backend/app/services/agent_runtime/delivery.py index 6b63740eb..c0c3c2f24 100644 --- a/backend/app/services/agent_runtime/delivery.py +++ b/backend/app/services/agent_runtime/delivery.py @@ -919,14 +919,30 @@ 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: + channel_delivery = stage_channel_delivery( + db, + run=run, + session=session, + message_id=message.id, + idempotency_key=request.idempotency_key, + clock=now, + ) receipt = DeliveryReceipt( tenant_id=run.tenant_id, run_id=run.id, diff --git a/backend/app/services/agent_runtime/session_context_background.py b/backend/app/services/agent_runtime/session_context_background.py index 7ebd073a9..d944949b7 100644 --- a/backend/app/services/agent_runtime/session_context_background.py +++ b/backend/app/services/agent_runtime/session_context_background.py @@ -128,9 +128,40 @@ async def resolve( ) if session.group_id is None: - raise SessionContextBackgroundError( - "session_compact_budget_unavailable", - "Group session has no group identity", + if session.source_channel != "feishu" or session.agent_id is None: + raise SessionContextBackgroundError( + "session_compact_budget_unavailable", + "External Group session has no owning Agent", + ) + agent_result = await db.execute( + select(Agent).where( + Agent.id == session.agent_id, + Agent.tenant_id == tenant_id, + Agent.status.in_(_ACTIVE_AGENT_STATUSES), + Agent.is_expired.is_(False), + Agent.deleted_at.is_(None), + ) + ) + agent = agent_result.scalar_one_or_none() + if agent is None: + raise SessionContextBackgroundError( + "session_compact_budget_unavailable", + "External Group Session Agent is unavailable", + ) + model = await resolve_active_agent_model(db, agent) + if model is None: + raise SessionContextBackgroundError( + "session_compact_budget_unavailable", + "External Group Session Agent has no active model", + ) + try: + threshold = _model_threshold(model, self._settings) + except ModelCapabilityError as exc: + raise SessionContextBackgroundError(exc.code, str(exc)) from exc + return SessionCompactPolicy( + source_agent_id=agent.id, + threshold_tokens=threshold, + contributing_model_ids=(model.id,), ) group_result = await db.execute( select(Group.id).where( @@ -411,19 +442,16 @@ def __init__( async def scan_once(self) -> int: async with self._session_factory() as db: - statement = ( - select(ChatSession.tenant_id, ChatSession.id) - .join( - Group, - (Group.id == ChatSession.group_id) - & (Group.tenant_id == ChatSession.tenant_id), - ) - .where( - ChatSession.deleted_at.is_(None), - ChatSession.last_message_at.is_not(None), - ChatSession.session_type == "group", - Group.deleted_at.is_(None), - sa.exists( + native_group = sa.and_( + ChatSession.group_id.is_not(None), + sa.exists( + select(1).where( + Group.id == ChatSession.group_id, + Group.tenant_id == ChatSession.tenant_id, + Group.deleted_at.is_(None), + ) + ), + sa.exists( select(1) .select_from(GroupMember) .join( @@ -445,6 +473,28 @@ async def scan_once(self) -> int: Agent.deleted_at.is_(None), ) ), + ) + external_feishu_group = sa.and_( + ChatSession.group_id.is_(None), + ChatSession.source_channel == "feishu", + ChatSession.agent_id.is_not(None), + sa.exists( + select(1).where( + Agent.id == ChatSession.agent_id, + Agent.tenant_id == ChatSession.tenant_id, + Agent.status.in_(_ACTIVE_AGENT_STATUSES), + Agent.is_expired.is_(False), + Agent.deleted_at.is_(None), + ) + ), + ) + statement = ( + select(ChatSession.tenant_id, ChatSession.id) + .where( + ChatSession.deleted_at.is_(None), + ChatSession.last_message_at.is_not(None), + ChatSession.session_type == "group", + sa.or_(native_group, external_feishu_group), ) .order_by(ChatSession.id) .limit(self._settings.AGENT_RUNTIME_SESSION_COMPACT_SCAN_BATCH_SIZE) diff --git a/backend/app/services/agent_runtime/session_context_compactor.py b/backend/app/services/agent_runtime/session_context_compactor.py index 81b8b44db..68c08b330 100644 --- a/backend/app/services/agent_runtime/session_context_compactor.py +++ b/backend/app/services/agent_runtime/session_context_compactor.py @@ -297,7 +297,7 @@ async def _resolve_models( "session_context_unavailable", "Session Compact target no longer exists", ) - if session.session_type == "group": + if session.session_type == "group" and session.group_id is not None: model = await resolve_multi_agent_compact_model( db, self._settings, @@ -308,6 +308,11 @@ async def _resolve_models( usage_agent_id=None, ) + if session.session_type == "group" and session.source_channel != "feishu": + raise SessionContextCompactorError( + "session_compact_model_unavailable", + "External Group Session channel is unsupported for compaction", + ) if session.agent_id is None or session.agent_id != request.source_agent_id: raise SessionContextCompactorError( "session_context_agent_mismatch", diff --git a/backend/app/services/agent_runtime/session_context_service.py b/backend/app/services/agent_runtime/session_context_service.py index 0d0a005e2..c93c6a2b0 100644 --- a/backend/app/services/agent_runtime/session_context_service.py +++ b/backend/app/services/agent_runtime/session_context_service.py @@ -301,6 +301,13 @@ def _message_scope(tenant_id: uuid.UUID, session_id: uuid.UUID): ChatMessage.conversation_id == sa_cast(ChatSession.id, String), ChatMessage.role.in_(_USER_VISIBLE_ROLES), ChatMessage.created_at.is_not(None), + ~and_( + ChatSession.session_type == "group", + ChatSession.group_id.is_(None), + ChatSession.source_channel == "feishu", + ChatMessage.role == "assistant", + func.lower(func.btrim(ChatMessage.content)) == "no_reply", + ), ) diff --git a/backend/app/services/agent_tools.py b/backend/app/services/agent_tools.py index 9a0d0c1ee..3e82cbf6e 100644 --- a/backend/app/services/agent_tools.py +++ b/backend/app/services/agent_tools.py @@ -3779,6 +3779,24 @@ async def _edit_file_outcome( ) +def _channel_cross_session_error(arguments: Mapping[str, object], session_id: str) -> str | None: + if not session_id: + return None + target_recipient_id = str(arguments.get("target_recipient_id") or "").strip() + target_member_id = str(arguments.get("target_member_id") or "").strip() + cross_session = bool( + target_member_id + or (target_recipient_id and target_recipient_id != session_id) + ) + if cross_session and arguments.get("cross_session_confirmed") is not True: + return ( + "Cross-Session channel delivery rejected. Normal replies are automatically " + "returned to the input Session. Set cross_session_confirmed=true only when the " + "user explicitly requested another person or group." + ) + return None + + async def execute_builtin_tool_outcome( tool_name: str, arguments: dict, @@ -3807,6 +3825,13 @@ async def execute_builtin_tool_outcome( path_error, "workspace_path_invalid", ) + if tool_name == "send_channel_message": + cross_session_error = _channel_cross_session_error(arguments, session_id) + if cross_session_error is not None: + return _typed_failure( + cross_session_error, + "cross_session_delivery_not_confirmed", + ) if ( tool_name in _WORKSPACE_SCOPED_FILE_TOOL_NAMES and arguments.get("workspace_scope", "agent") != "agent" @@ -4331,6 +4356,11 @@ async def execute_tool( "conversation confirmation." ) + if tool_name == "send_channel_message": + cross_session_error = _channel_cross_session_error(arguments, session_id) + if cross_session_error is not None: + return f"❌ {cross_session_error}" + path_error = _agent_relative_path_error(tool_name, arguments) if path_error is not None: return f"❌ {path_error}" diff --git a/backend/app/services/builtin_tool_definitions.py b/backend/app/services/builtin_tool_definitions.py index 826721928..d60385e89 100644 --- a/backend/app/services/builtin_tool_definitions.py +++ b/backend/app/services/builtin_tool_definitions.py @@ -554,7 +554,7 @@ { "name": "send_channel_message", "display_name": "Channel Message", - "description": "Send a message through an external channel. For a person, use query_directory then target_member_id. For a Feishu group, use query_directory(member_type='group') then target_recipient_id. Do not guess IDs.", + "description": "Send a proactive message through an external channel. Normal replies are automatically delivered to the current input Session and must not use this Tool. Use it only when the user explicitly asks to message another person or group. For a person, use query_directory then target_member_id. For a Feishu group, use query_directory(member_type='group') then target_recipient_id. Do not guess IDs.", "category": "communication", "icon": "💬", "is_default": False, @@ -569,6 +569,10 @@ "description": "Optional: specific external channel to use.", "enum": ["feishu", "dingtalk", "wecom", "slack", "teams", "microsoft_teams", "wechat"], }, + "cross_session_confirmed": { + "type": "boolean", + "description": "Set true only when the user explicitly requested sending to another person or group outside the current input Session.", + }, }, "required": ["message"], }, diff --git a/backend/tests/test_agent_runtime_chat_intake.py b/backend/tests/test_agent_runtime_chat_intake.py index 723989cb6..c3befa7eb 100644 --- a/backend/tests/test_agent_runtime_chat_intake.py +++ b/backend/tests/test_agent_runtime_chat_intake.py @@ -348,10 +348,16 @@ async def test_external_group_chat_uses_unified_session_without_native_group_sco assert message.user_id is None assert message.participant_id == participant.id command = start_run.await_args.args[0] - assert command.runtime_thread_id is None - assert command.scheduling_lane_key is None - assert command.scheduling_position_created_at is None - assert command.scheduling_position_id is None + assert command.runtime_thread_id == str(session.id) + assert command.scheduling_lane_key == ( + f"external_group_thread:{agent.tenant_id}:{session.id}" + ) + assert command.scheduling_position_created_at == message.created_at + assert command.scheduling_position_id == message.id + assert command.payload["context_cutoff"] == { + "message_id": str(message.id), + "created_at": message.created_at.isoformat(), + } assert command.delivery_target == { "kind": "session", "session_id": str(session.id), diff --git a/backend/tests/test_agent_runtime_delivery.py b/backend/tests/test_agent_runtime_delivery.py index fd4dee9ab..7d3fd478f 100644 --- a/backend/tests/test_agent_runtime_delivery.py +++ b/backend/tests/test_agent_runtime_delivery.py @@ -670,6 +670,73 @@ async def test_external_group_delivery_uses_channel_scope_without_native_members assert len(db.statements) == 5 +@pytest.mark.asyncio +async def test_exact_no_reply_suppresses_feishu_group_outbox() -> None: + tenant_id = uuid.uuid4() + agent_id = uuid.uuid4() + sender_user_id = uuid.uuid4() + session = ChatSession( + id=uuid.uuid4(), tenant_id=tenant_id, session_type="group", group_id=None, + agent_id=agent_id, user_id=sender_user_id, created_by_participant_id=uuid.uuid4(), + title="Feishu Group", source_channel="feishu", + external_conv_id="feishu_group_oc_123", is_group=True, + is_primary=False, deleted_at=None, + ) + run = _run( + tenant_id=tenant_id, session=session, agent_id=agent_id, + origin_user_id=sender_user_id, + delivery_target={ + "kind": "session", "session_id": str(session.id), + "channel_delivery": { + "version": 1, "channel": "feishu", + "target": {"receive_id": "oc_123", "receive_id_type": "chat_id"}, + }, + }, + ) + db = _RecordingDB(run, None, session, _agent(tenant_id, agent_id), _participant(agent_id)) + + receipt = await deliver_runtime_message( + db, _terminal_request(run, content=" no_reply "), clock=lambda: NOW, + ) + + assert receipt.status == "delivered" + assert _added(db, ChatMessage)[0].content == "no_reply" + assert _added(db, ChannelDelivery) == [] + assert run.delivery_status == "delivered" + + +@pytest.mark.asyncio +async def test_no_reply_with_visible_text_still_stages_feishu_group_outbox() -> None: + tenant_id = uuid.uuid4() + agent_id = uuid.uuid4() + sender_user_id = uuid.uuid4() + session = ChatSession( + id=uuid.uuid4(), tenant_id=tenant_id, session_type="group", group_id=None, + agent_id=agent_id, user_id=sender_user_id, created_by_participant_id=uuid.uuid4(), + title="Feishu Group", source_channel="feishu", + external_conv_id="feishu_group_oc_123", is_group=True, + is_primary=False, deleted_at=None, + ) + run = _run( + tenant_id=tenant_id, session=session, agent_id=agent_id, + origin_user_id=sender_user_id, + delivery_target={ + "kind": "session", "session_id": str(session.id), + "channel_delivery": { + "version": 1, "channel": "feishu", + "target": {"receive_id": "oc_123", "receive_id_type": "chat_id"}, + }, + }, + ) + db = _RecordingDB(run, None, session, _agent(tenant_id, agent_id), _participant(agent_id)) + + await deliver_runtime_message( + db, _terminal_request(run, content="我来处理\nNO_REPLY"), clock=lambda: NOW, + ) + + assert len(_added(db, ChannelDelivery)) == 1 + + @pytest.mark.asyncio async def test_duplicate_delivery_returns_the_stored_receipt_without_a_message() -> None: tenant_id = uuid.uuid4() diff --git a/backend/tests/test_feishu_channel_runtime.py b/backend/tests/test_feishu_channel_runtime.py index 1503f6901..125af583a 100644 --- a/backend/tests/test_feishu_channel_runtime.py +++ b/backend/tests/test_feishu_channel_runtime.py @@ -11,6 +11,7 @@ from app.api import feishu from app.services import channel_session +from app.services import agent_tools from app.services.agent_runtime.chat_intake import ChatRuntimeIntake from app.services.agent_runtime.contracts import RunHandle, RuntimeEventCursor @@ -164,8 +165,23 @@ async def enqueue(_db, **kwargs): assert session_call["created_by_user_id"] == user_id intake_call = calls["intake"] assert isinstance(intake_call, dict) - assert intake_call["content"] == "[发送者: Alice] Hello Feishu" + assert intake_call["content"] == ( + "[飞书发送者: Alice | user_id: feishu-user-1 | open_id: ou_sender] " + "Hello Feishu" + ) assert intake_call["display_content"] == "Hello Feishu" + assert intake_call["runtime_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." + ) assert intake_call["channel_delivery_target"] == { "receive_id": "oc_group_1", "receive_id_type": "chat_id", @@ -177,6 +193,36 @@ async def enqueue(_db, **kwargs): ) +@pytest.mark.asyncio +async def test_send_channel_message_rejects_unconfirmed_cross_session_target() -> None: + result = await agent_tools.execute_tool( + "send_channel_message", + { + "channel": "feishu", + "target_recipient_id": str(uuid.uuid4()), + "message": "wrong group", + }, + uuid.uuid4(), + uuid.uuid4(), + session_id=str(uuid.uuid4()), + ) + + assert result.startswith("❌ Cross-Session channel delivery rejected") + + typed = await agent_tools.execute_builtin_tool_outcome( + "send_channel_message", + { + "channel": "feishu", + "target_recipient_id": str(uuid.uuid4()), + "message": "wrong group", + }, + uuid.uuid4(), + uuid.uuid4(), + session_id=str(uuid.uuid4()), + ) + assert typed.error_code == "cross_session_delivery_not_confirmed" + + @pytest.mark.asyncio async def test_feishu_event_commits_runtime_before_provider_ack(monkeypatch) -> None: tenant_id = uuid.uuid4() @@ -224,7 +270,7 @@ async def accept(**kwargs): assert event_id in feishu._processed_events accepted = calls["accept"] assert isinstance(accepted, dict) - assert accepted["external_event_id"] == event_id + assert accepted["external_event_id"] == "om_message_1" @pytest.mark.asyncio diff --git a/frontend/src/components/ChannelConfig.tsx b/frontend/src/components/ChannelConfig.tsx index 2f9f32d2b..6ee30ff7d 100644 --- a/frontend/src/components/ChannelConfig.tsx +++ b/frontend/src/components/ChannelConfig.tsx @@ -231,7 +231,7 @@ const CHANNEL_REGISTRY: ChannelDef[] = [ ]; // ─── Feishu Permission JSON ───────────────────────────── -const FEISHU_PERM_BASIC_JSON = '{"scopes":{"tenant":["contact:contact.base:readonly","contact:user.base:readonly","contact:user.employee_id:readonly","contact:user.id:readonly","im:chat","im:message","im:message.group_at_msg:readonly","im:message.p2p_msg:readonly","im:message:send_as_bot","im:resource"],"user":[]}}'; +const FEISHU_PERM_BASIC_JSON = '{"scopes":{"tenant":["contact:contact.base:readonly","contact:user.base:readonly","contact:user.employee_id:readonly","contact:user.id:readonly","im:chat","im:message","im:message.group_at_msg:readonly","im:message.group_msg","im:message.p2p_msg:readonly","im:message:send_as_bot","im:resource"],"user":[]}}'; const FEISHU_PERM_BASIC_DISPLAY = `{ "scopes": { @@ -243,6 +243,7 @@ const FEISHU_PERM_BASIC_DISPLAY = `{ "im:chat", "im:message", "im:message.group_at_msg:readonly", + "im:message.group_msg", "im:message.p2p_msg:readonly", "im:message:send_as_bot", "im:resource" @@ -251,7 +252,7 @@ const FEISHU_PERM_BASIC_DISPLAY = `{ } }`; -const FEISHU_PERM_FULL_JSON = '{"scopes":{"tenant":["approval:approval","base:app:create","base:dashboard:create","base:field_group:create","bitable:app","bitable:app:readonly","board:whiteboard:node:create","calendar:calendar.event:create","calendar:calendar.event:delete","calendar:calendar.event:read","calendar:calendar.event:update","calendar:calendar.free_busy:read","calendar:calendar:readonly","contact:contact.base:readonly","contact:user.base:readonly","contact:user.employee_id:readonly","contact:user.id:readonly","docx:document","docx:document:create","drive:drive","im:chat","im:message","im:message.group_at_msg:readonly","im:message.p2p_msg:readonly","im:message:send_as_bot","im:resource","sheets:spreadsheet:create","slides:presentation:create","slides:presentation:write_only","wiki:wiki","wiki:wiki:readonly"],"user":[]}}'; +const FEISHU_PERM_FULL_JSON = '{"scopes":{"tenant":["approval:approval","base:app:create","base:dashboard:create","base:field_group:create","bitable:app","bitable:app:readonly","board:whiteboard:node:create","calendar:calendar.event:create","calendar:calendar.event:delete","calendar:calendar.event:read","calendar:calendar.event:update","calendar:calendar.free_busy:read","calendar:calendar:readonly","contact:contact.base:readonly","contact:user.base:readonly","contact:user.employee_id:readonly","contact:user.id:readonly","docx:document","docx:document:create","drive:drive","im:chat","im:message","im:message.group_at_msg:readonly","im:message.group_msg","im:message.p2p_msg:readonly","im:message:send_as_bot","im:resource","sheets:spreadsheet:create","slides:presentation:create","slides:presentation:write_only","wiki:wiki","wiki:wiki:readonly"],"user":[]}}'; const FEISHU_PERM_FULL_DISPLAY = `{ "scopes": { @@ -279,6 +280,7 @@ const FEISHU_PERM_FULL_DISPLAY = `{ "im:chat", "im:message", "im:message.group_at_msg:readonly", + "im:message.group_msg", "im:message.p2p_msg:readonly", "im:message:send_as_bot", "im:resource", diff --git a/specs/004-feishu-passive-listening/checklists/requirements.md b/specs/004-feishu-passive-listening/checklists/requirements.md new file mode 100644 index 000000000..28937a7c2 --- /dev/null +++ b/specs/004-feishu-passive-listening/checklists/requirements.md @@ -0,0 +1,36 @@ +# Specification Quality Checklist: 飞书群常驻 Agent V1 + +**Purpose**: Validate specification completeness and quality before proceeding to planning +**Created**: 2026-08-18 +**Feature**: [spec.md](../spec.md) + +## Content Quality + +- [x] No implementation details (languages, frameworks, APIs) +- [x] Focused on user value and business needs +- [x] Written for non-technical stakeholders +- [x] All mandatory sections completed + +## Requirement Completeness + +- [x] No [NEEDS CLARIFICATION] markers remain +- [x] Requirements are testable and unambiguous +- [x] Success criteria are measurable +- [x] Success criteria are technology-agnostic +- [x] All acceptance scenarios are defined +- [x] Edge cases are identified +- [x] Scope is clearly bounded +- [x] Dependencies and assumptions identified + +## Feature Readiness + +- [x] All functional requirements have clear acceptance criteria +- [x] User scenarios cover primary flows +- [x] Feature meets measurable outcomes defined in Success Criteria +- [x] No implementation details leak into specification + +## Notes + +- Validation iteration 1 passed all checks. +- The specification intentionally keeps Provider permission names and concrete code paths out of business requirements; those belong in the design artifact. +- User confirmation is required before proceeding to design and constitution review. diff --git a/specs/004-feishu-passive-listening/contracts/feishu-passive-listening.md b/specs/004-feishu-passive-listening/contracts/feishu-passive-listening.md new file mode 100644 index 000000000..2f992c23c --- /dev/null +++ b/specs/004-feishu-passive-listening/contracts/feishu-passive-listening.md @@ -0,0 +1,85 @@ +# Contract: 飞书群常驻 Agent V1 + +## 1. Inbound Event Contract + +接受条件： + +- event type 为现有飞书消息接收事件； +- message chat type 为 group； +- Provider `message_id` 非空； +- sender 为用户而非当前机器人； +- 消息类型属于现有已支持集合。 + +幂等键： + +```text +channel_message_id(agent_id, "feishu", provider_message_id) +``` + +同一键的重试必须收敛到同一 ChatMessage 和同一 Runtime source execution。 + +## 2. Silent Reply Contract + +常量： + +```text +NO_REPLY +``` + +规范化与判定： + +```text +silent := content.strip().casefold() == "no_reply" +``` + +不得使用 contains、suffix 或正则尾部匹配代替精确判定。 + +## 3. Suppression Scope + +只有以下条件全部成立才抑制： + +```text +delivery kind terminal +lifecycle status completed +channel feishu +receive_id_type chat_id +content exact silent token +``` + +结果： + +```text +Internal ChatMessage allowed +ChannelDelivery absent +Feishu provider call absent +Run terminal status completed +``` + +## 4. Non-silent Examples + +以下内容必须正常发送： + +```text +我来处理 +我来处理\nNO_REPLY +NO_REPLY：因为不相关 +`NO_REPLY` +请输出 NO_REPLY +``` + +## 5. Session Context Contract + +对飞书外部群 Session： + +- 入站用户消息始终可进入 pending/recent/compactable 选择。 +- 正常 Assistant 回复可进入选择。 +- 精确静默 Assistant 消息不进入模型可见 pending/recent/compactable 内容。 +- 原始数据库行不得删除。 +- 压缩水位线必须对应实际纳入 compact request 的最后一条消息位置。 + +## 6. Prohibited Changes + +- 不修改 Runtime `ModelIntent`、checkpoint lifecycle 或 finish schema。 +- 不写入或伪造原生 `group_id`。 +- 不将静默规则应用到飞书私聊或其他渠道。 +- 不在 Provider sender 收到 pending outbox 后才静默。 diff --git a/specs/004-feishu-passive-listening/data-model.md b/specs/004-feishu-passive-listening/data-model.md new file mode 100644 index 000000000..4dd0744a1 --- /dev/null +++ b/specs/004-feishu-passive-listening/data-model.md @@ -0,0 +1,83 @@ +# Data Model: 飞书群常驻 Agent V1 + +## 结论 + +本功能不新增表、不新增列、不新增迁移。所有事实继续由既有实体持有。 + +## Existing Entities + +### ChatSession + +飞书外部群形态保持： + +```text +tenant_id 必填，租户范围 +session_type group +group_id NULL；只为 Clawith 原生群保留 +agent_id 飞书 Channel 所属 Agent +source_channel feishu +external_conv_id 既有飞书外部会话映射，V1 不迁移 +is_group true +``` + +验证规则：不得把飞书 `chat_id` 写入 `group_id`；同一 Session 的 Agent 必须属于同一 tenant。 + +### ChatMessage + +- 入站用户消息：Provider `message_id` 经确定性映射得到本地 UUID。 +- 正常 Assistant 回复：保持现有落库行为。 +- 静默 Assistant 回复：允许保存精确 `NO_REPLY` 作为内部审计记录，但后续飞书群 Session Context 不把它当作有意义历史。 + +### AgentRun + +- 每条被接受飞书群消息对应一次现有 chat Run。 +- 不新增 lifecycle 或 completion mode。 +- 精确 `NO_REPLY` 仍是 completed Run。 +- 无外部 outbox 时沿用现有 settled delivery 行为。 + +### ChannelDelivery + +- 正常回答：创建一条 `pending` outbox，Provider sender 成功后转为 delivered。 +- 精确静默：不创建记录。 +- 失败/取消：不受 token 规则影响。 + +### SessionContextState + +- 继续以 `tenant_id + session_id` 持有滚动摘要、版本与 `covered_through_message_id`。 +- 飞书外部群的 `agent_id` 范围保持现有 group Session 规则；压缩模型归属通过 Session 的 `agent_id` 解析，不改变 state schema。 +- 水位线只在成功 CAS 提交有效摘要后前进。 + +## State Transitions + +### Normal reply + +```text +Inbound ChatMessage +→ AgentRun completed(content=normal text) +→ Assistant ChatMessage +→ ChannelDelivery pending +→ Provider delivered/failed +``` + +### Silent reply + +```text +Inbound ChatMessage +→ AgentRun completed(content=NO_REPLY) +→ Internal Assistant ChatMessage +→ no ChannelDelivery +→ no Provider call +``` + +### Session compaction + +```text +Messages after watermark reach threshold +→ lock Session +→ build compact request excluding exact silent Assistant rows +→ model returns candidate +→ CAS SessionContextState +→ watermark advances +``` + +失败时保持旧 state 和全部原始 ChatMessage。 diff --git a/specs/004-feishu-passive-listening/design.md b/specs/004-feishu-passive-listening/design.md new file mode 100644 index 000000000..71d5eafff --- /dev/null +++ b/specs/004-feishu-passive-listening/design.md @@ -0,0 +1,30 @@ +# Design: 飞书群常驻 Agent V1 + +**Status**: Implemented +**Detailed plan**: [plan.md](./plan.md) +**Research**: [research.md](./research.md) +**Data contract**: [data-model.md](./data-model.md) +**Behavior contract**: [contracts/feishu-passive-listening.md](./contracts/feishu-passive-listening.md) + +## Decision Summary + +1. 飞书应用增加全量群用户消息权限，事件入口保持不变。 +2. 入站消息以 Provider `message_id` 幂等，每条有效群消息进入现有 Durable Runtime。 +3. 模型无需发言时输出 token-only `NO_REPLY`；只做精确、大小写不敏感匹配。 +4. 不修改 Runtime 状态机。Run 和内部 Assistant ChatMessage 正常完成并可审计。 +5. 在创建飞书群 `ChannelDelivery` 之前过滤精确静默结果，因此没有 outbox，也不会调用飞书发送接口。 +6. 飞书外部群 Session 复用现有 Session Context summary/watermark/CAS；压缩模型来自 `session.agent_id`。 +7. 精确静默 Assistant 消息不进入后续模型可见 Session Context，但底层记录不删除。 +8. 不迁移外部渠道 ID，不统一其他渠道，不新增表、依赖或 checkpoint 状态。 + +## Critical Boundaries + +- 飞书 `chat_id` 永远不是 Clawith 原生 `group_id`。 +- `正文 + NO_REPLY` 必须发送，只有 token-only 才静默。 +- failed/cancelled/waiting 不属于静默结果。 +- 正常回答继续由现有 ChannelDelivery outbox 与 Provider receipt 保证。 +- 压缩失败不得推进水位线或删除历史。 + +## Constitution Verdict + +设计通过 [docs/constitution.md](../../docs/constitution.md) C1–C6 检查，无例外项。实现阶段必须先写回归测试，再修改代码。 diff --git a/specs/004-feishu-passive-listening/plan.md b/specs/004-feishu-passive-listening/plan.md new file mode 100644 index 000000000..2b1be0425 --- /dev/null +++ b/specs/004-feishu-passive-listening/plan.md @@ -0,0 +1,147 @@ +# Implementation Plan: 飞书群常驻 Agent V1 + +**Branch**: `004-feishu-passive-listening` | **Date**: 2026-08-18 | **Spec**: [spec.md](./spec.md) +**Input**: Feature specification from `/specs/004-feishu-passive-listening/spec.md` + +## Summary + +为现有飞书 Agent Channel 增加群内全部用户消息接收能力。每条被接受的群消息继续通过现有 `enqueue_channel_chat_runtime()` 创建 Durable Runtime Run；模型无需发言时输出 token-only `NO_REPLY`。Runtime 仍把它当作正常非空完成文本，产品侧仍可保留内部 Assistant `ChatMessage`，但外部渠道投递在创建 `ChannelDelivery` 之前识别“成功终态 + 飞书群路由 + 精确静默令牌”并抑制 Provider 出站。现有 Session Context 后台压缩扩展到 `group_id IS NULL` 的飞书外部群 Session，并使用该 Session 所属 Agent 的模型预算；不新增状态表、不改变 checkpoint 生命周期、不统一其他渠道。 + +## Technical Context + +**Language/Version**: Python 3.12 deployment baseline（package metadata >=3.11）；React 19 / strict TypeScript +**Primary Dependencies**: FastAPI、SQLAlchemy 2.x async ORM、LangGraph 1.2.x、Pydantic、httpx、React/Vite；不新增依赖 +**Storage**: PostgreSQL 15；复用 `chat_sessions`、`chat_messages`、`agent_runs`、`agent_run_events`、`channel_deliveries`、`session_context_states` +**Testing**: Pytest、Ruff；前端使用现有 test/build 命令 +**Target Platform**: Clawith backend/API/Runtime workers + 飞书自建应用机器人 +**Project Type**: Docker Compose monorepo web application +**Performance Goals**: 飞书事件回调只负责持久化消息与 Runtime Command 后返回；模型执行和外部投递保持异步。重复 Provider 消息不产生第二次 Run。长期群历史经滚动压缩后不进行无界全量装载。 +**Constraints**: V1 每条群消息均调用模型；精确 `NO_REPLY` 只抑制飞书群最终出站；不得影响失败/取消语义、飞书私聊、原生群和其他渠道；不得伪造原生 `group_id`。 +**Scale/Scope**: 单 Agent 多飞书群的长期 Session；本版不新增 Activation Gate、不统一企微/钉钉等渠道。 + +## Constitution Check + +*GATE: Passed before Phase 0 and re-checked after Phase 1.* + +| Gate | Result | Design evidence | +|---|---|---| +| Evidence Before Claims | PASS | 当前权限、入站幂等、ChannelDelivery 建立点和压缩器 `group_id` 限制均由源码与测试确认；OpenClaw 静默行为使用官方仓库与文档。 | +| Minimal Scoped Changes | PASS | 仅扩展飞书权限、飞书入站稳定 ID、外部投递静默过滤和飞书群压缩；不做跨渠道抽象或数据库迁移。 | +| Contract and State Ownership | PASS | 模型只拥有最终内容；Runtime checkpoint 不变；产品交付层决定是否建立飞书 outbox；Provider sender 仍拥有真实发送结果。 | +| Tests Prove Behavior | PASS | 计划先加入 token 精确匹配、零 outbox、普通回复、重复入站和外部群压缩水位线回归测试。 | +| Preserve Existing Work | PASS | 规格目录独立；现有 `docs/` 用户改动保持未触碰。 | +| C1 Runtime Boundary Isolation | PASS | 不增加 checkpoint 字段或第二状态机；API 仍只通过 Runtime Command Intake。 | +| C2 Multi-Tenant Scope | PASS | 新增查询分支必须同时按 `tenant_id`、Session、Agent 范围验证。 | +| C3 Idempotent Side Effects | PASS | 入站以飞书 `message_id` 幂等；静默路径不创建 `ChannelDelivery`；普通出站继续复用现有 outbox。 | +| C4 Wrapper Enforcement | PASS | 无新增前端 HTTP 请求；飞书发送继续通过现有 Provider sender。 | +| C5 DB/Performance | PASS | 无新表、无新 FK；压缩扫描保持批量扫描与现有 advisory lock/CAS。 | +| C6 Modularity | PASS | 静默识别作为小型纯函数；复用现有 Session Context policy/compactor/scanner。 | + +## Project Structure + +### Documentation (this feature) + +```text +specs/004-feishu-passive-listening/ +├── spec.md +├── design.md +├── plan.md +├── research.md +├── data-model.md +├── quickstart.md +├── contracts/ +│ └── feishu-passive-listening.md +├── checklists/ +│ └── requirements.md +└── tasks.md # 下一阶段生成 +``` + +### Source Code (repository root) + +```text +backend/app/ +├── api/feishu.py +├── services/ +│ ├── agent_runtime/ +│ │ ├── channel_delivery.py +│ │ ├── delivery.py +│ │ ├── model_step_service.py +│ │ ├── session_context_background.py +│ │ ├── session_context_compactor.py +│ │ └── session_context_service.py +│ └── llm/finish.py + +backend/tests/ +├── test_feishu_channel_runtime.py +├── test_agent_runtime_channel_delivery.py +├── test_agent_runtime_delivery.py +├── test_agent_runtime_session_context_background.py +├── test_agent_runtime_session_context_compactor.py +└── test_session_context_service.py + +frontend/src/ +└── components/ChannelConfig.tsx +``` + +**Structure Decision**: 保持现有 API → Channel Runtime Intake → Durable Runtime → Product Delivery → Provider Worker 分层。静默属于产品侧外部渠道投递过滤，不进入模型协议解析器或 Runtime graph；压缩属于现有 Session Context 子系统。 + +## Phase 0: Research Decisions + +详见 [research.md](./research.md)。已解决所有设计未知项：飞书全量权限、Provider 幂等键、OpenClaw 精确静默令牌、Clawith 投递截断点、飞书外部群压缩模型归属。 + +## Phase 1: Design + +### 1. 飞书入站 + +- 权限模板增加 `im:message.group_msg`；保留现有单聊、群 @ 与发送权限。 +- `im.message.receive_v1` 继续作为唯一事件入口。 +- Provider `message.message_id` 作为 `channel_message_id()` 的外部稳定输入；`event_id` 仅用于观测，不作为消息幂等权威事实。 +- 过滤机器人自身/机器人发送者；V1 处理飞书推送的用户消息，不扩展机器人间群消息。 +- API 在同一事务中持久化 `ChatMessage` 与 Runtime Command 后提交，模型执行不阻塞 Provider 回调。 + +### 2. 模型静默协议 + +- 仅飞书群 Run 的系统指令增加 token-only 规则。 +- 静默令牌常量为 `NO_REPLY`。 +- 识别函数只接受 `text.strip().casefold() == "no_reply"`；正文前后存在任何非空内容均不是静默。 +- 不修改 `finish`、`ModelIntent`、Verifier、checkpoint lifecycle 或 finalizer。 + +### 3. 出站抑制 + +- 在 `deliver_runtime_message()` 已确定实际 Session 和现有 `channel_delivery` route 后、调用 `stage_channel_delivery()` 前判定。 +- 必须同时满足：`kind=terminal`、`lifecycle_status=completed`、route channel 为 `feishu`、target `receive_id_type=chat_id`、内容为精确静默令牌。 +- 命中后仍保留内部 Assistant `ChatMessage` 和常规本地 delivery receipt，使 Run 能正常 settled；不创建 `ChannelDelivery`，因此 Provider worker 无工作项且不会调用飞书 API。 +- `agent_runs.delivery_status` 沿用“产品 Session 已投递”的既有含义：无 outbox 时为 `delivered`。审计通过终态内容为精确令牌且缺少对应 `channel_deliveries` 行证明有意静默；不新增数据库状态。 +- waiting、failed、cancelled 不进入静默判定，沿用既有投递策略。 + +### 4. 模型可见历史过滤 + +- 底层 `ChatMessage(content="NO_REPLY")` 保留用于审计。 +- 仅对 `session_type=group AND source_channel=feishu AND group_id IS NULL` 的 Session，Session Context 读取与 compactable 集合过滤精确静默 Assistant 消息。 +- 用户消息、正常 Assistant 回复、正文包含 `NO_REPLY` 的消息不被过滤。 +- 过滤不删除数据库行、不改变消息时间线水位线的权威位置；压缩水位线仍由最后一个实际纳入压缩的消息 ID 决定。 + +### 5. 飞书外部群 Session 压缩 + +- Session 判定：`session_type=group`、`group_id IS NULL`、`source_channel=feishu`、`agent_id IS NOT NULL`、未删除。 +- Policy resolver 验证 Session Agent 同租户、可用且未删除，使用该 Agent 的 active model 计算阈值；`source_agent_id=session.agent_id`。 +- LLMSessionContextCompactor 对此外部群使用该 Agent active model并记录 usage_agent_id；原生群仍使用 tenant multi-agent compact model。 +- Scanner 使用两个明确分支联合选择原生群候选和飞书外部群候选，保持现有批量游标、advisory lock 和 CAS 提交。 +- 不创建 `Group`/`GroupMember`，不写 `chat_sessions.group_id`，不修改外部会话 ID。 + +### 6. 测试顺序 + +1. 先加入精确静默识别与零 `ChannelDelivery` 回归测试。 +2. 加入飞书 `message_id` 重试幂等测试。 +3. 加入飞书外部群 policy/model selection/scanner 测试。 +4. 加入 Session Context 过滤和水位线测试。 +5. 实现最小代码变更。 +6. 运行 scoped pytest、Ruff、`scripts/arch-guard.sh`；前端权限常量改动后运行前端 test/build。 + +## Post-Design Constitution Re-check + +Phase 1 后仍全部 PASS。特别确认：静默不会修改 Runtime checkpoint state machine；压缩继续使用唯一 `session_context_states` 真相与 CAS；普通飞书发送继续由 `channel_deliveries` 和 Provider receipt 管理。 + +## Complexity Tracking + +无 Constitution 违规，无需例外批准。 diff --git a/specs/004-feishu-passive-listening/quickstart.md b/specs/004-feishu-passive-listening/quickstart.md new file mode 100644 index 000000000..1c767e05f --- /dev/null +++ b/specs/004-feishu-passive-listening/quickstart.md @@ -0,0 +1,58 @@ +# Quickstart: 飞书群常驻 Agent V1 验证 + +## Preconditions + +1. 使用测试租户和测试 Agent。 +2. 飞书自建应用开启机器人能力并订阅消息接收事件。 +3. 飞书应用取得并发布全量群用户消息权限。 +4. 将机器人加入隔离测试群。 + +## Local Contract Checks + +```bash +cd backend +.venv/bin/python -m pytest \ + tests/test_feishu_channel_runtime.py \ + tests/test_agent_runtime_channel_delivery.py \ + tests/test_agent_runtime_delivery.py \ + tests/test_agent_runtime_session_context_background.py \ + tests/test_agent_runtime_session_context_compactor.py \ + tests/test_session_context_service.py + +.venv/bin/ruff check \ + app/api/feishu.py \ + app/services/agent_runtime/channel_delivery.py \ + app/services/agent_runtime/delivery.py \ + app/services/agent_runtime/model_step_service.py \ + app/services/agent_runtime/session_context_background.py \ + app/services/agent_runtime/session_context_compactor.py \ + app/services/agent_runtime/session_context_service.py +``` + +如前端权限模板发生变化： + +```bash +cd frontend +npm test +npm run build +``` + +架构检查： + +```bash +scripts/arch-guard.sh +``` + +## Manual Scenarios + +1. 普通群消息、不 @Agent：确认产生一个入站 ChatMessage 和一个 Run。 +2. 重放同一 Provider message_id 三次：确认仍只有一个消息和一个 Run。 +3. 模型输出精确 `NO_REPLY`：确认内部 Run completed、无 ChannelDelivery、飞书群无消息。 +4. 模型输出 `正文\nNO_REPLY`：确认正常发送完整正文。 +5. 飞书私聊输出相同 token：确认本版未改变私聊路径。 +6. 生成超过压缩阈值的群历史：确认 SessionContextState 水位线前进，原始 ChatMessage 不减少。 + +## Evidence Boundary + +- 单元测试证明本地契约，不证明飞书权限已审批或线上事件实际到达。 +- 真实飞书验证必须分别提供：事件到达、Run 完成、无 outbox/有 outbox、Provider 发送结果和群内可见性的证据。 diff --git a/specs/004-feishu-passive-listening/research.md b/specs/004-feishu-passive-listening/research.md new file mode 100644 index 000000000..6cc64ed52 --- /dev/null +++ b/specs/004-feishu-passive-listening/research.md @@ -0,0 +1,73 @@ +# Research: 飞书群常驻 Agent V1 + +## Decision 1: 使用飞书消息事件接收全部群用户消息 + +**Decision**: 继续订阅 `im.message.receive_v1`，新增敏感权限 `im:message.group_msg`。 + +**Rationale**: 飞书官方事件根据应用权限决定推送范围；现有 `im:message.group_at_msg:readonly` 只能收到群内 @ 机器人消息，`im:message.group_msg` 才覆盖机器人所在群的全部用户消息。 + +**Alternatives considered**: + +- 定时调用历史消息 API：增加延迟、分页与重复读取复杂度，不适合实时常驻 Agent。 +- 保持群 @ 权限：无法实现普通消息进入上下文。 + +**Primary source**: [飞书接收消息事件](https://open.feishu.cn/document/server-docs/im-v1/message/events/receive?lang=zh-CN) + +## Decision 2: 入站幂等使用 Provider message_id + +**Decision**: 使用事件体中的飞书 `message.message_id` 生成本地稳定消息 ID；`event_id` 不作为消息幂等权威键。 + +**Rationale**: 飞书官方明确提示特殊情况下可能重复推送，应使用 `message_id` 去重而不是依赖 `event_id`。 + +**Alternatives considered**: + +- 继续优先 `event_id`：同一消息若以不同事件投递会重复执行。 +- 仅用内存集合：进程重启和多 worker 下不可靠。 + +## Decision 3: 采用 OpenClaw token-only 静默模式 + +**Decision**: 模型无需发言时输出精确 `NO_REPLY`；仅 token-only、大小写不敏感、允许首尾空白的结果静默。 + +**Rationale**: OpenClaw 当前将精确静默令牌从出站 payload 中过滤，并专门限制为 token-only，以避免吞掉正文末尾包含 `NO_REPLY` 的有效回答。 + +**Alternatives considered**: + +- 新增 Runtime `no_reply` intent：超过用户要求，扩大 checkpoint/Verifier/终态契约。 +- 空字符串：被现有 finish、node transition、Verifier 和 completed checkpoint 规则拒绝。 +- `endswith("NO_REPLY")`：存在吞掉有效正文的已知风险。 + +**Primary sources**: [OpenClaw agent loop](https://github.com/openclaw/openclaw/blob/main/docs/concepts/agent-loop.md), [OpenClaw tokens.ts](https://github.com/openclaw/openclaw/blob/main/src/auto-reply/tokens.ts) + +## Decision 4: 在外部 ChannelDelivery 建立前抑制 + +**Decision**: 保留正常 Runtime 完成和内部 Assistant ChatMessage，只跳过飞书群 `ChannelDelivery` 建立。 + +**Rationale**: 用户要求仅为“不发到飞书群”。该位置能确保 Provider worker 没有可发送 outbox，同时不修改 Runtime state machine，也不伪造发送失败。 + +**Alternatives considered**: + +- 在模型解析层吞掉：会触发非空 finish/Verifier 修复。 +- 在飞书 Sender 内丢弃：已经创建 pending outbox，容易产生状态与重试语义不一致。 +- 删除内部 ChatMessage：削弱审计且扩大 delivery 事务差异。 + +## Decision 5: 飞书外部群复用现有 Session Context + +**Decision**: 扩展现有 policy resolver、compactor model selection 和 scanner，使 `group_id IS NULL` 的飞书群 Session 使用 `session.agent_id` 对应模型预算。 + +**Rationale**: 当前外部飞书群已经是 `session_type=group`，但没有 Clawith 原生 `group_id`；现有 scanner inner join `groups`，因此不会压缩。复用现有水位线、advisory lock、CAS 和 summary schema 能避免第二套上下文状态。 + +**Alternatives considered**: + +- 将飞书 chat_id 写入原生 group_id：类型和领域都错误。 +- 为飞书新建上下文表：形成重复状态真相。 +- 暂不压缩：全量群消息会造成无界 pending history。 + +## Decision 6: V1 不统一其他渠道 + +**Decision**: 不迁移外部 ID，不建立统一 Conversation Adapter，不修改企微/钉钉/Slack/Teams/Discord。 + +**Rationale**: 用户明确选择先验证飞书临时版本，等更多渠道具备相同行为后再根据真实差异统一。 + +**Alternatives considered**: + +- 本次完成跨渠道统一：范围与迁移风险显著增加，且不是验证核心体验所必需。 diff --git a/specs/004-feishu-passive-listening/spec.md b/specs/004-feishu-passive-listening/spec.md new file mode 100644 index 000000000..63287dbc6 --- /dev/null +++ b/specs/004-feishu-passive-listening/spec.md @@ -0,0 +1,156 @@ +# Feature Specification: 飞书群常驻 Agent V1 + +**Feature Branch**: `004-feishu-passive-listening` +**Created**: 2026-08-18 +**Status**: Implemented and live-validated on 3010 +**Input**: 用户希望将 Agent 放入飞书群后接收群内全部用户消息；每条消息均进入 Agent 判断，需要参与时正常回复，不需要参与时输出精确静默令牌 `NO_REPLY`，且不得向飞书群发送该令牌或任何占位回复。长期复用的飞书群 Session 必须具备有界上下文和滚动压缩能力。 + +## User Scenarios & Testing *(mandatory)* + +### User Story 1 - Agent 旁听飞书群并按需发言 (Priority: P1) + +企业管理员将已配置的 Agent 机器人加入飞书群并授予读取群内全部用户消息的权限。此后，群成员无需每次 `@Agent`，群消息也能进入该 Agent 对应的群会话，由 Agent 结合职责和上下文判断是否需要发言。 + +**Why this priority**: 这是“常驻群成员”体验成立的前提；如果普通群消息无法进入 Agent，后续静默判断和长期上下文都没有意义。 + +**Independent Test**: 在机器人已加入且具有全量群消息权限的测试群发送一条不包含 `@Agent` 的普通用户消息，验证消息只被接收一次、进入正确的飞书群 Session，并触发一次 Agent 判断。 + +**Acceptance Scenarios**: + +1. **Given** Agent 机器人已加入飞书群且拥有全量用户消息权限，**When** 群成员发送一条不包含 `@Agent` 的文本消息，**Then** 系统持久化该消息并为对应 Agent 发起一次判断。 +2. **Given** 同一 Agent 同时存在飞书私聊和多个飞书群，**When** 任一群收到消息，**Then** 消息只进入该 Agent 与该飞书群对应的 Session，不进入私聊或其他群 Session。 +3. **Given** 飞书对同一消息重复推送事件，**When** 系统重复收到该消息，**Then** 只保留一条入站消息并只发起一次 Agent 判断。 +4. **Given** 消息由机器人自身或不在本版本支持范围内的发送者产生，**When** 系统收到事件，**Then** 不形成会导致机器人自我回复的循环。 + +--- + +### User Story 2 - 无需参与时保持群内静默 (Priority: P1) + +Agent 判断当前群消息不需要自己参与时，最终输出精确静默令牌 `NO_REPLY`。该次判断正常结束并保留必要的内部审计事实，但群成员看不到 `NO_REPLY`、空白消息、“无需回复”或其他占位内容。 + +**Why this priority**: 全量接收会显著增加 Agent 被调用的次数；如果每次调用都在群内发言，常驻 Agent 会造成严重干扰。 + +**Independent Test**: 让模型分别产生精确静默令牌、正常回答以及“正常回答后附带令牌”三种结果，验证只有精确静默令牌不产生飞书群出站调用。 + +**Acceptance Scenarios**: + +1. **Given** 一次飞书群判断的最终内容去除首尾空白后大小写不敏感地等于 `NO_REPLY`，**When** 判断正常完成，**Then** 系统不向该飞书群调用发送消息能力。 +2. **Given** 最终内容是非空正常回答，**When** 判断正常完成，**Then** 系统沿用现有可靠投递机制向原飞书群发送回答。 +3. **Given** 最终内容包含正常正文并在末尾出现 `NO_REPLY`，**When** 判断正常完成，**Then** 系统将整段内容视为正常回答，不得误判为静默。 +4. **Given** 最终内容为精确 `NO_REPLY`，**When** 查询内部运行记录，**Then** 可以确认该次判断已完成且出站被有意抑制，而不是模型失败或飞书发送失败。 +5. **Given** 飞书私聊、Clawith 原生群或其他外部渠道产生相同文本，**When** 判断完成，**Then** 本版本的飞书群静默规则不改变这些既有路径。 + +--- + +### User Story 3 - Agent 在长期群聊中保留可用上下文 (Priority: P1) + +同一飞书群 Session 可以长期接收大量群消息。Agent 每次判断都能获得近期原文与较早内容的滚动摘要，同时历史增长不会让每次判断读取无限消息或超过模型可用上下文。 + +**Why this priority**: 全量群消息会比 `@Agent` 模式更快累积历史；没有压缩会使成本、延迟和上下文溢出风险持续增长。 + +**Independent Test**: 在一个飞书群 Session 中生成超过压缩阈值的消息，验证压缩水位线前进、近期窗口仍保留原文、后续判断使用摘要加近期消息，且原始消息记录未被删除。 + +**Acceptance Scenarios**: + +1. **Given** 飞书群 Session 的待处理历史达到既有压缩条件，**When** 后台压缩执行，**Then** 系统推进该 Session 的上下文水位线并生成可供后续判断使用的滚动摘要。 +2. **Given** 较早消息已经进入滚动摘要，**When** 新群消息触发 Agent 判断，**Then** Agent 获得摘要、尚未压缩的消息和近期原文，而不是重新装载全部历史。 +3. **Given** Agent 多次输出精确 `NO_REPLY`，**When** 构造后续模型上下文，**Then** 纯静默输出不会作为有意义的群聊内容反复占用上下文预算。 +4. **Given** 飞书群属于外部渠道，**When** 执行上下文压缩，**Then** 系统使用该 Session 所属 Agent 的有效模型预算，不要求或伪造 Clawith 原生群身份。 +5. **Given** 上下文压缩失败，**When** 后续扫描再次执行，**Then** 原始消息仍然完整，水位线不错误前进，且失败可被运维人员识别。 + +--- + +### User Story 4 - 管理员能够正确开通飞书能力 (Priority: P2) + +配置 Agent 飞书渠道的用户能够看到并申请接收群内全部用户消息所需的权限，并了解这是敏感权限且需要在飞书侧发布后才能生效。 + +**Why this priority**: 服务端实现只有在飞书实际推送普通群消息时才能工作；权限遗漏会让功能表面启用但始终只收到 `@Agent` 消息。 + +**Independent Test**: 检查产品提供的飞书权限配置包含全量群消息权限，并验证未开通与已开通两种飞书应用配置下的可观察行为符合说明。 + +**Acceptance Scenarios**: + +1. **Given** 用户查看飞书渠道配置指南，**When** 复制或核对权限列表，**Then** 能看到接收群内全部用户消息所需的敏感权限及发布提示。 +2. **Given** 飞书应用尚未取得该权限，**When** 普通群消息未到达系统，**Then** 产品说明不会错误宣称全量监听已经生效。 + +### Edge Cases + +- 飞书重复推送同一 `message_id`，但 `event_id` 不同。 +- 同一群消息包含文字、富文本、图片或文件；沿用现有已支持类型，不因静默判断重复下载或重复入队。 +- 消息只有对机器人的 `@` 占位符，移除 mention 后没有可判断的正文。 +- 模型返回 `NO_REPLY`、`no_reply` 或带首尾空白的等价形式。 +- 模型返回 `NO_REPLY：因为……`、正文加 `NO_REPLY`、代码块中的 `NO_REPLY`；这些都不是精确静默结果。 +- Agent Run 成功但静默，必须与 Provider 调用失败区分。 +- Agent Run 失败、取消或等待外部输入时不得被静默规则误判为正常 `NO_REPLY`。 +- 高活跃群在压缩工作尚未完成时继续收到新消息；水位线必须保持单调且不跨越未纳入摘要的消息。 +- 飞书群 Session 的外部群标识不是 Clawith 原生 `group_id`，不得创建伪造的原生群记录。 +- 多租户中相同飞书群标识或相同 Agent 名称不得造成跨租户 Session、消息或摘要混用。 + +## Requirements *(mandatory)* + +### Functional Requirements + +- **FR-001**: 系统 MUST 能接收已授权飞书机器人所在群聊中的全部用户消息，而不仅是明确 `@机器人` 的消息。 +- **FR-002**: 系统 MUST 将每条支持的飞书群入站消息归入正确租户、正确 Agent 和正确外部群 Session。 +- **FR-003**: 系统 MUST 使用飞书消息自身的稳定标识保证入站幂等，不得仅依赖一次事件投递的标识。 +- **FR-004**: 系统 MUST 对机器人自身消息和重复消息实施循环与重复执行保护。 +- **FR-005**: 每条被接受的飞书群用户消息 MUST 进入一次 Agent 判断；V1 不引入独立 Activation Gate。 +- **FR-006**: 飞书群判断 Prompt MUST 明确告知 Agent：无需在群内发言时，最终内容只能是精确 `NO_REPLY`。 +- **FR-007**: 系统 MUST 仅在成功完成的飞书群判断最终内容去除首尾空白后大小写不敏感地精确等于 `NO_REPLY` 时抑制出站。 +- **FR-008**: 命中静默规则时，系统 MUST 不调用飞书群发送消息能力，且群内不得出现令牌、空白消息或替代占位内容。 +- **FR-009**: 系统 MUST NOT 使用后缀、子串或模糊匹配判断静默；包含任何其他可见正文的结果 MUST 按正常回答处理。 +- **FR-010**: 系统 MUST 保留足以区分“正常静默完成”“模型失败”“运行取消”和“飞书投递失败”的内部审计事实。 +- **FR-011**: V1 的静默抑制 MUST 仅作用于飞书群最终回复，不改变飞书私聊、Clawith 原生群和其他外部渠道的既有完成与投递行为。 +- **FR-012**: 系统 MUST 继续使用现有飞书群 Session，不得将飞书群 ID 写入或伪装为 Clawith 原生 `group_id`。 +- **FR-013**: 飞书群 Session MUST 使用现有 Session Context 机制维护滚动摘要、近期原文与压缩水位线，不新增另一套上下文状态机。 +- **FR-014**: 飞书外部群 Session 的压缩预算 MUST 来自该 Session 所属 Agent 的有效模型配置，而不是原生群成员列表。 +- **FR-015**: 纯 `NO_REPLY` Assistant 输出 MUST NOT 作为有意义的模型可见历史反复进入后续 Session Context，但底层审计记录可以保留。 +- **FR-016**: 压缩 MUST 保留原始消息记录；压缩失败时不得推进水位线或丢弃消息。 +- **FR-017**: 产品提供的飞书权限配置和指南 MUST 包含接收群内全部用户消息所需的敏感权限及飞书侧发布要求。 +- **FR-018**: 所有新增读取、写入、幂等与压缩操作 MUST 维持严格租户范围。 +- **FR-019**: V1 MUST NOT 统一或迁移飞书、企微、钉钉、Slack、Teams、Discord 的外部会话 ID 数据结构。 +- **FR-020**: V1 MUST NOT 新增依赖、创建第二套 Run 生命周期状态机或改变现有飞书出站成功的权威判定。 + +### Key Entities + +- **飞书群 Session**: 一个 Agent 与一个飞书群的长期逻辑会话；保留现有外部会话映射，不具有 Clawith 原生群身份。 +- **飞书入站消息**: 飞书推送的用户消息，包含稳定消息标识、发送者、群会话标识、消息类型和内容。 +- **Agent 判断 Run**: 针对一条被接受群消息执行的一次 Durable Runtime 判断；可能产生正常回答或精确静默令牌。 +- **静默结果**: 最终可见文本精确为 `NO_REPLY` 的成功完成结果；它只改变飞书群出站行为。 +- **Session Context**: 按 Session 保存的滚动摘要、压缩水位线和近期消息窗口；原始消息仍由消息记录持有。 +- **渠道投递记录**: 仅在需要真正向飞书发送内容时建立的可靠出站事实。 + +## Success Criteria *(mandatory)* + +### Measurable Outcomes + +- **SC-001**: 在具备全量群消息权限的测试群中，100% 的受支持普通用户消息能够进入正确 Session，且无需 `@Agent`。 +- **SC-002**: 对同一飞书消息进行至少 3 次重复事件投递时，系统只产生 1 条入站消息和 1 次 Agent 判断。 +- **SC-003**: 对精确 `NO_REPLY`、大小写变化和首尾空白共至少 6 个静默样例，飞书发送调用次数均为 0。 +- **SC-004**: 对正文包含或结尾附带 `NO_REPLY` 的至少 6 个非静默样例，均不被错误抑制。 +- **SC-005**: 正常回答仍通过现有可靠投递路径到达原飞书群，并能取得 Provider 成功或明确失败证据。 +- **SC-006**: 超过 Session 压缩阈值后，后续 Agent 判断不需要读取完整群历史；摘要水位线前进且近期原文仍可用。 +- **SC-007**: 压缩前后的原始飞书群消息数量和内容保持不变，压缩失败测试中水位线保持不变。 +- **SC-008**: 飞书私聊、原生群以及至少一个其他外部渠道的既有投递回归测试全部通过。 +- **SC-009**: 功能上线后，群内不会出现由精确静默令牌产生的 `NO_REPLY`、空白消息或“无需回复”占位内容。 + +## Assumptions + +- V1 仅覆盖飞书群；其他渠道继续维持当前接收与投递行为。 +- 飞书应用已经开启机器人能力、订阅现有消息接收事件，并由管理员申请及发布全量群消息敏感权限。 +- 每条被接受消息都调用现有 Agent 模型；独立低成本 Activation Gate、批处理与动态关注规则不在 V1 范围。 +- `@Agent` 是强相关信号，但 V1 不新增独立的 mention 调度状态机。 +- `NO_REPLY` 是模型控制令牌而不是面向群成员的内容；仅精确 token-only 结果触发静默。 +- V1 允许内部保留 `NO_REPLY` 审计记录，但不得将它作为有意义的后续群上下文。 +- 模型失败、取消和等待状态沿用现有 Runtime 语义；它们不是 `NO_REPLY`。 +- 现有飞书群 Session 映射继续通过既有渠道字段工作；本功能不迁移外部渠道身份结构。 + +## Out of Scope + +- 企微、钉钉、Slack、Teams、Discord 的全量群消息监听。 +- 跨渠道统一 Conversation Adapter 或新增渠道会话表。 +- Agent 动态修改 Activation Gate。 +- 在模型调用前进行独立相关性分类。 +- 用 `NO_REPLY` 抑制飞书私聊、Web Chat、原生群或其他渠道。 +- 删除内部静默运行记录或原始群消息。 +- 为本功能引入新的第三方依赖。 From 15b503a464f2f202de175c4ab78c49e95e41fb9e Mon Sep 17 00:00:00 2001 From: Y1fe1Zh0u Date: Wed, 19 Aug 2026 09:57:54 +0800 Subject: [PATCH 2/5] Keep external group conversations discoverable External group Sessions store the Agent creator as a placeholder user, so user-id-only filtering hid them from the admin-facing Other sessions list. Classify group and Agent sessions by their explicit session metadata before applying direct-session ownership filtering. Constraint: External Feishu groups must remain group_id=NULL and keep the existing placeholder user contract. Rejected: Move external groups into My sessions | that surface is intentionally restricted to writable Direct sessions. Confidence: high Scope-risk: narrow Directive: Do not use ChatSession.user_id to classify external group ownership. Tested: frontend npm test (112 passed); frontend npm run build Not-tested: 3010 deployment and live Feishu group visibility --- .../pages/agent-detail/AgentDetailPage.tsx | 17 ++----- .../pages/agent-detail/sessionVisibility.ts | 20 ++++++++ frontend/tests/sessionVisibility.test.mjs | 51 +++++++++++++++++++ 3 files changed, 74 insertions(+), 14 deletions(-) create mode 100644 frontend/src/pages/agent-detail/sessionVisibility.ts create mode 100644 frontend/tests/sessionVisibility.test.mjs diff --git a/frontend/src/pages/agent-detail/AgentDetailPage.tsx b/frontend/src/pages/agent-detail/AgentDetailPage.tsx index 09770d295..5c7d2a280 100644 --- a/frontend/src/pages/agent-detail/AgentDetailPage.tsx +++ b/frontend/src/pages/agent-detail/AgentDetailPage.tsx @@ -78,6 +78,7 @@ import { toolReconciliationsByCallId, waitingSessionActiveRunHint, } from './sessionRuntimeState'; +import { belongsInOtherSessions } from './sessionVisibility'; import { onboardingKickoffKey, shouldKickoffOnboarding } from './onboardingKickoff'; import { fetchAuth } from './utils/fetchAuth'; import { @@ -2617,10 +2618,6 @@ export default function AgentDetailPage() { /** Normalize IDs — API/JSON may use number vs string; loose equality was breaking "own session" detection. */ const sessionUserIdStr = (s: any) => (s?.user_id == null ? '' : String(s.user_id)); const viewerUserIdStr = () => (currentUser?.id == null ? '' : String(currentUser.id)); - const isAgentChatSession = (s: any) => - String(s?.source_channel || '').toLowerCase() === 'agent' || - String(s?.participant_type || '').toLowerCase() === 'agent'; - /** Ensure session shape from POST/list so P2P "mine" is never mistaken for read-only or agent thread. */ const normalizeChatSession = (sess: any) => { if (!sess || typeof sess !== 'object') return sess; @@ -2669,18 +2666,10 @@ export default function AgentDetailPage() { const isViewingOtherUsersSessions = canViewAllAgentChatSessions && chatScope === 'all'; - /** Sessions in scope=all that are not the current viewer's own P2P rows (for admin「其他用户」tab). - * Agent-to-agent sessions (source_channel === 'agent') store the creator's user_id, so we must - * exempt them from the user_id check — otherwise they'd always be hidden. */ + /** Sessions in scope=all that belong on the admin-facing "Other sessions" surface. */ const otherUsersSessions = useMemo(() => { const vu = viewerUserIdStr(); - return allSessions.filter((s: any) => { - // Always show agent-to-agent sessions in the "Other users" tab - if (isAgentChatSession(s)) return true; - const su = sessionUserIdStr(s); - if (vu && su === vu) return false; - return true; - }); + return allSessions.filter((s: any) => belongsInOtherSessions(s, vu)); }, [allSessions, currentUser?.id]); const othersListForPicker = otherUsersSessions; diff --git a/frontend/src/pages/agent-detail/sessionVisibility.ts b/frontend/src/pages/agent-detail/sessionVisibility.ts new file mode 100644 index 000000000..f23098f96 --- /dev/null +++ b/frontend/src/pages/agent-detail/sessionVisibility.ts @@ -0,0 +1,20 @@ +interface SessionVisibilityInput { + user_id?: string | number | null; + source_channel?: string | null; + participant_type?: string | null; + is_group?: boolean | null; +} + +export function belongsInOtherSessions( + session: SessionVisibilityInput, + viewerUserId: string, +): boolean { + const sourceChannel = String(session.source_channel || '').toLowerCase(); + const participantType = String(session.participant_type || '').toLowerCase(); + + if (session.is_group || participantType === 'group') return true; + if (sourceChannel === 'agent' || participantType === 'agent') return true; + + const sessionUserId = session.user_id == null ? '' : String(session.user_id); + return !viewerUserId || sessionUserId !== viewerUserId; +} diff --git a/frontend/tests/sessionVisibility.test.mjs b/frontend/tests/sessionVisibility.test.mjs new file mode 100644 index 000000000..38b11c9ce --- /dev/null +++ b/frontend/tests/sessionVisibility.test.mjs @@ -0,0 +1,51 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; + +import { belongsInOtherSessions } from '../src/pages/agent-detail/sessionVisibility.ts'; + +test('external group sessions stay visible when their placeholder owner is the viewer', () => { + assert.equal( + belongsInOtherSessions({ + user_id: 'viewer-1', + source_channel: 'feishu', + participant_type: 'group', + is_group: true, + }, 'viewer-1'), + true, + ); +}); + +test('agent sessions stay visible regardless of their stored user id', () => { + assert.equal( + belongsInOtherSessions({ + user_id: 'viewer-1', + source_channel: 'agent', + participant_type: 'agent', + }, 'viewer-1'), + true, + ); +}); + +test('the viewer own direct session stays out of other sessions', () => { + assert.equal( + belongsInOtherSessions({ + user_id: 'viewer-1', + source_channel: 'web', + participant_type: 'user', + is_group: false, + }, 'viewer-1'), + false, + ); +}); + +test('another user direct session remains visible', () => { + assert.equal( + belongsInOtherSessions({ + user_id: 'user-2', + source_channel: 'web', + participant_type: 'user', + is_group: false, + }, 'viewer-1'), + true, + ); +}); From 86da9ab7fdcb9bc6b4840c26770fcdd6147f37ea Mon Sep 17 00:00:00 2001 From: Y1fe1Zh0u Date: Wed, 19 Aug 2026 10:24:59 +0800 Subject: [PATCH 3/5] Acknowledge only Feishu group messages that earn a reply Preserve the provider message identity through the durable delivery route and add the GLANCE reaction only after a completed terminal response has passed exact NO_REPLY suppression. Reaction failures remain cosmetic so they cannot block or duplicate the durable text reply. The Feishu permission templates now request the matching write-only reaction scope. Constraint: The reaction must not run on intake, waiting states, direct chats, or exact NO_REPLY completions. Rejected: Add the reaction in the webhook handler | that runs before the Agent NO_REPLY decision. Rejected: Call Feishu from the Runtime delivery transaction | external side effects must stay behind the durable channel outbox. Confidence: high Scope-risk: narrow Directive: Keep source_message_id provider-native and gate future acknowledgement reactions on terminal completed delivery. Tested: 68 scoped backend tests; 112 frontend tests; frontend production build; scoped Ruff; git diff check; architecture guard Not-tested: Live Feishu group reaction rendering and newly approved application scope --- backend/app/api/feishu.py | 5 + .../agent_runtime/channel_delivery.py | 3 + .../channel_provider_delivery.py | 33 ++++++ .../app/services/agent_runtime/delivery.py | 12 ++ backend/app/services/feishu_service.py | 22 ++++ ...agent_runtime_channel_provider_delivery.py | 107 ++++++++++++++++++ backend/tests/test_agent_runtime_delivery.py | 1 + backend/tests/test_feishu_channel_runtime.py | 1 + backend/tests/test_feishu_service_api.py | 31 +++++ frontend/src/components/ChannelConfig.tsx | 6 +- 10 files changed, 219 insertions(+), 2 deletions(-) diff --git a/backend/app/api/feishu.py b/backend/app/api/feishu.py index 7b3f5998f..fec3fe71d 100644 --- a/backend/app/api/feishu.py +++ b/backend/app/api/feishu.py @@ -441,6 +441,11 @@ async def _accept_feishu_runtime_message( 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, diff --git a/backend/app/services/agent_runtime/channel_delivery.py b/backend/app/services/agent_runtime/channel_delivery.py index 96cdfd73f..24f5fa463 100644 --- a/backend/app/services/agent_runtime/channel_delivery.py +++ b/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, diff --git a/backend/app/services/agent_runtime/channel_provider_delivery.py b/backend/app/services/agent_runtime/channel_provider_delivery.py index f502f3a90..de6d3b1b6 100644 --- a/backend/app/services/agent_runtime/channel_provider_delivery.py +++ b/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, diff --git a/backend/app/services/agent_runtime/delivery.py b/backend/app/services/agent_runtime/delivery.py index c0c3c2f24..a851435be 100644 --- a/backend/app/services/agent_runtime/delivery.py +++ b/backend/app/services/agent_runtime/delivery.py @@ -935,6 +935,17 @@ async def deliver_runtime_message( ) 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, @@ -942,6 +953,7 @@ async def deliver_runtime_message( message_id=message.id, idempotency_key=request.idempotency_key, clock=now, + target_overrides=reaction_target_overrides, ) receipt = DeliveryReceipt( tenant_id=run.tenant_id, diff --git a/backend/app/services/feishu_service.py b/backend/app/services/feishu_service.py index 081deb415..fbdf40030 100644 --- a/backend/app/services/feishu_service.py +++ b/backend/app/services/feishu_service.py @@ -429,6 +429,28 @@ async def patch_message( data = self._parse_api_response(resp, stage=stage, message_id=message_id) return data + async def add_message_reaction( + self, + app_id: str, + app_secret: str, + message_id: str, + emoji_type: str, + stage: str = "add_message_reaction", + ) -> dict: + """Add one bot-identity reaction to an existing Feishu message.""" + async with httpx.AsyncClient() as client: + token_resp = await client.post( + FEISHU_APP_TOKEN_URL, + json={"app_id": app_id, "app_secret": app_secret}, + ) + app_token = token_resp.json().get("app_access_token", "") + resp = await client.post( + f"{FEISHU_SEND_MSG_URL}/{message_id}/reactions", + json={"reaction_type": {"emoji_type": emoji_type}}, + headers={"Authorization": f"Bearer {app_token}"}, + ) + return self._parse_api_response(resp, stage=stage, message_id=message_id) + async def resolve_open_id(self, app_id: str, app_secret: str, email: str | None = None, mobile: str | None = None) -> str | None: """Resolve a user's open_id for a specific app using email or mobile. diff --git a/backend/tests/test_agent_runtime_channel_provider_delivery.py b/backend/tests/test_agent_runtime_channel_provider_delivery.py index 6467607d4..78797bcbc 100644 --- a/backend/tests/test_agent_runtime_channel_provider_delivery.py +++ b/backend/tests/test_agent_runtime_channel_provider_delivery.py @@ -141,6 +141,113 @@ async def send_message(*args, **kwargs): assert calls["kwargs"]["stage"] == "runtime_channel_delivery" # type: ignore[index] +@pytest.mark.asyncio +async def test_feishu_group_delivery_reacts_to_source_message_before_reply( + monkeypatch, +) -> None: + calls: list[tuple[str, str]] = [] + + async def add_message_reaction(*_args, **kwargs): + calls.append(("reaction", kwargs["stage"])) + return {"code": 0, "data": {"reaction_id": "reaction-1"}} + + async def send_message(*_args, **kwargs): + calls.append(("message", kwargs["stage"])) + return {"code": 0, "data": {"message_id": "om-1"}} + + monkeypatch.setattr( + feishu_service.feishu_service, + "add_message_reaction", + add_message_reaction, + ) + monkeypatch.setattr(feishu_service.feishu_service, "send_message", send_message) + + await _sender(_config()).send( + _envelope( + "feishu", + { + "receive_id": "oc-1", + "receive_id_type": "chat_id", + "source_message_id": "om-source-1", + "reaction_emoji_type": "GLANCE", + }, + ) + ) + + assert calls == [ + ("reaction", "runtime_group_reply_reaction"), + ("message", "runtime_channel_delivery"), + ] + + +@pytest.mark.asyncio +async def test_feishu_group_delivery_continues_when_reaction_fails(monkeypatch) -> None: + sent = False + + async def add_message_reaction(*_args, **_kwargs): + raise RuntimeError("reaction unavailable") + + async def send_message(*_args, **_kwargs): + nonlocal sent + sent = True + return {"code": 0, "data": {"message_id": "om-1"}} + + monkeypatch.setattr( + feishu_service.feishu_service, + "add_message_reaction", + add_message_reaction, + ) + monkeypatch.setattr(feishu_service.feishu_service, "send_message", send_message) + + await _sender(_config()).send( + _envelope( + "feishu", + { + "receive_id": "oc-1", + "receive_id_type": "chat_id", + "source_message_id": "om-source-1", + "reaction_emoji_type": "GLANCE", + }, + ) + ) + + assert sent is True + + +@pytest.mark.asyncio +async def test_feishu_group_delivery_without_completed_reply_marker_skips_reaction( + monkeypatch, +) -> None: + reacted = False + + async def add_message_reaction(*_args, **_kwargs): + nonlocal reacted + reacted = True + + async def send_message(*_args, **_kwargs): + return {"code": 0, "data": {"message_id": "om-1"}} + + monkeypatch.setattr( + feishu_service.feishu_service, + "add_message_reaction", + add_message_reaction, + ) + monkeypatch.setattr(feishu_service.feishu_service, "send_message", send_message) + + await _sender(_config()).send( + _envelope( + "feishu", + { + "receive_id": "oc-1", + "receive_id_type": "chat_id", + "source_message_id": "om-source-1", + }, + ) + ) + + assert reacted is False + + @pytest.mark.asyncio async def test_dingtalk_delivery_uses_the_persisted_session_webhook(monkeypatch) -> None: client = _HTTPClient(_Response({"errcode": 0})) diff --git a/backend/tests/test_agent_runtime_delivery.py b/backend/tests/test_agent_runtime_delivery.py index 7d3fd478f..1a767e172 100644 --- a/backend/tests/test_agent_runtime_delivery.py +++ b/backend/tests/test_agent_runtime_delivery.py @@ -666,6 +666,7 @@ async def test_external_group_delivery_uses_channel_scope_without_native_members assert outbox[0].message_id == message.id assert outbox[0].channel == "feishu" assert outbox[0].target["receive_id"] == "oc_123" + assert outbox[0].target["reaction_emoji_type"] == "GLANCE" assert run.delivery_status == "pending" assert len(db.statements) == 5 diff --git a/backend/tests/test_feishu_channel_runtime.py b/backend/tests/test_feishu_channel_runtime.py index 125af583a..c9f1aecbe 100644 --- a/backend/tests/test_feishu_channel_runtime.py +++ b/backend/tests/test_feishu_channel_runtime.py @@ -185,6 +185,7 @@ async def enqueue(_db, **kwargs): assert intake_call["channel_delivery_target"] == { "receive_id": "oc_group_1", "receive_id_type": "chat_id", + "source_message_id": event_id, } assert intake_call["message_id"] == feishu.channel_message_id( agent_id, diff --git a/backend/tests/test_feishu_service_api.py b/backend/tests/test_feishu_service_api.py index 94361e84e..12144e6ab 100644 --- a/backend/tests/test_feishu_service_api.py +++ b/backend/tests/test_feishu_service_api.py @@ -73,6 +73,37 @@ async def test_patch_message_raises_when_business_code_nonzero(monkeypatch): ) +@pytest.mark.asyncio +async def test_add_message_reaction_uses_glance_emoji(monkeypatch): + client = _FakeAsyncClient() + calls: dict[str, object] = {} + + async def post(url, **kwargs): + if "app_access_token/internal" in url: + return _FakeResponse(200, {"app_access_token": "token_x"}) + calls["url"] = url + calls["kwargs"] = kwargs + return _FakeResponse(200, {"code": 0, "msg": "ok", "data": {}}) + + client.post = post + monkeypatch.setattr(feishu_service_module.httpx, "AsyncClient", lambda: client) + + await feishu_service_module.feishu_service.add_message_reaction( + "app_id", + "app_secret", + "om_source", + "GLANCE", + stage="unit_test_reaction", + ) + + assert calls["url"] == ( + "https://open.feishu.cn/open-apis/im/v1/messages/om_source/reactions" + ) + assert calls["kwargs"]["json"] == { # type: ignore[index] + "reaction_type": {"emoji_type": "GLANCE"} + } + + @pytest.mark.asyncio async def test_list_bot_chats_uses_app_identity_and_parses_groups(monkeypatch): client = _FakeAsyncClient( diff --git a/frontend/src/components/ChannelConfig.tsx b/frontend/src/components/ChannelConfig.tsx index 6ee30ff7d..be62d3cb0 100644 --- a/frontend/src/components/ChannelConfig.tsx +++ b/frontend/src/components/ChannelConfig.tsx @@ -231,7 +231,7 @@ const CHANNEL_REGISTRY: ChannelDef[] = [ ]; // ─── Feishu Permission JSON ───────────────────────────── -const FEISHU_PERM_BASIC_JSON = '{"scopes":{"tenant":["contact:contact.base:readonly","contact:user.base:readonly","contact:user.employee_id:readonly","contact:user.id:readonly","im:chat","im:message","im:message.group_at_msg:readonly","im:message.group_msg","im:message.p2p_msg:readonly","im:message:send_as_bot","im:resource"],"user":[]}}'; +const FEISHU_PERM_BASIC_JSON = '{"scopes":{"tenant":["contact:contact.base:readonly","contact:user.base:readonly","contact:user.employee_id:readonly","contact:user.id:readonly","im:chat","im:message","im:message.group_at_msg:readonly","im:message.group_msg","im:message.p2p_msg:readonly","im:message.reactions:write_only","im:message:send_as_bot","im:resource"],"user":[]}}'; const FEISHU_PERM_BASIC_DISPLAY = `{ "scopes": { @@ -245,6 +245,7 @@ const FEISHU_PERM_BASIC_DISPLAY = `{ "im:message.group_at_msg:readonly", "im:message.group_msg", "im:message.p2p_msg:readonly", + "im:message.reactions:write_only", "im:message:send_as_bot", "im:resource" ], @@ -252,7 +253,7 @@ const FEISHU_PERM_BASIC_DISPLAY = `{ } }`; -const FEISHU_PERM_FULL_JSON = '{"scopes":{"tenant":["approval:approval","base:app:create","base:dashboard:create","base:field_group:create","bitable:app","bitable:app:readonly","board:whiteboard:node:create","calendar:calendar.event:create","calendar:calendar.event:delete","calendar:calendar.event:read","calendar:calendar.event:update","calendar:calendar.free_busy:read","calendar:calendar:readonly","contact:contact.base:readonly","contact:user.base:readonly","contact:user.employee_id:readonly","contact:user.id:readonly","docx:document","docx:document:create","drive:drive","im:chat","im:message","im:message.group_at_msg:readonly","im:message.group_msg","im:message.p2p_msg:readonly","im:message:send_as_bot","im:resource","sheets:spreadsheet:create","slides:presentation:create","slides:presentation:write_only","wiki:wiki","wiki:wiki:readonly"],"user":[]}}'; +const FEISHU_PERM_FULL_JSON = '{"scopes":{"tenant":["approval:approval","base:app:create","base:dashboard:create","base:field_group:create","bitable:app","bitable:app:readonly","board:whiteboard:node:create","calendar:calendar.event:create","calendar:calendar.event:delete","calendar:calendar.event:read","calendar:calendar.event:update","calendar:calendar.free_busy:read","calendar:calendar:readonly","contact:contact.base:readonly","contact:user.base:readonly","contact:user.employee_id:readonly","contact:user.id:readonly","docx:document","docx:document:create","drive:drive","im:chat","im:message","im:message.group_at_msg:readonly","im:message.group_msg","im:message.p2p_msg:readonly","im:message.reactions:write_only","im:message:send_as_bot","im:resource","sheets:spreadsheet:create","slides:presentation:create","slides:presentation:write_only","wiki:wiki","wiki:wiki:readonly"],"user":[]}}'; const FEISHU_PERM_FULL_DISPLAY = `{ "scopes": { @@ -282,6 +283,7 @@ const FEISHU_PERM_FULL_DISPLAY = `{ "im:message.group_at_msg:readonly", "im:message.group_msg", "im:message.p2p_msg:readonly", + "im:message.reactions:write_only", "im:message:send_as_bot", "im:resource", "sheets:spreadsheet:create", From b3390946f80c54b33c9f3a92095001b8fe1fbaad Mon Sep 17 00:00:00 2001 From: Y1fe1Zh0u Date: Tue, 18 Aug 2026 16:50:13 +0800 Subject: [PATCH 4/5] Make Feishu contact search honor the configured bot scope MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Feishu user-search tool previously projected only pre-synced Clawith OrgMember rows, so a correctly configured bot could create calendar events while every attendee lookup returned zero. The tool now keeps the tenant-scoped projection as its first choice, then performs a bounded app-identity Contact v3 lookup across directly granted users and visible departments. Calendar attendee resolution privately reuses exact-name matches without exposing provider IDs to the model. Constraint: Provider open_id values must remain private to calendar execution Constraint: Local Tenant-scoped directory results remain authoritative when available Constraint: Bot authentication uses tenant_access_token; Feishu search/v1/user requires user_access_token Rejected: Treat empty local OrgMember results as missing Feishu permission | 3010 proved a valid token and one directly visible user Rejected: Persist live contacts from a read tool | introduces hidden cross-system writes and sync ownership Confidence: high Scope-risk: moderate Directive: Keep direct-user scope support even if department-based contact sync is expanded Tested: 140 scoped Feishu and Directory tests; focused Ruff; git diff --check; live 3010 read-only Contact API probe found 周逸飞 Not-tested: Deployed 3010 Agent Run and calendar invitation after this patch --- backend/app/services/agent_tools.py | 70 +++- .../app/services/builtin_tool_definitions.py | 2 +- backend/app/services/feishu_contact_search.py | 318 ++++++++++++++++++ ...test_agent_tools_typed_feishu_remaining.py | 104 +++++- backend/tests/test_feishu_contact_search.py | 174 ++++++++++ 5 files changed, 662 insertions(+), 6 deletions(-) create mode 100644 backend/app/services/feishu_contact_search.py create mode 100644 backend/tests/test_feishu_contact_search.py diff --git a/backend/app/services/agent_tools.py b/backend/app/services/agent_tools.py index 3e82cbf6e..2d816fb36 100644 --- a/backend/app/services/agent_tools.py +++ b/backend/app/services/agent_tools.py @@ -74,6 +74,7 @@ resolve_feishu_group_target, ) from app.services import agent_directory +from app.services.feishu_contact_search import search_feishu_contacts from app.services.workspace_collaboration import ( delete_workspace_file, move_workspace_path, @@ -18491,7 +18492,7 @@ async def _feishu_user_search_outcome( agent_id: uuid.UUID, arguments: dict, ) -> ToolExecutionOutcome: - """Project tenant-scoped Directory facts without exposing Provider IDs.""" + """Search synced contacts first, then the Agent app's live Feishu scope.""" query = arguments.get("query") if not isinstance(query, str) or not query.strip(): return _typed_failure( @@ -18582,6 +18583,47 @@ async def _feishu_user_search_outcome( "query_directory_failed", retryable=True, ) + if not members: + token, token_error = await _feishu_access_token_outcome(agent_id) + if token_error is not None or token is None: + return token_error or _typed_failure( + "Feishu did not return a tenant access token.", + "feishu_token_rejected", + ) + try: + live_matches, live_has_more = await search_feishu_contacts( + token, + query.strip(), + limit=limit, + offset=offset, + ) + except Exception as exc: + return _feishu_read_exception_outcome("user_search", exc) + live_members: list[dict[str, object]] = [] + for match in live_matches: + member = { + "display_name": match.display_name, + "source": "feishu_live", + } + if match.title: + member["title"] = match.title + live_members.append(member) + summary_payload = { + "query": query.strip(), + "returned_count": len(live_members), + "has_more": live_has_more, + "members": live_members, + } + return _typed_success( + _bounded_feishu_json(summary_payload), + metadata={ + "returned_count": len(live_members), + "has_more": live_has_more, + "limit": limit, + "offset": offset, + "source": "feishu_live", + }, + ) summary_payload = { "query": query.strip(), "returned_count": len(members), @@ -19447,7 +19489,31 @@ async def _feishu_open_id_for_visible_name( open_id = provider.get("open_id") if isinstance(open_id, str) and open_id and open_id not in exact_open_ids: exact_open_ids.append(open_id) - return exact_open_ids[0] if len(exact_open_ids) == 1 else None + if len(exact_open_ids) == 1: + return exact_open_ids[0] + if len(exact_open_ids) > 1: + return None + + token, token_error = await _feishu_access_token_outcome(agent_id) + if token_error is not None or token is None: + return None + live_matches, _ = await search_feishu_contacts( + token, + normalized_name, + limit=2, + offset=0, + exact_name=True, + ) + live_exact_open_ids = { + match.open_id + for match in live_matches + if match.display_name.casefold() == normalized_name.casefold() + } + return ( + next(iter(live_exact_open_ids)) + if len(live_exact_open_ids) == 1 + else None + ) async def _feishu_contacts_refresh(agent_id: uuid.UUID) -> None: diff --git a/backend/app/services/builtin_tool_definitions.py b/backend/app/services/builtin_tool_definitions.py index d60385e89..eb62e457d 100644 --- a/backend/app/services/builtin_tool_definitions.py +++ b/backend/app/services/builtin_tool_definitions.py @@ -2188,7 +2188,7 @@ { "name": "feishu_user_search", "display_name": "Feishu User Search", - "description": "Search the visible tenant directory for contactable Feishu colleagues. Returns stable member IDs and display facts only; use target_member_id with channel tools.", + "description": "Search colleagues visible to the Agent's configured Feishu app. Synced contacts include stable member IDs; live Feishu matches expose display facts only.", "category": "feishu", "icon": "🔍", "is_default": False, diff --git a/backend/app/services/feishu_contact_search.py b/backend/app/services/feishu_contact_search.py new file mode 100644 index 000000000..edbb2fc3b --- /dev/null +++ b/backend/app/services/feishu_contact_search.py @@ -0,0 +1,318 @@ +"""Bounded app-identity search over an Agent's visible Feishu contacts.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Mapping +from dataclasses import dataclass + +import httpx + +from app.services.feishu_service import feishu_service + +_API_BASE = "https://open.feishu.cn/open-apis/contact/v3" +_PAGE_SIZE = 50 +_MAX_DEPARTMENTS = 1_000 +_MAX_USER_PAGES = 2_000 +_CONCURRENCY = 10 + + +@dataclass(frozen=True, slots=True) +class FeishuContactMatch: + """One private Provider match; raw IDs must not enter model-visible output.""" + + open_id: str + display_name: str + title: str = "" + + +class FeishuContactSearchLimitError(RuntimeError): + """The Provider-visible directory exceeded the bounded search window.""" + + +def _body(payload: Mapping[str, object], *, stage: str) -> Mapping[str, object]: + data = payload.get("data") + if not isinstance(data, Mapping): + raise ValueError(f"Feishu {stage} returned an invalid data object") + return data + + +def _items(data: Mapping[str, object], *, stage: str) -> list[Mapping[str, object]]: + raw_items = data.get("items", []) + if not isinstance(raw_items, list): + raise ValueError(f"Feishu {stage} returned an invalid item list") + return [item for item in raw_items if isinstance(item, Mapping)] + + +def _next_page_token(data: Mapping[str, object]) -> str | None: + if data.get("has_more") is not True: + return None + token = data.get("page_token") + if not isinstance(token, str) or not token: + raise ValueError("Feishu pagination omitted page_token") + return token + + +def _department_id(item: Mapping[str, object]) -> str | None: + value = item.get("open_department_id") or item.get("department_id") + return value.strip() if isinstance(value, str) and value.strip() else None + + +def _contact( + item: Mapping[str, object], + query: str, + *, + exact_name: bool, +) -> FeishuContactMatch | None: + display_name = str(item.get("name") or "").strip() + searchable = ( + display_name, + str(item.get("en_name") or "").strip(), + str(item.get("email") or "").strip(), + ) + if exact_name: + matched = display_name.casefold() == query + else: + matched = any( + query in value.casefold() + for value in searchable + if value + ) + if not matched: + return None + open_id = item.get("open_id") or item.get("user_id") + if not isinstance(open_id, str) or not open_id.strip() or not display_name: + return None + return FeishuContactMatch( + open_id=open_id.strip(), + display_name=display_name, + title=str(item.get("title") or "").strip(), + ) + + +async def _get( + client: httpx.AsyncClient, + token: str, + url: str, + *, + params: dict[str, object], + stage: str, +) -> Mapping[str, object]: + response = await client.get( + url, + headers={"Authorization": f"Bearer {token}"}, + params=params, + ) + payload = feishu_service._parse_api_response(response, stage=stage) + if not isinstance(payload, Mapping): + raise ValueError(f"Feishu {stage} returned an invalid response") + return payload + + +async def _visible_department_ids( + client: httpx.AsyncClient, + token: str, +) -> list[str]: + department_ids: list[str] = [] + page_token: str | None = None + while True: + params: dict[str, object] = { + "department_id_type": "open_department_id", + "fetch_child": "true", + "page_size": _PAGE_SIZE, + } + if page_token: + params["page_token"] = page_token + payload = await _get( + client, + token, + f"{_API_BASE}/departments", + params=params, + stage="contact_departments", + ) + data = _body(payload, stage="contact_departments") + for item in _items(data, stage="contact_departments"): + department_id = _department_id(item) + if department_id and department_id not in department_ids: + department_ids.append(department_id) + if len(department_ids) > _MAX_DEPARTMENTS: + raise FeishuContactSearchLimitError( + "Feishu visible department count exceeded the search limit" + ) + page_token = _next_page_token(data) + if page_token is None: + break + + if "0" in department_ids: + page_token = None + while True: + params = { + "department_id_type": "open_department_id", + "fetch_child": "true", + "page_size": _PAGE_SIZE, + } + if page_token: + params["page_token"] = page_token + payload = await _get( + client, + token, + f"{_API_BASE}/departments/0/children", + params=params, + stage="contact_department_children", + ) + data = _body(payload, stage="contact_department_children") + for item in _items(data, stage="contact_department_children"): + department_id = _department_id(item) + if department_id and department_id not in department_ids: + department_ids.append(department_id) + if len(department_ids) > _MAX_DEPARTMENTS: + raise FeishuContactSearchLimitError( + "Feishu visible department count exceeded the search limit" + ) + page_token = _next_page_token(data) + if page_token is None: + break + return department_ids + + +async def _department_matches( + client: httpx.AsyncClient, + token: str, + department_id: str, + query: str, + page_budget: list[int], + *, + exact_name: bool, +) -> list[FeishuContactMatch]: + matches: list[FeishuContactMatch] = [] + page_token: str | None = None + while True: + page_budget[0] += 1 + if page_budget[0] > _MAX_USER_PAGES: + raise FeishuContactSearchLimitError( + "Feishu visible user pages exceeded the search limit" + ) + params: dict[str, object] = { + "department_id": department_id, + "department_id_type": "open_department_id", + "user_id_type": "open_id", + "page_size": _PAGE_SIZE, + } + if page_token: + params["page_token"] = page_token + payload = await _get( + client, + token, + f"{_API_BASE}/users/find_by_department", + params=params, + stage="contact_users", + ) + data = _body(payload, stage="contact_users") + for item in _items(data, stage="contact_users"): + contact = _contact(item, query, exact_name=exact_name) + if contact is not None: + matches.append(contact) + page_token = _next_page_token(data) + if page_token is None: + return matches + + +async def _independent_matches( + client: httpx.AsyncClient, + token: str, + query: str, + page_budget: list[int], + *, + exact_name: bool, +) -> list[FeishuContactMatch]: + """Read users granted directly in the app contact scope. + + Feishu's legacy list endpoint is the only app-identity endpoint that + exposes independently authorized users when no department is in scope. + """ + matches: list[FeishuContactMatch] = [] + page_token: str | None = None + while True: + page_budget[0] += 1 + if page_budget[0] > _MAX_USER_PAGES: + raise FeishuContactSearchLimitError( + "Feishu visible user pages exceeded the search limit" + ) + params: dict[str, object] = { + "department_id_type": "open_department_id", + "user_id_type": "open_id", + "page_size": 100, + } + if page_token: + params["page_token"] = page_token + payload = await _get( + client, + token, + f"{_API_BASE}/users", + params=params, + stage="contact_independent_users", + ) + data = _body(payload, stage="contact_independent_users") + for item in _items(data, stage="contact_independent_users"): + contact = _contact(item, query, exact_name=exact_name) + if contact is not None: + matches.append(contact) + page_token = _next_page_token(data) + if page_token is None: + return matches + + +async def search_feishu_contacts( + token: str, + query: str, + *, + limit: int, + offset: int, + exact_name: bool = False, +) -> tuple[list[FeishuContactMatch], bool]: + """Search the Agent app's visible Feishu directory without exposing IDs.""" + normalized_query = query.strip().casefold() + if not normalized_query: + return [], False + target_count = offset + limit + 1 + found_by_open_id: dict[str, FeishuContactMatch] = {} + page_budget = [0] + async with httpx.AsyncClient(timeout=20) as client: + for contact in await _independent_matches( + client, + token, + normalized_query, + page_budget, + exact_name=exact_name, + ): + found_by_open_id.setdefault(contact.open_id, contact) + department_ids = await _visible_department_ids(client, token) + for start in range(0, len(department_ids), _CONCURRENCY): + batch = department_ids[start : start + _CONCURRENCY] + batch_matches = await asyncio.gather( + *( + _department_matches( + client, + token, + department_id, + normalized_query, + page_budget, + exact_name=exact_name, + ) + for department_id in batch + ) + ) + for matches in batch_matches: + for contact in matches: + found_by_open_id.setdefault(contact.open_id, contact) + if len(found_by_open_id) >= target_count: + break + found = list(found_by_open_id.values()) + return found[offset : offset + limit], len(found) > offset + limit + + +__all__ = [ + "FeishuContactMatch", + "FeishuContactSearchLimitError", + "search_feishu_contacts", +] diff --git a/backend/tests/test_agent_tools_typed_feishu_remaining.py b/backend/tests/test_agent_tools_typed_feishu_remaining.py index 290f23b9c..185df29dc 100644 --- a/backend/tests/test_agent_tools_typed_feishu_remaining.py +++ b/backend/tests/test_agent_tools_typed_feishu_remaining.py @@ -2,10 +2,10 @@ from __future__ import annotations -from collections import defaultdict import json -from types import SimpleNamespace import uuid +from collections import defaultdict +from types import SimpleNamespace import httpx import pytest @@ -22,9 +22,9 @@ builtin_readiness, builtin_sensitive_paths, ) +from app.services.feishu_contact_search import FeishuContactMatch from app.services.feishu_service import feishu_service - F4_READ_TOOLS = frozenset( { "feishu_user_search", @@ -738,6 +738,16 @@ async def test_user_search_reuses_tenant_scoped_human_directory_window( }, ) + async def token(_agent_id): + return "tenant-token", None + + async def live_search(_token, _query, *, limit, offset): + assert (limit, offset) == (7, 3) + return [], False + + monkeypatch.setattr(agent_tools, "_feishu_access_token_outcome", token) + monkeypatch.setattr(agent_tools, "search_feishu_contacts", live_search) + assert_outcome( await execute( "feishu_user_search", @@ -833,6 +843,94 @@ async def test_user_search_returns_only_visible_contactable_feishu_members_witho assert forbidden not in serialized +@pytest.mark.asyncio +async def test_user_search_falls_back_to_agent_feishu_directory_without_exposing_open_id( + monkeypatch, +) -> None: + install_directory_payload( + monkeypatch, + { + "ok": True, + "has_more": False, + "members": [], + }, + ) + calls: list[tuple[str, str, int, int]] = [] + + async def token(_agent_id): + return "tenant-token", None + + async def live_search(token, query, *, limit, offset): + calls.append((token, query, limit, offset)) + return ( + [ + FeishuContactMatch( + open_id="ou-private-zhou", + display_name="周逸飞", + title="Engineer", + ) + ], + False, + ) + + monkeypatch.setattr(agent_tools, "_feishu_access_token_outcome", token) + monkeypatch.setattr(agent_tools, "search_feishu_contacts", live_search) + + outcome = assert_outcome( + await execute("feishu_user_search", {"query": "周逸飞"}), + "succeeded", + ) + payload = json.loads(outcome.summary or "") + + assert calls == [("tenant-token", "周逸飞", 20, 0)] + assert payload == { + "query": "周逸飞", + "returned_count": 1, + "has_more": False, + "members": [ + { + "display_name": "周逸飞", + "title": "Engineer", + "source": "feishu_live", + } + ], + } + assert "ou-private-zhou" not in (outcome.summary or "") + + +@pytest.mark.asyncio +async def test_calendar_name_resolution_uses_private_live_open_id_fallback( + monkeypatch, +) -> None: + async def directory(_agent_id, _arguments): + return {"ok": True, "members": [], "has_more": False} + + async def token(_agent_id): + return "tenant-token", None + + async def live_search(_token, _query, *, limit, offset, exact_name): + assert (limit, offset) == (2, 0) + assert exact_name is True + return ( + [ + FeishuContactMatch( + open_id="ou-private-zhou", + display_name="周逸飞", + ) + ], + False, + ) + + monkeypatch.setattr(agent_tools, "_query_directory_payload", directory) + monkeypatch.setattr(agent_tools, "_feishu_access_token_outcome", token) + monkeypatch.setattr(agent_tools, "search_feishu_contacts", live_search) + + assert ( + await agent_tools._feishu_open_id_for_visible_name(uuid.uuid4(), "周逸飞") + == "ou-private-zhou" + ) + + @pytest.mark.asyncio async def test_user_search_directory_failure_is_typed_retryable_read( monkeypatch, diff --git a/backend/tests/test_feishu_contact_search.py b/backend/tests/test_feishu_contact_search.py new file mode 100644 index 000000000..3c32ddb78 --- /dev/null +++ b/backend/tests/test_feishu_contact_search.py @@ -0,0 +1,174 @@ +"""Provider-bound tests for app-identity Feishu contact search.""" + +from __future__ import annotations + +import httpx +import pytest + +from app.services.feishu_contact_search import search_feishu_contacts +from app.services.feishu_service import FeishuAPIError + + +class FakeResponse: + def __init__(self, payload: dict, status_code: int = 200) -> None: + self._payload = payload + self.status_code = status_code + + def json(self) -> dict: + return self._payload + + +@pytest.mark.asyncio +async def test_searches_visible_departments_with_tenant_token(monkeypatch) -> None: + calls: list[tuple[str, dict, dict]] = [] + + class Client: + def __init__(self, *args, **kwargs): + del args, kwargs + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return False + + async def get(self, url, **kwargs): + calls.append((url, kwargs["params"], kwargs["headers"])) + if url.endswith("/users"): + return FakeResponse( + {"code": 0, "data": {"items": [], "has_more": False}} + ) + if url.endswith("/departments"): + return FakeResponse( + { + "code": 0, + "data": { + "items": [{"open_department_id": "0"}], + "has_more": False, + }, + } + ) + if url.endswith("/departments/0/children"): + return FakeResponse( + { + "code": 0, + "data": { + "items": [{"open_department_id": "od-engineering"}], + "has_more": False, + }, + } + ) + department_id = kwargs["params"]["department_id"] + items = ( + [ + { + "open_id": "ou-private-zhou", + "name": "周逸飞", + "en_name": "Yifei Zhou", + "title": "Engineer", + } + ] + if department_id == "od-engineering" + else [] + ) + return FakeResponse( + {"code": 0, "data": {"items": items, "has_more": False}} + ) + + monkeypatch.setattr(httpx, "AsyncClient", Client) + + matches, has_more = await search_feishu_contacts( + "tenant-token", + "周逸飞", + limit=20, + offset=0, + ) + + assert has_more is False + assert len(matches) == 1 + assert matches[0].open_id == "ou-private-zhou" + assert matches[0].display_name == "周逸飞" + assert matches[0].title == "Engineer" + assert all(headers == {"Authorization": "Bearer tenant-token"} for _, _, headers in calls) + assert any(url.endswith("/users/find_by_department") for url, _, _ in calls) + + +@pytest.mark.asyncio +async def test_searches_users_granted_directly_in_app_scope(monkeypatch) -> None: + class Client: + def __init__(self, *args, **kwargs): + del args, kwargs + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return False + + async def get(self, url, **_kwargs): + if url.endswith("/users"): + return FakeResponse( + { + "code": 0, + "data": { + "items": [ + { + "open_id": "ou-private-zhou", + "name": "周逸飞", + } + ], + "has_more": False, + }, + } + ) + if url.endswith("/departments"): + return FakeResponse( + {"code": 0, "data": {"items": [], "has_more": False}} + ) + raise AssertionError(f"unexpected URL: {url}") + + monkeypatch.setattr(httpx, "AsyncClient", Client) + + matches, has_more = await search_feishu_contacts( + "tenant-token", + "周逸飞", + limit=20, + offset=0, + exact_name=True, + ) + + assert has_more is False + assert [(match.display_name, match.open_id) for match in matches] == [ + ("周逸飞", "ou-private-zhou") + ] + + +@pytest.mark.asyncio +async def test_provider_rejection_is_not_converted_to_empty_results(monkeypatch) -> None: + class Client: + def __init__(self, *args, **kwargs): + del args, kwargs + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return False + + async def get(self, _url, **_kwargs): + return FakeResponse( + {"code": 40060, "msg": "no department authority"}, + status_code=400, + ) + + monkeypatch.setattr(httpx, "AsyncClient", Client) + + with pytest.raises(FeishuAPIError) as raised: + await search_feishu_contacts( + "tenant-token", + "周逸飞", + limit=20, + offset=0, + ) + + assert raised.value.code == 40060 From a51db2e2a0d0807507aca9c28f5cc807ec6d61c4 Mon Sep 17 00:00:00 2001 From: Y1fe1Zh0u Date: Wed, 19 Aug 2026 10:36:36 +0800 Subject: [PATCH 5/5] Keep external group conversations from holding user waits External Feishu groups share one public Session lane, so a waiting_user checkpoint can block unrelated group traffic and cannot be resumed by the Direct Chat scope contract. Mark the immutable Chat session type in Run input and repair user waits into public clarification replies without enabling native Group handoff tools. Constraint: External Feishu groups use session_type=group with group_id=NULL and must not gain native Clawith Group tools. Rejected: Resume external-group waiting_user Runs | one participant's wait would hold the shared public lane and consume unrelated messages. Confidence: high Scope-risk: narrow Directive: Public group Runs ask clarifying questions in their final reply and release the lane; only Direct Chat may retain waiting_user. Tested: 94 scoped Runtime, Chat intake, Channel, and Feishu tests; scoped Ruff; live 3010 new-Session calendar E2E. Not-tested: Full backend test suite and high-volume concurrent external-group traffic. --- .../app/services/agent_runtime/chat_intake.py | 1 + .../agent_runtime/model_step_service.py | 30 ++++++++++++++----- .../tests/test_agent_runtime_chat_intake.py | 1 + .../test_agent_runtime_model_step_service.py | 21 +++++++++++-- 4 files changed, 44 insertions(+), 9 deletions(-) diff --git a/backend/app/services/agent_runtime/chat_intake.py b/backend/app/services/agent_runtime/chat_intake.py index b4af54acc..9efa0ddd3 100644 --- a/backend/app/services/agent_runtime/chat_intake.py +++ b/backend/app/services/agent_runtime/chat_intake.py @@ -729,6 +729,7 @@ async def enqueue_chat_runtime( "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, **( diff --git a/backend/app/services/agent_runtime/model_step_service.py b/backend/app/services/agent_runtime/model_step_service.py index 494cace53..0f7b6f7a6 100644 --- a/backend/app/services/agent_runtime/model_step_service.py +++ b/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 diff --git a/backend/tests/test_agent_runtime_chat_intake.py b/backend/tests/test_agent_runtime_chat_intake.py index c3befa7eb..e5c94a101 100644 --- a/backend/tests/test_agent_runtime_chat_intake.py +++ b/backend/tests/test_agent_runtime_chat_intake.py @@ -358,6 +358,7 @@ async def test_external_group_chat_uses_unified_session_without_native_group_sco "message_id": str(message.id), "created_at": message.created_at.isoformat(), } + assert command.payload["chat_session_type"] == "group" assert command.delivery_target == { "kind": "session", "session_id": str(session.id), diff --git a/backend/tests/test_agent_runtime_model_step_service.py b/backend/tests/test_agent_runtime_model_step_service.py index ef038e32c..b1850a4b2 100644 --- a/backend/tests/test_agent_runtime_model_step_service.py +++ b/backend/tests/test_agent_runtime_model_step_service.py @@ -2309,7 +2309,24 @@ async def complete(*args, **kwargs): @pytest.mark.asyncio -async def test_group_run_repairs_waiting_user_instead_of_entering_unresumable_wait() -> None: +@pytest.mark.parametrize( + "group_input", + ( + {"group_context": {"group": {"group_id": str(uuid.uuid4())}}}, + { + "source_channel": "feishu", + "chat_session_type": "group", + "context_cutoff": { + "message_id": str(uuid.uuid4()), + "created_at": "2026-08-19T01:50:31+00:00", + }, + }, + ), + ids=("native-group", "external-feishu-group"), +) +async def test_group_run_repairs_waiting_user_instead_of_entering_unresumable_wait( + group_input: dict[str, object], +) -> None: tenant_id = uuid.uuid4() model = _model(tenant_id) agent = _agent(tenant_id) @@ -2319,7 +2336,7 @@ async def test_group_run_repairs_waiting_user_instead_of_entering_unresumable_wa session_context_version=1, recent_session_messages=state["snapshots"].recent_session_messages, related_run_summaries=(), - initial_input={"group_context": {"group": {"group_id": str(uuid.uuid4())}}}, + initial_input=group_input, ) async def complete(*args, **kwargs):