"""Tests for non-interactive mode HITL decision logic."""

import asyncio
import io
import signal
import sys
from collections.abc import AsyncIterator, Iterator, Sequence
from typing import TYPE_CHECKING, Any
from unittest.mock import AsyncMock, MagicMock, call, patch

import pytest
from langchain_core.messages import AIMessage
from rich.console import Console

if TYPE_CHECKING:
    from langchain_core.runnables import RunnableConfig
from rich.style import Style
from rich.text import Text

from deepagents_code.config import SHELL_ALLOW_ALL, ModelResult
from deepagents_code.non_interactive import (
    _MAX_HITL_ITERATIONS,
    HITLIterationLimitError,
    StreamState,
    ThreadUrlLookupState,
    _build_non_interactive_header,
    _collect_action_request_warnings,
    _make_hitl_decision,
    _process_ai_message,
    _run_agent_loop,
    _run_startup_command,
    _start_langsmith_thread_url_lookup,
    run_non_interactive,
)


@pytest.fixture
def console() -> Console:
    """Console that captures output."""
    return Console(quiet=True)


@pytest.fixture(autouse=True)
def skip_mcp_metadata_preload() -> Iterator[None]:
    """Keep non-MCP non-interactive tests from starting connector discovery."""
    with patch(
        "deepagents_code.main._preload_session_mcp_server_info",
        new_callable=AsyncMock,
        return_value=[],
    ):
        yield


class TestMakeHitlDecision:
    """Tests for _make_hitl_decision()."""

    def test_non_shell_action_approved(self, console: Console) -> None:
        """Non-shell actions should be auto-approved."""
        result = _make_hitl_decision(
            {"name": "read_file", "args": {"path": "/tmp/test"}}, console
        )
        assert result == {"type": "approve"}

    def test_shell_without_allow_list_rejected(self, console: Console) -> None:
        """Shell commands should be rejected when no allow-list is configured."""
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.shell_allow_list = None
            result = _make_hitl_decision(
                {"name": "execute", "args": {"command": "rm -rf /"}}, console
            )
            assert result["type"] == "reject"
            assert "not permitted" in result["message"]

    def test_shell_allowed_command_approved(self, console: Console) -> None:
        """Shell commands in the allow-list should be approved."""
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.shell_allow_list = ["ls", "cat", "grep"]
            result = _make_hitl_decision(
                {"name": "execute", "args": {"command": "ls -la"}}, console
            )
            assert result == {"type": "approve"}

    def test_shell_disallowed_command_rejected(self, console: Console) -> None:
        """Shell commands not in the allow-list should be rejected."""
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.shell_allow_list = ["ls", "cat", "grep"]
            result = _make_hitl_decision(
                {"name": "execute", "args": {"command": "rm -rf /"}}, console
            )
            assert result["type"] == "reject"
            assert "rm -rf /" in result["message"]
            assert "not in the allow-list" in result["message"]

    def test_shell_rejected_message_includes_allowed_commands(
        self, console: Console
    ) -> None:
        """Rejection message should list the allowed commands."""
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.shell_allow_list = ["ls", "cat"]
            result = _make_hitl_decision(
                {"name": "execute", "args": {"command": "whoami"}}, console
            )
            assert "ls" in result["message"]
            assert "cat" in result["message"]

    def test_empty_action_name_approved(self, console: Console) -> None:
        """Actions with empty name should be approved (non-shell)."""
        result = _make_hitl_decision({"name": "", "args": {}}, console)
        assert result == {"type": "approve"}

    def test_shell_piped_command_allowed(self, console: Console) -> None:
        """Piped shell commands where all segments are allowed should pass."""
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.shell_allow_list = ["ls", "grep"]
            result = _make_hitl_decision(
                {"name": "execute", "args": {"command": "ls | grep test"}}, console
            )
            assert result == {"type": "approve"}

    def test_shell_piped_command_with_disallowed_segment(
        self, console: Console
    ) -> None:
        """Piped commands with a disallowed segment should be rejected."""
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.shell_allow_list = ["ls"]
            result = _make_hitl_decision(
                {"name": "execute", "args": {"command": "ls | rm file"}}, console
            )
            assert result["type"] == "reject"

    def test_shell_dangerous_pattern_rejected(self, console: Console) -> None:
        """Dangerous patterns rejected even if base command is allowed."""
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.shell_allow_list = ["ls"]
            result = _make_hitl_decision(
                {"name": "execute", "args": {"command": "ls $(whoami)"}}, console
            )
            assert result["type"] == "reject"

    def test_shell_with_allow_all_approved(self, console: Console) -> None:
        """Shell commands should be approved when SHELL_ALLOW_ALL is set."""
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.shell_allow_list = SHELL_ALLOW_ALL
            result = _make_hitl_decision(
                {"name": "execute", "args": {"command": "rm -rf /"}}, console
            )
            assert result == {"type": "approve"}

    def test_execute_tool_gated_by_allow_list(self, console: Console) -> None:
        """The `execute` shell tool is gated by the allow-list."""
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.shell_allow_list = ["ls"]
            result = _make_hitl_decision(
                {"name": "execute", "args": {"command": "rm -rf /"}}, console
            )
            assert result["type"] == "reject"

    def test_collect_action_request_warnings_for_hidden_unicode(self) -> None:
        """Hidden Unicode in action args should generate warnings."""
        warnings = _collect_action_request_warnings(
            {"name": "execute", "args": {"command": "echo he\u200bllo"}}
        )
        assert warnings
        assert any("hidden Unicode" in warning for warning in warnings)

    def test_collect_action_request_warnings_for_suspicious_url(self) -> None:
        """Suspicious URLs in action args should generate warnings."""
        warnings = _collect_action_request_warnings(
            {"name": "fetch_url", "args": {"url": "https://аpple.com"}}
        )
        assert warnings
        assert any("URL warning" in warning for warning in warnings)

    def test_collect_action_request_warnings_nested_values(self) -> None:
        """Nested string values should be inspected recursively."""
        warnings = _collect_action_request_warnings(
            {
                "name": "fetch_url",
                "args": {"headers": {"Referer": "echo \u200bhello"}},
            }
        )
        assert warnings
        assert any("hidden Unicode" in warning for warning in warnings)


class TestBuildNonInteractiveHeader:
    """Tests for _build_non_interactive_header()."""

    def test_includes_agent_id(self) -> None:
        """Header should contain the agent identifier."""
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.model_name = None
            header = _build_non_interactive_header("my-agent", "abc123")
        assert "Agent: my-agent" in header.plain
        # Non-default agent should not have "(default)" label
        assert "(default)" not in header.plain

    def test_default_agent_label(self) -> None:
        """Header should show '(default)' for the default agent name."""
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.model_name = None
            header = _build_non_interactive_header("agent", "abc123")
        assert "Agent: agent (default)" in header.plain

    def test_includes_model_name(self) -> None:
        """Header should display model name when available."""
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.model_name = "gpt-5"
            header = _build_non_interactive_header("agent", "abc123")
        assert "Model: gpt-5" in header.plain

    def test_omits_model_when_none(self) -> None:
        """Header should not include model section when model_name is None."""
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.model_name = None
            header = _build_non_interactive_header("agent", "abc123")
        assert "Model:" not in header.plain

    def test_includes_thread_id(self) -> None:
        """Header should contain the thread ID."""
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.model_name = None
            header = _build_non_interactive_header("agent", "deadbeef")
        assert "Thread: deadbeef" in header.plain

    def test_thread_clickable_when_url_available(self) -> None:
        """Thread ID should be a hyperlink when LangSmith URL is available."""
        url = "https://smith.langchain.com/o/org/projects/p/proj/t/abc123"
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.model_name = None
            with patch(
                "deepagents_code.non_interactive.build_langsmith_thread_url",
                return_value=url,
            ):
                header = _build_non_interactive_header(
                    "agent",
                    "abc123",
                    include_thread_link=True,
                )
        # Find the span containing the thread ID and verify it has a link
        for start, end, style in header._spans:
            text = header.plain[start:end]
            if text == "abc123" and isinstance(style, Style) and style.link:
                assert style.link == url
                break
        else:
            pytest.fail("Thread ID span with hyperlink not found")

    def test_default_header_does_not_lookup_langsmith(self) -> None:
        """Header should skip LangSmith lookup unless explicitly enabled."""
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.model_name = None
            with patch(
                "deepagents_code.non_interactive.build_langsmith_thread_url",
            ) as mock_build_url:
                _build_non_interactive_header("agent", "abc123")

        mock_build_url.assert_not_called()


class TestSandboxTypeForwarding:
    """Test that sandbox_type is forwarded to start_server_and_get_agent."""

    async def test_sandbox_type_passed_to_server(self) -> None:
        """run_non_interactive should forward sandbox_type to the server."""
        mock_agent = MagicMock()
        mock_agent.astream = MagicMock(return_value=_async_iter([]))
        mock_server_proc = MagicMock()

        with (
            patch(
                "deepagents_code.non_interactive.create_model",
                return_value=ModelResult(
                    model=MagicMock(),
                    model_name="test-model",
                    provider="test",
                ),
            ),
            patch(
                "deepagents_code.non_interactive.generate_thread_id",
                return_value="test-thread",
            ),
            patch(
                "deepagents_code.non_interactive.settings",
            ) as mock_settings,
            patch(
                "deepagents_code.non_interactive.build_langsmith_thread_url",
                return_value=None,
            ),
            patch(
                "deepagents_code.server_manager.start_server_and_get_agent",
                new_callable=AsyncMock,
                return_value=(mock_agent, mock_server_proc, None),
            ) as mock_start_server,
        ):
            mock_settings.shell_allow_list = None
            mock_settings.has_tavily = False
            mock_settings.model_name = None

            await run_non_interactive(
                message="test task",
                sandbox_type="modal",
            )

        _, kwargs = mock_start_server.call_args
        assert kwargs["sandbox_type"] == "modal"
        assert kwargs["enable_interpreter"] is None

    async def test_sandbox_snapshot_name_passed_to_server(self) -> None:
        """`sandbox_snapshot_name` must reach `start_server_and_get_agent`."""
        mock_agent = MagicMock()
        mock_agent.astream = MagicMock(return_value=_async_iter([]))
        mock_server_proc = MagicMock()

        with (
            patch(
                "deepagents_code.non_interactive.create_model",
                return_value=ModelResult(
                    model=MagicMock(),
                    model_name="test-model",
                    provider="test",
                ),
            ),
            patch(
                "deepagents_code.non_interactive.generate_thread_id",
                return_value="test-thread",
            ),
            patch(
                "deepagents_code.non_interactive.settings",
            ) as mock_settings,
            patch(
                "deepagents_code.non_interactive.build_langsmith_thread_url",
                return_value=None,
            ),
            patch(
                "deepagents_code.server_manager.start_server_and_get_agent",
                new_callable=AsyncMock,
                return_value=(mock_agent, mock_server_proc, None),
            ) as mock_start_server,
        ):
            mock_settings.shell_allow_list = None
            mock_settings.has_tavily = False
            mock_settings.model_name = None

            await run_non_interactive(
                message="test task",
                sandbox_type="langsmith",
                sandbox_snapshot_name="my-snap",
            )

        _, kwargs = mock_start_server.call_args
        assert kwargs["sandbox_snapshot_name"] == "my-snap"


class TestQuietMode:
    """Tests for --quiet flag in run_non_interactive."""

    @pytest.mark.parametrize(
        ("quiet", "expected_kwargs"),
        [
            pytest.param(True, {"stderr": True}, id="quiet-redirects-to-stderr"),
            pytest.param(False, {}, id="default-uses-stdout"),
        ],
    )
    async def test_console_creation(
        self, quiet: bool, expected_kwargs: dict[str, object]
    ) -> None:
        """Console should use stderr when quiet=True, stdout otherwise."""
        mock_console = MagicMock(spec=Console)
        mock_agent = MagicMock()
        mock_agent.astream = MagicMock(return_value=_async_iter([]))
        mock_server_proc = MagicMock()

        with (
            patch(
                "deepagents_code.non_interactive.Console",
                return_value=mock_console,
            ) as mock_console_cls,
            patch(
                "deepagents_code.non_interactive.create_model",
                return_value=ModelResult(
                    model=MagicMock(),
                    model_name="test-model",
                    provider="test",
                ),
            ),
            patch(
                "deepagents_code.non_interactive.generate_thread_id",
                return_value="test-thread",
            ),
            patch(
                "deepagents_code.non_interactive.settings",
            ) as mock_settings,
            patch(
                "deepagents_code.non_interactive.build_langsmith_thread_url",
                return_value=None,
            ),
            patch(
                "deepagents_code.server_manager.start_server_and_get_agent",
                new_callable=AsyncMock,
                return_value=(mock_agent, mock_server_proc, None),
            ),
        ):
            mock_settings.shell_allow_list = None
            mock_settings.has_tavily = False
            mock_settings.model_name = None

            await run_non_interactive(message="test", quiet=quiet)

        mock_console_cls.assert_called_once_with(**expected_kwargs)

    async def test_quiet_stdout_contains_only_agent_text(self) -> None:
        """In quiet mode, stdout should have only agent text."""
        # Build a fake AI message with a text block followed by a tool-call block
        ai_msg = MagicMock(spec=AIMessage)
        ai_msg.content_blocks = [
            {"type": "text", "text": "Hello from agent"},
            {"type": "tool_call_chunk", "name": "read_file", "id": "tc1", "index": 0},
        ]
        stream_chunks = [
            # 3-tuple: (namespace, stream_mode, data)
            ("", "messages", (ai_msg, {})),
        ]

        stdout_buf = io.StringIO()
        stderr_buf = io.StringIO()

        mock_agent = MagicMock()
        mock_agent.astream = MagicMock(return_value=_async_iter(stream_chunks))
        mock_server_proc = MagicMock()

        with (
            patch(
                "deepagents_code.non_interactive.create_model",
                return_value=ModelResult(
                    model=MagicMock(),
                    model_name="test-model",
                    provider="test",
                ),
            ),
            patch(
                "deepagents_code.non_interactive.generate_thread_id",
                return_value="test-thread",
            ),
            patch(
                "deepagents_code.non_interactive.settings",
            ) as mock_settings,
            patch(
                "deepagents_code.non_interactive.build_langsmith_thread_url",
                return_value=None,
            ),
            patch(
                "deepagents_code.server_manager.start_server_and_get_agent",
                new_callable=AsyncMock,
                return_value=(mock_agent, mock_server_proc, None),
            ),
            patch.object(sys, "stdout", stdout_buf),
            patch.object(sys, "stderr", stderr_buf),
        ):
            mock_settings.shell_allow_list = None
            mock_settings.has_tavily = False
            mock_settings.model_name = None

            await run_non_interactive(message="test", quiet=True)

        stdout = stdout_buf.getvalue()
        stderr = stderr_buf.getvalue()

        # Agent response text goes to stdout
        assert "Hello from agent" in stdout
        # Diagnostic messages should NOT be on stdout
        assert "Calling tool" not in stdout
        assert "Task completed" not in stdout
        assert "Running task" not in stdout
        # Tool notifications still go to stderr
        assert "Calling tool" in stderr or "read_file" in stderr
        # Header and completion messages are fully suppressed in quiet mode
        assert "Task completed" not in stderr
        assert "Running task" not in stderr


class TestNoStreamMode:
    """Tests for --no-stream flag in run_non_interactive."""

    async def test_no_stream_buffers_output(self) -> None:
        """In no-stream mode, stdout should receive text only after completion."""
        # Build two text chunks to verify buffering vs streaming
        ai_msg1 = MagicMock(spec=AIMessage)
        ai_msg1.content_blocks = [{"type": "text", "text": "Hello "}]
        ai_msg2 = MagicMock(spec=AIMessage)
        ai_msg2.content_blocks = [{"type": "text", "text": "world"}]

        stream_chunks = [
            ("", "messages", (ai_msg1, {})),
            ("", "messages", (ai_msg2, {})),
        ]

        stdout_writes: list[str] = []

        class TrackingStringIO(io.StringIO):
            """StringIO that records each write call separately."""

            def write(self, s: str) -> int:
                stdout_writes.append(s)
                return super().write(s)

        stdout_buf = TrackingStringIO()

        mock_agent = MagicMock()
        mock_agent.astream = MagicMock(return_value=_async_iter(stream_chunks))
        mock_server_proc = MagicMock()

        with (
            patch(
                "deepagents_code.non_interactive.create_model",
                return_value=ModelResult(
                    model=MagicMock(),
                    model_name="test-model",
                    provider="test",
                ),
            ),
            patch(
                "deepagents_code.non_interactive.generate_thread_id",
                return_value="test-thread",
            ),
            patch(
                "deepagents_code.non_interactive.settings",
            ) as mock_settings,
            patch(
                "deepagents_code.non_interactive.build_langsmith_thread_url",
                return_value=None,
            ),
            patch(
                "deepagents_code.server_manager.start_server_and_get_agent",
                new_callable=AsyncMock,
                return_value=(mock_agent, mock_server_proc, None),
            ),
            patch.object(sys, "stdout", stdout_buf),
        ):
            mock_settings.shell_allow_list = None
            mock_settings.has_tavily = False
            mock_settings.model_name = None

            await run_non_interactive(message="test", quiet=True, stream=False)

        stdout = stdout_buf.getvalue()
        assert "Hello world" in stdout

        # Verify the text was NOT written incrementally — the first
        # text write should contain the full concatenated response
        text_writes = [w for w in stdout_writes if w != "\n"]
        assert len(text_writes) == 1
        assert text_writes[0] == "Hello world"

    async def test_stream_mode_writes_incrementally(self) -> None:
        """Default stream mode should write text chunks as they arrive."""
        ai_msg1 = MagicMock(spec=AIMessage)
        ai_msg1.content_blocks = [{"type": "text", "text": "Hello "}]
        ai_msg2 = MagicMock(spec=AIMessage)
        ai_msg2.content_blocks = [{"type": "text", "text": "world"}]

        stream_chunks = [
            ("", "messages", (ai_msg1, {})),
            ("", "messages", (ai_msg2, {})),
        ]

        stdout_writes: list[str] = []

        class TrackingStringIO(io.StringIO):
            """StringIO that records each write call separately."""

            def write(self, s: str) -> int:
                stdout_writes.append(s)
                return super().write(s)

        stdout_buf = TrackingStringIO()

        mock_agent = MagicMock()
        mock_agent.astream = MagicMock(return_value=_async_iter(stream_chunks))
        mock_server_proc = MagicMock()

        with (
            patch(
                "deepagents_code.non_interactive.create_model",
                return_value=ModelResult(
                    model=MagicMock(),
                    model_name="test-model",
                    provider="test",
                ),
            ),
            patch(
                "deepagents_code.non_interactive.generate_thread_id",
                return_value="test-thread",
            ),
            patch(
                "deepagents_code.non_interactive.settings",
            ) as mock_settings,
            patch(
                "deepagents_code.non_interactive.build_langsmith_thread_url",
                return_value=None,
            ),
            patch(
                "deepagents_code.server_manager.start_server_and_get_agent",
                new_callable=AsyncMock,
                return_value=(mock_agent, mock_server_proc, None),
            ),
            patch.object(sys, "stdout", stdout_buf),
        ):
            mock_settings.shell_allow_list = None
            mock_settings.has_tavily = False
            mock_settings.model_name = None

            await run_non_interactive(message="test", quiet=True, stream=True)

        stdout = stdout_buf.getvalue()
        assert "Hello world" in stdout

        # Verify text was written incrementally (two separate writes)
        text_writes = [w for w in stdout_writes if w != "\n"]
        assert len(text_writes) == 2
        assert text_writes[0] == "Hello "
        assert text_writes[1] == "world"


class TestFastFollowLangsmithLink:
    """Tests for best-effort fast-follow LangSmith link output."""

    async def test_prints_link_when_lookup_ready(self) -> None:
        """Should print LangSmith link before completion when ready."""
        mock_console = MagicMock(spec=Console)
        ready_state = ThreadUrlLookupState()
        ready_state.done.set()
        ready_state.url = (
            "https://smith.langchain.com/o/org/projects/p/proj/t/test-thread"
        )

        mock_agent = MagicMock()
        mock_agent.astream = MagicMock(return_value=_async_iter([]))
        mock_server_proc = MagicMock()

        with (
            patch(
                "deepagents_code.non_interactive.Console",
                return_value=mock_console,
            ),
            patch(
                "deepagents_code.non_interactive.create_model",
                return_value=ModelResult(
                    model=MagicMock(),
                    model_name="test-model",
                    provider="test",
                ),
            ),
            patch(
                "deepagents_code.non_interactive.generate_thread_id",
                return_value="test-thread",
            ),
            patch(
                "deepagents_code.non_interactive.settings",
            ) as mock_settings,
            patch(
                "deepagents_code.non_interactive._start_langsmith_thread_url_lookup",
                return_value=ready_state,
            ),
            patch(
                "deepagents_code.server_manager.start_server_and_get_agent",
                new_callable=AsyncMock,
                return_value=(mock_agent, mock_server_proc, None),
            ),
        ):
            mock_settings.shell_allow_list = None
            mock_settings.has_tavily = False
            mock_settings.model_name = None

            await run_non_interactive(message="test", quiet=False)

        printed = [
            str(call.args[0]) for call in mock_console.print.call_args_list if call.args
        ]
        assert any("View in LangSmith:" in line for line in printed)

    async def test_skips_link_when_lookup_not_ready(self) -> None:
        """Should not wait for or print link when lookup is still in flight."""
        mock_console = MagicMock(spec=Console)
        pending_state = ThreadUrlLookupState()
        pending_state.url = (
            "https://smith.langchain.com/o/org/projects/p/proj/t/test-thread"
        )

        mock_agent = MagicMock()
        mock_agent.astream = MagicMock(return_value=_async_iter([]))
        mock_server_proc = MagicMock()

        with (
            patch(
                "deepagents_code.non_interactive.Console",
                return_value=mock_console,
            ),
            patch(
                "deepagents_code.non_interactive.create_model",
                return_value=ModelResult(
                    model=MagicMock(),
                    model_name="test-model",
                    provider="test",
                ),
            ),
            patch(
                "deepagents_code.non_interactive.generate_thread_id",
                return_value="test-thread",
            ),
            patch(
                "deepagents_code.non_interactive.settings",
            ) as mock_settings,
            patch(
                "deepagents_code.non_interactive._start_langsmith_thread_url_lookup",
                return_value=pending_state,
            ),
            patch(
                "deepagents_code.server_manager.start_server_and_get_agent",
                new_callable=AsyncMock,
                return_value=(mock_agent, mock_server_proc, None),
            ),
        ):
            mock_settings.shell_allow_list = None
            mock_settings.has_tavily = False
            mock_settings.model_name = None

            await run_non_interactive(message="test", quiet=False)

        printed = [
            str(call.args[0]) for call in mock_console.print.call_args_list if call.args
        ]
        assert not any("View in LangSmith:" in line for line in printed)

    async def test_skips_link_when_lookup_done_but_url_none(self) -> None:
        """Should not print link when lookup completed but URL is None."""
        mock_console = MagicMock(spec=Console)
        done_no_url = ThreadUrlLookupState()
        done_no_url.done.set()

        mock_agent = MagicMock()
        mock_agent.astream = MagicMock(return_value=_async_iter([]))
        mock_server_proc = MagicMock()

        with (
            patch(
                "deepagents_code.non_interactive.Console",
                return_value=mock_console,
            ),
            patch(
                "deepagents_code.non_interactive.create_model",
                return_value=ModelResult(
                    model=MagicMock(),
                    model_name="test-model",
                    provider="test",
                ),
            ),
            patch(
                "deepagents_code.non_interactive.generate_thread_id",
                return_value="test-thread",
            ),
            patch(
                "deepagents_code.non_interactive.settings",
            ) as mock_settings,
            patch(
                "deepagents_code.non_interactive._start_langsmith_thread_url_lookup",
                return_value=done_no_url,
            ),
            patch(
                "deepagents_code.server_manager.start_server_and_get_agent",
                new_callable=AsyncMock,
                return_value=(mock_agent, mock_server_proc, None),
            ),
        ):
            mock_settings.shell_allow_list = None
            mock_settings.has_tavily = False
            mock_settings.model_name = None

            await run_non_interactive(message="test", quiet=False)

        printed = [
            str(call.args[0]) for call in mock_console.print.call_args_list if call.args
        ]
        assert not any("View in LangSmith:" in line for line in printed)

    async def test_quiet_mode_skips_thread_url_lookup(self) -> None:
        """Should not start LangSmith URL lookup when quiet=True."""
        mock_agent = MagicMock()
        mock_agent.astream = MagicMock(return_value=_async_iter([]))
        mock_server_proc = MagicMock()

        with (
            patch(
                "deepagents_code.non_interactive.Console",
                return_value=MagicMock(spec=Console),
            ),
            patch(
                "deepagents_code.non_interactive.create_model",
                return_value=ModelResult(
                    model=MagicMock(),
                    model_name="test-model",
                    provider="test",
                ),
            ),
            patch(
                "deepagents_code.non_interactive.generate_thread_id",
                return_value="test-thread",
            ),
            patch(
                "deepagents_code.non_interactive.settings",
            ) as mock_settings,
            patch(
                "deepagents_code.non_interactive._start_langsmith_thread_url_lookup",
            ) as mock_lookup,
            patch(
                "deepagents_code.server_manager.start_server_and_get_agent",
                new_callable=AsyncMock,
                return_value=(mock_agent, mock_server_proc, None),
            ),
        ):
            mock_settings.shell_allow_list = None
            mock_settings.has_tavily = False
            mock_settings.model_name = None

            await run_non_interactive(message="test", quiet=True)

        mock_lookup.assert_not_called()


class TestStartLangsmithThreadUrlLookup:
    """Tests for _start_langsmith_thread_url_lookup."""

    def test_sets_url_on_success(self) -> None:
        """Should populate state.url when build succeeds."""
        url = "https://smith.langchain.com/o/org/projects/p/proj/t/tid"
        with patch(
            "deepagents_code.non_interactive.build_langsmith_thread_url",
            return_value=url,
        ):
            state = _start_langsmith_thread_url_lookup("tid")
            assert state.done.wait(timeout=2.0)
        assert state.url == url

    def test_signals_done_on_exception(self) -> None:
        """Should signal done and leave url as None when build raises."""
        with patch(
            "deepagents_code.non_interactive.build_langsmith_thread_url",
            side_effect=RuntimeError("boom"),
        ):
            state = _start_langsmith_thread_url_lookup("tid")
            assert state.done.wait(timeout=2.0)
        assert state.url is None

    def test_signals_done_when_url_is_none(self) -> None:
        """Should signal done when build returns None."""
        with patch(
            "deepagents_code.non_interactive.build_langsmith_thread_url",
            return_value=None,
        ):
            state = _start_langsmith_thread_url_lookup("tid")
            assert state.done.wait(timeout=2.0)
        assert state.url is None


class TestShellAllowListDecisionLogic:
    """Tests for shell allow-list → auto_approve / interrupt_shell_only."""

    @pytest.mark.parametrize(
        (
            "shell_allow_list",
            "expected_auto",
            "expected_shell_only",
            "expected_allow_list",
        ),
        [
            pytest.param(
                None,
                True,
                False,
                None,
                id="no-allow-list-auto-approves",
            ),
            pytest.param(
                ["ls", "cat"],
                False,
                True,
                ["ls", "cat"],
                id="restrictive-list-interrupts-shell-only",
            ),
            pytest.param(
                SHELL_ALLOW_ALL,
                True,
                False,
                None,
                id="allow-all-auto-approves",
            ),
        ],
    )
    async def test_shell_auto_approve_branches(
        self,
        shell_allow_list: list[str] | None,
        expected_auto: bool,
        expected_shell_only: bool,
        expected_allow_list: list[str] | None,
    ) -> None:
        """Verify start_server_and_get_agent receives correct flags."""
        mock_agent = MagicMock()
        mock_agent.astream = MagicMock(return_value=_async_iter([]))
        mock_server_proc = MagicMock()

        with (
            patch(
                "deepagents_code.non_interactive.create_model",
                return_value=ModelResult(
                    model=MagicMock(),
                    model_name="test-model",
                    provider="test",
                ),
            ),
            patch(
                "deepagents_code.non_interactive.generate_thread_id",
                return_value="test-thread",
            ),
            patch(
                "deepagents_code.non_interactive.settings",
            ) as mock_settings,
            patch(
                "deepagents_code.non_interactive.build_langsmith_thread_url",
                return_value=None,
            ),
            patch(
                "deepagents_code.server_manager.start_server_and_get_agent",
                new_callable=AsyncMock,
                return_value=(mock_agent, mock_server_proc, None),
            ) as mock_start_server,
        ):
            mock_settings.shell_allow_list = shell_allow_list
            mock_settings.has_tavily = False
            mock_settings.model_name = None

            await run_non_interactive(message="test task")

        _, kwargs = mock_start_server.call_args
        assert kwargs["auto_approve"] is expected_auto
        assert kwargs["interrupt_shell_only"] is expected_shell_only
        assert kwargs["shell_allow_list"] == expected_allow_list


class TestNonInteractivePrompt:
    """Tests that run_non_interactive passes interactive=False."""

    async def test_passes_interactive_false(self) -> None:
        mock_agent = MagicMock()
        mock_agent.astream = MagicMock(return_value=_async_iter([]))
        mock_server_proc = MagicMock()

        with (
            patch(
                "deepagents_code.non_interactive.create_model",
                return_value=ModelResult(
                    model=MagicMock(),
                    model_name="test-model",
                    provider="test",
                ),
            ),
            patch(
                "deepagents_code.non_interactive.generate_thread_id",
                return_value="test-thread",
            ),
            patch(
                "deepagents_code.non_interactive.settings",
            ) as mock_settings,
            patch(
                "deepagents_code.non_interactive.build_langsmith_thread_url",
                return_value=None,
            ),
            patch(
                "deepagents_code.server_manager.start_server_and_get_agent",
                new_callable=AsyncMock,
                return_value=(mock_agent, mock_server_proc, None),
            ) as mock_start_server,
        ):
            mock_settings.shell_allow_list = None
            mock_settings.has_tavily = False
            mock_settings.model_name = None

            await run_non_interactive(message="do the thing")

        _, kwargs = mock_start_server.call_args
        assert kwargs["interactive"] is False

    async def test_initial_skill_wraps_prompt_and_metadata(self) -> None:
        """Headless skill execution should send wrapped prompt + `__skill`."""
        mock_agent = MagicMock()
        mock_agent.astream = MagicMock(return_value=_async_iter([]))
        mock_server_proc = MagicMock()
        skill = {
            "name": "code-review",
            "description": "Review code changes",
            "path": "/skills/code-review/SKILL.md",
            "license": None,
            "compatibility": None,
            "metadata": {},
            "allowed_tools": [],
            "source": "user",
        }

        with (
            patch(
                "deepagents_code.non_interactive.create_model",
                return_value=ModelResult(
                    model=MagicMock(),
                    model_name="test-model",
                    provider="test",
                ),
            ),
            patch(
                "deepagents_code.non_interactive.generate_thread_id",
                return_value="test-thread",
            ),
            patch(
                "deepagents_code.non_interactive.settings",
            ) as mock_settings,
            patch(
                "deepagents_code.non_interactive.build_langsmith_thread_url",
                return_value=None,
            ),
            patch(
                "deepagents_code.skills.invocation.discover_skills_and_roots",
                return_value=([skill], []),
            ),
            patch(
                "deepagents_code.skills.load.load_skill_content",
                return_value="# Instructions\nDo stuff",
            ),
            patch(
                "deepagents_code.server_manager.start_server_and_get_agent",
                new_callable=AsyncMock,
                return_value=(mock_agent, mock_server_proc, None),
            ),
        ):
            mock_settings.shell_allow_list = None
            mock_settings.has_tavily = False
            mock_settings.model_name = None

            await run_non_interactive(
                message="review this patch",
                initial_skill="code-review",
                quiet=True,
            )

        stream_input = mock_agent.astream.call_args.args[0]
        user_msg = stream_input["messages"][0]
        assert "I'm invoking the skill `code-review`." in user_msg["content"]
        assert "**User request:** review this patch" in user_msg["content"]
        assert user_msg["additional_kwargs"]["__skill"]["name"] == "code-review"
        assert user_msg["additional_kwargs"]["__skill"]["args"] == "review this patch"

    async def test_initial_skill_missing_returns_error_without_starting_server(
        self,
    ) -> None:
        """Missing headless skill should fail before the server starts."""
        with (
            patch(
                "deepagents_code.non_interactive.create_model",
                return_value=ModelResult(
                    model=MagicMock(),
                    model_name="test-model",
                    provider="test",
                ),
            ),
            patch(
                "deepagents_code.non_interactive.settings",
            ) as mock_settings,
            patch(
                "deepagents_code.skills.invocation.discover_skills_and_roots",
                return_value=([], []),
            ),
            patch(
                "deepagents_code.server_manager.start_server_and_get_agent",
                new_callable=AsyncMock,
            ) as mock_start_server,
        ):
            mock_settings.shell_allow_list = None
            mock_settings.has_tavily = False
            mock_settings.model_name = None

            result = await run_non_interactive(
                message="review this patch",
                initial_skill="missing-skill",
                quiet=True,
            )

        assert result == 1
        mock_start_server.assert_not_awaited()


def _make_interrupt_chunk(interrupt_id: str = "i1") -> tuple:
    """Return a stream chunk that triggers one HITL interrupt.

    The interrupt value is a dict that would normally need to pass the
    HITLRequest Pydantic validator. Tests that call this helper must also
    patch `_HITL_REQUEST_ADAPTER.validate_python` to pass-through so that
    validation is bypassed and `state.interrupt_occurred` is set correctly.
    """
    interrupt = MagicMock()
    interrupt.id = interrupt_id
    interrupt.value = {
        "action_requests": [{"name": "read_file", "args": {"path": "/tmp/f"}}]
    }
    return ("", "updates", {"__interrupt__": [interrupt]})


def _make_looping_agent() -> MagicMock:
    """Return a mock agent whose astream always yields one interrupt chunk."""
    chunk = _make_interrupt_chunk()
    mock_agent = MagicMock()
    mock_agent.astream = MagicMock(side_effect=lambda *_, **__: _async_iter([chunk]))
    return mock_agent


class TestMaxTurns:
    """Tests for max_turns parameter in _run_agent_loop."""

    async def test_raises_after_user_limit(self) -> None:
        """HITLIterationLimitError is raised after max_turns HITL iterations."""
        agent = _make_looping_agent()
        console = Console(quiet=True)
        file_op_tracker = MagicMock()
        file_op_tracker.complete_with_message.return_value = None
        config: RunnableConfig = {"configurable": {"thread_id": "t1"}}

        with (
            patch(
                "deepagents_code.non_interactive.dispatch_hook", new_callable=AsyncMock
            ),
            patch("deepagents_code.non_interactive.dispatch_hook_fire_and_forget"),
            patch("deepagents_code.non_interactive.settings") as mock_settings,
            patch(
                "deepagents_code.non_interactive._HITL_REQUEST_ADAPTER"
            ) as mock_adapter,
        ):
            mock_settings.shell_allow_list = None
            mock_settings.model_name = ""
            mock_adapter.validate_python.side_effect = lambda v: v
            with pytest.raises(HITLIterationLimitError) as exc_info:
                await _run_agent_loop(
                    agent,
                    "task",
                    config,
                    console,
                    file_op_tracker,
                    quiet=True,
                    max_turns=1,
                )
        assert "--max-turns 1" in str(exc_info.value)

    async def test_error_message_names_user_flag(self) -> None:
        """Error message references --max-turns and tells the user how to fix it."""
        agent = _make_looping_agent()
        console = Console(quiet=True)
        file_op_tracker = MagicMock()
        file_op_tracker.complete_with_message.return_value = None
        config: RunnableConfig = {"configurable": {"thread_id": "t1"}}

        with (
            patch(
                "deepagents_code.non_interactive.dispatch_hook", new_callable=AsyncMock
            ),
            patch("deepagents_code.non_interactive.dispatch_hook_fire_and_forget"),
            patch("deepagents_code.non_interactive.settings") as mock_settings,
            patch(
                "deepagents_code.non_interactive._HITL_REQUEST_ADAPTER"
            ) as mock_adapter,
        ):
            mock_settings.shell_allow_list = None
            mock_settings.model_name = ""
            mock_adapter.validate_python.side_effect = lambda v: v
            with pytest.raises(HITLIterationLimitError) as exc_info:
                await _run_agent_loop(
                    agent,
                    "task",
                    config,
                    console,
                    file_op_tracker,
                    quiet=True,
                    max_turns=2,
                )
        msg = str(exc_info.value)
        assert "--max-turns 2" in msg
        assert "Increase --max-turns" in msg

    async def test_max_turns_above_default_is_honored(self) -> None:
        """User's --max-turns overrides the internal safety default (no clamp)."""
        # Pin the no-clamp invariant: with _MAX_HITL_ITERATIONS patched to 2
        # and max_turns=4, the loop must run 4 astream calls (1 initial + 3
        # HITL resumes) before the guard trips. If max_turns were clamped by
        # the internal default, we'd see only 2 calls.
        agent = _make_looping_agent()
        console = Console(quiet=True)
        file_op_tracker = MagicMock()
        file_op_tracker.complete_with_message.return_value = None
        config: RunnableConfig = {"configurable": {"thread_id": "t1"}}

        with (
            patch(
                "deepagents_code.non_interactive.dispatch_hook", new_callable=AsyncMock
            ),
            patch("deepagents_code.non_interactive.dispatch_hook_fire_and_forget"),
            patch("deepagents_code.non_interactive.settings") as mock_settings,
            patch(
                "deepagents_code.non_interactive._HITL_REQUEST_ADAPTER"
            ) as mock_adapter,
            patch("deepagents_code.non_interactive._MAX_HITL_ITERATIONS", 2),
        ):
            mock_settings.shell_allow_list = None
            mock_settings.model_name = ""
            mock_adapter.validate_python.side_effect = lambda v: v
            with pytest.raises(HITLIterationLimitError) as exc_info:
                await _run_agent_loop(
                    agent,
                    "task",
                    config,
                    console,
                    file_op_tracker,
                    quiet=True,
                    max_turns=4,
                )
        msg = str(exc_info.value)
        assert "Exceeded 4 agentic turns" in msg
        assert "--max-turns 4" in msg
        assert "internal safety default" not in msg
        assert agent.astream.call_count == 4

    async def test_no_max_turns_uses_internal_default(self) -> None:
        """Omitting max_turns falls back to the internal safety default."""
        agent = _make_looping_agent()
        console = Console(quiet=True)
        file_op_tracker = MagicMock()
        file_op_tracker.complete_with_message.return_value = None
        config: RunnableConfig = {"configurable": {"thread_id": "t1"}}

        with (
            patch(
                "deepagents_code.non_interactive.dispatch_hook", new_callable=AsyncMock
            ),
            patch("deepagents_code.non_interactive.dispatch_hook_fire_and_forget"),
            patch("deepagents_code.non_interactive.settings") as mock_settings,
            patch(
                "deepagents_code.non_interactive._HITL_REQUEST_ADAPTER"
            ) as mock_adapter,
            patch("deepagents_code.non_interactive._MAX_HITL_ITERATIONS", 1),
        ):
            mock_settings.shell_allow_list = None
            mock_settings.model_name = ""
            mock_adapter.validate_python.side_effect = lambda v: v
            with pytest.raises(HITLIterationLimitError) as exc_info:
                await _run_agent_loop(
                    agent,
                    "task",
                    config,
                    console,
                    file_op_tracker,
                    quiet=True,
                    max_turns=None,
                )
        msg = str(exc_info.value)
        assert "internal safety default of 1" in msg

    async def test_max_turns_forwarded_from_run_non_interactive(self) -> None:
        """run_non_interactive passes max_turns through to _run_agent_loop."""
        mock_agent = MagicMock()
        mock_agent.astream = MagicMock(return_value=_async_iter([]))
        mock_server_proc = MagicMock()

        with (
            patch(
                "deepagents_code.non_interactive.create_model",
                return_value=ModelResult(
                    model=MagicMock(),
                    model_name="test-model",
                    provider="test",
                ),
            ),
            patch(
                "deepagents_code.non_interactive.generate_thread_id",
                return_value="test-thread",
            ),
            patch("deepagents_code.non_interactive.settings") as mock_settings,
            patch(
                "deepagents_code.non_interactive.build_langsmith_thread_url",
                return_value=None,
            ),
            patch(
                "deepagents_code.non_interactive._run_agent_loop",
                new_callable=AsyncMock,
            ) as mock_loop,
            patch(
                "deepagents_code.server_manager.start_server_and_get_agent",
                new_callable=AsyncMock,
                return_value=(mock_agent, mock_server_proc, None),
            ),
        ):
            mock_settings.shell_allow_list = None
            mock_settings.has_tavily = False
            mock_settings.model_name = None

            await run_non_interactive(message="task", max_turns=7)

        _, kwargs = mock_loop.call_args
        assert kwargs.get("max_turns") == 7

    async def test_max_turns_none_forwarded_by_default(self) -> None:
        """run_non_interactive forwards max_turns=None when not supplied."""
        mock_agent = MagicMock()
        mock_agent.astream = MagicMock(return_value=_async_iter([]))
        mock_server_proc = MagicMock()

        with (
            patch(
                "deepagents_code.non_interactive.create_model",
                return_value=ModelResult(
                    model=MagicMock(),
                    model_name="test-model",
                    provider="test",
                ),
            ),
            patch(
                "deepagents_code.non_interactive.generate_thread_id",
                return_value="test-thread",
            ),
            patch("deepagents_code.non_interactive.settings") as mock_settings,
            patch(
                "deepagents_code.non_interactive.build_langsmith_thread_url",
                return_value=None,
            ),
            patch(
                "deepagents_code.non_interactive._run_agent_loop",
                new_callable=AsyncMock,
            ) as mock_loop,
            patch(
                "deepagents_code.server_manager.start_server_and_get_agent",
                new_callable=AsyncMock,
                return_value=(mock_agent, mock_server_proc, None),
            ),
        ):
            mock_settings.shell_allow_list = None
            mock_settings.has_tavily = False
            mock_settings.model_name = None

            await run_non_interactive(message="task")

        _, kwargs = mock_loop.call_args
        assert kwargs.get("max_turns") is None

    async def test_honors_full_user_budget_before_raising(self) -> None:
        """With max_turns=N, exactly N agentic turns run before the guard trips.

        Pins the counting semantics: the initial stream is turn 1, each HITL
        resume adds one more turn, and the guard trips on the (N+1)-th call.
        Flipping the check from `>=` to `>` would allow N+1 astream calls.
        """
        agent = _make_looping_agent()
        console = Console(quiet=True)
        file_op_tracker = MagicMock()
        file_op_tracker.complete_with_message.return_value = None
        config: RunnableConfig = {"configurable": {"thread_id": "t1"}}

        with (
            patch(
                "deepagents_code.non_interactive.dispatch_hook", new_callable=AsyncMock
            ),
            patch("deepagents_code.non_interactive.dispatch_hook_fire_and_forget"),
            patch("deepagents_code.non_interactive.settings") as mock_settings,
            patch(
                "deepagents_code.non_interactive._HITL_REQUEST_ADAPTER"
            ) as mock_adapter,
        ):
            mock_settings.shell_allow_list = None
            mock_settings.model_name = ""
            mock_adapter.validate_python.side_effect = lambda v: v
            with pytest.raises(HITLIterationLimitError):
                await _run_agent_loop(
                    agent,
                    "task",
                    config,
                    console,
                    file_op_tracker,
                    quiet=True,
                    max_turns=3,
                )
        assert agent.astream.call_count == 3  # 1 initial + 2 HITL resumes = 3 turns

    async def test_limit_hit_returns_exit_code_124(self) -> None:
        """run_non_interactive returns 124 when --max-turns is exhausted.

        A dedicated exit code (matching GNU `timeout`) lets CI distinguish
        budget exhaustion from generic failures, which still return 1.
        """
        looping_agent = _make_looping_agent()
        mock_server_proc = MagicMock()

        async def fake_run_agent_loop(*_args: Any, **_kwargs: Any) -> None:  # noqa: RUF029
            msg = "Exceeded 1 agentic turns (--max-turns 1)."
            raise HITLIterationLimitError(msg)

        with (
            patch(
                "deepagents_code.non_interactive.create_model",
                return_value=ModelResult(
                    model=MagicMock(),
                    model_name="test-model",
                    provider="test",
                ),
            ),
            patch(
                "deepagents_code.non_interactive.generate_thread_id",
                return_value="test-thread",
            ),
            patch("deepagents_code.non_interactive.settings") as mock_settings,
            patch(
                "deepagents_code.non_interactive.build_langsmith_thread_url",
                return_value=None,
            ),
            patch(
                "deepagents_code.non_interactive._run_agent_loop",
                new=fake_run_agent_loop,
            ),
            patch(
                "deepagents_code.server_manager.start_server_and_get_agent",
                new_callable=AsyncMock,
                return_value=(looping_agent, mock_server_proc, None),
            ),
        ):
            mock_settings.shell_allow_list = None
            mock_settings.has_tavily = False
            mock_settings.model_name = None

            result = await run_non_interactive(message="task", max_turns=1)

        assert result == 124


class TestRunStartupCommand:
    """Tests for `_run_startup_command` (`--startup-cmd`)."""

    async def test_successful_command_prints_stdout(self) -> None:
        """Exit 0 with stdout — output should be routed through the console."""
        buf = io.StringIO()
        console = Console(file=buf, width=200, highlight=False)

        await _run_startup_command("echo hello-startup", console, quiet=False)

        output = buf.getvalue()
        assert "Running startup command: echo hello-startup" in output
        assert "hello-startup" in output
        assert "Warning" not in output

    @pytest.mark.skipif(sys.platform == "win32", reason="`false` is POSIX-only")
    async def test_non_zero_exit_warns_but_does_not_raise(self) -> None:
        """Non-zero exit emits a yellow warning and keeps the session alive."""
        buf = io.StringIO()
        console = Console(file=buf, width=200, highlight=False)

        # `false` is guaranteed to exit 1 on POSIX.
        await _run_startup_command("false", console, quiet=False)

        output = buf.getvalue()
        assert "Warning" in output
        assert "exited with code 1" in output

    @pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell redirection")
    async def test_stderr_routed_through_console(self) -> None:
        """Commands that write to stderr should render under the dim stream."""
        buf = io.StringIO()
        console = Console(file=buf, width=200, highlight=False)

        await _run_startup_command("sh -c 'echo oops 1>&2'", console, quiet=False)

        output = buf.getvalue()
        assert "oops" in output

    async def test_stdout_with_brackets_does_not_raise(self) -> None:
        """Shell output with `[...]` must not be parsed as Rich markup."""
        buf = io.StringIO()
        console = Console(file=buf, width=200, highlight=False)

        # Unbalanced/unknown markup would raise `MarkupError` if parsed.
        await _run_startup_command(
            "printf '[INFO] starting [1/3]\\n'", console, quiet=False
        )

        output = buf.getvalue()
        assert "[INFO] starting [1/3]" in output
        assert "Warning" not in output

    async def test_quiet_mode_suppresses_header(self) -> None:
        """In quiet mode, the "Running" header should not appear."""
        buf = io.StringIO()
        console = Console(file=buf, width=200, highlight=False)

        await _run_startup_command("echo hi", console, quiet=True)

        output = buf.getvalue()
        assert "Running startup command" not in output
        assert "hi" in output

    async def test_launch_failure_warns(self) -> None:
        """Unlaunchable commands warn instead of crashing."""
        buf = io.StringIO()
        console = Console(file=buf, width=200, highlight=False)

        with patch(
            "asyncio.create_subprocess_shell",
            side_effect=OSError("boom"),
        ):
            await _run_startup_command("whatever", console, quiet=False)

        output = buf.getvalue()
        assert "Warning" in output
        assert "failed to launch" in output

    async def test_timeout_kills_process_group_on_posix(self) -> None:
        """Timeouts should terminate the whole POSIX process group."""
        buf = io.StringIO()
        console = Console(file=buf, width=200, highlight=False)

        mock_proc = AsyncMock()
        mock_proc.communicate = AsyncMock(side_effect=TimeoutError())
        mock_proc.wait = AsyncMock()
        mock_proc.returncode = None
        mock_proc.pid = 12345
        mock_proc.kill = MagicMock()

        with (
            patch("asyncio.create_subprocess_shell", return_value=mock_proc),
            patch.object(sys, "platform", "darwin"),
            patch("os.getpgid", return_value=12345),
            patch("os.killpg") as mock_killpg,
        ):
            await _run_startup_command("sleep 999", console, quiet=False)

        mock_killpg.assert_called_once_with(12345, signal.SIGTERM)
        mock_proc.kill.assert_not_called()
        assert "timed out" in buf.getvalue()

    async def test_cancellation_kills_process_group_on_posix(self) -> None:
        """Outer cancellation should still clean up the startup process group."""
        buf = io.StringIO()
        console = Console(file=buf, width=200, highlight=False)

        mock_proc = AsyncMock()
        mock_proc.communicate = AsyncMock(side_effect=asyncio.CancelledError())
        mock_proc.wait = AsyncMock()
        mock_proc.returncode = None
        mock_proc.pid = 12345
        mock_proc.kill = MagicMock()

        with (
            patch("asyncio.create_subprocess_shell", return_value=mock_proc),
            patch.object(sys, "platform", "darwin"),
            patch("os.getpgid", return_value=12345),
            patch("os.killpg") as mock_killpg,
            pytest.raises(asyncio.CancelledError),
        ):
            await _run_startup_command("sleep 999", console, quiet=False)

        mock_killpg.assert_called_once_with(12345, signal.SIGTERM)
        mock_proc.kill.assert_not_called()
        assert "timed out" not in buf.getvalue()

    async def test_timeout_escalates_to_sigkill_when_sigterm_ignored(self) -> None:
        """If SIGTERM + 5s wait also times out, SIGKILL must follow."""
        buf = io.StringIO()
        console = Console(file=buf, width=200, highlight=False)

        # First `communicate` raises TimeoutError (hit 60s limit).
        # First `wait` raises TimeoutError (hit 5s post-SIGTERM grace).
        # Second `wait` returns normally (post-SIGKILL reap).
        mock_proc = AsyncMock()
        mock_proc.communicate = AsyncMock(side_effect=TimeoutError())
        mock_proc.wait = AsyncMock(side_effect=[TimeoutError(), None])
        mock_proc.returncode = None
        mock_proc.pid = 12345
        mock_proc.kill = MagicMock()

        with (
            patch("asyncio.create_subprocess_shell", return_value=mock_proc),
            patch.object(sys, "platform", "darwin"),
            patch("os.getpgid", return_value=12345),
            patch("os.killpg") as mock_killpg,
        ):
            await _run_startup_command("sleep 999", console, quiet=False)

        assert mock_killpg.call_args_list == [
            call(12345, signal.SIGTERM),
            call(12345, signal.SIGKILL),
        ]
        mock_proc.kill.assert_not_called()
        assert "timed out" in buf.getvalue()

    async def test_timeout_uses_proc_kill_on_windows(self) -> None:
        """Windows has no process groups; fall back to `proc.kill()`."""
        buf = io.StringIO()
        console = Console(file=buf, width=200, highlight=False)

        mock_proc = AsyncMock()
        mock_proc.communicate = AsyncMock(side_effect=TimeoutError())
        mock_proc.wait = AsyncMock()
        mock_proc.returncode = None
        mock_proc.pid = 12345
        mock_proc.kill = MagicMock()

        with (
            patch("asyncio.create_subprocess_shell", return_value=mock_proc),
            patch.object(sys, "platform", "win32"),
            patch("os.killpg") as mock_killpg,
        ):
            await _run_startup_command("sleep 999", console, quiet=False)

        mock_proc.kill.assert_called_once()
        mock_killpg.assert_not_called()
        assert "timed out" in buf.getvalue()

    async def test_empty_command_is_not_executed(self) -> None:
        """Whitespace-only `--startup-cmd` should be treated as unset."""
        buf = io.StringIO()
        console = Console(file=buf, width=200, highlight=False)

        with patch(
            "asyncio.create_subprocess_shell",
            new=AsyncMock(),
        ) as mock_spawn:
            # `run_non_interactive` strips and skips when empty; replicate
            # that contract here by not calling through when stripped empty.
            command = "   "
            if command.strip():
                await _run_startup_command(command.strip(), console, quiet=False)

        mock_spawn.assert_not_called()
        assert buf.getvalue() == ""


class TestProcessAiMessageStats:
    """`_process_ai_message` threads the active provider into usage stats.

    Guards the wiring between `settings.model_provider` and
    `SessionStats.record_request` — the per-model API is unit-tested in
    isolation elsewhere, but these confirm the call site actually forwards the
    configured provider.
    """

    def test_records_provider_from_settings(self, console: Console) -> None:
        """Split input/output usage records the configured provider."""
        state = StreamState()
        message = AIMessage(
            content="",
            usage_metadata={
                "input_tokens": 100,
                "output_tokens": 50,
                "total_tokens": 150,
            },
        )
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.model_name = "gpt-5.5"
            mock_settings.model_provider = "openai"
            _process_ai_message(message, state, console)

        assert state.stats.per_model["openai", "gpt-5.5"].input_tokens == 100
        assert state.stats.per_model["openai", "gpt-5.5"].output_tokens == 50

    def test_records_provider_on_total_only_fallback(self, console: Console) -> None:
        """Total-only usage (no split) still forwards the provider."""
        state = StreamState()
        message = AIMessage(
            content="",
            usage_metadata={
                "input_tokens": 0,
                "output_tokens": 0,
                "total_tokens": 150,
            },
        )
        with patch("deepagents_code.non_interactive.settings") as mock_settings:
            mock_settings.model_name = "gpt-5.5"
            mock_settings.model_provider = "openai"
            _process_ai_message(message, state, console)

        assert state.stats.per_model["openai", "gpt-5.5"].input_tokens == 150


async def _async_iter(items: Sequence[object]) -> AsyncIterator[object]:  # noqa: RUF029
    """Create an async iterator from a list for testing."""
    for item in items:
        yield item
