tracer-cloud--opensre/tests/interactive_shell/test_terminal_runtime_turns.py

"""Turn-focused tests for interactive shell terminal runtime dispatch helpers."""

from __future__ import annotations

import asyncio
import io

import pytest
from rich.console import Console

from core.llm.types import AgentLLMResponse, ToolCall
from surfaces.interactive_shell.runtime.core.turn_accounting import (
    ToolCallingTurnResult,
)
from surfaces.interactive_shell.runtime.shell_turn_execution import execute_shell_turn
from surfaces.interactive_shell.runtime.turn_host import run_agent_turn_queue
from surfaces.interactive_shell.runtime.utils import input_policy as loop_input_policy
from surfaces.interactive_shell.session import Session
from tests.core.agent.orchestration.action_execution_test_harness import (
    FakeActionLLM,
)
from tools.interactive_shell.actions import (
    investigation as _investigation_tool,
)
from tools.interactive_shell.actions import (
    slash as _slash_tool,
)


def test_turn_needs_exclusive_stdin_for_bare_integration_menu(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(loop_input_policy, "repl_tty_interactive", lambda: True)
    session = Session()

    assert loop_input_policy.turn_needs_exclusive_stdin("/integrations", session) is True
    assert loop_input_policy.turn_needs_exclusive_stdin("/investigate", session) is True
    assert loop_input_policy.turn_needs_exclusive_stdin("/mcp", session) is True
    assert loop_input_policy.turn_needs_exclusive_stdin("/model", session) is True
    assert loop_input_policy.turn_needs_exclusive_stdin("/theme", session) is True

    assert loop_input_policy.turn_needs_exclusive_stdin("/integrations list", session) is False
    assert loop_input_policy.turn_needs_exclusive_stdin("/theme blue", session) is True
    assert loop_input_policy.turn_needs_exclusive_stdin("/verify", session) is True
    assert loop_input_policy.turn_needs_exclusive_stdin("/verify datadog", session) is False

    # Gating is literal-/slash only: bare command words are not recognized.
    assert loop_input_policy.turn_needs_exclusive_stdin("integrations", session) is False
    assert loop_input_policy.turn_needs_exclusive_stdin("integrations list", session) is False
    assert loop_input_policy.turn_needs_exclusive_stdin("verify", session) is False


def test_turn_needs_exclusive_stdin_false_for_investigate_with_target(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Queued menu selections run as ``/investigate <target>`` without blocking the prompt."""
    monkeypatch.setattr(loop_input_policy, "repl_tty_interactive", lambda: True)
    session = Session()

    assert loop_input_policy.turn_needs_exclusive_stdin("/investigate generic", session) is False
    assert loop_input_policy.turn_needs_exclusive_stdin("/investigate alert.json", session) is False


def test_turn_needs_exclusive_stdin_for_exit_commands(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(loop_input_policy, "repl_tty_interactive", lambda: True)
    session = Session()

    assert loop_input_policy.turn_needs_exclusive_stdin("/exit", session) is True
    assert loop_input_policy.turn_needs_exclusive_stdin("/quit", session) is True
    # Bare command words are not recognized under literal-/slash gating.
    assert loop_input_policy.turn_needs_exclusive_stdin("quit", session) is False


def test_turn_needs_exclusive_stdin_for_update(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """``/update`` hits the network; block the next prompt until output is printed."""
    monkeypatch.setattr(loop_input_policy, "repl_tty_interactive", lambda: True)
    session = Session()

    assert loop_input_policy.turn_needs_exclusive_stdin("/update", session) is True
    # Bare command words are not recognized under literal-/slash gating.
    assert loop_input_policy.turn_needs_exclusive_stdin("update", session) is False


def test_turn_needs_exclusive_stdin_for_integration_setup(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(loop_input_policy, "repl_tty_interactive", lambda: True)
    session = Session()

    assert loop_input_policy.turn_needs_exclusive_stdin("/integrations setup", session) is True
    assert loop_input_policy.turn_needs_exclusive_stdin("/mcp connect github", session) is True
    # Bare command words are not recognized under literal-/slash gating.
    assert (
        loop_input_policy.turn_needs_exclusive_stdin("integrations setup datadog", session) is False
    )


def test_turn_needs_exclusive_stdin_for_integration_remove(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """``remove``/``disconnect`` drive a native inline picker that reads raw
    stdin; the REPL must block the next prompt so keystrokes and CPR responses
    do not leak into the prompt buffer."""
    monkeypatch.setattr(loop_input_policy, "repl_tty_interactive", lambda: True)
    session = Session()

    assert loop_input_policy.turn_needs_exclusive_stdin("/integrations remove", session) is True
    assert (
        loop_input_policy.turn_needs_exclusive_stdin("/integrations remove github", session) is True
    )
    assert loop_input_policy.turn_needs_exclusive_stdin("/mcp disconnect", session) is True
    assert loop_input_policy.turn_needs_exclusive_stdin("/mcp disconnect github", session) is True
    # Bare command words are not recognized under literal-/slash gating.
    assert (
        loop_input_policy.turn_needs_exclusive_stdin("integrations remove github", session) is False
    )


def test_turn_needs_exclusive_stdin_for_onboard(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """``/onboard`` is an interactive wizard; the REPL must wait for it to
    finish before reading the next prompt so the wizard subprocess has
    exclusive stdin and can drive its own questionary widgets.
    """
    monkeypatch.setattr(loop_input_policy, "repl_tty_interactive", lambda: True)
    session = Session()

    assert loop_input_policy.turn_needs_exclusive_stdin("/onboard", session) is True
    # Args don't change the exclusive-stdin requirement.
    assert loop_input_policy.turn_needs_exclusive_stdin("/onboard local_llm", session) is True
    # Bare command words are not recognized under literal-/slash gating.
    assert loop_input_policy.turn_needs_exclusive_stdin("onboard", session) is False


def test_turn_needs_exclusive_stdin_for_config(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """``/config`` delegates to a subprocess; block the next prompt until output
    is printed so config lines do not overlap the pinned input bar.
    """
    monkeypatch.setattr(loop_input_policy, "repl_tty_interactive", lambda: True)
    session = Session()

    assert loop_input_policy.turn_needs_exclusive_stdin("/config", session) is True
    assert loop_input_policy.turn_needs_exclusive_stdin("/config show", session) is True
    assert (
        loop_input_policy.turn_needs_exclusive_stdin(
            "/config set interactive.layout pinned",
            session,
        )
        is True
    )


def test_queued_literal_quit_requests_runtime_exit() -> None:
    async def _scenario() -> None:
        from surfaces.interactive_shell.runtime.core.state import ReplState

        state = ReplState()
        session = Session()
        console = Console(file=io.StringIO(), force_terminal=False, highlight=False)

        async def _run_turn(text: str) -> None:
            await asyncio.to_thread(
                execute_shell_turn,
                text,
                session,
                console,
                recorder=None,
                confirm_fn=None,
                is_tty=None,
                request_exit=state.request_exit,
            )

        worker = asyncio.create_task(run_agent_turn_queue(state=state, run_turn=_run_turn))
        await state.queue.put("/quit")
        await asyncio.wait_for(state.queue.join(), timeout=1)
        await asyncio.wait_for(worker, timeout=1)

        assert state.exit_requested is True

    asyncio.run(_scenario())


def test_execute_shell_turn_nitro_prompt_uses_cli_agent_actions(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    nitro_prompt = (
        "I want to deploy OpenSRE on a remote EC2 Nitro instance, and then I want to send\n"
        'it an investigation. Can you please deploy the instance and send it "hello world"?'
    )
    action_calls: list[str] = []
    llm_calls: list[str] = []

    def _fake_execute_cli_actions(
        text: str,
        _session: Session,
        _console: Console,
        **kwargs: object,
    ) -> ToolCallingTurnResult:
        action_calls.append(text)
        return ToolCallingTurnResult(
            planned_count=2,
            executed_count=2,
            executed_success_count=2,
            has_unhandled_clause=False,
            handled=True,
        )

    def _fake_answer_shell_question(
        text: str,
        _session: Session,
        _console: Console,
        **kwargs: object,
    ) -> None:
        llm_calls.append(text)

    session = Session()
    console = Console(file=io.StringIO(), force_terminal=False, highlight=False)
    execute_shell_turn(
        nitro_prompt,
        session,
        console,
        recorder=None,
        confirm_fn=None,
        is_tty=None,
        execute_actions=_fake_execute_cli_actions,
        answer_agent=_fake_answer_shell_question,
    )

    assert action_calls == [nitro_prompt]
    assert llm_calls == []


def test_execute_shell_turn_nitro_prompt_executes_remote_then_investigation(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    nitro_prompt = (
        "I want to deploy OpenSRE on a remote EC2 Nitro instance, and then I want to send\n"
        'it an investigation. Can you please deploy the instance and send it "hello world"?'
    )
    call_order: list[str] = []

    def _fake_dispatch(
        command: str,
        session: Session,
        console: Console,
        **_kwargs: object,
    ) -> bool:
        call_order.append(f"slash:{command}")
        session.record("slash", command, ok=True)
        console.print(f"ran {command}")
        return True

    def _fake_run_text_investigation(
        alert_text: str,
        _session: Session,
        _console: Console,
        **_kwargs: object,
    ) -> None:
        call_order.append(f"investigation:{alert_text}")

    monkeypatch.setattr(
        "surfaces.interactive_shell.runtime.action_turn.default_llm_factory",
        lambda: FakeActionLLM(
            [
                AgentLLMResponse(
                    content="",
                    tool_calls=[
                        ToolCall(
                            id="call_remote",
                            name="slash_invoke",
                            input={"command": "/remote", "args": []},
                        ),
                        ToolCall(
                            id="call_investigate",
                            name="investigation_start",
                            input={"alert_text": "hello world"},
                        ),
                    ],
                    raw_content=None,
                )
            ]
        ),
    )
    monkeypatch.setattr(_slash_tool, "dispatch_slash", _fake_dispatch)
    monkeypatch.setattr(_investigation_tool, "run_text_investigation", _fake_run_text_investigation)

    session = Session()
    console = Console(file=io.StringIO(), force_terminal=False, highlight=False)
    execute_shell_turn(
        nitro_prompt,
        session,
        console,
        recorder=None,
        confirm_fn=None,
        is_tty=None,
    )

    assert call_order == ["slash:/remote", "investigation:hello world"]


class TestDispatchSpinnerBehavior:
    @pytest.mark.parametrize(
        "text",
        [
            "/history",
            "/tests",
            "/model show",
        ],
    )
    def test_slash_dispatches_do_not_show_assistant_spinner(self, text: str) -> None:
        assert loop_input_policy.turn_should_show_spinner(text, Session()) is False

    @pytest.mark.parametrize(
        "text",
        [
            "why did this fail?",
            "explain deploy",
            # Bare command words and opensre passthrough are no longer treated as
            # literal commands, so the spinner shows while the planner runs.
            "tests",
            "help",
            "opensre investigate -i alert.json",
        ],
    )
    def test_non_slash_dispatches_show_assistant_spinner(self, text: str) -> None:
        assert loop_input_policy.turn_should_show_spinner(text, Session()) is True