test_agent_and_compact_rpc_e2e.py

"""E2E tests for Agent Selection and Session Compaction RPC APIs."""

import uuid

import pytest

from copilot import CopilotClient, RuntimeConnection
from copilot.rpc import AgentSelectRequest
from copilot.session import PermissionHandler

from .testharness import CLI_PATH, E2ETestContext

pytestmark = pytest.mark.asyncio(loop_scope="module")


class TestAgentSelectionRpc:
    @pytest.mark.asyncio
    async def test_should_list_available_custom_agents(self):
        """Test listing available custom agents via RPC."""
        client = CopilotClient(connection=RuntimeConnection.for_stdio(path=CLI_PATH))

        try:
            await client.start()
            session = await client.create_session(
                on_permission_request=PermissionHandler.approve_all,
                custom_agents=[
                    {
                        "name": "test-agent",
                        "display_name": "Test Agent",
                        "description": "A test agent",
                        "prompt": "You are a test agent.",
                    },
                    {
                        "name": "another-agent",
                        "display_name": "Another Agent",
                        "description": "Another test agent",
                        "prompt": "You are another agent.",
                    },
                ],
            )

            result = await session.rpc.agent.list()
            assert result.agents is not None
            assert len(result.agents) == 2
            assert result.agents[0].name == "test-agent"
            assert result.agents[0].display_name == "Test Agent"
            assert result.agents[0].description == "A test agent"
            assert result.agents[1].name == "another-agent"

            await session.disconnect()
            await client.stop()
        finally:
            await client.force_stop()

    @pytest.mark.asyncio
    async def test_should_return_null_when_no_agent_is_selected(self):
        """Test getCurrent returns null when no agent is selected."""
        client = CopilotClient(connection=RuntimeConnection.for_stdio(path=CLI_PATH))

        try:
            await client.start()
            session = await client.create_session(
                on_permission_request=PermissionHandler.approve_all,
                custom_agents=[
                    {
                        "name": "test-agent",
                        "display_name": "Test Agent",
                        "description": "A test agent",
                        "prompt": "You are a test agent.",
                    }
                ],
            )

            result = await session.rpc.agent.get_current()
            assert result.agent is None

            await session.disconnect()
            await client.stop()
        finally:
            await client.force_stop()

    @pytest.mark.asyncio
    async def test_should_select_and_get_current_agent(self):
        """Test selecting an agent and verifying getCurrent returns it."""
        client = CopilotClient(connection=RuntimeConnection.for_stdio(path=CLI_PATH))

        try:
            await client.start()
            session = await client.create_session(
                on_permission_request=PermissionHandler.approve_all,
                custom_agents=[
                    {
                        "name": "test-agent",
                        "display_name": "Test Agent",
                        "description": "A test agent",
                        "prompt": "You are a test agent.",
                    }
                ],
            )

            # Select the agent
            select_result = await session.rpc.agent.select(AgentSelectRequest(name="test-agent"))
            assert select_result.agent is not None
            assert select_result.agent.name == "test-agent"
            assert select_result.agent.display_name == "Test Agent"

            # Verify getCurrent returns the selected agent
            current_result = await session.rpc.agent.get_current()
            assert current_result.agent is not None
            assert current_result.agent.name == "test-agent"

            await session.disconnect()
            await client.stop()
        finally:
            await client.force_stop()

    @pytest.mark.asyncio
    async def test_should_deselect_current_agent(self):
        """Test deselecting the current agent."""
        client = CopilotClient(connection=RuntimeConnection.for_stdio(path=CLI_PATH))

        try:
            await client.start()
            session = await client.create_session(
                on_permission_request=PermissionHandler.approve_all,
                custom_agents=[
                    {
                        "name": "test-agent",
                        "display_name": "Test Agent",
                        "description": "A test agent",
                        "prompt": "You are a test agent.",
                    }
                ],
            )

            # Select then deselect
            await session.rpc.agent.select(AgentSelectRequest(name="test-agent"))
            await session.rpc.agent.deselect()

            # Verify no agent is selected
            current_result = await session.rpc.agent.get_current()
            assert current_result.agent is None

            await session.disconnect()
            await client.stop()
        finally:
            await client.force_stop()

    @pytest.mark.asyncio
    async def test_should_return_empty_list_when_no_custom_agents_configured(self):
        """Test listing agents returns no custom agents when none configured."""
        client = CopilotClient(connection=RuntimeConnection.for_stdio(path=CLI_PATH))

        try:
            await client.start()
            session = await client.create_session(
                on_permission_request=PermissionHandler.approve_all
            )

            result = await session.rpc.agent.list()
            # The CLI may return built-in/default agents even when no custom agents
            # are configured. Verify no custom test agents appear in the list.
            custom_names = {"test-agent", "another-agent"}
            for agent in result.agents:
                assert agent.name not in custom_names, (
                    f"Expected no custom agents, but found {agent.name!r}"
                )

            await session.disconnect()
            await client.stop()
        finally:
            await client.force_stop()

    @pytest.mark.asyncio
    async def test_should_call_agent_reload(self):
        """Test reloading agents via RPC."""
        client = CopilotClient(connection=RuntimeConnection.for_stdio(path=CLI_PATH))
        reload_agent = {
            "name": f"reload-test-agent-{uuid.uuid4().hex}",
            "display_name": "Reload Agent",
            "description": "An agent used to validate reload",
            "prompt": "You are a reload test agent.",
        }

        try:
            await client.start()
            session = await client.create_session(
                on_permission_request=PermissionHandler.approve_all,
                custom_agents=[reload_agent],
            )

            before = await session.rpc.agent.list()
            _assert_reload_agent(before.agents, reload_agent)

            result = await session.rpc.agent.reload()
            assert result.agents is not None
            current = await session.rpc.agent.list()
            assert _agent_summaries(result.agents) == _agent_summaries(current.agents)

            await session.disconnect()
            await client.stop()
        finally:
            await client.force_stop()


def _assert_reload_agent(agents, expected):
    matches = [agent for agent in agents if agent.name == expected["name"]]
    assert len(matches) == 1
    assert matches[0].display_name == expected["display_name"]
    assert matches[0].description == expected["description"]


def _agent_summaries(agents):
    return sorted((agent.name, agent.display_name) for agent in agents)


class TestSessionCompactionRpc:
    @pytest.mark.asyncio
    async def test_should_compact_session_history_after_messages(self, ctx: E2ETestContext):
        """Test compacting session history via RPC."""
        session = await ctx.client.create_session(
            on_permission_request=PermissionHandler.approve_all
        )

        # Send a message to create some history
        await session.send_and_wait("What is 2+2?")

        # Compact the session
        result = await session.rpc.history.compact()
        assert isinstance(result.success, bool)
        assert isinstance(result.tokens_removed, (int, float))
        assert isinstance(result.messages_removed, (int, float))

        await session.disconnect()