Runtime

The multi-command app runner: duho.app and duho.run_command, re-exported at the top level. duho.utf8_stdio, which main and app call first, is below.

duho.runtime

Multi-command app runner: wire discovered commands into a runnable app.

This is the driver layer that turns a set of :class:~duho.discovery.Command objects (class commands -- Cmd subclasses -- and module commands -- :class:~duho.discovery.ModuleCommand) into a real subcommand app:

• build a top-level parser for a root Cmd/Args (global options);
• add a subparsers tree and register every command under it;
• parse argv (with _passthrough_ and the nested-help / shared-namespace behaviors preserved);
• dispatch exactly one selected command through the lifecycle init -> main -> success / finally_ with a shared context.

Composed on the shipped parser, not a parallel one. The whole point of this layer is that it reuses duho's existing _parser_/_initparser_/"#cls" machinery rather than introducing a second parser class. The parser behaviors clients rely on are carried over on that path:

• Parent-arg inheritance -- every subcommand parser is built with argparse parents=[<root parser>] so global/root options appear on each subcommand.
• Shared namespace -- class commands already carry the "#cls" deepest-selection contract (_initparser_), which yields one merged instance of the deepest selected class. A module command's parsed instance stays the ROOT instance (plus any fields a module register hook, or its own declared Args class, added directly) -- it is never itself constructed as a duho class, unlike a class command.
• Nested-help suppression -- the optional two-pass prepass uses the existing :func:duho.parsers.prerun_parse (quiet=True), which detaches the subparsers action and silences every terminal action (help, version, print-completion, help-agents) and any parse error for the duration of the call, restoring all of it before returning. No hand-patching.

• register hook -- a module command may define register(parser, args) (or the arity-tolerant register(parser, args, logger)) to add arguments directly on the argparse object of its subcommand.

• _passthrough_ -- argv after the first literal -- is captured by the root parser's patched parse_known_args and reaches the dispatched command.

No target fan-out / thread pools live here -- a single command is dispatched. Parallel fan-out is in :mod:duho.fanout.

__all__ = ['run_command', 'app', 'run', 'accepts_positional', 'build_command_parser'] module-attribute

accepts_positional(func, count)

True if func can be called with count positional arguments.

That holds when it declares at least count positional parameters (with or without defaults) or a *args catch-all. A keyword-only parameter does not count. func is read on its own signature, not the one a functools.wraps wrapper points to. A callable whose signature cannot be read (a builtin, a C callable) gives False, so a caller never supplies an argument it cannot take.

Source code in src/duho/runtime/_arity.py

def accepts_positional(func: _ty.Callable[..., object], count: int) -> bool:
    """True if ``func`` can be called with ``count`` positional arguments.

    That holds when it declares at least ``count`` positional parameters (with
    or without defaults) or a ``*args`` catch-all. A keyword-only parameter does
    not count. ``func`` is read on its own signature, not the one a
    ``functools.wraps`` wrapper points to. A callable whose signature cannot be
    read (a builtin, a C callable) gives ``False``, so a caller never supplies
    an argument it cannot take.
    """
    try:
        params = _inspect.signature(func, follow_wrapped=False).parameters
    except (TypeError, ValueError):
        return False
    positional = 0
    for param in params.values():
        if param.kind is _inspect.Parameter.VAR_POSITIONAL:
            return True
        if param.kind in (
            _inspect.Parameter.POSITIONAL_ONLY,
            _inspect.Parameter.POSITIONAL_OR_KEYWORD,
        ):
            positional += 1
    return positional >= count

app(root=None, *, commands=None, source=None, entry_points=None, argv=None, name=None, description=None, env=None, config=None, setup_logging=True, dispatch=None, mcp=None, mcp_command=None, utf8_stdio=None, on_error=None, adapter=None)

Build a multi-command app, parse argv, and dispatch one command.

root is a Cmd/Args/LoggingArgs subclass supplying the app's global options (None -> a bare data root, for an app whose commands all come from discovery). The BASE command set is resolved by precedence (:func:_resolve_commands): commands > discover_commands(source) > discover_entry_points(entry_points) > root._subcommands_. env.paths("CMDS_PATH", ty=Path) then ALWAYS merges on top of whichever base was used -- a layer, not a branch reachable only when no other source is given -- extending the base rather than replacing it, with a discovered command overriding a same-named base command (logged, never silent).

Additive, not exclusive. root's own declared _subcommands_ are ALWAYS registered too (app reuses the subparsers action root_cls._parser_() already built for them), no matter what commands/source/entry_points was passed -- passing one of those does not remove or replace a root's built-ins, it only adds alongside them (see :func:_resolve_commands for the exact contract). Give root no _subcommands_ of its own (or pass root=None) for an app whose ONLY commands are the ones explicitly resolved here.

entry_points is an installed-distribution entry-point group name (e.g. "myapp.commands"): every entry point advertised in that group by an installed distribution is loaded and registered as a subcommand, so a separately-installed plugin package can contribute commands without the app knowing about it. Loading is resilient -- a broken plugin warns and is skipped, the rest still load. importlib.metadata is imported lazily, so an app that does not use entry_points= never pays its import cost.

Each command is registered under a title="command" subparsers action:

• a class command via its own _parser_(subparsers, parents=[root]) -- the shipped path, so "#cls" deepest-selection and any nested _subcommands_ keep working, with global options inherited via parents=;
• a module command as a subparser (help/description from the module docstring), inheriting global options via parents=; if the module defines register(parser, args) (or register(parser, args, logger)) it is called so the module adds its own arguments directly.

Parsing goes through the root parser's patched parse_known_args (from _initparser_), so "#cls" selection, _passthrough_ capture, and the layered instance construction all apply. When setup_logging, stderr logging is initialised and verbosity applied -- identical to duho.main, including its fallback for a plain Cmd command selected under a LoggingArgs root (see _setup_instance_logging).

Config/env thread-down. Before parsing, env/config-file defaults are layered onto the root and every class command's fields (precedence CLI > env

config > class default): config (or, if omitted, a Cli root's _config_) is loaded once; its top-level keys apply to the root and each [<subcommand>] table to that command. This is app()'s analogue of the _apply_default_layers call duho.main/duho.parse make -- needed here because commands come from sources not reachable via root._subcommands_. The resolved env (if any) is attached to the dispatched instance as the sandwich-named _env_ handle, so a command can read app-wide settings via self._env_.

The selected command is dispatched via :func:run_command; its return is this function's return (success -> 0, a main returning 2 -> 2, and any OTHER non-None value a command returns passes straight through unchanged -- hence the Any return type, not int). Discovery is resilient: a single unimportable command drops out with a warning and the rest still run.

The dispatch seam. app owns discovery, parser build, command registration, config/env thread-down, parsing, and logging setup. The final "run the one selected command" step is the ONE point a consumer can override: pass dispatch to replace it. The callable receives the resolved :class:~duho.discovery.Command (a Cmd subclass for a class command; the :class:~duho.discovery.ModuleCommand for a module command) and the parsed instance, and must return an int exit code, which becomes app's return. A dispatch may call :func:run_command itself (the default when dispatch is None), fan the command out over targets via :mod:duho.fanout, build a per-invocation context threaded ahead of args, or anything else -- everything app already resolved (the same command and instance the default path would run) is reused rather than re-derived. When dispatch is None the behavior is byte-identical to calling :func:run_command directly, so existing callers are unaffected.

source may also be a list or tuple of sources (a later source replaces an earlier command of the same name; see :func:~duho.discovery.discover_commands). on_error(source, exc) is the per-command error policy: it is passed to discovery (source is then the file or module that failed) and called when registering one command raises (source is then the command); returning skips that command, raising aborts. None keeps the default: discovery skips only ImportError/NotImplementedError with a warning and registration errors propagate.

adapter(entrypoint) is applied to a module command's entrypoint by the default run step (see :func:run_command). It cannot be combined with dispatch, which replaces that step: a custom dispatch passes adapter to :func:run_command itself (ValueError otherwise).

MCP launch trigger. Checked FIRST, before argv is parsed or anything else here runs: a <PREFIX>MCP/<NAME>_MCP environment variable (name derived from env's prefix, else from root/name/argv[0]/root's class name -- see duho.args._mcp_env_var_name) set to "stdio" serves this app's FULL resolved tree (class and module commands alike) as an MCP server over stdio instead of running any command, returning the server's own exit code. mcp=False (or root's own _mcp_ = False) disables this trigger entirely -- the variable, if set, is left untouched and a normal run proceeds. See :func:duho.args._maybe_serve_mcp_trigger for the full contract (env var removal, unsupported-transport handling, lazy duho.mcp import).

Opt-in MCP subcommand (mcp_command). None (the default) uses root's own _mcp_command_ class attribute (False unless declared); an explicit True/False/str here wins over it. See :func:_resolve_mcp_command_name for the exact name/validation rules. When resolved to a name, duho.mcp.McpCmd is registered under it like any other class command -- it goes through the SAME collision/override accounting every other resolved command does (:func:_register_commands), except a collision with an EXISTING name is a build-time ValueError here (an explicit opt-in must not silently lose to it), and it requires this app to already have at least one other subcommand (a subparsers action that would ONLY ever offer mcp is not a meaningful CLI). duho.mcp is imported only once a name is actually resolved.

UTF-8 stdio. :func:duho.utf8_stdio runs FIRST, before the MCP launch trigger and before anything else here, unless opted out -- utf8_stdio=False here, or root's own _utf8_stdio_ = False when this kwarg is left at its default None (a bare root=None root is treated as opted in, matching _mcp_'s own default-root handling). Opted out, duho does not touch stdio at all.

Source code in src/duho/runtime/_app.py

def app(
    root: _ty.Optional[type] = None,
    *,
    commands: _ty.Optional[_ty.Sequence[_Command]] = None,
    source: _ty.Union[str, _Path, _ty.Sequence[_ty.Union[str, _Path]], None] = None,
    entry_points: _ty.Optional[str] = None,
    argv: _ty.Optional[_ty.Sequence[str]] = None,
    name: _ty.Optional[str] = None,
    description: _ty.Optional[str] = None,
    env: _ty.Optional[_Env] = None,
    config: _ty.Optional[_ty.Union[str, _Path]] = None,
    setup_logging: bool = True,
    dispatch: _ty.Optional[_ty.Callable[[_Command, object], _ty.Any]] = None,
    mcp: _ty.Optional[bool] = None,
    mcp_command: _ty.Optional[_ty.Union[str, bool]] = None,
    utf8_stdio: _ty.Optional[bool] = None,
    on_error: _ty.Optional[_ty.Callable[[object, BaseException], object]] = None,
    adapter: _ty.Optional[
        _ty.Callable[
            [_ty.Callable[..., object]], _ty.Optional[_ty.Callable[..., object]]
        ]
    ] = None,
) -> _ty.Any:
    """Build a multi-command app, parse ``argv``, and dispatch one command.

    ``root`` is a ``Cmd``/``Args``/``LoggingArgs`` subclass supplying the app's
    global options (``None`` -> a bare data root, for an app whose commands all
    come from discovery). The BASE command set is resolved by precedence
    (:func:`_resolve_commands`): ``commands`` > ``discover_commands(source)`` >
    ``discover_entry_points(entry_points)`` > ``root._subcommands_``.
    ``env.paths("CMDS_PATH", ty=Path)`` then ALWAYS merges on top of whichever
    base was used -- a layer, not a branch reachable only when no other source
    is given -- extending the base rather than replacing it, with a discovered
    command overriding a same-named base command (logged, never silent).

    **Additive, not exclusive.** ``root``'s own declared
    ``_subcommands_`` are ALWAYS registered too (``app`` reuses the
    subparsers action ``root_cls._parser_()`` already built for them), no
    matter what ``commands``/``source``/``entry_points`` was passed --
    passing one of those does not remove or replace a root's built-ins, it
    only adds alongside them (see :func:`_resolve_commands` for the exact
    contract). Give ``root`` no ``_subcommands_`` of its own (or pass
    ``root=None``) for an app whose ONLY commands are the ones explicitly
    resolved here.

    ``entry_points`` is an installed-distribution entry-point **group** name
    (e.g. ``"myapp.commands"``): every entry point advertised in that group by an
    installed distribution is loaded and registered as a subcommand, so a
    separately-installed plugin package can contribute commands without the app
    knowing about it. Loading is resilient -- a broken plugin warns and is
    skipped, the rest still load. ``importlib.metadata`` is imported lazily, so
    an app that does not use ``entry_points=`` never pays its import cost.

    Each command is registered under a ``title="command"`` subparsers action:

    * a **class command** via its own ``_parser_(subparsers, parents=[root])`` --
      the shipped path, so ``"#cls"`` deepest-selection and any nested
      ``_subcommands_`` keep working, with global options inherited via
      ``parents=``;
    * a **module command** as a subparser (help/description from the module
      docstring), inheriting global options via ``parents=``; if the module
      defines ``register(parser, args)`` (or ``register(parser, args, logger)``)
      it is called so the module adds its own arguments directly.

    Parsing goes through the root parser's patched ``parse_known_args`` (from
    ``_initparser_``), so ``"#cls"`` selection, ``_passthrough_`` capture, and
    the layered instance construction all apply. When ``setup_logging``,
    stderr logging is initialised and verbosity applied -- identical to
    ``duho.main``, including its fallback for a plain ``Cmd`` command
    selected under a ``LoggingArgs`` root (see ``_setup_instance_logging``).

    **Config/env thread-down.** Before parsing, env/config-file defaults are
    layered onto the root and every class command's fields (precedence CLI > env
    > config > class default): ``config`` (or, if omitted, a ``Cli`` root's
    ``_config_``) is loaded once; its top-level keys apply to the root and each
    ``[<subcommand>]`` table to that command. This is app()'s analogue of the
    ``_apply_default_layers`` call ``duho.main``/``duho.parse`` make -- needed
    here because commands come from sources not reachable via
    ``root._subcommands_``. The resolved ``env`` (if any) is attached to the
    dispatched instance as the sandwich-named ``_env_`` handle, so a command can
    read app-wide settings via ``self._env_``.

    The selected command is dispatched via :func:`run_command`; its return is
    this function's return (success -> ``0``, a ``main`` returning ``2`` ->
    ``2``, and any OTHER non-``None`` value a command returns passes straight
    through unchanged -- hence the ``Any`` return type, not ``int``). Discovery
    is resilient: a single unimportable command drops out with a warning and
    the rest still run.

    **The ``dispatch`` seam.** ``app`` owns discovery, parser build, command
    registration, config/env thread-down, parsing, and logging setup. The final
    "run the one selected command" step is the ONE point a consumer can override:
    pass ``dispatch`` to replace it. The callable receives the resolved
    :class:`~duho.discovery.Command` (a ``Cmd`` subclass for a class command; the
    :class:`~duho.discovery.ModuleCommand` for a module command) and the parsed
    ``instance``, and must return an ``int`` exit code, which becomes ``app``'s
    return. A dispatch may call :func:`run_command` itself (the default when
    ``dispatch is None``), fan the command out over targets via
    :mod:`duho.fanout`, build a per-invocation context threaded ahead of args, or
    anything else -- everything ``app`` already resolved (the same ``command`` and
    ``instance`` the default path would run) is reused rather than re-derived. When
    ``dispatch`` is ``None`` the behavior is byte-identical to calling
    :func:`run_command` directly, so existing callers are unaffected.

    ``source`` may also be a list or tuple of sources (a later source replaces
    an earlier command of the same name; see :func:`~duho.discovery.discover_commands`).
    ``on_error(source, exc)`` is the per-command error policy: it is passed to
    discovery (``source`` is then the file or module that failed) and called
    when registering one command raises (``source`` is then the command); returning
    skips that command, raising aborts. ``None`` keeps the default: discovery skips
    only ``ImportError``/``NotImplementedError`` with a warning and registration
    errors propagate.

    ``adapter(entrypoint)`` is applied to a module command's entrypoint by the
    default run step (see :func:`run_command`). It cannot be combined with
    ``dispatch``, which replaces that step: a custom dispatch passes ``adapter``
    to :func:`run_command` itself (``ValueError`` otherwise).

    **MCP launch trigger.** Checked FIRST, before ``argv``
    is parsed or anything else here runs: a ``<PREFIX>MCP``/``<NAME>_MCP``
    environment variable (name derived from ``env``'s prefix, else from
    ``root``/``name``/``argv[0]``/``root``'s class name -- see
    ``duho.args._mcp_env_var_name``) set to ``"stdio"`` serves this app's
    FULL resolved tree (class and module commands alike) as an MCP server
    over stdio instead of running any command, returning the server's own
    exit code. ``mcp=False`` (or ``root``'s own ``_mcp_ = False``) disables
    this trigger entirely -- the variable, if set, is left untouched and a
    normal run proceeds. See :func:`duho.args._maybe_serve_mcp_trigger` for
    the full contract (env var removal, unsupported-transport handling,
    lazy ``duho.mcp`` import).

    **Opt-in MCP subcommand** (``mcp_command``). ``None`` (the
    default) uses ``root``'s own ``_mcp_command_`` class attribute
    (``False`` unless declared); an explicit ``True``/``False``/``str`` here
    wins over it. See :func:`_resolve_mcp_command_name` for the exact
    name/validation rules. When resolved to a name, ``duho.mcp.McpCmd`` is
    registered under it like any other class command -- it goes through the
    SAME collision/override accounting every other resolved command does
    (:func:`_register_commands`), except a collision with an EXISTING name
    is a build-time ``ValueError`` here (an explicit opt-in must not
    silently lose to it), and it requires this app to already have at least
    one other subcommand (a subparsers action that would ONLY ever offer
    ``mcp`` is not a meaningful CLI). ``duho.mcp`` is imported only once a
    name is actually resolved.

    **UTF-8 stdio.** :func:`duho.utf8_stdio` runs FIRST, before the MCP
    launch trigger and before anything else here, unless opted out --
    ``utf8_stdio=False`` here, or ``root``'s own ``_utf8_stdio_ = False``
    when this kwarg is left at its default ``None`` (a bare ``root=None``
    root is treated as opted in, matching ``_mcp_``'s own default-root
    handling). Opted out, duho does not touch stdio at all.
    """
    _token = _duho_logging._enter_traceback_env(root)
    try:
        root_cls_for_mcp = root if root is not None else _Args
        if (
            utf8_stdio
            if utf8_stdio is not None
            else getattr(root_cls_for_mcp, "_utf8_stdio_", True)
        ):
            _compat.utf8_stdio()

        mcp_enabled = (
            mcp if mcp is not None else getattr(root_cls_for_mcp, "_mcp_", True)
        )
        if mcp_enabled:

            def _mcp_core_for_this_app() -> object:
                from .. import mcp as _mcp_module

                return _mcp_module._core_for_app(
                    root,
                    commands=commands,
                    source=source,
                    entry_points=entry_points,
                    argv=argv,
                    name=name,
                    description=description,
                    env=env,
                    config=config,
                    dispatch=dispatch,
                    on_error=on_error,
                    adapter=adapter,
                )

            served = _maybe_serve_mcp_trigger(
                root_cls_for_mcp,
                env=env,
                name=name,
                core_factory=_mcp_core_for_this_app,
            )
            if served is not None:
                return served

        config_source = None
        if config is None and root is not None:
            try:
                config, config_source = _class_config_location(root, argv)
            except Exception as exc:
                _end_if_reported(exc, _app_name(root, name), root)
                raise

        run = _default_run(dispatch, adapter)
        # Names CMDS_PATH overrode, logged by `_run_app` once logging is set up; also
        # lets `_register_commands` treat a collision on that name as this override.
        cmds_path_overridden: set[str] = set()
        resolved_commands = _resolve_commands(
            root,
            commands,
            source,
            env,
            entry_points,
            overridden=cmds_path_overridden,
            on_error=on_error,
        )

        mcp_cls = _build_mcp_command_class(
            root,
            mcp_command,
            _existing_command_names(root, resolved_commands),
            has_other_subcommand=bool(resolved_commands)
            or bool(getattr(root, "_subcommands_", None)),
        )
        if mcp_cls is not None:
            resolved_commands = list(resolved_commands) + [mcp_cls]

        completion_cls = _build_completion_command_class(
            root,
            _existing_command_names(root, resolved_commands),
            has_other_subcommand=bool(resolved_commands)
            or bool(getattr(root, "_subcommands_", None)),
        )
        if completion_cls is not None:
            resolved_commands = list(resolved_commands) + [completion_cls]

        parser, base_parser, root_cls, raw_config, prepass_args = _prepare_app_parser(
            root, name, description, config, argv, resolved_commands, config_source
        )

        subparsers, registry, notices = _register_commands(
            root,
            resolved_commands,
            parser,
            base_parser,
            root_cls,
            prepass_args,
            cmds_path_overridden,
            inherited_config_hint=config is not None or _names_config_source(root_cls),
            on_error=on_error,
        )

        required_root_actions = _finalize_command_tree(
            parser, subparsers, root_cls, registry, raw_config
        )

        # Lets `duho.mcp.serve_running_app`, called from a dispatched command, serve
        # this already-built tree. Set only around the dispatch.
        mcp_dispatch = _make_post_parse_dispatch(
            env, root_cls, notices, cmds_path_overridden, run
        )
        token = _compat._MCP_CONTEXT.set(("app", parser, root_cls, mcp_dispatch))
        try:
            return _run_app(
                parser,
                argv,
                env,
                setup_logging,
                root_cls,
                required_root_actions,
                cmds_path_overridden,
                notices,
                run,
            )
        finally:
            _compat._MCP_CONTEXT.reset(token)
    finally:
        _duho_logging._exit_traceback_env(_token)

build_command_parser(command, root=None)

Build the subparser :func:app would build for one resolved command.

command is a Cmd subclass or a :class:~duho.discovery.ModuleCommand; root is the app's root class, whose options the subparser inherits (a bare Args when None). The parser is built into a throwaway parent, and a module command's register hook runs with None for the parsed globals. Whatever the build raises propagates, so a caller can drop a command that cannot build; TypeError for anything that is not a command.

Source code in src/duho/runtime/_register.py

def build_command_parser(
    command: _Command, root: _ty.Optional[type] = None
) -> _argparse.ArgumentParser:
    """Build the subparser :func:`app` would build for one resolved command.

    ``command`` is a ``Cmd`` subclass or a :class:`~duho.discovery.ModuleCommand`;
    ``root`` is the app's root class, whose options the subparser inherits (a bare
    ``Args`` when ``None``). The parser is built into a throwaway parent, and a
    module command's ``register`` hook runs with ``None`` for the parsed globals.
    Whatever the build raises propagates, so a caller can drop a command that
    cannot build; ``TypeError`` for anything that is not a command.
    """
    root_cls = root if root is not None else _Args
    base_parser = root_cls._parser_(  # type: ignore[attr-defined]
        add_help=False, _skip_subcommands_=True
    )
    subparsers = _argparse.ArgumentParser(add_help=False).add_subparsers()
    if _is_class_command(command):
        return _register_class_command(subparsers, _ty.cast(type, command), base_parser)
    if _is_module_command(command):
        module_command = _ty.cast(_ModuleCommand, command)
        _register_module_command(
            subparsers, module_command, base_parser, None, root_cls
        )
        return subparsers.choices[module_command._parsername_]
    raise TypeError(
        f"build_command_parser(): expected a Cmd subclass or a discovered "
        f"ModuleCommand, got {command!r} ({type(command).__name__})"
    )

run(root, argv=None, *, exit=True, **app_kwargs)

run(
    root: type,
    argv: _ty.Optional[_ty.Sequence[str]] = None,
    *,
    exit: _ty.Literal[True] = True,
    **app_kwargs: _ty.Any
) -> _ty.NoReturn

run(
    root: type,
    argv: _ty.Optional[_ty.Sequence[str]] = None,
    *,
    exit: _ExitPolicy,
    **app_kwargs: _ty.Any
) -> int

Run root as the program: dispatch, print the answer, and exit.

Calls duho.main(root, argv), or duho.app(root, argv=argv, **app_kwargs) when keywords are given, then sys.exit with the status. None is 0; a Result prints its rendered value (if not None) and exits with its code; any other int is the status; any other value is printed (a str as it is, else JSON) with status 0. KeyboardInterrupt exits 130, silently unless DUHO_TRACEBACK is set, which re-raises it; a closed stdout (a BrokenPipeError, or on Windows OSError(EINVAL)), from the command, from printing or from the final flush, exits 1 silently either way. A SystemExit passes through untouched.

exit says when to end the process instead of returning the status: True always; False never, so a SystemExit becomes its status (None is 0, an int itself, anything else is printed to standard error and is 1), --help and --version included; "parse" only for a SystemExit raised while the command line is parsed (--help, --version, a usage error). A callable is called as f(status, error), error being the SystemExit, KeyboardInterrupt or BrokenPipeError caught (else None), and returns one of those.

Source code in src/duho/runtime/_program.py

def run(
    root: type,
    argv: _ty.Optional[_ty.Sequence[str]] = None,
    *,
    exit: _ExitPolicy = True,
    **app_kwargs: _ty.Any,
) -> int:
    """Run ``root`` as the program: dispatch, print the answer, and exit.

    Calls ``duho.main(root, argv)``, or ``duho.app(root, argv=argv, **app_kwargs)``
    when keywords are given, then ``sys.exit`` with the status. ``None`` is 0; a
    ``Result`` prints its rendered ``value`` (if not ``None``) and exits with its
    code; any other ``int`` is the status; any other value is printed (a ``str``
    as it is, else JSON) with status 0. ``KeyboardInterrupt`` exits 130, silently
    unless ``DUHO_TRACEBACK`` is set, which re-raises it; a closed stdout (a
    ``BrokenPipeError``, or on Windows ``OSError(EINVAL)``), from the command, from
    printing or from the final flush, exits 1 silently either way. A
    ``SystemExit`` passes through untouched.

    ``exit`` says when to end the process instead of returning the status:
    ``True`` always; ``False`` never, so a ``SystemExit`` becomes its status
    (``None`` is 0, an ``int`` itself, anything else is printed to standard error
    and is 1), ``--help`` and ``--version`` included; ``"parse"`` only for a
    ``SystemExit`` raised while the command line is parsed (``--help``,
    ``--version``, a usage error). A callable is called as ``f(status, error)``,
    ``error`` being the ``SystemExit``, ``KeyboardInterrupt`` or
    ``BrokenPipeError`` caught (else ``None``), and returns one of those.
    """
    _token = _duho_logging._enter_traceback_env(root)
    try:
        error: _ty.Optional[BaseException] = None
        try:
            if app_kwargs:
                from ._app import app as _app

                outcome = _app(root, argv=argv, **app_kwargs)
            else:
                from ..args._entry import main as _main

                outcome = _main(root, argv)
            status = _status_of(outcome)
        except SystemExit as exc:
            if exit is True:
                raise
            error, status = exc, _exit_status(exc.code)
        except KeyboardInterrupt as exc:
            if _traceback_enabled():
                raise
            error, status = exc, 130
        except OSError as exc:
            if not _stdout_closed(exc):
                raise
            _silence_stdout()
            error, status = _as_broken_pipe(exc), 1
        else:
            closed = _flush_stdout()
            if closed is not None:
                error, status = closed, 1
        policy = exit(status, error) if callable(exit) else exit
        if policy == "parse":
            policy = getattr(error, "_duho_parser_exit_", False)
        if policy is True:
            _sys.exit(status)
        return status
    finally:
        _duho_logging._exit_traceback_env(_token)

run_command(command, instance, *, context=None, adapter=None)

Dispatch one already-resolved command against a parsed instance.

instance is the parsed args/command instance produced by parsing (for a class command it IS the command; for a module command it is the root/parent instance carrying the parsed globals). Returns what the command returns: None becomes 0, an int propagates, and anything else (e.g. a JSON-serialisable object) passes through unchanged.

• Class command (a Cmd): the parsed instance is itself the command, so this calls instance() (Cmd.__call__ is the entrypoint). Parsing already owns building the instance; there is no separate parse here.
• Module command (:class:ModuleCommand): runs the lifecycle -- ctx = command.init(instance) (identity/no-op default returning None), then command.main(instance) and command.success(ctx, instance) inside a try whose finally always runs command.finally_(ctx, instance). If context is passed it overrides the init result (the driver builds the context once and threads it in). main's return value (None -> 0) is the result; an exception from main propagates after finally_ runs.

adapter(entrypoint), when given, is called with a module command's entrypoint (:attr:ModuleCommand.entrypoint) and returns the callable to call with instance in its place; a falsy return keeps the entrypoint. It never applies to the lifecycle hooks, nor to a class command.

No separate logger argument is threaded: hooks read instance._logger_. For a module command, THIS driver ensures it is present before any hook runs: when instance has no _logger_ of its own (a plain root, not LoggingArgs-based), it is set to module_command._logger_for(instance) (the "duho" fallback) so a hook written against the documented args._logger_ convention never hits AttributeError. Setting the attribute is best-effort: a root whose _logger_ is a read-only property simply keeps using its own resolution.

A module command's init/main/success/finally_ are never awaited (unlike a class command's __call__, see :func:_maybe_await): a coroutine returned by any of them is closed immediately and raises TypeError (see :func:_reject_coroutine), rather than silently never running.

Source code in src/duho/runtime/_run.py

def run_command(
    command: _Command,
    instance: object,
    *,
    context: object = None,
    adapter: _ty.Optional[
        _ty.Callable[
            [_ty.Callable[..., object]], _ty.Optional[_ty.Callable[..., object]]
        ]
    ] = None,
) -> _ty.Any:
    """Dispatch one already-resolved command against a parsed ``instance``.

    ``instance`` is the parsed args/command instance produced by parsing (for a
    class command it IS the command; for a module command it is the root/parent
    instance carrying the parsed globals). Returns what the command returns:
    ``None`` becomes ``0``, an ``int`` propagates, and anything else (e.g. a
    JSON-serialisable object) passes through unchanged.

    * **Class command** (a ``Cmd``): the parsed ``instance`` is itself the
      command, so this calls ``instance()`` (``Cmd.__call__`` is the entrypoint).
      Parsing already owns building the instance; there is no separate parse here.
    * **Module command** (:class:`ModuleCommand`): runs the lifecycle --
      ``ctx = command.init(instance)`` (identity/no-op default returning
      ``None``), then ``command.main(instance)`` and ``command.success(ctx,
      instance)`` inside a ``try`` whose ``finally`` always runs
      ``command.finally_(ctx, instance)``. If ``context`` is passed it overrides
      the ``init`` result (the driver builds the context once and threads it in).
      ``main``'s return value (``None`` -> ``0``) is the result; an
      exception from ``main`` propagates after ``finally_`` runs.

    ``adapter(entrypoint)``, when given, is called with a module command's
    entrypoint (:attr:`ModuleCommand.entrypoint`) and returns the callable to
    call with ``instance`` in its place; a falsy return keeps the entrypoint.
    It never applies to the lifecycle hooks, nor to a class command.

    No separate ``logger`` argument is threaded: hooks read ``instance._logger_``.
    For a module command, THIS driver ensures it is present before any hook
    runs: when ``instance`` has no ``_logger_`` of its own (a plain root, not
    ``LoggingArgs``-based), it is set to ``module_command._logger_for(instance)``
    (the ``"duho"`` fallback) so a hook written against the documented
    ``args._logger_`` convention never hits ``AttributeError``. Setting
    the attribute is best-effort: a root whose ``_logger_`` is a read-only
    property simply keeps using its own resolution.

    A module command's ``init``/``main``/``success``/``finally_`` are never
    awaited (unlike a class command's ``__call__``, see :func:`_maybe_await`):
    a coroutine returned by any of them is closed immediately and raises
    ``TypeError`` (see :func:`_reject_coroutine`), rather than silently never
    running.
    """
    # A command file from a scanned directory has its own copy of the builtins.
    _refresh_sibling_builtins()
    if _is_module_command(command):
        module_command = _ty.cast(_ModuleCommand, command)
        if not isinstance(getattr(instance, "_logger_", None), _logging.Logger):
            try:
                instance._logger_ = module_command._logger_for(instance)  # type: ignore[attr-defined]
            except (
                Exception
            ):  # pragma: no cover - a property-bearing root may refuse the write
                pass
        ctx = context if context is not None else module_command.init(instance)
        _reject_coroutine(ctx, "%s init()" % _command_name(command))
        try:
            adapted = adapter(module_command.entrypoint) if adapter else None
            result = adapted(instance) if adapted else module_command.main(instance)
            _reject_coroutine(result, "%s main()" % _command_name(command))
            # `success` is the SUCCESS hook: run it only when main reported
            # success (None or exit code 0), not for a non-zero exit code.
            if result is None or result == 0:
                success_result = module_command.success(ctx, instance)
                _reject_coroutine(
                    success_result, "%s success()" % _command_name(command)
                )
        finally:
            # A raising `finally_` must not mask the original exception (if main
            # raised) nor the real exit code: log and swallow its error.
            try:
                fin_result = module_command.finally_(ctx, instance)
                _reject_coroutine(fin_result, "%s finally_()" % _command_name(command))
            except Exception:
                _LOGGER.exception(
                    "finally_ hook for command %r raised; ignoring",
                    _command_name(command),
                )
        return 0 if result is None else result

    # Class command: the instance is the command. An `async def __call__` gets its
    # own `asyncio.run` per call, so each fan-out worker has an independent loop.
    result = _maybe_await(instance())  # type: ignore[operator]
    return 0 if result is None else result

duho.utf8_stdio(streams=None)

Reconfigure text streams to UTF-8 in place; return the names switched.

streams defaults to stdout and stderr; tests pass fakes. A stream is left alone when PYTHONIOENCODING is set, UTF-8 mode is on, it has no reconfigure(), it is a tty, or it is already UTF-8. Errors policy: surrogateescape for "stdout", backslashreplace otherwise. Never raises: a failing reconfigure() skips that stream.

Source code in src/duho/_compat.py

def utf8_stdio(
    streams: _ty.Optional[_ty.Mapping[str, _ty.Any]] = None,
) -> list[str]:
    """Reconfigure text streams to UTF-8 in place; return the names switched.

    ``streams`` defaults to stdout and stderr; tests pass fakes. A stream is
    left alone when ``PYTHONIOENCODING`` is set, UTF-8 mode is on, it has no
    ``reconfigure()``, it is a tty, or it is already UTF-8. Errors policy:
    ``surrogateescape`` for ``"stdout"``, ``backslashreplace`` otherwise. Never
    raises: a failing ``reconfigure()`` skips that stream.
    """
    if streams is None:
        streams = {"stdout": _sys.stdout, "stderr": _sys.stderr}

    if _os.environ.get("PYTHONIOENCODING"):
        return []
    if getattr(_sys.flags, "utf8_mode", 0):
        return []

    switched: list[str] = []
    for stream_name, stream in streams.items():
        reconfigure = getattr(stream, "reconfigure", None)
        if reconfigure is None:
            continue
        try:
            is_tty = stream.isatty()
        except Exception:
            is_tty = False
        if is_tty:
            continue
        encoding = getattr(stream, "encoding", None)
        if encoding:
            try:
                if _codecs.lookup(encoding).name == "utf-8":
                    continue
            except LookupError:
                pass
        errors = "surrogateescape" if stream_name == "stdout" else "backslashreplace"
        try:
            reconfigure(encoding="utf-8", errors=errors)
        except (OSError, ValueError):
            continue
        switched.append(stream_name)
    return switched