fix(api): use batch memory ingest to prevent seq ordering race condition

api/app/core/memory/memory_service.py

@@ -162,6 +162,35 @@ async def ingest_agent_message(
             language=language,
         )
 
+    @staticmethod
+    async def ingest_agent_messages(
+        conversation_id: str,
+        messages: List[Any],
+        app_id: str,
+        config_id: str = "",
+        workspace_id: str = "",
+        end_user_id: str = "",
+        language: str = "zh",
+    ) -> bool:
+        """批量 Agent 消息摄入：一次事务写入 + 一次滑动窗口派发。
+
+        同一回合的 user + assistant 应通过此入口一次派发，避免两次
+        fire-and-forget 造成的 seq 分配竞态（顺序颠倒）。
+
+        每条 message 需带以下属性：
+            .id, .role, .content, .created_at, .meta_data, .should_memorize
+        """
+        from app.core.memory.pipelines.dispatcher import ingest_agent_messages
+        return await ingest_agent_messages(
+            conversation_id=conversation_id,
+            messages=messages,
+            app_id=app_id,
+            config_id=config_id,
+            workspace_id=workspace_id,
+            end_user_id=end_user_id,
+            language=language,
+        )
+
     @staticmethod
     async def ingest_workflow_messages(
         messages: List[dict],

api/app/core/memory/pipelines/dispatcher.py

@@ -566,6 +566,149 @@ async def ingest_agent_message(
     return True
 
 
+async def ingest_agent_messages(
+    conversation_id: str,
+    messages: List["Any"],
+    app_id: str,
+    config_id: str = "",
+    workspace_id: str = "",
+    end_user_id: str = "",
+    language: str = "zh",
+) -> bool:
+    """批量 Agent 消息摄入：一次事务写入 memory_messages + 一次滑动窗口派发。
+
+    专用于同一回合的多条消息（典型场景：user + assistant 一起入队），确保：
+    1. 一次 pg_advisory_xact_lock，seq 按 messages 顺序连续递增（消除并发派发的
+       seq 分配竞态，避免 assistant.seq < user.seq 顺序颠倒）
+    2. 一次事务提交，一次滑动窗口派发，一次 refresh_active_key。
+
+    每条 message 需带以下属性：
+        .id, .role, .content, .created_at, .meta_data, .should_memorize
+
+    Returns:
+        True 表示成功写入至少一条，False 表示跳过(门禁未开或全部内容为空)
+    """
+    if not messages:
+        return False
+
+    if not await check_memory_enabled(app_id):
+        return False
+
+    # 构建 write_batch 输入 + 记录原始 role/should_memorize（用于后续 Fast Write）。
+    # write_batch 内部会过滤空 content，此处用相同规则同步 role_and_flag 列表，
+    # 保证之后与 written[] zip 对齐。
+    batch_inputs: list[dict] = []
+    role_and_flag: list[tuple[str, bool]] = []
+    for m in messages:
+        files = None
+        if hasattr(m, "meta_data") and m.meta_data:
+            files = m.meta_data.get("files")
+
+        dialog_at: Optional[str] = None
+        if hasattr(m, "created_at") and m.created_at:
+            _created = m.created_at
+            if isinstance(_created, datetime):
+                _created = _created.replace(tzinfo=timezone.utc) if _created.tzinfo is None else _created
+                dialog_at = _created.isoformat()
+            elif isinstance(_created, str):
+                dialog_at = _created
+
+        content_str = str(m.content or "")
+        should_memorize = bool(getattr(m, "should_memorize", True))
+
+        batch_inputs.append({
+            "role": m.role,
+            "content": m.content,
+            "original_message_id": m.id,
+            "created_at": m.created_at,
+            "should_memorize": should_memorize,
+            "files": files,
+            "dialog_at": dialog_at,
+        })
+
+        if content_str.strip():
+            role_and_flag.append((str(m.role), should_memorize))
+
+    # 写 memory_messages。original_message_id 外键指向 messages 表，而 messages 可能
+    # 经 BatchPersistQueue 攒批延迟落库——本表先 commit 时外键引用的行尚不存在，
+    # 会抛 IntegrityError（ForeignKeyViolation）。捕获后延迟重试，最终一致。
+    written: List[dict] = []
+    for attempt in range(AGENT_MESSAGE_FK_RETRY):
+        try:
+            with get_db_context() as db:
+                repo = MemoryMessageRepository(db)
+                written = repo.write_batch(
+                    conversation_id=str(conversation_id),
+                    messages=batch_inputs,
+                    end_user_id=end_user_id,
+                    source=MemoryMessageSource.AGENT,
+                )
+                if not written:
+                    return False
+                db.commit()
+            break
+        except IntegrityError as exc:
+            orig = getattr(exc, "orig", None)
+            is_fk = orig is not None and "ForeignKeyViolation" in type(orig).__name__
+            if not is_fk:
+                raise
+            if attempt >= AGENT_MESSAGE_FK_RETRY - 1:
+                # 重试耗尽：基本可断定 messages 行已永久缺失（落库失败/任务丢失），
+                # 该条记忆静默丢失，升级为 error 并带固定聚合标记，供日志平台配置告警。
+                logger.error(
+                    "MEMORY_MESSAGES_FK_EXHAUSTED retries=%d conv=%s end_user=%s "
+                    "batch_size=%d app_id=%s err=%s",
+                    AGENT_MESSAGE_FK_RETRY, conversation_id, end_user_id,
+                    len(batch_inputs), app_id, exc,
+                )
+                raise
+            logger.warning(
+                "ingest_agent_messages FK 冲突（messages 尚未落库），重试 %d/%d: conv=%s, err=%s",
+                attempt + 1, AGENT_MESSAGE_FK_RETRY, conversation_id, exc,
+            )
+            await asyncio.sleep(AGENT_MESSAGE_FK_RETRY_BASE_DELAY_S * (attempt + 1))
+
+    await refresh_active_key(conversation_id)
+    mark_conversation_pending(conversation_id)
+
+    # Fast Write 派发（隔离：任一条失败不阻断整批，也不阻断下面的滑动窗口派发）。
+    # 权限取原始 role / should_memorize；Agent 路径有应用级门禁 require_app_gate=True。
+    # zip(strict=True)：len 不一致直接报错兜底（batch_inputs 与 write_batch 的空 content
+    # 过滤规则一致，理论上不会不齐；出现即上游数据异常，进 except 走隔离降级）。
+    try:
+        for (role, should_memorize), written_msg in zip(role_and_flag, written, strict=True):
+            await safe_push_fast_write(
+                role=role,
+                should_memorize=should_memorize,
+                app_id=app_id,
+                require_app_gate=True,
+                end_user_id=end_user_id,
+                target_message=written_msg,
+                config_id=config_id,
+                workspace_id=workspace_id,
+                conversation_id=str(conversation_id),
+                message_seq=written_msg["message_seq"],
+                language=language,
+                source=MemoryMessageSource.AGENT.value,
+            )
+    except Exception as e:
+        logger.warning(
+            "[FastDispatcher] agent fast write dispatch loop failed "
+            "(isolated, normal flow unaffected): conv=%s, end_user_id=%s, "
+            "batch=%s, written=%s, err=%s",
+            conversation_id, end_user_id, len(role_and_flag), len(written), e,
+        )
+
+    await check_sliding_window_and_dispatch(
+        conversation_id=str(conversation_id),
+        config_id=config_id,
+        end_user_id=end_user_id,
+        workspace_id=workspace_id,
+        language=language,
+    )
+    return True
+
+
 # ──────────────────────────────────────────────
 # 入口3: Workflow 消息摄入
 # ──────────────────────────────────────────────

api/app/services/app_chat_service.py

@@ -1603,24 +1603,39 @@ async def load_annotation_context():
                 ))
                 save_messages_enqueued = True
 
-                # 记忆写入 + 派发：复用 conversation_service.dispatch_memory_sync
+                # 记忆写入 + 派发：改为批量派发，一次 write_batch 分配连续 seq，
+                # 消除原先两次 fire-and-forget 并发引发的 seq 顺序颠倒问题。
                 result_row = await self.db.execute(
                     select(Conversation).where(Conversation.id == conversation_id)
                 )
                 conv = result_row.scalar_one_or_none()
                 if conv:
                     now = datetime.now(timezone.utc)
-
-                    for m in [
-                        {"id": user_message_id, "role": "user", "content": message, "meta_data": human_meta, "should_memorize": memory},
-                        {"id": message_id, "role": "assistant", "content": full_content, "meta_data": assistant_meta, "should_memorize": True},
-                    ]:
-                        memorize = m.pop("should_memorize")
-                        self.conversation_service.dispatch_memory_sync(
-                            message=SimpleNamespace(conversation_id=conversation_id, created_at=now, **m),
+                    asyncio.create_task(
+                        self.conversation_service.dispatch_memory_batch(
+                            messages=[
+                                SimpleNamespace(
+                                    id=user_message_id,
+                                    conversation_id=conversation_id,
+                                    role="user",
+                                    content=message,
+                                    meta_data=human_meta,
+                                    created_at=now,
+                                    should_memorize=memory,
+                                ),
+                                SimpleNamespace(
+                                    id=message_id,
+                                    conversation_id=conversation_id,
+                                    role="assistant",
+                                    content=full_content,
+                                    meta_data=assistant_meta,
+                                    created_at=now,
+                                    should_memorize=True,
+                                ),
+                            ],
                             conversation=conv,
-                            should_memorize=memorize,
                         )
+                    )
 
                 # Enqueue agent execution after messages so the FK is satisfied
                 # within the same batch (messages commit first, then execution).

api/app/services/conversation_service.py

@@ -1,4 +1,5 @@
 """会话服务"""
+import asyncio
 import uuid
 from types import SimpleNamespace
 from datetime import timedelta
@@ -291,6 +292,9 @@ def add_message(
             self.db.commit()
             self.db.refresh(message)
 
+            # TODO(messages-memory-decoupling Phase 2): 迁移完所有成对调用点后，移除
+            # 此处 dispatch_memory_sync 调用，改由业务层显式调 dispatch_memory_batch。
+            # 参见 .kiro/specs/messages-memory-decoupling/design.md
             if sync_memory:
                 self.dispatch_memory_sync(message, conversation, should_memorize)
 
@@ -532,7 +536,7 @@ async def _run():
                     logger.warning(
                         f"[ConversationService] dispatch_agent_message 异步执行失败: "
                         f"conv={_conversation_id}, err={exc}",
-                        exc_info=exc,
+                        exc_info=True,
                     )
 
             loop = asyncio.get_event_loop()
@@ -547,6 +551,53 @@ async def _run():
                 exc_info=True,
             )
 
+    async def dispatch_memory_batch(
+        self,
+        messages: List[Any],
+        conversation: Conversation,
+    ) -> None:
+        """批量派发同一回合的多条消息到记忆系统（async 线性实现）。
+
+        与 dispatch_memory_sync 的区别：一次 write_batch（一次 pg_advisory_xact_lock
+        + 一次事务）分配连续 seq，一次滑动窗口派发。批内 seq 严格 user < assistant。
+
+        本方法**本身不 fire-and-forget**。调用方决定：
+            - fire-and-forget：`asyncio.create_task(svc.dispatch_memory_batch(...))`
+            - 阻塞等待：`await svc.dispatch_memory_batch(...)`
+        主 chat 流程默认走 fire-and-forget，避免拖累流式响应。
+
+        Args:
+            messages: 消息列表，元素需带 .id / .conversation_id / .role /
+                .content / .created_at / .meta_data / .should_memorize 属性。
+                所有消息应属于同一对话，seq 分配顺序 = 列表顺序。
+            conversation: 所属 Conversation 实例（提供 workspace_id / app_id / user_id）。
+        """
+        # 数据形状规范化（None → {}, 缺失字段默认等）由下游 dispatcher.ingest_agent_messages
+        # 统一处理；空/异常输入由下游 if not messages / if not written 双重拦截；
+        # 本方法只做纯派发（薄派发层），不重复防御。
+        from app.db import get_async_db_context
+        from app.core.memory.memory_service import MemoryService
+
+        try:
+            async with get_async_db_context() as db:
+                config_id = await MemoryConfigService(db).get_workspace_active_config_id_async(
+                    conversation.workspace_id
+                )
+            await MemoryService.ingest_agent_messages(
+                conversation_id=str(messages[0].conversation_id) if messages else "",
+                messages=messages,
+                app_id=str(conversation.app_id),
+                config_id=str(config_id),
+                workspace_id=str(conversation.workspace_id),
+                end_user_id=str(conversation.user_id) if conversation.user_id else "",
+            )
+        except Exception as exc:
+            logger.warning(
+                f"[ConversationService] dispatch_memory_batch 执行失败: "
+                f"conv={conversation.id}, batch={len(messages)}, err={exc}",
+                exc_info=True,
+            )
+
     def get_messages(
             self,
             conversation_id: uuid.UUID,