From 83d157689c5d1dc4a2a10d8176e81dff97e2b9c6 Mon Sep 17 00:00:00 2001 From: chentang Date: Thu, 18 Dec 2025 14:38:19 +0800 Subject: [PATCH 1/2] fix bugs: try to fix bugs in _submit_web_logs --- src/memos/mem_scheduler/base_scheduler.py | 37 ++++++++++------------- 1 file changed, 16 insertions(+), 21 deletions(-) diff --git a/src/memos/mem_scheduler/base_scheduler.py b/src/memos/mem_scheduler/base_scheduler.py index 81defaa0f..9ab356f1d 100644 --- a/src/memos/mem_scheduler/base_scheduler.py +++ b/src/memos/mem_scheduler/base_scheduler.py @@ -846,28 +846,23 @@ def _submit_web_logs( f"[DIAGNOSTIC] base_scheduler._submit_web_logs called. Message to publish: {message.model_dump_json(indent=2)}" ) - if self.rabbitmq_config is None: - logger.info( - "[DIAGNOSTIC] base_scheduler._submit_web_logs: RabbitMQ config not loaded; skipping publish." - ) - return - - for message in messages: - message_info = message.debug_info() - logger.info(f"[DIAGNOSTIC] base_scheduler._submit_web_logs: submitted {message_info}") + try: + for message in messages: + # Always call publish; the publisher now caches when offline and flushes after reconnect + logger.info( + f"[DIAGNOSTIC] base_scheduler._submit_web_logs: enqueue publish {message.model_dump_json(indent=2)}" + ) + self.rabbitmq_publish_message(message=message.to_dict()) + logger.info( + "[DIAGNOSTIC] base_scheduler._submit_web_logs: publish dispatched " + "item_id=%s task_id=%s label=%s", + message.item_id, + message.task_id, + message.label, + ) + except Exception as e: + logger.error(f"[DIAGNOSTIC] base_scheduler._submit_web_logs failed: {e}", exc_info=True) - # Always call publish; the publisher now caches when offline and flushes after reconnect - logger.info( - f"[DIAGNOSTIC] base_scheduler._submit_web_logs: enqueue publish {message_info}" - ) - self.rabbitmq_publish_message(message=message.to_dict()) - logger.info( - "[DIAGNOSTIC] base_scheduler._submit_web_logs: publish dispatched " - "item_id=%s task_id=%s label=%s", - message.item_id, - message.task_id, - message.label, - ) logger.debug( f"{len(messages)} submitted. {self._web_log_message_queue.qsize()} in queue. additional_log_info: {additional_log_info}" ) From e50c56cf817cb6d63b8e8e882aeaa4de12c444b8 Mon Sep 17 00:00:00 2001 From: chentang Date: Thu, 18 Dec 2025 15:00:28 +0800 Subject: [PATCH 2/2] fix bugs: try to address bugs --- src/memos/mem_scheduler/base_scheduler.py | 13 +++++-------- .../webservice_modules/rabbitmq_service.py | 6 ++++-- 2 files changed, 9 insertions(+), 10 deletions(-) diff --git a/src/memos/mem_scheduler/base_scheduler.py b/src/memos/mem_scheduler/base_scheduler.py index 9ab356f1d..1e0ecaadb 100644 --- a/src/memos/mem_scheduler/base_scheduler.py +++ b/src/memos/mem_scheduler/base_scheduler.py @@ -842,12 +842,7 @@ def _submit_web_logs( messages = [messages] # transform single message to list for message in messages: - logger.info( - f"[DIAGNOSTIC] base_scheduler._submit_web_logs called. Message to publish: {message.model_dump_json(indent=2)}" - ) - - try: - for message in messages: + try: # Always call publish; the publisher now caches when offline and flushes after reconnect logger.info( f"[DIAGNOSTIC] base_scheduler._submit_web_logs: enqueue publish {message.model_dump_json(indent=2)}" @@ -860,8 +855,10 @@ def _submit_web_logs( message.task_id, message.label, ) - except Exception as e: - logger.error(f"[DIAGNOSTIC] base_scheduler._submit_web_logs failed: {e}", exc_info=True) + except Exception as e: + logger.error( + f"[DIAGNOSTIC] base_scheduler._submit_web_logs failed: {e}", exc_info=True + ) logger.debug( f"{len(messages)} submitted. {self._web_log_message_queue.qsize()} in queue. additional_log_info: {additional_log_info}" diff --git a/src/memos/mem_scheduler/webservice_modules/rabbitmq_service.py b/src/memos/mem_scheduler/webservice_modules/rabbitmq_service.py index a8a09760c..db8320879 100644 --- a/src/memos/mem_scheduler/webservice_modules/rabbitmq_service.py +++ b/src/memos/mem_scheduler/webservice_modules/rabbitmq_service.py @@ -7,6 +7,8 @@ from pathlib import Path from queue import Empty +from pyglet.libs.win32.constants import FALSE + from memos.configs.mem_scheduler import AuthConfig, RabbitMQConfig from memos.context.context import ContextThread from memos.dependency import require_python_package @@ -325,14 +327,14 @@ def rabbitmq_publish_message(self, message: dict): f"[DIAGNOSTIC] Publishing {label} message in Cloud Env. " f"Exchange: {exchange_name}, Routing Key: '{routing_key}'." ) - logger.info(f" - Message Content: {json.dumps(message, indent=2)}") + logger.info(f" - Message Content: {json.dumps(message, indent=2, ensure_ascii=FALSE)}") elif label == "knowledgeBaseUpdate": # Original diagnostic logging for knowledgeBaseUpdate if NOT in cloud env logger.info( f"[DIAGNOSTIC] Publishing knowledgeBaseUpdate message (Local Env). " f"Current configured Exchange: {exchange_name}, Routing Key: '{routing_key}'." ) - logger.info(f" - Message Content: {json.dumps(message, indent=2)}") + logger.info(f" - Message Content: {json.dumps(message, indent=2, ensure_ascii=FALSE)}") with self._rabbitmq_lock: logger.info(