fix: propagate memory tenant id in task collect (#15837)

### What problem does this PR solve?
Propagate `tenant_id` from memory task messages into task collection so
refactored task execution can build a valid context.

### Type of change
- [x] Bug Fix (non-breaking change which fixes an issue)

api/apps/services/memory_api_service.py

@@ -131,6 +131,20 @@ async def update_memory(memory_id: str, new_memory_setting: dict):
     """
     current_memory = _require_memory_access(memory_id)
 
+    def _normalize_memory_type(value):
+        if value is None:
+            return []
+        if isinstance(value, int):
+            return sorted(get_memory_type_human(value))
+        if isinstance(value, list):
+            return sorted(str(v).strip().lower() for v in value if str(v).strip())
+        return sorted(str(value).strip().lower().split(","))
+
+    def _normalize_str(value):
+        if value is None:
+            return ""
+        return str(value).strip()
+
     update_dict = {}
     # check name length
     if "name" in new_memory_setting:
@@ -186,8 +200,21 @@ async def update_memory(memory_id: str, new_memory_setting: dict):
     memory_dict.update({"memory_type": get_memory_type_human(current_memory.memory_type)})
     to_update = {}
     for k, v in update_dict.items():
-        if isinstance(v, list) and set(memory_dict[k]) != set(v):
-            to_update[k] = v
+        if k == "memory_type":
+            current_value = _normalize_memory_type(memory_dict.get(k))
+            new_value = _normalize_memory_type(v)
+            if current_value != new_value:
+                to_update[k] = new_value
+        elif k == "embd_id":
+            current_value = _normalize_str(memory_dict.get(k))
+            new_value = _normalize_str(v)
+            if current_value != new_value:
+                to_update[k] = new_value
+        elif isinstance(v, list):
+            current_value = sorted(str(item).strip() for item in memory_dict.get(k, []))
+            new_value = sorted(str(item).strip() for item in v)
+            if current_value != new_value:
+                to_update[k] = v
         elif memory_dict[k] != v:
             to_update[k] = v
 

api/db/__init__.py

@@ -17,6 +17,8 @@
 from enum import IntEnum
 from enum import StrEnum
 
+from common.constants import PipelineTaskType
+
 
 class UserTenantRole(StrEnum):
     OWNER = 'owner'
@@ -59,14 +61,6 @@ class CanvasCategory(StrEnum):
     DataFlow = "dataflow_canvas"
 
 
-class PipelineTaskType(StrEnum):
-    PARSE = "Parse"
-    DOWNLOAD = "Download"
-    RAPTOR = "RAPTOR"
-    GRAPH_RAG = "GraphRAG"
-    MINDMAP = "Mindmap"
-
-
 VALID_PIPELINE_TASK_TYPES = {PipelineTaskType.PARSE, PipelineTaskType.DOWNLOAD, PipelineTaskType.RAPTOR, PipelineTaskType.GRAPH_RAG, PipelineTaskType.MINDMAP}
 
 

api/db/joint_services/memory_message_service.py

@@ -14,6 +14,7 @@
 #  limitations under the License.
 #
 import logging
+from datetime import datetime
 from typing import List
 
 from common import settings
@@ -349,8 +350,9 @@ async def queue_save_to_memory_task(memory_ids: list[str], message_dict: dict):
             "doc_id": _memory_id,
             "task_type": "memory",
             "progress": 0.0,
+            "begin_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
             "digest": str(_source_id)
-        }
+    }
 
     not_found_memory = []
     failed_memory = []
@@ -387,6 +389,7 @@ async def queue_save_to_memory_task(memory_ids: list[str], message_dict: dict):
             "task_id": task["id"],
             "task_type": task["task_type"],
             "memory_id": memory_id,
+            "tenant_id": memory.tenant_id,
             "source_id": raw_message_id,
             "message_dict": message_dict
         }

api/db/services/pipeline_operation_log_service.py

@@ -20,14 +20,14 @@ from datetime import datetime, timedelta
 
 from peewee import fn
 
-from api.db import VALID_PIPELINE_TASK_TYPES, PipelineTaskType
+from api.db import VALID_PIPELINE_TASK_TYPES
 from api.db.db_models import DB, Document, PipelineOperationLog
 from api.db.services.canvas_service import UserCanvasService
 from api.db.services.common_service import CommonService
 from api.db.services.document_service import DocumentService
 from api.db.services.knowledgebase_service import KnowledgebaseService
 from api.db.services.task_service import GRAPH_RAPTOR_FAKE_DOC_ID, TaskService
-from common.constants import TaskStatus
+from common.constants import PipelineTaskType, TaskStatus
 from common.misc_utils import get_uuid
 from common.time_utils import current_timestamp, datetime_format
 

api/db/services/task_service.py

@@ -343,8 +343,10 @@ class TaskService(CommonService):
                         ((prog == -1) | (prog > cls.model.progress))))
                     ).execute()
 
-        process_duration = (datetime.now() - task.begin_at).total_seconds()
-        cls.model.update(process_duration=process_duration).where(cls.model.id == id).execute()
+        begin_at = task.begin_at
+        if begin_at is not None:
+            process_duration = (datetime.now() - begin_at).total_seconds()
+            cls.model.update(process_duration=process_duration).where(cls.model.id == id).execute()
 
     @classmethod
     @DB.connection_context()

rag/svr/task_executor.py

@@ -256,6 +256,8 @@ async def collect():
         task["kb_id"] = msg.get("kb_id", "")
     if task_type[:6] == "memory":
         task["memory_id"] = msg["memory_id"]
+        if msg.get("tenant_id"):
+            task["tenant_id"] = msg["tenant_id"]
         task["source_id"] = msg["source_id"]
         task["message_dict"] = msg["message_dict"]
     return redis_msg, task