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.

compatibility/consumers/atlas_click/src/atlas_click/cli.py

@@ -2,8 +2,8 @@
 
 from __future__ import annotations
 
-import click
 import base_cli
+import click
 
 
 @click.group(name="atlas-consumer", help="Inventory resources managed by Atlas.")

compatibility/consumers/atlas_click/tests/test_consumer.py

@@ -4,7 +4,6 @@
 from pathlib import Path
 
 import base_cli
-
 from atlas_click.cli import command
 
 

compatibility/consumers/beacon_typer/src/beacon_typer/cli.py

@@ -5,7 +5,6 @@
 import base_cli
 import typer
 
-
 cli = typer.Typer(help="Deploy Beacon services with typed parameters.")
 
 

compatibility/consumers/beacon_typer/tests/test_consumer.py

@@ -4,7 +4,6 @@
 from pathlib import Path
 
 import base_cli
-
 from beacon_typer.cli import command
 
 

compatibility/consumers/cinder_automation/src/cinder_automation/cli.py

@@ -2,9 +2,10 @@
 
 from __future__ import annotations
 
-import click
-import base_cli
+from typing import Any
 
+import base_cli
+import click
 
 app = base_cli.App(
     name="cinder-consumer",
@@ -27,7 +28,7 @@
     default="json",
     show_default=True,
 )
-def reconcile(ctx: base_cli.Context, target: str, output_format: str) -> None:
+def reconcile(ctx: base_cli.Context[Any, Any, Any], target: str, output_format: str) -> None:
     """Publish the result of one idempotent reconciliation step."""
 
     action = "would-reconcile" if ctx.dry_run else "reconciled"

compatibility/consumers/cinder_automation/tests/test_consumer.py

@@ -4,7 +4,6 @@
 from pathlib import Path
 
 import base_cli
-
 from cinder_automation.cli import app
 
 

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

@@ -60,6 +60,21 @@ exceptions. Interrupts and other non-success outcomes are therefore visible as
 errors while retaining the existing `base_cli.outcome` attribute for detailed
 dashboard filtering.
 
+## 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

docs/testing.md

@@ -55,3 +55,22 @@ The full gate writes a machine-readable result to
 `$BASE_CLI_VALIDATION_RESULT` (or `/tmp/base-cli-validation-result.json`). If
 Node.js is unavailable, the result is marked `partial`, the gate exits with
 status `2`, and it cannot be reported as an authoritative pass.
+
+Examples and compatibility consumers use the fully parameterized context type
+when strict mypy checks a callback directly:
+
+```python
+from typing import Any
+import base_cli
+
+def main(ctx: base_cli.Context[Any, Any, Any]) -> None:
+    ...
+```
+
+### Consumer source quality
+
+The style gate runs Ruff over the entire repository with its standard generated-file
+exclusions. The typing gate checks every Git-visible Python source outside `lib/`
+(checked separately), `scripts/` (validation tools), and `tests/` (test harnesses)
+with strict mypy. This includes example and compatibility consumer packages and
+new top-level source directories; untracked sources are included during development.

examples/automation_observability_app/src/automation_observability_app/cli.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 import base_cli
 import click
 
@@ -35,7 +37,7 @@
 )
 @base_cli.option("--api-token", hidden=True, help="Optional secret for a real adapter.")
 def run(
-    ctx: base_cli.Context,
+    ctx: base_cli.Context[Any, Any, Any],
     target: str,
     output_format: str,
     api_token: str | None,

examples/minimal_cli/src/minimal_cli/cli.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 import base_cli
 
 app = base_cli.App(
@@ -13,7 +15,7 @@
 
 @app.command()
 @base_cli.option("--name", required=True, help="Name to greet.")
-def greet(ctx: base_cli.Context, name: str) -> None:
+def greet(ctx: base_cli.Context[Any, Any, Any], name: str) -> None:
     """Print a deterministic greeting."""
 
     ctx.log.info("greeting requested for %s", name)

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

@@ -8,6 +8,7 @@
 import warnings
 from io import TextIOWrapper
 from pathlib import Path
+from threading import RLock
 from typing import BinaryIO, TextIO, cast
 
 try:
@@ -43,6 +44,7 @@
     "error": logging.ERROR,
     "critical": logging.CRITICAL,
 }
+_CONFIGURE_LOGGER_LOCK = RLock()
 
 
 # pylint: disable=too-many-arguments
@@ -57,12 +59,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,37 +79,53 @@ 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
-    for handler in list(logger.handlers):
-        handler.close()
-        logger.removeHandler(handler)
-
-    user_stream = stream if stream is not None else sys.stderr
-    user_handler = logging.StreamHandler(user_stream)
-    user_handler.setLevel(stream_level)
-    user_handler.setFormatter(
-        _handler_formatter(
-            formatter,
-            use_color=_use_color(user_stream),
-            json_logs=json_logs,
-            run_id=run_id,
-        )
-    )
-    logger.addHandler(user_handler)
-
-    if log_file is not None:
-        file_handler = SecureLogFileHandler(log_file, encoding="utf-8")
-        file_handler.setLevel(logging.DEBUG)
-        file_handler.setFormatter(
+    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
+    externally_routed = foreign_handlers or parent_configured
+    if logger.level == logging.NOTSET:
+        logger.setLevel(logging.DEBUG)
+        logger._base_cli_level = logging.DEBUG  # type: ignore[attr-defined]
+    if propagate is not None:
+        logger.propagate = propagate
+    else:
+        logger.propagate = externally_routed
+    with _CONFIGURE_LOGGER_LOCK:
+        for handler in list(logger.handlers):
+            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)
+        user_handler.setLevel(stream_level)
+        user_handler.setFormatter(
             _handler_formatter(
                 formatter,
-                use_color=False,
+                use_color=_use_color(user_stream),
                 json_logs=json_logs,
                 run_id=run_id,
             )
         )
-        logger.addHandler(file_handler)
+        user_handler._base_cli_owned = True  # type: ignore[attr-defined]
+        logger.addHandler(user_handler)
+
+        if log_file is not None:
+            file_handler = SecureLogFileHandler(log_file, encoding="utf-8")
+            file_handler.setLevel(logging.DEBUG)
+            file_handler.setFormatter(
+                _handler_formatter(
+                    formatter,
+                    use_color=False,
+                    json_logs=json_logs,
+                    run_id=run_id,
+                )
+            )
+            file_handler._base_cli_owned = True  # type: ignore[attr-defined]
+            logger.addHandler(file_handler)
     return logger
 
 

scripts/validate_consumer_typing.py

@@ -0,0 +1,48 @@
+"""Type-check every example and compatibility consumer at strict settings."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+
+def main() -> int:
+    root = Path(__file__).resolve().parents[1]
+    # Git discovery also includes new, untracked source directories. Infrastructure
+    # scripts/tests have separate runtime gates; all product/example code is typed.
+    result = subprocess.run(
+        ["git", "ls-files", "--cached", "--others", "--exclude-standard", "--", "*.py"],
+        cwd=root,
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    files = sorted(set(result.stdout.splitlines()))
+    excluded = {"lib", "scripts", "tests"}
+    sources = [name for name in files if Path(name).parts[0] not in excluded]
+    if not sources:
+        raise RuntimeError("No consumer Python sources found")
+    print("Consumer typing sources:")
+    print("\n".join(f"- {source}" for source in sources))
+    return subprocess.run(
+        [sys.executable, "-m", "mypy", "--strict", "--explicit-package-bases", *sources],
+        cwd=root,
+        env={
+            **os.environ,
+            "MYPYPATH": os.pathsep.join(
+                str(p)
+                for p in [
+                    root / "lib/python",
+                    *sorted(root.glob("examples/*/src")),
+                    *sorted(root.glob("compatibility/consumers/*/src")),
+                ]
+            ),
+        },
+        check=False,
+    ).returncode
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/full_validate.sh

@@ -43,12 +43,15 @@ run_typing() {
   require_commands python mypy
   python -m mypy --strict examples/typed_consumer.py
   python -m mypy --strict lib/python/base_cli
+  # Discover all Python sources so new top-level directories cannot escape checks.
+  python scripts/validate_consumer_typing.py
 }
 
 run_style() {
   require_commands ruff
-  ruff format --check lib/python/base_cli scripts examples tests
-  ruff check lib/python/base_cli scripts examples tests
+  # Markdown examples are validated by the dedicated documentation gate.
+  ruff format --check --exclude "*.md" .
+  ruff check .
 }
 
 run_contracts() {

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,