feat: add explicit async command adapter

CHANGELOG.md

@@ -17,6 +17,8 @@ and versions are tracked in the repo-root `VERSION` file.
   compatibility matrix, and document the dependency update policy.
 - Move PyYAML behind the optional `base-cli[yaml]` extra and provide an
   actionable installation hint when YAML configuration or output is selected.
+- Add an explicit `App.async_command()` adapter and `run_async()` helper for
+  deterministic async callbacks without changing the synchronous core.
 
 ### Added
 

docs/consumer-profiles.md

@@ -238,6 +238,8 @@ generic parameter is reserved for a future compatibility boundary and is not
 part of the 0.4.x API.
 
 The core lifecycle is synchronous by design. Native `async def` callbacks and
-callbacks that return awaitables are rejected with an actionable error. An
-adapter that owns an event loop may run asynchronous work explicitly at its
-boundary and return a normal synchronous callback result to base-cli.
+callbacks that return awaitables are rejected with an actionable error. Use
+`@app.async_command()` (or `base_cli.run_async()` in a consumer adapter) when
+an application needs asynchronous work. The adapter owns one event loop for
+the invocation, preserves normal context/logging/exit-code handling, and
+rejects nested event loops so cancellation and cleanup remain deterministic.

docs/index.md

@@ -64,6 +64,9 @@ application.
   project discovery or configuration policy.
 - Use the [Typer adapter](typer-adapter.md) to bring an existing Typer command
   tree under the same lifecycle.
+- Use `@app.async_command()` when a command calls async APIs; see the
+  [consumer profile contract](consumer-profiles.md) for loop ownership and
+  cancellation rules.
 - Review the [JSON contracts](json-contracts.md) and [output contracts](output-contracts.md)
   before building automation around command output.
 

lib/python/base_cli/__init__.py

@@ -28,6 +28,7 @@ def _resolve_version() -> str:
 __version__ = _resolve_version()
 
 from . import (
+    asyncio_adapter,
     command_filters,
     command_protocol,
     deprecations,
@@ -48,6 +49,7 @@ def _resolve_version() -> str:
     option,
     run_app,
 )
+from .asyncio_adapter import run_async
 from .attachment import (
     AttachmentAdapter,
     AttachmentContextFactory,
@@ -149,6 +151,7 @@ def _resolve_version() -> str:
 
 __all__ = [
     "App",
+    "asyncio_adapter",
     "__version__",
     "AttachmentAdapter",
     "AttachmentContextFactory",
@@ -213,6 +216,7 @@ def _resolve_version() -> str:
     "render_inspection_json",
     "testing",
     "argument",
+    "run_async",
     "attach",
     "attach_typer",
     "command",

lib/python/base_cli/app.py

@@ -9,7 +9,7 @@
 import sys
 import time
 import traceback
-from collections.abc import Callable, Iterable
+from collections.abc import Awaitable, Callable, Iterable
 from contextlib import redirect_stdout
 from contextvars import ContextVar, Token
 from dataclasses import dataclass
@@ -34,6 +34,7 @@
     prune_log_files,
     prune_run_bundles,
 )
+from .asyncio_adapter import run_async
 from .attachment import AttachmentContract
 from .config import ConfigSnapshot
 from .context import Context, recover_current_context, reset_current_context, set_current_context
@@ -619,6 +620,32 @@ def decorator(func: Callable[_P, _R]) -> Callable[_P, _R]:
 
         return decorator
 
+    def async_command(
+        self,
+        *command_args: Any,
+        **command_kwargs: Any,
+    ) -> Callable[[Callable[_P, Awaitable[_R]]], Callable[_P, _R]]:
+        """Register an async callback through the explicit asyncio adapter.
+
+        The callback remains an ordinary Click command from the lifecycle's
+        perspective: ``run_async`` owns one event loop for the invocation,
+        waits for the callback, and returns its normal synchronous result for
+        exit-code normalization. Native ``@app.command`` callbacks remain
+        synchronous and continue to reject unadapted coroutines.
+        """
+
+        def decorator(func: Callable[_P, Awaitable[_R]]) -> Callable[_P, _R]:
+            if not inspect.iscoroutinefunction(func):
+                raise TypeError("@app.async_command() requires an async def callback.")
+
+            @functools.wraps(func)
+            def synchronous_callback(*args: _P.args, **kwargs: _P.kwargs) -> _R:
+                return run_async(func(*args, **kwargs))
+
+            return self.command(*command_args, **command_kwargs)(synchronous_callback)
+
+        return decorator
+
     def subcommand(
         self,
         *command_args: Any,

lib/python/base_cli/asyncio_adapter.py

@@ -0,0 +1,39 @@
+"""Explicit asyncio support for commands that opt into an async boundary."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable
+from typing import TypeVar
+
+__all__ = ["run_async"]
+
+_T = TypeVar("_T")
+
+
+async def _await_result(awaitable: Awaitable[_T]) -> _T:
+    return await awaitable
+
+
+def run_async(awaitable: Awaitable[_T]) -> _T:
+    """Run one awaitable with an adapter-owned event loop.
+
+    ``base_cli`` invokes this helper from synchronous Click callbacks. The
+    adapter owns the loop for the duration of the command, and ``asyncio.run``
+    cancels pending tasks and closes the loop before returning. Calling it from
+    an already-running event loop is rejected so a consumer cannot accidentally
+    create nested-loop behavior with ambiguous cancellation or signal rules.
+    """
+
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        return asyncio.run(_await_result(awaitable))
+
+    close = getattr(awaitable, "close", None)
+    if callable(close):
+        close()
+    raise RuntimeError(
+        "base_cli.run_async() owns the event loop for a CLI invocation and "
+        "cannot be called while another event loop is running."
+    )

tests/test_api_stability.py

@@ -9,6 +9,7 @@
 EXPECTED_FACADE_EXPORTS = frozenset(
     {
         "App",
+        "asyncio_adapter",
         "__version__",
         "AttachmentAdapter",
         "AttachmentContextFactory",
@@ -110,6 +111,7 @@
         "redact_json_value",
         "resolve_output_format",
         "run_app",
+        "run_async",
         "success_envelope",
         "RuntimeBinding",
         "ServicesT",

tests/test_public_api.py

@@ -8,6 +8,7 @@
 
 import base_cli
 from base_cli import (
+    asyncio_adapter,
     attachment,
     command_filters,
     command_protocol,
@@ -88,6 +89,8 @@ def test_facade_exports_supported_modules_functions_and_types(self) -> None:
         self.assertIs(base_cli.json_contracts, json_contracts)
         self.assertIs(base_cli.deprecations, deprecations)
         self.assertIs(base_cli.experimental, experimental)
+        self.assertIs(base_cli.asyncio_adapter, asyncio_adapter)
+        self.assertEqual(set(asyncio_adapter.__all__), {"run_async"})
         self.assertTrue(issubclass(base_cli.ConfigurationError, ValueError))
 
     def test_module_all_surfaces_are_explicit(self) -> None:

tests/test_typed_contracts.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import importlib.util
 import tempfile
 import unittest
@@ -12,6 +13,42 @@
 
 @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed")
 class TypedContractTests(unittest.TestCase):
+    def test_async_command_runs_with_adapter_owned_event_loop(self) -> None:
+        app = base_cli.App(name="async-adapter", log_to_file=False)
+        seen: list[str] = []
+
+        @app.async_command()
+        async def command(_context: base_cli.Context[Any, Any, Any]) -> int:
+            await asyncio.sleep(0)
+            seen.append("called")
+            return 0
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            result = invoke(app, [], home=Path(tmpdir))
+
+        self.assertEqual(result.exit_code, 0, result.output)
+        self.assertEqual(seen, ["called"])
+
+    def test_async_command_requires_async_callback(self) -> None:
+        app = base_cli.App(name="async-adapter-type", log_to_file=False)
+
+        with self.assertRaisesRegex(TypeError, "requires an async def"):
+
+            @app.async_command()
+            def command(_context: base_cli.Context[Any, Any, Any]) -> int:
+                return 0
+
+    def test_run_async_rejects_nested_event_loop_and_closes_coroutine(self) -> None:
+        async def pending() -> None:
+            await asyncio.sleep(0)
+
+        async def outer() -> None:
+            coroutine = pending()
+            with self.assertRaisesRegex(RuntimeError, "owns the event loop"):
+                base_cli.run_async(coroutine)
+
+        asyncio.run(outer())
+
     def test_attachment_contract_is_public_and_preserves_command_identity(self) -> None:
         import click