diff --git a/tests/test_utils_logging_config.py b/tests/test_utils_logging_config.py index c3b4c4c..25ff048 100644 --- a/tests/test_utils_logging_config.py +++ b/tests/test_utils_logging_config.py @@ -175,6 +175,22 @@ class TestFilesAndRotation(IsolatedLoggerTestCase): stat.side_effect = stat_failing_for_logs self.assertEqual(self.manager.get_log_files(), []) + def test_truncate_does_not_roll_the_log_over_while_deleting_backups(self): + # With a file limit smaller than one record, any log line written + # while the backups are being deleted rolls the main log over and + # resurrects prompt_manager.log.1. The handler's own stream must also + # be reset, or its stale offset reports the old size, rolls over on the + # next record anyway and pads the file with zeros on POSIX. + self._rotate(max_file_size=400, backup_count=2, messages=10) + result = self.manager.truncate_logs() + self.assertEqual(result["errors"], []) + self.logger.info("after truncate") + leftovers = sorted(p.name for p in pathlib.Path(self.log_dir).iterdir()) + self.assertEqual(leftovers, ["prompt_manager.log"]) + main_log = pathlib.Path(self.log_dir, "prompt_manager.log").read_bytes() + self.assertNotIn(b"\x00", main_log) + self.assertIn(b"after truncate", main_log) + def test_truncate_clears_main_log_backups_and_buffer(self): self._rotate() result = self.manager.truncate_logs() diff --git a/utils/logging_config.py b/utils/logging_config.py index 023bae7..e4aa69c 100644 --- a/utils/logging_config.py +++ b/utils/logging_config.py @@ -13,7 +13,6 @@ import collections import logging import logging.handlers import os -import json import threading from datetime import datetime from typing import Dict, Any, Optional, List @@ -93,7 +92,8 @@ class PromptManagerLogger: # Custom formatter formatter = logging.Formatter( - "%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s", + "%(asctime)s - %(name)s - %(levelname)s - " + "%(filename)s:%(lineno)d - %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) @@ -270,6 +270,36 @@ class PromptManagerLogger: self.logger.error(f"Error reading log file {filename}: {e}") raise + def _truncate_main_log(self, log_file: Path) -> None: + """Empty the main log through the handler that writes to it. + + Opening the file a second time with ``"w"`` leaves the handler's own + stream at its old offset: later records land past the new end + (zero-padded on POSIX) and ``tell()`` keeps reporting the old size, so + the next record rolls the file over. Truncating the handler's stream + keeps position and size in step. + """ + handled = False + target = os.path.normcase(os.path.realpath(str(log_file))) + for handler in self.logger.handlers: + stream = getattr(handler, "stream", None) + base = getattr(handler, "baseFilename", None) + if stream is None or base is None: + continue + if os.path.normcase(os.path.realpath(base)) != target: + continue + handler.acquire() + try: + stream.flush() + stream.seek(0) + stream.truncate() + handled = True + finally: + handler.release() + if not handled: + with open(log_file, "w", encoding="utf-8", errors="replace") as f: + f.write("") + def truncate_logs(self) -> Dict[str, Any]: """Truncate all log files. @@ -283,28 +313,34 @@ class PromptManagerLogger: """ results = {"truncated": [], "errors": []} - for log_file in self.log_dir.glob("prompt_manager.log*"): + # Collect first and stay silent until every file has been handled: + # a log line written mid-way can roll the just-emptied main log over + # and recreate the backup that was deleted a moment earlier. + log_files = sorted(self.log_dir.glob("prompt_manager.log*")) + for log_file in log_files: try: if log_file.name == "prompt_manager.log": - # For main log file, just clear it with safe encoding - with open(log_file, "w", encoding="utf-8", errors="replace") as f: - f.write("") + self._truncate_main_log(log_file) else: # For rotated files, delete them log_file.unlink() results["truncated"].append(log_file.name) - self.logger.info(f"Truncated log file: {log_file.name}") except Exception as e: - error_msg = f"Failed to truncate {log_file.name}: {str(e)}" - results["errors"].append(error_msg) - self.logger.error(error_msg) + results["errors"].append( + f"Failed to truncate {log_file.name}: {str(e)}" + ) # Clear memory buffer with self._buffer_lock: self._log_buffer.clear() + if results["truncated"]: + self.logger.info("Truncated log files: %s", ", ".join(results["truncated"])) + for error_msg in results["errors"]: + self.logger.error(error_msg) + return results def update_config(self, new_config: Dict[str, Any]):