fix: reject nested run_app invocations

CHANGELOG.md

@@ -13,6 +13,8 @@ and versions are tracked in the repo-root `VERSION` file.
 
 ### Fixed
 
+- Reject recursive and concurrent in-process `run_app()` calls before they can
+  replace another invocation's stdout or logging handlers.
 - Preserve explicit application identities losslessly while using
   collision-resistant, path-safe runtime namespace components.
 - Give `BatteriesIncludedConfigLoader.cli_name` a documented identity role by

README.md

@@ -746,6 +746,13 @@ flag; otherwise the message says that diagnostic context was unavailable.
 Embedding code that needs the original exception can pass the keyword-only
 `reraise_unexpected=True` argument to `run_app()`.
 
+`run_app()` is non-reentrant and allows only one active invocation per process.
+Nested or concurrent calls fail fast before entering Click or replacing the
+active stdout/logging handlers. Put reusable command behavior in an ordinary
+function and call that function from another callback; use a separate process
+when an independent CLI invocation is required. This does not change the
+framework's supported multi-process logging and runtime coordination.
+
 | Command result or exception | `outcome` | Exit code | Default message |
 | --- | --- | ---: | --- |
 | `None` or returned `0` | `success` | 0 | none |

lib/python/base_cli/_run.py

@@ -10,6 +10,7 @@
 import traceback
 from collections.abc import Callable, Mapping
 from contextlib import redirect_stdout
+from threading import Lock
 from typing import Any, TextIO, cast
 
 from ._app_core import (
@@ -33,6 +34,7 @@
 from .redaction import option_aliases_from_decls
 
 _MAX_JSON_CAPTURE_BYTES = 8 * 1_048_576
+_RUN_APP_LOCK = Lock()
 
 
 class JsonCaptureLimitError(RuntimeError):
@@ -100,11 +102,50 @@ def run_app(
     *,
     reraise_unexpected: bool = False,
 ) -> int:
-    """Run an App, registered command, or attached Click tree and return its status."""
+    """Run an App, registered command, or attached Click tree and return its status.
+
+    ``run_app`` is a process-wide boundary: recursive or concurrent calls fail
+    fast before entering Click or changing stdout and logging handlers.
+    """
 
     if not isinstance(app, App):
         app = get_command_app(app)
 
+    if not _RUN_APP_LOCK.acquire(blocking=False):
+        active_state = _INVOCATION_STATE.get()
+        if active_state is not None:
+            identity = getattr(active_state.owner_app, "name", app.name)
+            print(
+                f"ERROR: Nested run_app() for CLI identity '{identity}' is not supported; "
+                "call the command logic directly instead.",
+                file=sys.stderr,
+            )
+        else:
+            print(
+                "ERROR: Concurrent run_app() calls in one process are not supported; "
+                "invoke each CLI in a separate process or serialize calls.",
+                file=sys.stderr,
+            )
+        return ExitCode.FAILURE
+
+    try:
+        return _run_app_invocation(
+            app,
+            argv,
+            reraise_unexpected=reraise_unexpected,
+        )
+    finally:
+        _RUN_APP_LOCK.release()
+
+
+def _run_app_invocation(
+    app: App,
+    argv: list[str] | None = None,
+    *,
+    reraise_unexpected: bool = False,
+) -> int:
+    """Run one process-wide invocation after acquiring the output/runtime boundary."""
+
     try:
         click = _require_click()
     except RuntimeError as exc:

tests/test_run_app_reentrancy.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+import io
+import tempfile
+import threading
+import unittest
+from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from typing import Any
+from unittest import mock
+
+import base_cli
+from base_cli.logging import SecureLogFileHandler
+
+
+def _outer_callback(app: base_cli.App, nested_statuses: list[int]) -> Callable[..., None]:
+    @base_cli.option("--fail-inner", is_flag=True)
+    def main(ctx: base_cli.Context[Any, Any, Any], fail_inner: bool) -> None:
+        ctx.log.info("outer-before")
+        inner_argv = ["--unknown"] if fail_inner else []
+        nested_statuses.append(base_cli.run_app(app, inner_argv))
+        ctx.log.info("outer-after")
+
+    return main
+
+
+def _file_handler_close_tracker(
+    close_calls: list[SecureLogFileHandler],
+    original_close: Callable[[SecureLogFileHandler], None],
+) -> Callable[[SecureLogFileHandler], None]:
+    def track_close(handler: SecureLogFileHandler) -> None:
+        close_calls.append(handler)
+        original_close(handler)
+
+    return track_close
+
+
+class RunAppReentrancyTests(unittest.TestCase):
+    def test_same_identity_nested_success_and_failure_attempts_leave_outer_logging_intact(self) -> None:
+        for nested_args in ([], ["--unknown"]):
+            with self.subTest(nested_args=nested_args), tempfile.TemporaryDirectory() as tmpdir:
+                home = Path(tmpdir)
+                nested_statuses: list[int] = []
+                app = base_cli.App(name=f"nested-{len(nested_args)}-{home.name}")
+                app.command()(_outer_callback(app, nested_statuses))
+
+                close_calls: list[SecureLogFileHandler] = []
+                original_close = SecureLogFileHandler.close
+                track_close = _file_handler_close_tracker(close_calls, original_close)
+
+                outer_args = ["--fail-inner"] if nested_args else []
+                with mock.patch.object(SecureLogFileHandler, "close", new=track_close):
+                    result = base_cli.testing.invoke(app, outer_args, home=home)
+                    metadata_files = list((home / ".cache").glob("**/run.json"))
+
+                self.assertEqual(result.exit_code, 0, result.output)
+                self.assertEqual(nested_statuses, [base_cli.ExitCode.FAILURE])
+                self.assertIn("Nested run_app()", result.stderr)
+                self.assertEqual(len(metadata_files), 1)
+                log_text = (metadata_files[0].parent / "logs" / "primary.log").read_text(encoding="utf-8")
+                self.assertIn("outer-before", log_text)
+                self.assertIn("outer-after", log_text)
+                self.assertEqual(log_text.count("outer-before"), 1)
+                self.assertEqual(log_text.count("outer-after"), 1)
+                self.assertEqual(len(close_calls), 1)
+
+    def test_concurrent_in_process_invocation_fails_fast_without_entering_second_command(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            started = threading.Event()
+            release = threading.Event()
+            callback_calls: list[str] = []
+            stderr = io.StringIO()
+            app = base_cli.App(
+                name=f"concurrent-{Path(tmpdir).name}",
+                log_to_file=False,
+                profile=base_cli.CliProfile.generic(cache_root=Path(tmpdir) / "cache"),
+            )
+
+            @app.command()
+            def main(ctx: base_cli.Context[Any, Any, Any]) -> None:
+                del ctx
+                callback_calls.append("entered")
+                started.set()
+                release.wait(timeout=5)
+
+            with mock.patch("sys.stderr", stderr), ThreadPoolExecutor(max_workers=1) as executor:
+                first = executor.submit(base_cli.run_app, app, [])
+                self.assertTrue(started.wait(timeout=2))
+                second_status = base_cli.run_app(app, [])
+                release.set()
+                first_status = first.result(timeout=2)
+
+        self.assertEqual(first_status, base_cli.ExitCode.SUCCESS)
+        self.assertEqual(second_status, base_cli.ExitCode.FAILURE)
+        self.assertEqual(callback_calls, ["entered"])
+        self.assertIn("Concurrent run_app()", stderr.getvalue())
+
+
+if __name__ == "__main__":
+    unittest.main()