diff --git a/backend/app/api/channels.py b/backend/app/api/channels.py index 5723d7d7..4659084b 100644 --- a/backend/app/api/channels.py +++ b/backend/app/api/channels.py @@ -19,6 +19,7 @@ resume_binding_ingress, wait_binding_ingress_stopped, ) +from app.channels.adapters.base import is_group_conv_id from app.channels.adapters.feishu import ( FeishuPermanentError, validate_feishu_credentials, @@ -1186,7 +1187,6 @@ def list_channel_conversations( ).all() message_counts = {session_id: count for session_id, count in count_rows} - group_prefix = f"{binding.channel}_group_" conversations: list[ChannelConversationRead] = [] for chat_session in page: last_message = db.exec( @@ -1202,7 +1202,9 @@ def list_channel_conversations( external_conv_id=external_conv_id, display_name=identity_names.get(chat_session.user_id) or user_names.get(chat_session.user_id), - is_group=bool(external_conv_id and external_conv_id.startswith(group_prefix)), + is_group=bool( + external_conv_id and is_group_conv_id(binding.channel, external_conv_id) + ), agent_id=chat_session.agent_id, agent_name=agent_name_map.get(chat_session.agent_id), message_count=message_counts.get(chat_session.id, 0), diff --git a/backend/app/channels/adapters/base.py b/backend/app/channels/adapters/base.py index 143c4abf..58c6c147 100644 --- a/backend/app/channels/adapters/base.py +++ b/backend/app/channels/adapters/base.py @@ -43,6 +43,19 @@ def external_conv_id(self) -> str: return f"{self.channel}_p2p_{self.from_user_id}" +def is_group_conv_id(channel: str, external_conv_id: str) -> bool: + """判定 external_conv_id 是否为群会话。 + + 兼容三种格式:无 scope 的 {channel}_group_x、带 scope 的 {channel}_{scope}_group_x、 + 以及成员级 {channel}_{scope}_group_x#member;不看 channel 之后的 scope 内容。 + """ + prefix = f"{channel}_" + if not external_conv_id.startswith(prefix): + return False + rest = external_conv_id[len(prefix):] + return rest.startswith("group_") or "_group_" in rest + + class ChannelAdapter(Protocol): """渠道适配器协议:归一化 + 出站 + 可选 typing + ingress 生命周期。""" diff --git a/backend/app/channels/adapters/wecom.py b/backend/app/channels/adapters/wecom.py index 6ee01764..c90cf2ac 100644 --- a/backend/app/channels/adapters/wecom.py +++ b/backend/app/channels/adapters/wecom.py @@ -57,11 +57,11 @@ def normalize_wecom_frame(frame: dict[str, Any], *, account_scope: str = "") -> chat_id = str(body.get("chatid") or "").strip() chattype = str(body.get("chattype") or "").strip() if chattype == "group" and not chat_id: - # 群消息缺 chatid 时按私聊降级,避免群会话退化为每人一个会话 - logger.warning("企微群消息缺少 chatid,按私聊降级处理 msgid=%s", body.get("msgid")) - chattype = "single" - # 官方文档：chatid 仅群聊返回 - is_group = bool(chat_id) + # 群消息缺 chatid 无法构成稳定群会话:丢弃,绝不降级混入成员私聊会话 + logger.warning("企微群消息缺少 chatid,丢弃 msgid=%s", body.get("msgid")) + return None + # 以 chattype 为准(官方文档 chatid 仅群聊返回,但字段存在性不等于会话类型) + is_group = chattype == "group" headers = frame.get("headers") or {} event_id = str(body.get("msgid") or body.get("msg_id") or headers.get("req_id") or "").strip() if not event_id: @@ -72,10 +72,10 @@ def normalize_wecom_frame(frame: dict[str, Any], *, account_scope: str = "") -> event_id=event_id, from_user_id=from_user_id, to_user_id=str(body.get("aibotid") or "").strip(), - session_id=chat_id or from_user_id, - group_id=chat_id, + session_id=(chat_id or from_user_id) if is_group else from_user_id, + group_id=chat_id if is_group else "", # 企微无 context_token 概念：发送仅需 chatid，占位保持内核必填语义 - context_token=chat_id or from_user_id, + context_token=(chat_id or from_user_id) if is_group else from_user_id, text=text, is_group=is_group, raw=frame, diff --git a/backend/app/core/agent_loop.py b/backend/app/core/agent_loop.py index 481605c4..e3186cb1 100644 --- a/backend/app/core/agent_loop.py +++ b/backend/app/core/agent_loop.py @@ -27,6 +27,7 @@ resource_ref_from_row, runtime_snapshot_from_package, ) +from app.channels.adapters.base import is_group_conv_id from app.channels.service_outbox import stage_channel_delivery from app.core.agent_identity_prompt import AgentIdentityPrompt from app.core.cancellation import clear_chat_turn_cancelled, is_chat_turn_cancelled @@ -1566,14 +1567,19 @@ def stream_failure_response( ) return self._finish_stale_completed_skill(request.tenant_id, chat_session, skills) - memory_context = [ - memory_read(row) - for row in self.memory.context_memories( - request.tenant_id, - request.user_id, - agent_id=chat_session.agent_id, - ) - ] + # 群记忆红线:群聊会话不检索个人记忆(私人偏好绝不在群里被念出来) + memory_context = ( + [] + if is_group_conv_id(chat_session.channel or "", chat_session.external_conv_id or "") + else [ + memory_read(row) + for row in self.memory.context_memories( + request.tenant_id, + request.user_id, + agent_id=chat_session.agent_id, + ) + ] + ) if memory_context: self.events.record( request.tenant_id, @@ -2260,14 +2266,19 @@ def status(phase: str, payload: dict[str, object] | None = None) -> None: user_message_id=user_message.id, ) self._finish_stale_completed_skill(request.tenant_id, chat_session, skills) - memory_context = [ - memory_read(row) - for row in self.memory.context_memories( - request.tenant_id, - request.user_id, - agent_id=chat_session.agent_id, - ) - ] + # 群记忆红线:群聊会话不检索个人记忆(私人偏好绝不在群里被念出来) + memory_context = ( + [] + if is_group_conv_id(chat_session.channel or "", chat_session.external_conv_id or "") + else [ + memory_read(row) + for row in self.memory.context_memories( + request.tenant_id, + request.user_id, + agent_id=chat_session.agent_id, + ) + ] + ) if memory_context: self.events.record( request.tenant_id, diff --git a/backend/app/memory/jobs.py b/backend/app/memory/jobs.py index 4fe112d5..ab1b5ecf 100644 --- a/backend/app/memory/jobs.py +++ b/backend/app/memory/jobs.py @@ -5,6 +5,7 @@ from sqlmodel import Session, select from app.async_jobs import AsyncJob, enqueue_async_job +from app.channels.adapters.base import is_group_conv_id from app.db import engine from app.db.models import AgentEvent, ChatSession, Message, ModelConfig from app.memory.service import MemoryService, memory_read @@ -63,6 +64,9 @@ def run_memory_capture_job(payload: dict[str, Any]) -> None: ) db.commit() return + # 群记忆红线:群聊会话不沉淀任何记忆(成员隐私不得混入群账号/跨成员共享) + if is_group_conv_id(chat_session.channel or "", chat_session.external_conv_id or ""): + return user_events = db.exec( select(AgentEvent) diff --git a/backend/tests/test_channel_wechat.py b/backend/tests/test_channel_wechat.py index 32716efd..329b0b5e 100644 --- a/backend/tests/test_channel_wechat.py +++ b/backend/tests/test_channel_wechat.py @@ -970,3 +970,18 @@ def handler(request: httpx.Request) -> httpx.Response: random_ids = [payload["msg"]["client_id"] for payload in sent] assert len(set(random_ids)) == 3 assert all(not cid.startswith("staffdeck:msg_1") for cid in random_ids) + + +def test_is_group_conv_id_formats() -> None: + from app.channels.adapters.base import is_group_conv_id + + # 无 scope 旧格式 / 带 scope / 成员级(#member)三种群格式均判真 + assert is_group_conv_id("wechat", "wechat_group_room_1") is True + assert is_group_conv_id("wecom", "wecom_corpA_group_wr_1") is True + assert is_group_conv_id("wecom", "wecom_corpA_group_wr_1#zhangsan") is True + assert is_group_conv_id("feishu", "feishu_group_oc_1:thread:ot_1") is True + # 私聊不误判 + assert is_group_conv_id("wechat", "wechat_p2p_user_1") is False + assert is_group_conv_id("wecom", "wecom_corpA_p2p_zhangsan") is False + # channel 不匹配不误判 + assert is_group_conv_id("wechat", "wecom_group_wr_1") is False diff --git a/backend/tests/test_channel_wecom.py b/backend/tests/test_channel_wecom.py index cec2d19f..00246892 100644 --- a/backend/tests/test_channel_wecom.py +++ b/backend/tests/test_channel_wecom.py @@ -1652,16 +1652,14 @@ def slow_process_inbound(binding, inbound, *, db_engine=None, staged_event_pk=No manager.stop_binding(binding_id) -def test_group_without_chatid_degrades_to_p2p_with_warning(caplog) -> None: +def test_group_without_chatid_dropped_with_warning(caplog) -> None: import logging frame = _text_frame(chattype="group", chatid="") with caplog.at_level(logging.WARNING, logger="app.channels.adapters.wecom"): inbound = normalize_wecom_frame(frame) - assert inbound is not None - # chattype=group 但缺 chatid:降级私聊,不退化为"每人一个群会话" - assert inbound.is_group is False - assert inbound.external_conv_id == "wecom_p2p_zhangsan" + # chattype=group 但缺 chatid:无法构成稳定群会话,丢弃,绝不降级混入私聊 + assert inbound is None assert any("缺少 chatid" in record.message for record in caplog.records) # 有 chatid 的群形态不变 @@ -1670,6 +1668,15 @@ def test_group_without_chatid_degrades_to_p2p_with_warning(caplog) -> None: assert group.external_conv_id == "wecom_group_wr_1" +def test_single_chattype_with_chatid_is_p2p() -> None: + """chattype=single 却带 chatid 的异常帧:以 chattype 为准按私聊处理。""" + inbound = normalize_wecom_frame(_text_frame(chattype="single", chatid="wr_ignored")) + assert inbound is not None + assert inbound.is_group is False + assert inbound.group_id == "" + assert inbound.external_conv_id == "wecom_p2p_zhangsan" + + # ---------- 断开超时主动告警 ---------- diff --git a/backend/tests/test_memory_service.py b/backend/tests/test_memory_service.py index d2a4c85d..1c26151b 100644 --- a/backend/tests/test_memory_service.py +++ b/backend/tests/test_memory_service.py @@ -2,7 +2,16 @@ from sqlmodel import Session, SQLModel, create_engine, select from app.api.memories import clear_my_memories, list_memories -from app.db.models import AgentProfile, ChatSession, MemoryRecord, Message, ModelConfig, Tenant, User +from app.db.models import ( + AgentEvent, + AgentProfile, + ChatSession, + MemoryRecord, + Message, + ModelConfig, + Tenant, + User, +) from app.llm.client import LLMClient from app.memory.jobs import _conversation_messages_for_turn from app.memory.service import MemoryService, memory_rows_for_read @@ -507,3 +516,105 @@ def _test_session() -> Session: ) SQLModel.metadata.create_all(engine) return Session(engine) + + +def test_memory_job_skips_group_sessions(monkeypatch) -> None: + """群记忆红线:群聊会话的后台记忆任务直接跳过,不调用 capture、不写入。""" + from app.memory.jobs import run_memory_capture_job + + engine = create_engine( + "sqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + SQLModel.metadata.create_all(engine) + with Session(engine) as db: + db.add(Tenant(id="tenant_demo", name="Demo")) + db.add( + ModelConfig( + id="model_1", + tenant_id="tenant_demo", + name="m", + model="gpt-x", + api_key_encrypted="x", + ) + ) + db.add( + ChatSession( + id="s_group", + tenant_id="tenant_demo", + user_id="user_group", + agent_id="agent_1", + channel="wecom", + external_conv_id="wecom_corpA_group_wr_1", + ) + ) + db.add( + ChatSession( + id="s_p2p", + tenant_id="tenant_demo", + user_id="user_web", + agent_id="agent_1", + channel="wecom", + external_conv_id="wecom_corpA_p2p_zhangsan", + ) + ) + db.commit() + monkeypatch.setattr("app.memory.jobs.engine", engine) + calls: list[str] = [] + + def fake_capture(self, request, chat_session, step_result, tool_result, model_config, messages): # noqa: ANN001 + calls.append(chat_session.id) + return [] + + monkeypatch.setattr(MemoryService, "capture_turn", fake_capture) + + with Session(engine) as db: + # 私聊会话补一条用户消息事件与对应消息历史(capture 前置要求) + db.add( + AgentEvent( + tenant_id="tenant_demo", + session_id="s_p2p", + event_type="user_message_received", + payload_json={"turn_id": "t1"}, + ) + ) + db.add( + Message( + tenant_id="tenant_demo", + session_id="s_p2p", + role="user", + content="hi", + metadata_json={"turn_id": "t1"}, + ) + ) + db.add( + Message( + tenant_id="tenant_demo", + session_id="s_p2p", + role="assistant", + content="你好", + metadata_json={"turn_id": "t1"}, + ) + ) + db.commit() + + def _payload(session_id: str) -> dict: + return { + "request": ChatTurnRequest( + tenant_id="tenant_demo", message="hi", user_id="u", agent_id="agent_1" + ).model_dump(mode="json"), + "session_id": session_id, + "model_config_id": "model_1", + "step_result": StepAgentResult().model_dump(mode="json"), + "tool_result": None, + } + + # 群会话:跳过 + run_memory_capture_job(_payload("s_group")) + assert calls == [] + with Session(engine) as db: + assert db.exec(select(MemoryRecord)).all() == [] + # 私聊会话:正常进入 capture(不误伤) + run_memory_capture_job(_payload("s_p2p")) + assert calls == ["s_p2p"]