Revert "fix(memory,task): add Redis fair lock for ordered memory writes"

api/app/tasks.py

@@ -1,4 +1,5 @@
 import asyncio
+import hashlib
 import os
 import re
 import shutil
@@ -37,10 +38,12 @@ from app.db import get_db, get_db_context
 from app.models import Document, File, Knowledge
 from app.models.end_user_model import EndUser
 from app.schemas import document_schema, file_schema
+from app.schemas.model_schema import ModelInfo
 from app.services.memory_agent_service import MemoryAgentService, get_end_user_connected_config
 from app.services.memory_forget_service import MemoryForgetService
+from app.services.memory_perceptual_service import MemoryPerceptualService
 from app.utils.config_utils import resolve_config_id
-from app.utils.redis_lock import RedisFairLock
+from app.utils.redis_lock import RedisLock
 
 logger = get_logger(__name__)
 
@@ -1145,28 +1148,8 @@ def write_message_task(
             logger.info(f"[CELERY WRITE] Write completed successfully: {result}")
             return result
 
-    redis_client = get_sync_redis_client()
-    lock = None
-    if redis_client is not None:
-        lock = RedisFairLock(
-            key=f"memory_write:{end_user_id}",
-            redis_client=redis_client,
-            expire=120,
-            timeout=300,
-            auto_renewal=True,
-        )
-        if not lock.acquire():
-            logger.warning(f"[CELERY WRITE] 获取锁超时，跳过本次写入: end_user_id={end_user_id}")
-            return {
-                "status": "SKIPPED",
-                "error": "acquire lock timeout",
-                "end_user_id": end_user_id,
-                "config_id": str(config_id),
-                "elapsed_time": time.time() - start_time,
-                "task_id": self.request.id,
-            }
-
     try:
+        # 尝试获取现有事件循环，如果不存在则创建新的
         loop = set_asyncio_event_loop()
 
         result = loop.run_until_complete(_run())
@@ -1175,6 +1158,7 @@ def write_message_task(
         logger.info(f"[CELERY WRITE] Task completed successfully "
                     f"- elapsed_time={elapsed_time:.2f}s, task_id={self.request.id}")
 
+        # 记录该用户最后一次 write_message 成功的时间，供时间轴筛选使用
         try:
             _r = get_sync_redis_client()
             if _r is not None:
@@ -1215,12 +1199,9 @@ def write_message_task(
             "elapsed_time": elapsed_time,
             "task_id": self.request.id
         }
-    finally:
-        if lock is not None:
-            try:
-                lock.release()
-            except Exception as e:
-                logger.warning(f"[CELERY WRITE] 释放锁失败: {e}")
+
+
+# unused task
 # @celery_app.task(name="app.core.memory.agent.health.check_read_service")
 # def check_read_service_task() -> Dict[str, str]:
 #     """Call read_service and write latest status to Redis.
@@ -2898,6 +2879,3 @@ def init_community_clustering_for_users(self, end_user_ids: List[str], workspace
             "elapsed_time": time.time() - start_time,
             "task_id": self.request.id,
         }
-
-
-# unused task