fix(channels): 群聊止血——企微 is_group 以 chattype 为准，群记忆红线（不写不读）

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),

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 生命周期。"""
 

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,

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,

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)

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

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"
+
+
 # ---------- 断开超时主动告警 ----------
 
 

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"]