fix: preserve consumer logging ownership and routing

CHANGELOG.md

@@ -30,6 +30,7 @@ and versions are tracked in the repo-root `VERSION` file.
 
 ### Fixed
 
+- Preserve consumer-owned logging handlers, explicit levels, and parent routing across CLI invocations (#387).
 - Validate nested configuration mappings before merge/provenance traversal,
   reject recursive or excessively deep values with source-aware errors, and
   continue to accept shared YAML aliases.

docs/api-reference.md

@@ -1030,7 +1030,7 @@ base_cli.command(...)
 
 ### `configure_logger`
 **Kind:** function  
-**Signature:** `configure_logger(cli_name: 'str', log_file: 'Path | None', debug: 'bool', *, quiet: 'bool' = False, stream: 'TextIO | None' = None, formatter: 'logging.Formatter | None' = None, json_logs: 'bool' = False, run_id: 'str | None' = None, log_level: 'str | None' = None) -> 'logging.Logger'`
+**Signature:** `configure_logger(cli_name: 'str', log_file: 'Path | None', debug: 'bool', *, quiet: 'bool' = False, stream: 'TextIO | None' = None, formatter: 'logging.Formatter | None' = None, json_logs: 'bool' = False, run_id: 'str | None' = None, log_level: 'str | None' = None, propagate: 'bool | None' = None) -> 'logging.Logger'`
 
 **Behavior:** Configure user-facing and persistent handlers for a CLI logger.
 

docs/integrations.md

@@ -53,6 +53,21 @@ paths, and secrets are never attached. A missing API package, invalid provider,
 or failing exporter is logged at debug level and treated as a no-op; it cannot
 change the command's exit status or cleanup behavior.
 
+## Logger ownership
+
+The lifecycle uses `base_cli.<cli_name>` and owns only the handlers it creates.
+Consumer handlers remain attached and open after configuration and cleanup.
+An explicit consumer logger level, or configuration on the `base_cli` parent,
+is preserved, including `logging.config.dictConfig()` routing. Such levels may
+filter records before the lifecycle handlers see them. Configure the consumer
+logger at DEBUG if the persistent handler should receive every record.
+
+Unconfigured CLI loggers use DEBUG with propagation disabled to avoid duplicate
+terminal output. `configure_logger(..., propagate=True)` explicitly enables host
+routing; `False` disables it, and the default `None` preserves consumer routing.
+Use a consumer handler or configure the `base_cli` parent before invoking an App
+when embedding it in a host with centralized logging.
+
 ## Log timestamp environment variable
 
 Set `BASE_CLI_LOG_UTC=1` to make the default text formatter use UTC

lib/python/base_cli/context.py

@@ -156,6 +156,8 @@ def _cleanup_resources(self, *, preserve_temp_ownership: bool) -> None:
         elif not preserve_temp_ownership:
             self._close_owned_temp_descriptor()
         for handler in list(self.log.handlers):
+            if not getattr(handler, "_base_cli_owned", False):
+                continue
             try:
                 handler.flush()
             except BaseException as exc:  # pylint: disable=broad-exception-caught

lib/python/base_cli/logging.py

@@ -57,12 +57,15 @@ def configure_logger(
     json_logs: bool = False,
     run_id: str | None = None,
     log_level: str | None = None,
+    propagate: bool | None = None,
 ) -> logging.Logger:
     """Configure user-facing and persistent handlers for a CLI logger.
 
     ``log_level`` optionally selects the user-stream threshold from DEBUG,
     INFO, WARNING, ERROR, or CRITICAL. The persistent file handler remains at
     DEBUG. When omitted, the existing ``debug`` and ``quiet`` policy applies.
+    Consumer handlers and configured levels are preserved. ``propagate=None``
+    preserves consumer routing; unconfigured loggers default to no propagation.
     """
     normalized_log_level = log_level.lower() if log_level is not None else None
     if normalized_log_level is not None and normalized_log_level not in _CONFIGURED_LOG_LEVELS:
@@ -74,11 +77,28 @@ def configure_logger(
         else _CONFIGURED_LOG_LEVELS[normalized_log_level]
     )
     logger = logging.getLogger(f"base_cli.{cli_name}")
-    logger.setLevel(logging.DEBUG)
-    logger.propagate = False
+    foreign_handlers = any(not getattr(handler, "_base_cli_owned", False) for handler in logger.handlers)
+    parent = logger.parent
+    parent_configured = False
+    while parent is not None and parent is not logging.root:
+        parent_configured |= bool(parent.handlers) or parent.level != logging.NOTSET
+        parent = parent.parent
+    configured = (
+        foreign_handlers
+        or parent_configured
+        or (logger.level != logging.NOTSET and logger.level != getattr(logger, "_base_cli_level", None))
+    )
+    if not configured:
+        logger.setLevel(logging.DEBUG)
+        logger._base_cli_level = logging.DEBUG  # type: ignore[attr-defined]
+    if propagate is not None:
+        logger.propagate = propagate
+    elif not configured:
+        logger.propagate = False
     for handler in list(logger.handlers):
-        handler.close()
-        logger.removeHandler(handler)
+        if getattr(handler, "_base_cli_owned", False):
+            handler.close()
+            logger.removeHandler(handler)
 
     user_stream = stream if stream is not None else sys.stderr
     user_handler = logging.StreamHandler(user_stream)
@@ -91,6 +111,7 @@ def configure_logger(
             run_id=run_id,
         )
     )
+    user_handler._base_cli_owned = True  # type: ignore[attr-defined]
     logger.addHandler(user_handler)
 
     if log_file is not None:
@@ -104,6 +125,7 @@ def configure_logger(
                 run_id=run_id,
             )
         )
+        file_handler._base_cli_owned = True  # type: ignore[attr-defined]
         logger.addHandler(file_handler)
     return logger
 

tests/test_app_lifecycle.py

@@ -15,6 +15,8 @@
 
 
 class _BrokenHandler(logging.Handler):
+    _base_cli_owned = True
+
     def emit(self, record: logging.LogRecord) -> None:
         del record
 
@@ -205,6 +207,8 @@ def test_handler_removal_interruption_uses_direct_detach_fallback(self) -> None:
             logger = logging.Logger("isolated-handler-removal")
             logger.addHandler(logging.NullHandler())
             logger.addHandler(logging.NullHandler())
+            for handler in logger.handlers:
+                handler._base_cli_owned = True
             logger.removeHandler = mock.Mock(side_effect=KeyboardInterrupt())
             context = base_cli.Context(
                 cli_name="handler-removal-interrupt",

tests/test_cleanup_security.py

@@ -30,7 +30,9 @@ def _context(
     ) -> tuple[base_cli.Context, io.StringIO]:
         stream = io.StringIO()
         logger = logging.Logger(f"cleanup-security-{id(stream)}", level=logging.DEBUG)
-        logger.addHandler(logging.StreamHandler(stream))
+        handler = logging.StreamHandler(stream)
+        handler._base_cli_owned = True
+        logger.addHandler(handler)
         context = base_cli.Context(
             cli_name="cleanup-security",
             run_id=run_id,

tests/test_logger_ownership.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import io
+import logging
+from pathlib import Path
+from unittest.mock import Mock
+
+import base_cli
+from base_cli.testing import invoke
+
+
+def test_consumer_handler_and_level_survive_invocation(tmp_path: Path) -> None:
+    logger = logging.getLogger("base_cli.owned-handler-test")
+    sink = logging.StreamHandler(io.StringIO())
+    sink.close = Mock()
+    logger.addHandler(sink)
+    logger.setLevel(logging.INFO)
+    logger.propagate = True
+    app = base_cli.App(name="owned-handler-test", version="1")
+
+    @app.command()
+    def main(ctx: base_cli.Context) -> None:
+        ctx.log.info("consumer message")
+
+    try:
+        assert invoke(app, [], home=tmp_path).exit_code == 0
+        assert logger.handlers == [sink]
+        assert logger.level == logging.INFO
+        assert logger.propagate
+        sink.close.assert_not_called()
+        assert "consumer message" in sink.stream.getvalue()
+    finally:
+        logger.removeHandler(sink)
+
+
+def test_default_logger_does_not_duplicate_through_root() -> None:
+    stream = io.StringIO()
+    root_sink = logging.StreamHandler(stream)
+    logging.root.addHandler(root_sink)
+    try:
+        logger = base_cli.configure_logger("default-no-duplicates", None, False, stream=stream)
+        logger.warning("one warning")
+        assert stream.getvalue().count("one warning") == 1
+    finally:
+        logging.root.removeHandler(root_sink)
+        for handler in list(logger.handlers):
+            handler.close()
+            logger.removeHandler(handler)
+
+
+def test_parent_consumer_configuration_is_preserved() -> None:
+    parent = logging.getLogger("base_cli.host")
+    stream = io.StringIO()
+    sink = logging.StreamHandler(stream)
+    parent.addHandler(sink)
+    try:
+        logger = base_cli.configure_logger("host.child", None, False, stream=io.StringIO())
+        logger.warning("host record")
+        assert logger.propagate
+        assert "host record" in stream.getvalue()
+    finally:
+        parent.removeHandler(sink)
+        for handler in list(logger.handlers):
+            handler.close()
+            logger.removeHandler(handler)