tracer-cloud--opensre/surfaces/interactive_shell/runtime/input/actions.py

"""Pure input-action decisions for the interactive shell controller."""

from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass

from surfaces.interactive_shell.runtime.core.turn_detection import (
    looks_like_cancel_request,
    looks_like_confirmation_answer,
)
from surfaces.interactive_shell.runtime.input.events import (
    InputCancelled,
    InputClosed,
    InputEvent,
    InputSubmitted,
)

QUEUE_DURING_CONFIRMATION_WARNING = (
    "[dim](type y/N to confirm the pending action; your input has been queued for after)[/]"
)


@dataclass(frozen=True)
class ShellInputSnapshot:
    exit_requested: bool
    dispatch_running: bool
    awaiting_confirmation: bool


@dataclass(frozen=True)
class IgnoreInput:
    pass


@dataclass(frozen=True)
class CloseShell:
    pass


@dataclass(frozen=True)
class CancelTurn:
    submitted_text: str | None = None


@dataclass(frozen=True)
class DeliverConfirmation:
    text: str


@dataclass(frozen=True)
class SubmitTurn:
    text: str
    wait_until_idle: bool = False
    warning: str | None = None


InputAction = IgnoreInput | CloseShell | CancelTurn | DeliverConfirmation | SubmitTurn


def decide_input_action(
    event: InputEvent,
    snapshot: ShellInputSnapshot,
    *,
    needs_exclusive_stdin: Callable[[str], bool],
) -> InputAction:
    """Interpret one prompt event without mutating runtime state."""
    match event:
        case InputClosed():
            return CloseShell()
        case InputCancelled():
            return CancelTurn()
        case InputSubmitted(text):
            if snapshot.exit_requested or not text:
                return IgnoreInput()

            stripped = text.strip()
            if not stripped:
                return IgnoreInput()

            if snapshot.dispatch_running and looks_like_cancel_request(stripped):
                return CancelTurn(submitted_text=stripped)

            if snapshot.awaiting_confirmation:
                if looks_like_confirmation_answer(stripped):
                    return DeliverConfirmation(text=stripped)
                return SubmitTurn(
                    text=stripped,
                    warning=QUEUE_DURING_CONFIRMATION_WARNING,
                )

            return SubmitTurn(
                text=stripped,
                wait_until_idle=needs_exclusive_stdin(stripped),
            )
    raise AssertionError(f"Unhandled input event: {event!r}")


__all__ = [
    "CancelTurn",
    "CloseShell",
    "DeliverConfirmation",
    "IgnoreInput",
    "InputAction",
    "QUEUE_DURING_CONFIRMATION_WARNING",
    "ShellInputSnapshot",
    "SubmitTurn",
    "decide_input_action",
]