From 9571bf0a0c9154841a0a93e4681f848d6ce8fca1 Mon Sep 17 00:00:00 2001 From: TZZheng Date: Sun, 12 Jul 2026 15:49:59 -0500 Subject: [PATCH 1/3] fix(logging): make setup_logging idempotent (#749) --- src/lingtai/kernel/logging.py | 67 +++++++++++++++++--- tests/test_kernel_logging.py | 113 ++++++++++++++++++++++++++++++++++ 2 files changed, 172 insertions(+), 8 deletions(-) create mode 100644 tests/test_kernel_logging.py diff --git a/src/lingtai/kernel/logging.py b/src/lingtai/kernel/logging.py index 781907ea3..ea1b06ee8 100644 --- a/src/lingtai/kernel/logging.py +++ b/src/lingtai/kernel/logging.py @@ -5,6 +5,30 @@ from pathlib import Path _logger: logging.Logger | None = None +_HANDLER_OWNER_ATTR = "_lingtai_handler" +_CONSOLE_HANDLER = "console" +_FILE_HANDLER = "file" + + +def _owned_handlers(logger: logging.Logger, kind: str) -> list[logging.Handler]: + return [ + handler + for handler in logger.handlers + if getattr(handler, _HANDLER_OWNER_ATTR, None) == kind + ] + + +def _replace_owned_handler( + logger: logging.Logger, + handlers: list[logging.Handler], + replacement: logging.Handler, +) -> logging.Handler: + for handler in handlers: + logger.removeHandler(handler) + handler.close() + logger.addHandler(replacement) + return replacement + def setup_logging( verbose: bool = False, @@ -15,29 +39,56 @@ def setup_logging( Args: verbose: If True, set console to DEBUG; otherwise INFO. - log_dir: Directory for log files. None = no file logging. + log_dir: Directory for log files. None leaves file logging unchanged. logger_name: Logger name (default: "lingtai"). """ global _logger logger = logging.getLogger(logger_name) logger.setLevel(logging.DEBUG) - if not logger.handlers: + console_handlers = _owned_handlers(logger, _CONSOLE_HANDLER) + if console_handlers and isinstance(console_handlers[0], logging.StreamHandler): + ch = console_handlers[0] + for duplicate in console_handlers[1:]: + logger.removeHandler(duplicate) + duplicate.close() + else: ch = logging.StreamHandler() - ch.setLevel(logging.DEBUG if verbose else logging.INFO) + setattr(ch, _HANDLER_OWNER_ATTR, _CONSOLE_HANDLER) + ch = _replace_owned_handler(logger, console_handlers, ch) + ch.setLevel(logging.DEBUG if verbose else logging.INFO) + if ch.formatter is None: fmt = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s", datefmt="%H:%M:%S") ch.setFormatter(fmt) - logger.addHandler(ch) if log_dir is not None: log_path = Path(log_dir) log_path.mkdir(parents=True, exist_ok=True) - fh = logging.FileHandler(log_path / "agent.log") + filename = (log_path / "agent.log").resolve() + file_handlers = _owned_handlers(logger, _FILE_HANDLER) + if file_handlers and isinstance(file_handlers[0], logging.FileHandler): + fh = file_handlers[0] + for duplicate in file_handlers[1:]: + logger.removeHandler(duplicate) + duplicate.close() + if Path(fh.baseFilename) != filename: + fh.acquire() + try: + if fh.stream is not None: + fh.stream.close() + fh.baseFilename = str(filename) + fh.stream = fh._open() + finally: + fh.release() + else: + fh = logging.FileHandler(filename) + setattr(fh, _HANDLER_OWNER_ATTR, _FILE_HANDLER) + fh = _replace_owned_handler(logger, file_handlers, fh) fh.setLevel(logging.DEBUG) - fmt = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s") - fh.setFormatter(fmt) - logger.addHandler(fh) + if fh.formatter is None: + fmt = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s") + fh.setFormatter(fmt) _logger = logger return logger diff --git a/tests/test_kernel_logging.py b/tests/test_kernel_logging.py new file mode 100644 index 000000000..7c84fd42f --- /dev/null +++ b/tests/test_kernel_logging.py @@ -0,0 +1,113 @@ +"""Regression tests for package-internal logging setup.""" + +from __future__ import annotations + +import logging + +from lingtai.kernel.logging import setup_logging + + +def _close_handlers(logger: logging.Logger) -> None: + for handler in logger.handlers[:]: + logger.removeHandler(handler) + handler.close() + + +def test_setup_logging_preserves_application_handlers_and_is_idempotent(tmp_path): + logger = logging.getLogger("test.lingtai.logging.idempotent") + _close_handlers(logger) + application_handler = logging.StreamHandler() + logger.addHandler(application_handler) + + try: + first = setup_logging(log_dir=tmp_path, logger_name=logger.name) + owned = [ + handler + for handler in logger.handlers + if getattr(handler, "_lingtai_handler", None) + ] + + second = setup_logging(log_dir=tmp_path, logger_name=logger.name) + + assert first is second is logger + assert application_handler in logger.handlers + assert len(logger.handlers) == 3 + assert [ + handler + for handler in logger.handlers + if getattr(handler, "_lingtai_handler", None) + ] == owned + finally: + _close_handlers(logger) + + +def test_setup_logging_retargets_owned_file_handler(tmp_path): + logger = logging.getLogger("test.lingtai.logging.retarget") + _close_handlers(logger) + first_dir = tmp_path / "first" + second_dir = tmp_path / "second" + + try: + setup_logging(log_dir=first_dir, logger_name=logger.name) + file_handler = next( + handler + for handler in logger.handlers + if getattr(handler, "_lingtai_handler", None) == "file" + ) + + setup_logging(log_dir=second_dir, logger_name=logger.name) + logger.info("new target") + file_handler.flush() + + assert next( + handler + for handler in logger.handlers + if getattr(handler, "_lingtai_handler", None) == "file" + ) is file_handler + assert not (first_dir / "agent.log").read_text() + assert "new target" in (second_dir / "agent.log").read_text() + finally: + _close_handlers(logger) + + +def test_setup_logging_updates_console_level_and_none_keeps_file_handler(tmp_path): + logger = logging.getLogger("test.lingtai.logging.reconfigure") + _close_handlers(logger) + + try: + setup_logging(verbose=False, log_dir=tmp_path, logger_name=logger.name) + console_handler = next( + handler + for handler in logger.handlers + if getattr(handler, "_lingtai_handler", None) == "console" + ) + file_handler = next( + handler + for handler in logger.handlers + if getattr(handler, "_lingtai_handler", None) == "file" + ) + + setup_logging(verbose=True, log_dir=None, logger_name=logger.name) + + assert console_handler.level == logging.DEBUG + assert file_handler in logger.handlers + finally: + _close_handlers(logger) + + +def test_setup_logging_closes_duplicate_owned_file_handler(tmp_path): + logger = logging.getLogger("test.lingtai.logging.duplicate") + _close_handlers(logger) + + try: + setup_logging(log_dir=tmp_path, logger_name=logger.name) + duplicate = logging.FileHandler(tmp_path / "duplicate.log") + setattr(duplicate, "_lingtai_handler", "file") + logger.addHandler(duplicate) + + setup_logging(log_dir=tmp_path, logger_name=logger.name) + + assert duplicate not in logger.handlers + assert duplicate._closed + finally: + _close_handlers(logger) From 994ea286cdb8f7fa10c09d02e44aec0e8bc2291b Mon Sep 17 00:00:00 2001 From: TZZheng Date: Sun, 12 Jul 2026 16:05:51 -0500 Subject: [PATCH 2/3] fix(logging): serialize concurrent setup (#749) --- src/lingtai/kernel/logging.py | 91 ++++++++++++++++++----------------- tests/test_kernel_logging.py | 44 +++++++++++++++++ 2 files changed, 91 insertions(+), 44 deletions(-) diff --git a/src/lingtai/kernel/logging.py b/src/lingtai/kernel/logging.py index ea1b06ee8..cc3763371 100644 --- a/src/lingtai/kernel/logging.py +++ b/src/lingtai/kernel/logging.py @@ -2,9 +2,11 @@ from __future__ import annotations import logging +import threading from pathlib import Path _logger: logging.Logger | None = None +_SETUP_LOCK = threading.RLock() _HANDLER_OWNER_ATTR = "_lingtai_handler" _CONSOLE_HANDLER = "console" _FILE_HANDLER = "file" @@ -43,55 +45,56 @@ def setup_logging( logger_name: Logger name (default: "lingtai"). """ global _logger - logger = logging.getLogger(logger_name) - logger.setLevel(logging.DEBUG) + with _SETUP_LOCK: + logger = logging.getLogger(logger_name) + logger.setLevel(logging.DEBUG) - console_handlers = _owned_handlers(logger, _CONSOLE_HANDLER) - if console_handlers and isinstance(console_handlers[0], logging.StreamHandler): - ch = console_handlers[0] - for duplicate in console_handlers[1:]: - logger.removeHandler(duplicate) - duplicate.close() - else: - ch = logging.StreamHandler() - setattr(ch, _HANDLER_OWNER_ATTR, _CONSOLE_HANDLER) - ch = _replace_owned_handler(logger, console_handlers, ch) - ch.setLevel(logging.DEBUG if verbose else logging.INFO) - if ch.formatter is None: - fmt = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s", - datefmt="%H:%M:%S") - ch.setFormatter(fmt) - - if log_dir is not None: - log_path = Path(log_dir) - log_path.mkdir(parents=True, exist_ok=True) - filename = (log_path / "agent.log").resolve() - file_handlers = _owned_handlers(logger, _FILE_HANDLER) - if file_handlers and isinstance(file_handlers[0], logging.FileHandler): - fh = file_handlers[0] - for duplicate in file_handlers[1:]: + console_handlers = _owned_handlers(logger, _CONSOLE_HANDLER) + if console_handlers and isinstance(console_handlers[0], logging.StreamHandler): + ch = console_handlers[0] + for duplicate in console_handlers[1:]: logger.removeHandler(duplicate) duplicate.close() - if Path(fh.baseFilename) != filename: - fh.acquire() - try: - if fh.stream is not None: - fh.stream.close() - fh.baseFilename = str(filename) - fh.stream = fh._open() - finally: - fh.release() else: - fh = logging.FileHandler(filename) - setattr(fh, _HANDLER_OWNER_ATTR, _FILE_HANDLER) - fh = _replace_owned_handler(logger, file_handlers, fh) - fh.setLevel(logging.DEBUG) - if fh.formatter is None: - fmt = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s") - fh.setFormatter(fmt) + ch = logging.StreamHandler() + setattr(ch, _HANDLER_OWNER_ATTR, _CONSOLE_HANDLER) + ch = _replace_owned_handler(logger, console_handlers, ch) + ch.setLevel(logging.DEBUG if verbose else logging.INFO) + if ch.formatter is None: + fmt = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s", + datefmt="%H:%M:%S") + ch.setFormatter(fmt) + + if log_dir is not None: + log_path = Path(log_dir) + log_path.mkdir(parents=True, exist_ok=True) + filename = (log_path / "agent.log").resolve() + file_handlers = _owned_handlers(logger, _FILE_HANDLER) + if file_handlers and isinstance(file_handlers[0], logging.FileHandler): + fh = file_handlers[0] + for duplicate in file_handlers[1:]: + logger.removeHandler(duplicate) + duplicate.close() + if Path(fh.baseFilename) != filename: + fh.acquire() + try: + if fh.stream is not None: + fh.stream.close() + fh.baseFilename = str(filename) + fh.stream = fh._open() + finally: + fh.release() + else: + fh = logging.FileHandler(filename) + setattr(fh, _HANDLER_OWNER_ATTR, _FILE_HANDLER) + fh = _replace_owned_handler(logger, file_handlers, fh) + fh.setLevel(logging.DEBUG) + if fh.formatter is None: + fmt = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s") + fh.setFormatter(fmt) - _logger = logger - return logger + _logger = logger + return logger def get_logger() -> logging.Logger: diff --git a/tests/test_kernel_logging.py b/tests/test_kernel_logging.py index 7c84fd42f..d83d3cc59 100644 --- a/tests/test_kernel_logging.py +++ b/tests/test_kernel_logging.py @@ -3,7 +3,9 @@ from __future__ import annotations import logging +import threading +import lingtai.kernel.logging as kernel_logging from lingtai.kernel.logging import setup_logging @@ -111,3 +113,45 @@ def test_setup_logging_closes_duplicate_owned_file_handler(tmp_path): assert duplicate._closed finally: _close_handlers(logger) + + +def test_setup_logging_serializes_concurrent_handler_discovery(monkeypatch): + logger = logging.getLogger("test.lingtai.logging.concurrent") + _close_handlers(logger) + discovery_barrier = threading.Barrier(2) + original_owned_handlers = kernel_logging._owned_handlers + + def synchronized_discovery(target, kind): + handlers = original_owned_handlers(target, kind) + if kind == "console" and not handlers: + try: + discovery_barrier.wait(timeout=0.2) + except threading.BrokenBarrierError: + pass + return handlers + + monkeypatch.setattr(kernel_logging, "_owned_handlers", synchronized_discovery) + errors = [] + + def run_setup(): + try: + setup_logging(logger_name=logger.name) + except Exception as exc: + errors.append(exc) + + threads = [ + threading.Thread(target=run_setup) + for _ in range(2) + ] + + try: + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=2) + + assert all(not thread.is_alive() for thread in threads) + assert not errors + assert len(original_owned_handlers(logger, "console")) == 1 + finally: + _close_handlers(logger) From e6e99ad7f2ba6796e19d0884107ff3a9f9a2e571 Mon Sep 17 00:00:00 2001 From: TZZheng Date: Sun, 12 Jul 2026 16:26:43 -0500 Subject: [PATCH 3/3] fix(logging): retarget file handler transactionally --- src/lingtai/kernel/logging.py | 20 +++++------ tests/test_kernel_logging.py | 64 ++++++++++++++++++++++++++++++++--- 2 files changed, 69 insertions(+), 15 deletions(-) diff --git a/src/lingtai/kernel/logging.py b/src/lingtai/kernel/logging.py index cc3763371..22ed3f140 100644 --- a/src/lingtai/kernel/logging.py +++ b/src/lingtai/kernel/logging.py @@ -72,18 +72,16 @@ def setup_logging( file_handlers = _owned_handlers(logger, _FILE_HANDLER) if file_handlers and isinstance(file_handlers[0], logging.FileHandler): fh = file_handlers[0] - for duplicate in file_handlers[1:]: - logger.removeHandler(duplicate) - duplicate.close() if Path(fh.baseFilename) != filename: - fh.acquire() - try: - if fh.stream is not None: - fh.stream.close() - fh.baseFilename = str(filename) - fh.stream = fh._open() - finally: - fh.release() + replacement = logging.FileHandler(filename) + setattr(replacement, _HANDLER_OWNER_ATTR, _FILE_HANDLER) + replacement.setLevel(fh.level) + replacement.setFormatter(fh.formatter) + fh = _replace_owned_handler(logger, file_handlers, replacement) + else: + for duplicate in file_handlers[1:]: + logger.removeHandler(duplicate) + duplicate.close() else: fh = logging.FileHandler(filename) setattr(fh, _HANDLER_OWNER_ATTR, _FILE_HANDLER) diff --git a/tests/test_kernel_logging.py b/tests/test_kernel_logging.py index d83d3cc59..d13b2bb2f 100644 --- a/tests/test_kernel_logging.py +++ b/tests/test_kernel_logging.py @@ -59,19 +59,75 @@ def test_setup_logging_retargets_owned_file_handler(tmp_path): setup_logging(log_dir=second_dir, logger_name=logger.name) logger.info("new target") - file_handler.flush() - - assert next( + replacement = next( handler for handler in logger.handlers if getattr(handler, "_lingtai_handler", None) == "file" - ) is file_handler + ) + replacement.flush() + + assert replacement is not file_handler + assert file_handler._closed assert not (first_dir / "agent.log").read_text() assert "new target" in (second_dir / "agent.log").read_text() finally: _close_handlers(logger) +def test_setup_logging_failed_retarget_preserves_handler_and_retry_recovers(tmp_path): + logger = logging.getLogger("test.lingtai.logging.retarget_failure") + _close_handlers(logger) + first_dir = tmp_path / "first" + blocked_dir = tmp_path / "blocked" + blocked_dir.mkdir() + obstacle = blocked_dir / "agent.log" + obstacle.mkdir() + + try: + setup_logging(log_dir=first_dir, logger_name=logger.name) + file_handler = next( + handler + for handler in logger.handlers + if getattr(handler, "_lingtai_handler", None) == "file" + ) + original_filename = file_handler.baseFilename + original_stream = file_handler.stream + original_formatter = file_handler.formatter + original_level = file_handler.level + + try: + setup_logging(log_dir=blocked_dir, logger_name=logger.name) + except IsADirectoryError: + pass + else: + raise AssertionError("retargeting to a directory should fail") + + assert file_handler in logger.handlers + assert file_handler.baseFilename == original_filename + assert file_handler.stream is original_stream + assert file_handler.formatter is original_formatter + assert file_handler.level == original_level + logger.info("old target remains usable") + file_handler.flush() + assert "old target remains usable" in (first_dir / "agent.log").read_text() + + obstacle.rmdir() + setup_logging(log_dir=blocked_dir, logger_name=logger.name) + replacement = next( + handler + for handler in logger.handlers + if getattr(handler, "_lingtai_handler", None) == "file" + ) + logger.info("retry reached new target") + replacement.flush() + + assert replacement is not file_handler + assert file_handler._closed + assert "retry reached new target" in obstacle.read_text() + finally: + _close_handlers(logger) + + def test_setup_logging_updates_console_level_and_none_keeps_file_handler(tmp_path): logger = logging.getLogger("test.lingtai.logging.reconfigure") _close_handlers(logger)