fix: truncating logs no longer rolls the main log over or pads it with zeros

truncate_logs emptied prompt_manager.log through a second open(), leaving the
rotating handler's stream at its old offset: the next record was written past
the new end (zero padding on POSIX) and tell() still reported the old size, so
the file rolled over and resurrected the backup that had just been deleted.
It also logged once per file from inside the deletion loop, which could do the
same with a small size limit. The main log is now truncated through the
handler's own stream and a single summary line is logged afterwards.
This commit is contained in:
Vito Sansevero
2026-10-08 18:01:34 -07:00
parent 9d6eb728fc
commit f99e76a135
2 changed files with 62 additions and 10 deletions
+16
View File
@@ -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()
+46 -10
View File
@@ -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]):