tracer-cloud--opensre/surfaces/interactive_shell/runtime/subprocess_runner/opensre_cli_runner.py

"""OpenSRE CLI command runner — surface adapter over tools.interactive_shell.cli."""

from __future__ import annotations

from collections.abc import Callable

from rich.console import Console

from surfaces.interactive_shell.runtime.subprocess_runner.repl_presenter import make_repl_presenter
from surfaces.interactive_shell.session import Session
from surfaces.interactive_shell.ui import DIM, WARNING
from tools.interactive_shell.cli import (
    INTERACTIVE_OPENSRE_COMMAND_PATHS,
    OPENSRE_BLOCKED_SUBCOMMANDS,
    OpensreCommandClass,
    OpensreExecutionMode,
    OpensreExecutionPlan,
    OpensreRunOutcome,
    OpensreRunResult,
    _run_foreground_via_presenter,
    _run_streaming_via_presenter,
    build_opensre_cli_argv,
    build_opensre_execution_plan,
    classify_opensre_command,
    interactive_wizard_handoff_response_text,
    is_interactive_wizard,
    opensre_confirmation_reason,
)
from tools.interactive_shell.cli import (
    run_opensre_cli_command as _run_opensre_cli_command,
)
from tools.interactive_shell.cli import (
    run_opensre_cli_command_result as _run_opensre_cli_command_result,
)

# Backward-compatible aliases for tests and slash parity.
_INTERACTIVE_OPENSRE_COMMAND_PATHS = INTERACTIVE_OPENSRE_COMMAND_PATHS
_OPENSRE_BLOCKED_SUBCOMMANDS = OPENSRE_BLOCKED_SUBCOMMANDS


def _is_interactive_wizard(tokens: list[str]) -> bool:
    return is_interactive_wizard(tokens)


def _classify_opensre_command(tokens: list[str]) -> str:
    return classify_opensre_command(tokens)


def _opensre_confirmation_reason(tokens: list[str]) -> str:
    return opensre_confirmation_reason(tokens)


def _build_opensre_execution_plan(tokens: list[str]) -> OpensreExecutionPlan:
    return build_opensre_execution_plan(tokens)


def print_interactive_wizard_handoff(console: Console, command_str: str) -> None:
    console.print(
        f"[{WARNING}]`opensre {command_str}` is an interactive wizard "
        "that needs a full terminal.[/]"
    )
    console.print(
        f"[{DIM}]Type [bold]/{command_str}[/bold] directly in this shell to launch it.[/]"
    )


def run_opensre_cli_command(
    args: str,
    session: Session,
    console: Console,
    *,
    confirm_fn: Callable[[str], str] | None = None,
    is_tty: bool | None = None,
) -> bool:
    presenter = make_repl_presenter(
        session,
        console,
        confirm_fn=confirm_fn,
        is_tty=is_tty,
        action_already_listed=True,
    )
    return _run_opensre_cli_command(args, presenter)


def run_opensre_cli_command_result(
    args: str,
    session: Session,
    console: Console,
    *,
    confirm_fn: Callable[[str], str] | None = None,
    is_tty: bool | None = None,
) -> OpensreRunResult:
    presenter = make_repl_presenter(
        session,
        console,
        confirm_fn=confirm_fn,
        is_tty=is_tty,
        action_already_listed=True,
    )
    return _run_opensre_cli_command_result(args, presenter)


# Foreground helpers kept for monkeypatch tests that patch subprocess_runner paths.
def _run_opensre_foreground(
    argv_list: list[str],
    display_command: str,
    session: Session,
    console: Console,
) -> None:
    presenter = make_repl_presenter(session, console, action_already_listed=True)
    _run_foreground_via_presenter(
        presenter,
        argv_list=argv_list,
        display_command=display_command,
    )


def _run_opensre_foreground_streaming(
    argv_list: list[str],
    display_command: str,
    session: Session,
    console: Console,
) -> None:
    presenter = make_repl_presenter(session, console, action_already_listed=True)
    _run_streaming_via_presenter(
        presenter,
        argv_list=argv_list,
        display_command=display_command,
    )


__all__ = [
    "OpensreCommandClass",
    "OpensreExecutionMode",
    "OpensreExecutionPlan",
    "OpensreRunOutcome",
    "OpensreRunResult",
    "_INTERACTIVE_OPENSRE_COMMAND_PATHS",
    "_OPENSRE_BLOCKED_SUBCOMMANDS",
    "_build_opensre_execution_plan",
    "_classify_opensre_command",
    "_is_interactive_wizard",
    "_opensre_confirmation_reason",
    "_run_opensre_foreground",
    "_run_opensre_foreground_streaming",
    "build_opensre_cli_argv",
    "interactive_wizard_handoff_response_text",
    "print_interactive_wizard_handoff",
    "run_opensre_cli_command",
    "run_opensre_cli_command_result",
]